From 2c712592dde499588ed516b773105712063b9f7a Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 22:40:46 +0300 Subject: [PATCH] fix: use TAI-time-based grain index alignment in passthrough Instead of chasing headIndex from mxlFlowReaderGetRuntimeInfo (which points to the NEXT grain to be written, causing perpetual TOO_EARLY), the passthrough now uses mxlTimestampToIndex + mxlGetNsUntilIndex for proper timing alignment, matching the pattern used by mxl-gst-sink. Key changes: - realign() computes read_index from current TAI time minus 1 grain delay - process() uses mxlSleepForNs to wait until the target grain is due - on_add_reader fetches grain_rate from mxlFlowConfigInfo - Removed separate write_index_ (uses grain_info.index for writer) --- nodes/passthrough/src/passthrough_node.cpp | 34 ++++++++++++++++------ nodes/passthrough/src/passthrough_node.hpp | 7 +++-- 2 files changed, 29 insertions(+), 12 deletions(-) diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index ad9407d..8c08c97 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -2,6 +2,7 @@ #include #include +#include #include @@ -19,10 +20,14 @@ 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; - mxlFlowRuntimeInfo runtime{}; - mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); - read_index_ = runtime.headIndex; - spdlog::info("Passthrough: reader added on video_in, starting at index {}", read_index_); + + mxlFlowConfigInfo config{}; + mxlFlowReaderGetConfigInfo(*reader_, &config); + grain_rate_ = config.common.grainRate; + + realign(); + spdlog::info("Passthrough: reader added on video_in, grain_rate={}/{}, read_index={}", + grain_rate_.numerator, grain_rate_.denominator, read_index_); } } @@ -40,21 +45,32 @@ void PassthroughNode::on_remove_reader(const std::string& port_id) { } } +void PassthroughNode::realign() { + auto now = mxlGetTime(); + auto current_index = mxlTimestampToIndex(&grain_rate_, now); + constexpr int64_t read_delay_grains = 1; + read_index_ = current_index - read_delay_grains; + aligned_ = true; +} + void PassthroughNode::process() { - if (!reader_ || !writer_) { + if (!reader_ || !writer_ || !aligned_) { return; } + auto wait_ns = mxlGetNsUntilIndex(read_index_, &grain_rate_); + if (wait_ns > 0 && wait_ns < 500000000ULL) { + mxlSleepForNs(wait_ns); + } + mxlGrainInfo grain_info{}; uint8_t* payload = nullptr; auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload); if (status != MXL_STATUS_OK) { 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; + spdlog::warn("Passthrough: index {} out of range ({}), realigning", read_index_, static_cast(status)); + realign(); } return; } diff --git a/nodes/passthrough/src/passthrough_node.hpp b/nodes/passthrough/src/passthrough_node.hpp index ec4a6a4..9ee6769 100644 --- a/nodes/passthrough/src/passthrough_node.hpp +++ b/nodes/passthrough/src/passthrough_node.hpp @@ -4,7 +4,6 @@ #include -#include #include namespace dmf_node { @@ -34,12 +33,14 @@ public: nlohmann::json status() const override; private: + void realign(); + std::optional reader_; std::optional writer_; - mxlInstance mxl_instance_ = nullptr; + mxlRational grain_rate_{50, 1}; uint64_t read_index_ = 0; - uint64_t write_index_ = 0; uint64_t grains_processed_ = 0; + bool aligned_ = false; }; } // namespace dmf_node