Initial commit: testpattern → fakesink pipeline over MXL shared memory

- NodeBase, Signal, FlowDef, V210 shared headers
- testpattern node: SMPTE 75% color bars writer at 25fps
- fakesink node: non-blocking MXL reader with per-second stats
- studio-manager: FlowGraph data model, graph-driven fork/exec launcher
- mxl pinned as submodule at 0ae1dc5

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
JohannesItten
2026-06-23 13:35:01 +03:00
commit ca682eaf0a
17 changed files with 1269 additions and 0 deletions
+4
View File
@@ -0,0 +1,4 @@
add_executable(dmf-node-fakesink main.cpp)
target_compile_features(dmf-node-fakesink PRIVATE cxx_std_20)
target_link_libraries(dmf-node-fakesink PRIVATE dmf-shared)
install(TARGETS dmf-node-fakesink RUNTIME DESTINATION bin)
+87
View File
@@ -0,0 +1,87 @@
#include <chrono>
#include <string>
#include <mxl/flow.h>
#include <mxl/time.h>
#include "NodeBase.hpp"
class FakeSinkNode : public dmf::NodeBase {
void run() override {
const auto flow_info = config().at("flow_id");
const auto flow_id = flow_info.at("id").get<std::string>();
const int fps_num = flow_info.value("fps_num", 25);
const int fps_den = flow_info.value("fps_den", 1);
log("flow=%s", flow_id.c_str());
log("waiting for flow to become active...");
bool active = false;
while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
mxlIsFlowActive(instance(), flow_id.c_str(), &active);
if (!active) mxlSleepForNs(100'000'000);
}
if (!dmf::g_running) return;
log("flow active — starting read");
mxlFlowReader reader{};
mxlStatus st = mxlCreateFlowReader(instance(), flow_id.c_str(), nullptr, &reader);
if (st != MXL_STATUS_OK) {
log("mxlCreateFlowReader failed (status=%d)", st);
return;
}
const mxlRational rate = {fps_num, fps_den};
uint64_t index = mxlGetCurrentIndex(&rate);
uint64_t frame_count = 0;
uint64_t invalid_count = 0;
uint64_t late_count = 0;
auto wall_start = std::chrono::steady_clock::now();
auto last_log_time = wall_start;
while (dmf::g_running.load(std::memory_order_relaxed)) {
mxlGrainInfo grain{};
uint8_t* buf = nullptr;
st = mxlFlowReaderGetGrainNonBlocking(reader, index, &grain, &buf);
if (st == MXL_STATUS_OK) {
frame_count++;
if (grain.flags & MXL_GRAIN_FLAG_INVALID) invalid_count++;
index++;
} else if (st == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
mxlSleepForNs(1'000'000); // 1 ms poll
} else if (st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
late_count++;
// Jump to the most recent frame in the ring buffer
mxlFlowRuntimeInfo ri{};
mxlFlowReaderGetRuntimeInfo(reader, &ri);
index = ri.headIndex;
} else {
log("unexpected status=%d on index=%llu", st, index);
break;
}
// Log stats every second (wall clock) — avoid per-frame fprintf
auto now = std::chrono::steady_clock::now();
if (std::chrono::duration<double>(now - last_log_time).count() >= 1.0) {
const double elapsed = std::chrono::duration<double>(now - wall_start).count();
log("frames=%llu invalid=%llu late=%llu avg=%.2f fps",
frame_count, invalid_count, late_count,
static_cast<double>(frame_count) / elapsed);
last_log_time = now;
}
}
log("stopped — total frames=%llu invalid=%llu late=%llu",
frame_count, invalid_count, late_count);
mxlReleaseFlowReader(instance(), reader);
}
};
int main() {
FakeSinkNode node;
return node.execute();
}