diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index 8c08c97..e8acb1c 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -25,9 +25,14 @@ void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader re 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_); + 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", + grain_rate_.numerator, grain_rate_.denominator, read_index_, READ_DELAY_GRAINS); } } @@ -45,32 +50,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_ || !aligned_) { return; } - auto wait_ns = mxlGetNsUntilIndex(read_index_, &grain_rate_); - if (wait_ns > 0 && wait_ns < 500000000ULL) { - mxlSleepForNs(wait_ns); + auto now = mxlGetTime(); + if (delivery_deadline_ > now) { + auto sleep_ns = delivery_deadline_ - now; + if (sleep_ns < 500000000ULL) { + mxlSleepUntil(delivery_deadline_); + } } mxlGrainInfo grain_info{}; uint8_t* payload = nullptr; - auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload); + auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 5000000ULL, &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) { - spdlog::warn("Passthrough: index {} out of range ({}), realigning", read_index_, static_cast(status)); - realign(); + auto old_index = read_index_; + 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_); } return; } @@ -79,8 +84,10 @@ void PassthroughNode::process() { uint8_t* out_payload = nullptr; 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 {}: {}", grain_info.index, static_cast(status)); - read_index_ = grain_info.index + 1; + 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; } @@ -91,7 +98,8 @@ void PassthroughNode::process() { out_grain.flags = grain_info.flags; mxlFlowWriterCommitGrain(*writer_, &out_grain); - read_index_ = grain_info.index + 1; + 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 9ee6769..8045c4a 100644 --- a/nodes/passthrough/src/passthrough_node.hpp +++ b/nodes/passthrough/src/passthrough_node.hpp @@ -33,12 +33,13 @@ public: nlohmann::json status() const override; private: - void realign(); + static constexpr int64_t READ_DELAY_GRAINS = 2; std::optional reader_; 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; };