From 3e1477c56c6db8d7ad27adba5924ad3dd32c213a Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 22:22:46 +0300 Subject: [PATCH] fix: LWS HTTP connection leak, grain index alignment, and reader head tracking - control_server: add Connection: close header + always return -1 after HTTP response to force-close connection. Without this, lws_service() blocks forever after the first POST /cmd, freezing the node process loop. - passthrough: initialize read_index_ to runtime.headIndex when reader is added (prevents TOO_EARLY errors on first read) - passthrough: handle MXL_ERR_OUT_OF_RANGE_TOO_EARLY in addition to TOO_LATE (both jump to head index) - passthrough: use grain_info.index as writer index instead of separate write_index_ counter (MXL writers must use TAI-based grain indices) - passthrough: reduce grain read timeout from 100ms to 20ms for tighter loop with LWS poll - node_runner: add 'status' command handler that sends node status back via control server --- libs/dmf-node/src/control_server.cpp | 4 +++- libs/dmf-node/src/node_runner.cpp | 8 +++++++ nodes/passthrough/src/passthrough_node.cpp | 28 ++++++++++++---------- 3 files changed, 26 insertions(+), 14 deletions(-) diff --git a/libs/dmf-node/src/control_server.cpp b/libs/dmf-node/src/control_server.cpp index cfb1d1d..acdd6b5 100644 --- a/libs/dmf-node/src/control_server.cpp +++ b/libs/dmf-node/src/control_server.cpp @@ -47,12 +47,14 @@ static int send_http_json(struct lws* wsi, const std::string& status, const std: auto hdr = "HTTP/1.1 " + status + "\r\n" "Content-Type: application/json\r\n" "Content-Length: " + std::to_string(body.size()) + "\r\n" + "Connection: close\r\n" "\r\n"; std::vector buf(LWS_PRE + hdr.size() + body.size()); std::memcpy(buf.data() + LWS_PRE, hdr.data(), hdr.size()); std::memcpy(buf.data() + LWS_PRE + hdr.size(), body.data(), body.size()); lws_write(wsi, buf.data() + LWS_PRE, hdr.size() + body.size(), LWS_WRITE_HTTP); - return lws_http_transaction_completed(wsi) ? -1 : 0; + lws_http_transaction_completed(wsi); + return -1; } struct PerSession { diff --git a/libs/dmf-node/src/node_runner.cpp b/libs/dmf-node/src/node_runner.cpp index 266e04c..974eda6 100644 --- a/libs/dmf-node/src/node_runner.cpp +++ b/libs/dmf-node/src/node_runner.cpp @@ -144,6 +144,14 @@ int NodeRunner::exec(std::unique_ptr node) { } }); + control_server->register_command("status", [&](const nlohmann::json& /*msg*/) { + nlohmann::json resp; + resp["event"] = "status"; + resp["node_id"] = node_id_; + resp["data"] = node->status(); + control_server->send_event(resp); + }); + control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) { spdlog::info("Shutdown command received"); g_running = false; diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index 1d52bba..ad9407d 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -19,7 +19,10 @@ void PassthroughNode::on_add_writer(const std::string& port_id, mxlFlowWriter wr void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) { if (port_id == "video_in") { reader_ = reader; - spdlog::info("Passthrough: reader added on video_in"); + mxlFlowRuntimeInfo runtime{}; + mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); + read_index_ = runtime.headIndex; + spdlog::info("Passthrough: reader added on video_in, starting at index {}", read_index_); } } @@ -45,26 +48,23 @@ void PassthroughNode::process() { mxlGrainInfo grain_info{}; uint8_t* payload = nullptr; - auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 100000000ULL, &grain_info, &payload); + auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload); if (status != MXL_STATUS_OK) { - if (status == MXL_ERR_TIMEOUT) { - return; - } - if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE) { + if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { mxlFlowRuntimeInfo runtime{}; mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); + spdlog::warn("Passthrough: index {} out of range, jumping to head {}", read_index_, runtime.headIndex); read_index_ = runtime.headIndex; - return; } return; } mxlGrainInfo out_grain{}; uint8_t* out_payload = nullptr; - status = mxlFlowWriterOpenGrain(*writer_, write_index_, &out_grain, &out_payload); + status = mxlFlowWriterOpenGrain(*writer_, grain_info.index, &out_grain, &out_payload); if (status != MXL_STATUS_OK) { - spdlog::warn("Passthrough: failed to open output grain at index {}: {}", write_index_, static_cast(status)); - read_index_++; + spdlog::warn("Passthrough: failed to open output grain at index {}: {}", grain_info.index, static_cast(status)); + read_index_ = grain_info.index + 1; return; } @@ -75,9 +75,12 @@ void PassthroughNode::process() { out_grain.flags = grain_info.flags; mxlFlowWriterCommitGrain(*writer_, &out_grain); - read_index_++; - write_index_++; + read_index_ = grain_info.index + 1; grains_processed_++; + + if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) { + spdlog::info("Passthrough: grain #{}, index={}, size={}", grains_processed_, grain_info.index, grain_info.grainSize); + } } nlohmann::json PassthroughNode::status() const { @@ -85,7 +88,6 @@ nlohmann::json PassthroughNode::status() const { {"type", "passthrough"}, {"grains_processed", grains_processed_}, {"read_index", read_index_}, - {"write_index", write_index_}, {"has_reader", reader_.has_value()}, {"has_writer", writer_.has_value()}, };