From d677b654f0b3314c5f0a78490c76bac3386af6a2 Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 22:51:57 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20non-blocking=20grain=20timing=20?= =?UTF-8?q?=E2=80=94=20use=20mxlGetNsUntilIndex=20instead=20of=20mxlSleepU?= =?UTF-8?q?ntil?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mxlSleepUntil blocks the entire thread including LWS poll, causing the control server to become unresponsive and grains to be missed. Instead, check mxlGetNsUntilIndex and skip process() if the grain isn't due yet (>2ms away). The node loop spins with poll(1) keeping LWS responsive, and only attempts a grain read when it's nearly due. --- nodes/passthrough/src/passthrough_node.cpp | 15 ++----- nodes/passthrough/src/passthrough_node.hpp | 1 - test.sh | 51 ++++++++++++++++++++++ 3 files changed, 55 insertions(+), 12 deletions(-) create mode 100755 test.sh diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index e8acb1c..244c049 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -28,7 +28,6 @@ void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader re auto now = mxlGetTime(); auto current_index = mxlTimestampToIndex(&grain_rate_, now); read_index_ = current_index - READ_DELAY_GRAINS; - delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); aligned_ = true; spdlog::info("Passthrough: reader added, grain_rate={}/{}, read_index={}, delay={} grains", @@ -55,12 +54,9 @@ void PassthroughNode::process() { return; } - auto now = mxlGetTime(); - if (delivery_deadline_ > now) { - auto sleep_ns = delivery_deadline_ - now; - if (sleep_ns < 500000000ULL) { - mxlSleepUntil(delivery_deadline_); - } + auto ns_until = mxlGetNsUntilIndex(read_index_, &grain_rate_); + if (ns_until > 2000000ULL) { + return; } mxlGrainInfo grain_info{}; @@ -70,10 +66,9 @@ void PassthroughNode::process() { if (status != MXL_STATUS_OK) { if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { auto old_index = read_index_; - now = mxlGetTime(); + auto now = mxlGetTime(); auto current_index = mxlTimestampToIndex(&grain_rate_, now); read_index_ = current_index - READ_DELAY_GRAINS; - delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); spdlog::warn("Passthrough: index {} out of range ({}), realigned to {}", old_index, static_cast(status), read_index_); } @@ -87,7 +82,6 @@ void PassthroughNode::process() { spdlog::warn("Passthrough: failed to open output grain at index {}: {}", grain_info.index, static_cast(status)); read_index_++; - delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); return; } @@ -99,7 +93,6 @@ void PassthroughNode::process() { mxlFlowWriterCommitGrain(*writer_, &out_grain); read_index_++; - delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); grains_processed_++; if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) { diff --git a/nodes/passthrough/src/passthrough_node.hpp b/nodes/passthrough/src/passthrough_node.hpp index 8045c4a..c4e519a 100644 --- a/nodes/passthrough/src/passthrough_node.hpp +++ b/nodes/passthrough/src/passthrough_node.hpp @@ -39,7 +39,6 @@ private: std::optional writer_; mxlRational grain_rate_{50, 1}; uint64_t read_index_ = 0; - uint64_t delivery_deadline_ = 0; uint64_t grains_processed_ = 0; bool aligned_ = false; }; diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..838812e --- /dev/null +++ b/test.sh @@ -0,0 +1,51 @@ +#!/bin/bash +set -e + +ENGINE_PORT=${ENGINE_PORT:-9000} +MXL_DOMAIN=${MXL_DOMAIN:-/tmp/dmf-mxl} +FLOW_ID=${FLOW_ID:-a0000001-0000-0000-0000-000000000001} +BASE_URL="http://127.0.0.1:${ENGINE_PORT}" + +log() { echo "=== $1 ==="; } + +log "Adding passthrough node" +curl -s -X POST "${BASE_URL}/api/graph/nodes" \ + -H "Content-Type: application/json" \ + -d '{"type":"passthrough","id":"pass1"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Starting graph" +curl -s -X POST "${BASE_URL}/api/graph/start" | python3 -m json.tool 2>/dev/null || echo "" + +sleep 1 + +log "Connecting input to flow ${FLOW_ID}" +curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/connect-input" \ + -H "Content-Type: application/json" \ + -d "{\"port_id\":\"video_in\",\"flow_id\":\"${FLOW_ID}\"}" | python3 -m json.tool 2>/dev/null || echo "" + +log "Connecting output (creates new MXL flow)" +OUTPUT=$(curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/connect-output" \ + -H "Content-Type: application/json" \ + -d '{"port_id":"video_out"}') +echo "$OUTPUT" | python3 -m json.tool 2>/dev/null || echo "$OUTPUT" + +OUTPUT_FLOW_ID=$(echo "$OUTPUT" | python3 -c "import sys,json; print(json.load(sys.stdin)['flow_id'])" 2>/dev/null) + +if [ -n "$OUTPUT_FLOW_ID" ]; then + log "Output flow ID: ${OUTPUT_FLOW_ID}" + log "To verify with mxl-gst-sink:" + echo " mxl-gst-sink -d ${MXL_DOMAIN} -v ${OUTPUT_FLOW_ID}" +fi + +sleep 3 + +log "Checking MXL domain" +mxl-info --domain "${MXL_DOMAIN}" 2>/dev/null || true + +log "Sending status command" +curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/command" \ + -H "Content-Type: application/json" \ + -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Graph state" +curl -s "${BASE_URL}/api/graph" | python3 -m json.tool 2>/dev/null || echo ""