fix: use mxlSleepUntil delivery deadline pattern for grain timing
Match the mxl-gst-sink pattern: sleep until the next grain's delivery deadline using mxlSleepUntil, then read with a short 5ms timeout. Uses 2-grain read delay (40ms at 50fps) for buffering headroom. Previous approach of mxlSleepForNs + 20ms GetGrain timeout caused the passthrough to fall behind: each iteration took ~25ms (poll+sleep+read), missing grains and perpetually chasing headIndex via realign.
This commit is contained in:
@@ -25,9 +25,14 @@ void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader re
|
|||||||
mxlFlowReaderGetConfigInfo(*reader_, &config);
|
mxlFlowReaderGetConfigInfo(*reader_, &config);
|
||||||
grain_rate_ = config.common.grainRate;
|
grain_rate_ = config.common.grainRate;
|
||||||
|
|
||||||
realign();
|
auto now = mxlGetTime();
|
||||||
spdlog::info("Passthrough: reader added on video_in, grain_rate={}/{}, read_index={}",
|
auto current_index = mxlTimestampToIndex(&grain_rate_, now);
|
||||||
grain_rate_.numerator, grain_rate_.denominator, read_index_);
|
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() {
|
void PassthroughNode::process() {
|
||||||
if (!reader_ || !writer_ || !aligned_) {
|
if (!reader_ || !writer_ || !aligned_) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
auto wait_ns = mxlGetNsUntilIndex(read_index_, &grain_rate_);
|
auto now = mxlGetTime();
|
||||||
if (wait_ns > 0 && wait_ns < 500000000ULL) {
|
if (delivery_deadline_ > now) {
|
||||||
mxlSleepForNs(wait_ns);
|
auto sleep_ns = delivery_deadline_ - now;
|
||||||
|
if (sleep_ns < 500000000ULL) {
|
||||||
|
mxlSleepUntil(delivery_deadline_);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
mxlGrainInfo grain_info{};
|
mxlGrainInfo grain_info{};
|
||||||
uint8_t* payload = nullptr;
|
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_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) {
|
||||||
spdlog::warn("Passthrough: index {} out of range ({}), realigning", read_index_, static_cast<int>(status));
|
auto old_index = read_index_;
|
||||||
realign();
|
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<int>(status), read_index_);
|
||||||
}
|
}
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -79,8 +84,10 @@ void PassthroughNode::process() {
|
|||||||
uint8_t* out_payload = nullptr;
|
uint8_t* out_payload = nullptr;
|
||||||
status = mxlFlowWriterOpenGrain(*writer_, grain_info.index, &out_grain, &out_payload);
|
status = mxlFlowWriterOpenGrain(*writer_, grain_info.index, &out_grain, &out_payload);
|
||||||
if (status != MXL_STATUS_OK) {
|
if (status != MXL_STATUS_OK) {
|
||||||
spdlog::warn("Passthrough: failed to open output grain at index {}: {}", grain_info.index, static_cast<int>(status));
|
spdlog::warn("Passthrough: failed to open output grain at index {}: {}",
|
||||||
read_index_ = grain_info.index + 1;
|
grain_info.index, static_cast<int>(status));
|
||||||
|
read_index_++;
|
||||||
|
delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -91,7 +98,8 @@ void PassthroughNode::process() {
|
|||||||
out_grain.flags = grain_info.flags;
|
out_grain.flags = grain_info.flags;
|
||||||
mxlFlowWriterCommitGrain(*writer_, &out_grain);
|
mxlFlowWriterCommitGrain(*writer_, &out_grain);
|
||||||
|
|
||||||
read_index_ = grain_info.index + 1;
|
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) {
|
||||||
|
|||||||
@@ -33,12 +33,13 @@ public:
|
|||||||
nlohmann::json status() const override;
|
nlohmann::json status() const override;
|
||||||
|
|
||||||
private:
|
private:
|
||||||
void realign();
|
static constexpr int64_t READ_DELAY_GRAINS = 2;
|
||||||
|
|
||||||
std::optional<mxlFlowReader> reader_;
|
std::optional<mxlFlowReader> reader_;
|
||||||
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;
|
||||||
};
|
};
|
||||||
|
|||||||
Reference in New Issue
Block a user