fix: separate grain processing thread from LWS event loop
The fundamental issue: mxlSleepUntil/mxlSleepForNs blocks the entire thread, making LWS unresponsive and causing control commands to hang. Meanwhile, LWS poll(1) adds latency that makes grain timing unreliable. Solution: run node->process() in a dedicated thread that can freely use mxlSleepUntil for precise grain timing, while the main thread runs control_server->poll(10) for LWS event handling. Passthrough now uses mxlSleepUntil(deadline) matching the mxl-gst-sink Cursor pattern, with 2-grain read delay for buffering.
This commit is contained in:
@@ -170,20 +170,19 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
|
|||||||
running_ = true;
|
running_ = true;
|
||||||
spdlog::info("Node '{}' entering process loop", node_id_);
|
spdlog::info("Node '{}' entering process loop", node_id_);
|
||||||
|
|
||||||
while (g_running && running_) {
|
std::thread process_thread([&]() {
|
||||||
control_server->poll(1);
|
while (g_running && running_) {
|
||||||
node->process();
|
node->process();
|
||||||
|
|
||||||
static int status_counter = 0;
|
|
||||||
if (++status_counter % 100 == 0) {
|
|
||||||
nlohmann::json status_event;
|
|
||||||
status_event["event"] = "status";
|
|
||||||
status_event["node_id"] = node_id_;
|
|
||||||
status_event["data"] = node->status();
|
|
||||||
control_server->send_event(status_event);
|
|
||||||
}
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
while (g_running && running_) {
|
||||||
|
control_server->poll(10);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
running_ = false;
|
||||||
|
process_thread.join();
|
||||||
|
|
||||||
spdlog::info("Node '{}' shutting down", node_id_);
|
spdlog::info("Node '{}' shutting down", node_id_);
|
||||||
mxlDestroyInstance(mxl_instance_);
|
mxlDestroyInstance(mxl_instance_);
|
||||||
return 0;
|
return 0;
|
||||||
|
|||||||
@@ -51,13 +51,12 @@ void PassthroughNode::on_remove_reader(const std::string& port_id) {
|
|||||||
|
|
||||||
void PassthroughNode::process() {
|
void PassthroughNode::process() {
|
||||||
if (!reader_ || !writer_ || !aligned_) {
|
if (!reader_ || !writer_ || !aligned_) {
|
||||||
|
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
auto ns_until = mxlGetNsUntilIndex(read_index_, &grain_rate_);
|
auto deadline = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
|
||||||
if (ns_until > 2000000ULL) {
|
mxlSleepUntil(deadline);
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
mxlGrainInfo grain_info{};
|
mxlGrainInfo grain_info{};
|
||||||
uint8_t* payload = nullptr;
|
uint8_t* payload = nullptr;
|
||||||
|
|||||||
@@ -5,6 +5,7 @@
|
|||||||
#include <mxl/flow.h>
|
#include <mxl/flow.h>
|
||||||
|
|
||||||
#include <optional>
|
#include <optional>
|
||||||
|
#include <thread>
|
||||||
|
|
||||||
namespace dmf_node {
|
namespace dmf_node {
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user