fix: non-blocking grain timing — use mxlGetNsUntilIndex instead of mxlSleepUntil
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.
This commit is contained in:
@@ -28,7 +28,6 @@ void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader re
|
|||||||
auto now = mxlGetTime();
|
auto now = mxlGetTime();
|
||||||
auto current_index = mxlTimestampToIndex(&grain_rate_, now);
|
auto current_index = mxlTimestampToIndex(&grain_rate_, now);
|
||||||
read_index_ = current_index - READ_DELAY_GRAINS;
|
read_index_ = current_index - READ_DELAY_GRAINS;
|
||||||
delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
|
|
||||||
aligned_ = true;
|
aligned_ = true;
|
||||||
|
|
||||||
spdlog::info("Passthrough: reader added, grain_rate={}/{}, read_index={}, delay={} grains",
|
spdlog::info("Passthrough: reader added, grain_rate={}/{}, read_index={}, delay={} grains",
|
||||||
@@ -55,12 +54,9 @@ void PassthroughNode::process() {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
auto now = mxlGetTime();
|
auto ns_until = mxlGetNsUntilIndex(read_index_, &grain_rate_);
|
||||||
if (delivery_deadline_ > now) {
|
if (ns_until > 2000000ULL) {
|
||||||
auto sleep_ns = delivery_deadline_ - now;
|
return;
|
||||||
if (sleep_ns < 500000000ULL) {
|
|
||||||
mxlSleepUntil(delivery_deadline_);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
mxlGrainInfo grain_info{};
|
mxlGrainInfo grain_info{};
|
||||||
@@ -70,10 +66,9 @@ void PassthroughNode::process() {
|
|||||||
if (status != MXL_STATUS_OK) {
|
if (status != MXL_STATUS_OK) {
|
||||||
if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
|
if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
|
||||||
auto old_index = read_index_;
|
auto old_index = read_index_;
|
||||||
now = mxlGetTime();
|
auto now = mxlGetTime();
|
||||||
auto current_index = mxlTimestampToIndex(&grain_rate_, now);
|
auto current_index = mxlTimestampToIndex(&grain_rate_, now);
|
||||||
read_index_ = current_index - READ_DELAY_GRAINS;
|
read_index_ = current_index - READ_DELAY_GRAINS;
|
||||||
delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
|
|
||||||
spdlog::warn("Passthrough: index {} out of range ({}), realigned to {}",
|
spdlog::warn("Passthrough: index {} out of range ({}), realigned to {}",
|
||||||
old_index, static_cast<int>(status), read_index_);
|
old_index, static_cast<int>(status), read_index_);
|
||||||
}
|
}
|
||||||
@@ -87,7 +82,6 @@ void PassthroughNode::process() {
|
|||||||
spdlog::warn("Passthrough: failed to open output grain at index {}: {}",
|
spdlog::warn("Passthrough: failed to open output grain at index {}: {}",
|
||||||
grain_info.index, static_cast<int>(status));
|
grain_info.index, static_cast<int>(status));
|
||||||
read_index_++;
|
read_index_++;
|
||||||
delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
|
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -99,7 +93,6 @@ void PassthroughNode::process() {
|
|||||||
mxlFlowWriterCommitGrain(*writer_, &out_grain);
|
mxlFlowWriterCommitGrain(*writer_, &out_grain);
|
||||||
|
|
||||||
read_index_++;
|
read_index_++;
|
||||||
delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
|
|
||||||
grains_processed_++;
|
grains_processed_++;
|
||||||
|
|
||||||
if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) {
|
if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) {
|
||||||
|
|||||||
@@ -39,7 +39,6 @@ private:
|
|||||||
std::optional<mxlFlowWriter> writer_;
|
std::optional<mxlFlowWriter> writer_;
|
||||||
mxlRational grain_rate_{50, 1};
|
mxlRational grain_rate_{50, 1};
|
||||||
uint64_t read_index_ = 0;
|
uint64_t read_index_ = 0;
|
||||||
uint64_t delivery_deadline_ = 0;
|
|
||||||
uint64_t grains_processed_ = 0;
|
uint64_t grains_processed_ = 0;
|
||||||
bool aligned_ = false;
|
bool aligned_ = false;
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -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 ""
|
||||||
Reference in New Issue
Block a user