itten/dmf-studio-rnd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

removed sync groups

Browse Source
This commit is contained in:
JohannesItten
2026-07-09 23:21:59 +03:00
parent 1f133507bf
commit f9e7fe79d4
1 changed files with 73 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nodes/pip/main.cpp
+73 -63
View File
@@ -106,84 +106,95 @@ class PiPNode : public dmf::NodeBase {
std::vector<uint16_t> Cb0(inset_w / 2), Cb1(inset_w / 2);
std::vector<uint16_t> 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<uint8_t> inset_cache(
static_cast<size_t>(inset_stride) * static_cast<size_t>(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<size_t>(bg_stride) * static_cast<size_t>(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<size_t>(inset_stride) * static_cast<size_t>(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<size_t>(bg_stride) * static_cast<size_t>(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);