From f9e7fe79d47536f8a9797fdefe2d373bf4c24336 Mon Sep 17 00:00:00 2001 From: JohannesItten Date: Thu, 9 Jul 2026 23:21:59 +0300 Subject: [PATCH] removed sync groups --- nodes/pip/main.cpp | 136 ++++++++++++++++++++++++--------------------- 1 file changed, 73 insertions(+), 63 deletions(-) diff --git a/nodes/pip/main.cpp b/nodes/pip/main.cpp index 580ba61..cf9e500 100644 --- a/nodes/pip/main.cpp +++ b/nodes/pip/main.cpp @@ -106,84 +106,95 @@ class PiPNode : public dmf::NodeBase { std::vector Cb0(inset_w / 2), Cb1(inset_w / 2); std::vector Cr0(inset_w / 2), Cr1(inset_w / 2); - // --- Sync group: blocks until both inputs have grain N --- - mxlFlowSynchronizationGroup sync_group{}; - mxlCreateFlowSynchronizationGroup(instance(), &sync_group); - mxlFlowSynchronizationGroupAddReader(sync_group, bg_reader); - mxlFlowSynchronizationGroupAddReader(sync_group, inset_reader); - - // --- Clock --- + // --- Clock: driven by background (inset follows best-effort) --- const mxlRational rate = {fps_num, fps_den}; uint64_t index = mxlGetCurrentIndex(&rate); log("start index=%llu", index); + // Cache the last good inset frame so we can composite even when the + // hardware input hasn't delivered a frame at the exact current index. + std::vector inset_cache( + static_cast(inset_stride) * static_cast(inset_h), 0); + bool have_inset = false; + uint64_t frame_count = 0, stall_count = 0; bool fatal = false; while (dmf::g_running.load(std::memory_order_relaxed)) { - // WaitForDataAt expects TAI nanoseconds, not a grain index. - const uint64_t tai_ns = mxlIndexToTimestamp(&rate, index); - const mxlStatus st = mxlFlowSynchronizationGroupWaitForDataAt( - sync_group, tai_ns, 200'000'000); + // Background is the master clock. + mxlGrainInfo bg_grain{}; + uint8_t* bg_buf = nullptr; + const mxlStatus bg_st = mxlFlowReaderGetGrain( + bg_reader, index, 80'000'000, &bg_grain, &bg_buf); - if (st == MXL_STATUS_OK) { - mxlGrainInfo bg_grain{}, inset_grain{}; - uint8_t* bg_buf = nullptr; + if (bg_st == MXL_STATUS_OK && bg_buf) { + // Try to read inset at the same index. Short timeout: hardware + // inputs often land a frame behind, so we don't wait long — + // falling back to the cached frame is acceptable. + mxlGrainInfo inset_grain{}; uint8_t* inset_buf = nullptr; + mxlStatus in_st = mxlFlowReaderGetGrain( + inset_reader, index, 8'000'000, &inset_grain, &inset_buf); - // Use blocking reads: sync group returns OK when headIndex >= expectedIndex, - // but the writer may not have set validSlices yet. Blocking read handles - // that race by waiting for validSlices == totalSlices. - const mxlStatus bg_st = mxlFlowReaderGetGrain( - bg_reader, index, 40'000'000, &bg_grain, &bg_buf); - const mxlStatus in_st = mxlFlowReaderGetGrain( - inset_reader, index, 40'000'000, &inset_grain, &inset_buf); - - if (bg_st != MXL_STATUS_OK || in_st != MXL_STATUS_OK || !bg_buf || !inset_buf) { - log("grain read after sync OK: bg=%s inset=%s at index=%llu", - dmf::mxl_status_str(bg_st), dmf::mxl_status_str(in_st), index); - index++; - } else { - mxlGrainInfo out_grain{}; - uint8_t* out_buf = nullptr; - const mxlStatus wst = mxlFlowWriterOpenGrain( - out_writer, index, &out_grain, &out_buf); - if (wst != MXL_STATUS_OK) { - log("writer OpenGrain failed (%s) at index=%llu", - dmf::mxl_status_str(wst), index); - } else { - std::memcpy(out_buf, bg_buf, - static_cast(bg_stride) * static_cast(bg_h)); - - dmf::v210::scale_and_overlay( - inset_buf, inset_stride, inset_w, inset_h, - out_buf, out_stride, - pip_x, pip_y, clamped_w, clamped_h, - Y0, Y1, Cb0, Cb1, Cr0, Cr1); - - out_grain.flags = (bg_grain.flags | inset_grain.flags) & MXL_GRAIN_FLAG_INVALID; - out_grain.validSlices = out_grain.totalSlices; - mxlFlowWriterCommitGrain(out_writer, &out_grain); - frame_count++; - if (frame_count % 25 == 0) - log("heartbeat frames=%llu stalls=%llu index=%llu", - frame_count, stall_count, index); - } - index++; + if (in_st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) { + // Inset has moved ahead — grab its latest committed frame. + mxlFlowRuntimeInfo ri{}; + mxlFlowReaderGetRuntimeInfo(inset_reader, &ri); + in_st = mxlFlowReaderGetGrain( + inset_reader, ri.headIndex, 8'000'000, &inset_grain, &inset_buf); + stall_count++; } - } else if (st == MXL_ERR_TIMEOUT || - st == MXL_ERR_OUT_OF_RANGE_TOO_EARLY || - st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) { + if (in_st == MXL_STATUS_OK && inset_buf) { + std::memcpy(inset_cache.data(), inset_buf, + static_cast(inset_stride) * static_cast(inset_h)); + have_inset = true; + } + // TOO_EARLY (8 ms elapsed, inset not ready): use cached frame — imperceptible. + + mxlGrainInfo out_grain{}; + uint8_t* out_buf = nullptr; + const mxlStatus wst = mxlFlowWriterOpenGrain( + out_writer, index, &out_grain, &out_buf); + if (wst == MXL_STATUS_OK) { + std::memcpy(out_buf, bg_buf, + static_cast(bg_stride) * static_cast(bg_h)); + if (have_inset) { + dmf::v210::scale_and_overlay( + inset_cache.data(), inset_stride, inset_w, inset_h, + out_buf, out_stride, + pip_x, pip_y, clamped_w, clamped_h, + Y0, Y1, Cb0, Cb1, Cr0, Cr1); + } + out_grain.flags = bg_grain.flags & MXL_GRAIN_FLAG_INVALID; + out_grain.validSlices = out_grain.totalSlices; + mxlFlowWriterCommitGrain(out_writer, &out_grain); + frame_count++; + if (frame_count % 25 == 0) + log("heartbeat frames=%llu stalls=%llu index=%llu", + frame_count, stall_count, index); + } else { + log("writer OpenGrain failed (%s) at index=%llu", + dmf::mxl_status_str(wst), index); + } + index++; + + } else if (bg_st == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { + // Background stalled longer than 80 ms — skip this index. stall_count++; - const uint64_t current = mxlGetCurrentIndex(&rate); - const uint64_t next = (current > index) ? current : index + 1; - log("sync stall (%s) at index=%llu → jumping to %llu frames=%llu", - dmf::mxl_status_str(st), index, next, frame_count); - index = next; + log("bg stall (TOO_EARLY) at index=%llu frames=%llu", index, frame_count); + index++; + + } else if (bg_st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) { + stall_count++; + mxlFlowRuntimeInfo ri{}; + mxlFlowReaderGetRuntimeInfo(bg_reader, &ri); + log("bg TOO_LATE at index=%llu → jumping to %llu frames=%llu", + index, ri.headIndex, frame_count); + index = ri.headIndex; } else { - log("sync fatal (%s) at index=%llu", dmf::mxl_status_str(st), index); + log("bg fatal (%s) at index=%llu", dmf::mxl_status_str(bg_st), index); fatal = true; break; } @@ -193,7 +204,6 @@ class PiPNode : public dmf::NodeBase { fatal ? "fatal error" : "shutdown signal", frame_count, stall_count, index); - mxlReleaseFlowSynchronizationGroup(instance(), sync_group); mxlReleaseFlowReader(instance(), bg_reader); mxlReleaseFlowReader(instance(), inset_reader); mxlReleaseFlowWriter(instance(), out_writer);