MTL implementation md

Reviewed-on: #6

CMakeLists.txt

@@ -113,6 +113,8 @@ if(DECKLINK_SDK_DIR)
 endif()
 
 add_subdirectory(nodes/videoin)
+add_subdirectory(nodes/pip)
+add_subdirectory(nodes/gaindb)
 
 # ── Asio standalone (needed by Crow; no Boost) ───────────────────────────────
 FetchContent_Declare(asio_fc

combiner.md

@@ -1,138 +0,0 @@
-# AV Combiner Node — Implementation Steps
-
-Takes video from one upstream node and audio from another, outputs both as new MXL flows.
-Does NOT need a sync group — audio and video run at different rates and are handled independently.
-
----
-
-## Step 1 — Frontend (`dmf-studio-ui/src/nodeTypes.ts`)
-
-Add the node definition. Port IDs use `video_in-in` / `video_out-out` pattern so that
-`handleToPort()` produces distinct keys (`video_in_flow_id` vs `video_out_flow_id`).
-
-```typescript
-avcombiner: {
-  type: 'avcombiner',
-  label: 'AV Combiner',
-  ports: [
-    { id: 'video_in-in',   kind: 'video', direction: 'in'  },
-    { id: 'audio_in-in',   kind: 'audio', direction: 'in'  },
-    { id: 'video_out-out', kind: 'video', direction: 'out' },
-    { id: 'audio_out-out', kind: 'audio', direction: 'out' },
-  ],
-  params: [],
-},
-```
-
-Config keys the node receives:
-- `video_in_flow_id.id`  — input video flow UUID
-- `audio_in_flow_id.id`  — input audio flow UUID
-- `video_out_flow_id.id` — output video flow UUID
-- `audio_out_flow_id.id` — output audio flow UUID
-
----
-
-## Step 2 — CMake
-
-Create `nodes/avcombiner/CMakeLists.txt`:
-
-```cmake
-add_executable(dmf-node-avcombiner main.cpp)
-target_compile_features(dmf-node-avcombiner PRIVATE cxx_std_20)
-target_link_libraries(dmf-node-avcombiner PRIVATE dmf-shared)
-install(TARGETS dmf-node-avcombiner RUNTIME DESTINATION bin)
-```
-
-In the root `CMakeLists.txt`, add alongside the other nodes:
-
-```cmake
-add_subdirectory(nodes/avcombiner)
-```
-
----
-
-## Step 3 — `nodes/avcombiner/main.cpp`
-
-Structure (follow the same patterns as `ndiout`):
-
-```
-1. Check video_in_flow_id and video_out_flow_id both present — exit if not
-2. Check audio_in_flow_id and audio_out_flow_id (optional — audio is optional)
-
-3. Wait for video input flow active  (mxlIsFlowActive loop, 100ms sleep)
-4. Create video input reader         (mxlCreateFlowReader)
-5. Get video config info             (mxlFlowReaderGetConfigInfo → video_stride)
-6. Read video format from flow_def   (dmf::read_video_flow_info(domain(), flow_id))
-   → width, height, fps_num, fps_den
-
-7. If has_audio:
-   Wait for audio input flow active
-   Create audio input reader
-   Read audio format from flow_def   (dmf::read_audio_flow_info(domain(), flow_id))
-   → sample_rate, channels, samples_per_grain
-
-8. Create video output writer        (mxlCreateFlowWriter with make_video_flow_def)
-   → video_out_stride from configInfo.discrete.sliceSizes[0]
-
-9. If has_audio:
-   Create audio output writer        (mxlCreateFlowWriter with make_audio_flow_def)
-
-10. Init clocks:
-    video_index = mxlGetCurrentIndex(&video_rate)
-    audio_index = mxlGetCurrentIndex(&audio_rate)   // if has_audio
-
-11. Main loop (same pattern as ndiout):
-
-    // Audio — non-blocking
-    if (has_audio) {
-        mxlFlowReaderGetSamplesNonBlocking(audio_in_reader, audio_index, samples_per_grain, &in_slice)
-        if OK:
-            mxlFlowWriterOpenSamples(audio_out_writer, audio_index, samples_per_grain, &out_slice)
-            memcpy each channel fragment (frag0, frag1 wrap pattern)
-            mxlFlowWriterCommitSamples
-            audio_index += samples_per_grain
-        if TOO_LATE: jump to headIndex
-    }
-
-    // Video — non-blocking
-    mxlFlowReaderGetGrainNonBlocking(video_in_reader, video_index, &grain, &in_buf)
-    if OK:
-        mxlFlowWriterOpenGrain(video_out_writer, video_index, &out_grain, &out_buf)
-        memcpy(out_buf, in_buf, video_out_stride * height)
-        mxlFlowWriterCommitGrain
-        video_index++
-    if TOO_EARLY: mxlSleepForNs(1ms)
-    if TOO_LATE:  jump to headIndex
-```
-
-### Audio memcpy pattern (wrapped ring buffer)
-
-MXL audio slices can wrap around the ring buffer — always copy both fragments:
-
-```cpp
-const size_t frag0 = in_slice.base.fragments[0].size / sizeof(float);
-const size_t frag1 = in_slice.base.fragments[1].size / sizeof(float);
-for (int c = 0; c < channels; ++c) {
-    const auto* src0 = reinterpret_cast<const float*>(
-        static_cast<const uint8_t*>(in_slice.base.fragments[0].pointer) + c * in_slice.stride);
-    auto* dst0 = reinterpret_cast<float*>(
-        static_cast<uint8_t*>(out_slice.base.fragments[0].pointer) + c * out_slice.stride);
-    std::memcpy(dst0, src0, frag0 * sizeof(float));
-    if (frag1 > 0) {
-        const auto* src1 = reinterpret_cast<const float*>(
-            static_cast<const uint8_t*>(in_slice.base.fragments[1].pointer) + c * in_slice.stride);
-        auto* dst1 = reinterpret_cast<float*>(
-            static_cast<uint8_t*>(out_slice.base.fragments[1].pointer) + c * out_slice.stride);
-        std::memcpy(dst1, src1, frag1 * sizeof(float));
-    }
-}
-```
-
----
-
-## Reference nodes
-
-- Input flow setup (wait + reader + flow_def read): `nodes/ndiout/main.cpp`, `nodes/decklinkout/main.cpp`
-- Output flow setup (writer creation): `nodes/ndiin/main.cpp`, `nodes/decklinkin/main.cpp`
-- Audio memcpy pattern: `nodes/ndiout/main.cpp` lines 172–190
-- `make_video_flow_def` / `make_audio_flow_def` / `read_video_flow_info` / `read_audio_flow_info`: `shared/FlowDef.hpp`

gain-check.json

@@ -0,0 +1,24 @@
+{
+  "nodes": [
+    { "id": "testpattern", "type": "testpattern", "params": { "pattern": "bars" } },
+    { "id": "gaindb",      "type": "gaindb",      "params": {} },
+    { "id": "ndiout",      "type": "ndiout",      "params": {} }
+  ],
+  "edges": [
+    {
+      "from": "testpattern", "from_port": "video_flow_id",
+      "to":   "ndiout",      "to_port":   "video_flow_id",
+      "format": { "kind": "video", "width": 1920, "height": 1080, "fps_num": 25, "fps_den": 1 }
+    },
+    {
+      "from": "testpattern", "from_port": "audio_flow_id",
+      "to":   "gaindb",      "to_port":   "audio_in_flow_id",
+      "format": { "kind": "audio", "sample_rate": 48000, "channels": 2, "bit_depth": 32 }
+    },
+    {
+      "from": "gaindb", "from_port": "audio_out_flow_id",
+      "to":   "ndiout", "to_port":   "audio_flow_id",
+      "format": { "kind": "audio", "sample_rate": 48000, "channels": 2, "bit_depth": 32 }
+    }
+  ]
+}

mtl.md

@@ -0,0 +1,81 @@
+# ST 2110-20 receiver with Intel MTL
+
+The MTL pattern is structurally identical to `DeckLinkReceiver` — callback-based, MTL manages frame buffers, you drain them into MXL.
+
+## MTL initialization (once per process)
+
+```cpp
+mtl_init_params p{};
+p.ports[MTL_PORT_P] = "0000:31:00.0"; // PCI address or netdev for AF_XDP
+p.num_ports = 1;
+p.flags = MTL_FLAG_BIND_NUMA;
+// For AF_XDP mode (no DPDK hugepages needed):
+p.transport = MTL_TRANSPORT_AF_XDP;
+
+mtl_handle dev = mtl_init(&p);
+```
+
+## Create RX session
+
+```cpp
+st20_rx_ops ops{};
+ops.port.num_port = 1;
+memcpy(ops.port.sip_addr[MTL_PORT_P], multicast_ip, 4); // join this group
+ops.port.udp_port[MTL_PORT_P] = 20000;
+ops.width  = 1280;
+ops.height = 720;
+ops.fps    = ST_FPS_P25;
+ops.fmt    = ST20_FMT_YUV_422_10BIT; // RFC 4175 packed — see note below
+ops.framebuff_cnt = 3;
+ops.notify_frame_ready = on_frame_ready; // your callback
+ops.priv = this;
+
+st20_rx_handle rx = st20_rx_create(dev, &ops);
+```
+
+## Callback — replaces `wait_for_frame()`
+
+```cpp
+static int on_frame_ready(void* priv, void* frame, st20_rx_frame_meta* meta) {
+    auto* self = static_cast<St2110InNode*>(priv);
+    // frame points to a complete assembled frame — MTL did the reassembly
+    const uint64_t index = mxlGetCurrentIndex(&self->video_rate);
+    mxlGrainInfo grain{};
+    uint8_t* buf = nullptr;
+    if (mxlFlowWriterOpenGrain(self->writer, index, &grain, &buf) == MXL_STATUS_OK) {
+        memcpy(buf, frame, self->frame_size); // ← see format note
+        grain.validSlices = grain.totalSlices;
+        mxlFlowWriterCommitGrain(self->writer, &grain);
+    }
+    st20_rx_put_framebuff(self->rx, frame); // return buffer to MTL pool
+    return 0;
+}
+```
+
+## Format conversion: RFC 4175 vs V210
+
+ST 2110-20 wire format is RFC 4175 packed 10-bit — **not V210**. They encode the same YCbCr 4:2:2 10-bit data differently:
+- RFC 4175: 5 bytes per 2 pixels, big-endian packed
+- V210: 4 bytes per 3 luma + 2 chroma, little-endian with padding bits
+
+MXL in this project uses V210. Check MTL's `output_fmt` option — newer MTL versions support `ST_FRAME_FMT_V210` as the output format, which means MTL does the conversion internally. If your version doesn't have it, you'll need a small RFC4175→V210 conversion step before the `memcpy`.
+
+Check `st_frame_fmt` enum in MTL headers for what's available.
+
+## Teardown
+
+```cpp
+st20_rx_free(rx);
+mtl_uninit(dev);
+```
+
+## Main loop
+
+The callback is called from MTL's internal thread (like DeckLink's), so the MXL write happens inside the callback rather than in the main loop. The main loop just blocks on `g_running`:
+
+```cpp
+while (dmf::g_running.load(std::memory_order_relaxed))
+    mxlSleepForNs(10'000'000);
+```
+
+MTL drives the pacing, exactly like DeckLink hardware does.

nodes/decklinkin/main.cpp

@@ -95,23 +95,29 @@ class DeckLinkInNode : public dmf::NodeBase {
         std::vector<float>   audio_buf(static_cast<size_t>(max_audio_samples) * static_cast<size_t>(channels));
 
         // --- Clock ---
+        // video_index is determined AFTER each hardware frame arrives so it reflects
+        // the actual TAI slot the frame landed in.  wait_for_frame() is the natural
+        // pacer — no separate sleep needed.
         const mxlRational video_rate = {fps_num, fps_den};
         const mxlRational audio_rate = {receiver.audio_info.sample_rate, 1};
-        uint64_t          video_index = mxlGetCurrentIndex(&video_rate);
         uint64_t audio_index = has_audio ? mxlGetCurrentIndex(&audio_rate) : 0;
-        log("start video_index=%llu", static_cast<unsigned long long>(video_index));
+        uint64_t frame_count = 0, drop_count = 0;
+        log("ready, waiting for first frame...");
 
         // --- Capture loop ---
         while (dmf::g_running.load(std::memory_order_relaxed)) {
             int samples_written = 0;
 
-            // DeckLink delivers one video frame + accompanying audio per callback.
+            // Blocks until DeckLink hardware delivers a frame — this IS the pacing.
             if (!receiver.wait_for_frame(
                     frame_buf.data(), video_stride, width, height,
                     (has_audio && audio_writer) ? audio_buf.data() : nullptr,
                     max_audio_samples,
                     (has_audio && audio_writer) ? &samples_written : nullptr)) break;
 
+            // Resolve TAI index now — after the frame arrived, not before.
+            const uint64_t video_index = mxlGetCurrentIndex(&video_rate);
+
             // Video grain
             mxlGrainInfo grain{};
             uint8_t*     video_buf_ptr = nullptr;
@@ -121,9 +127,17 @@ class DeckLinkInNode : public dmf::NodeBase {
                 grain.flags       = 0;
                 grain.validSlices = grain.totalSlices;
                 mxlFlowWriterCommitGrain(video_writer, &grain);
+                frame_count++;
+                if (frame_count % 25 == 0)
+                    log("heartbeat frames=%llu  drops=%llu  index=%llu",
+                        frame_count, drop_count, video_index);
+            } else {
+                drop_count++;
+                log("OpenGrain failed (%s) at index=%llu  drops=%llu",
+                    dmf::mxl_status_str(vst), video_index, drop_count);
             }
 
-            // Audio samples (same fragment-wrap pattern as videoin)
+            // Audio samples
             if (has_audio && audio_writer && samples_written > 0) {
                 mxlMutableWrappedMultiBufferSlice slice{};
                 mxlStatus ast = mxlFlowWriterOpenSamples(
@@ -153,14 +167,9 @@ class DeckLinkInNode : public dmf::NodeBase {
                 }
                 audio_index += static_cast<uint64_t>(samples_written);
             }
-
-            // Pace video to the MXL clock
-            const uint64_t ns = mxlGetNsUntilIndex(video_index + 1, &video_rate);
-            if (ns > 0 && ns < 2'000'000'000ULL) mxlSleepForNs(ns);
-            video_index = mxlGetCurrentIndex(&video_rate);
         }
 
-        log("stopped at video_index=%llu", static_cast<unsigned long long>(video_index));
+        log("stopped  frames=%llu  drops=%llu", frame_count, drop_count);
         mxlReleaseFlowWriter(instance(), video_writer);
         if (audio_writer) mxlReleaseFlowWriter(instance(), audio_writer);
     }

nodes/gaindb/CMakeLists.txt

@@ -0,0 +1,4 @@
+add_executable(dmf-node-gaindb main.cpp)
+target_compile_features(dmf-node-gaindb PRIVATE cxx_std_20)
+target_link_libraries(dmf-node-gaindb PRIVATE dmf-shared)
+install(TARGETS dmf-node-gaindb RUNTIME DESTINATION bin)

nodes/gaindb/main.cpp

@@ -0,0 +1,153 @@
+#include <cmath>
+#include <cstring>
+#include <string>
+#include <vector>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+#include "NodeBase.hpp"
+#include "FlowDef.hpp"
+
+class GainDbNode : public dmf::NodeBase {
+    void run() override {
+        if (!config().contains("audio_in_flow_id"))  { log("no audio input connected");  return; }
+        if (!config().contains("audio_out_flow_id")) { log("no audio output connected"); return; }
+
+        const auto in_id  = config().at("audio_in_flow_id").at("id").get<std::string>();
+        const auto out_id = config().at("audio_out_flow_id").at("id").get<std::string>();
+
+        const float gain_db     = config().value("gain_db", 0.0f);
+        const float gain_linear = std::pow(10.0f, gain_db / 20.0f);
+        log("gain=%.2f dB  (x%.4f linear)", gain_db, gain_linear);
+
+        // --- wait for input flow ---
+        log("waiting for flow %s...", in_id.c_str());
+        bool active = false;
+        while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
+            mxlIsFlowActive(instance(), in_id.c_str(), &active);
+            if (!active) mxlSleepForNs(100'000'000);
+        }
+        if (!dmf::g_running) return;
+
+        // --- create reader ---
+        mxlFlowReader in_reader{};
+        if (mxlCreateFlowReader(instance(), in_id.c_str(), "", &in_reader) != MXL_STATUS_OK) {
+            log("mxlCreateFlowReader failed"); return;
+        }
+
+        // --- read format from upstream flow_def ---
+        const auto fi = dmf::read_audio_flow_info(domain(), in_id);
+        const int sample_rate       = fi.sample_rate;
+        const int channels          = fi.channels;
+        const int samples_per_grain = fi.samples_per_grain;
+        log("audio: %d Hz  %dch  %d samples/grain", sample_rate, channels, samples_per_grain);
+
+        // --- create output writer (same format as input) ---
+        mxlFlowWriter     out_writer{};
+        mxlFlowConfigInfo out_cfg{};
+        bool created = false;
+        // gr_num/gr_den = sample_rate/samples_per_grain (e.g. 48000/1920 = 25/1)
+        const mxlStatus wst = mxlCreateFlowWriter(
+            instance(),
+            dmf::make_audio_flow_def(out_id, node_id(),
+                sample_rate, channels, /*bit_depth=*/32,
+                /*gr_num=*/sample_rate, /*gr_den=*/samples_per_grain).c_str(),
+            "", &out_writer, &out_cfg, &created);
+        if (wst != MXL_STATUS_OK) {
+            log("mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(wst));
+            mxlReleaseFlowReader(instance(), in_reader);
+            return;
+        }
+        log("output ready  buffer=%u samples", out_cfg.continuous.bufferLength);
+
+        // temp flat buffer for one channel — handles ring wrap on both in and out slices
+        std::vector<float> temp(static_cast<size_t>(samples_per_grain));
+
+        const mxlRational audio_rate = {sample_rate, 1};
+        uint64_t audio_index = mxlGetCurrentIndex(&audio_rate);
+        log("start index=%llu", audio_index);
+
+        uint64_t grain_count = 0, late_count = 0;
+
+        while (dmf::g_running.load(std::memory_order_relaxed)) {
+            mxlWrappedMultiBufferSlice in_slice{};
+            const mxlStatus rst = mxlFlowReaderGetSamplesNonBlocking(
+                in_reader, audio_index,
+                static_cast<size_t>(samples_per_grain), &in_slice);
+
+            if (rst == MXL_STATUS_OK) {
+                mxlMutableWrappedMultiBufferSlice out_slice{};
+                const mxlStatus ost = mxlFlowWriterOpenSamples(
+                    out_writer, audio_index,
+                    static_cast<size_t>(samples_per_grain), &out_slice);
+
+                if (ost == MXL_STATUS_OK) {
+                    const size_t in_f0  = in_slice.base.fragments[0].size  / sizeof(float);
+                    const size_t in_f1  = in_slice.base.fragments[1].size  / sizeof(float);
+                    const size_t out_f0 = out_slice.base.fragments[0].size / sizeof(float);
+                    const size_t out_f1 = out_slice.base.fragments[1].size / sizeof(float);
+
+                    for (int c = 0; c < channels; ++c) {
+                        // flatten input channel c into temp
+                        const auto* s0 = reinterpret_cast<const float*>(
+                            static_cast<const uint8_t*>(in_slice.base.fragments[0].pointer)
+                            + c * in_slice.stride);
+                        std::memcpy(temp.data(), s0, in_f0 * sizeof(float));
+                        if (in_f1 > 0) {
+                            const auto* s1 = reinterpret_cast<const float*>(
+                                static_cast<const uint8_t*>(in_slice.base.fragments[1].pointer)
+                                + c * in_slice.stride);
+                            std::memcpy(temp.data() + in_f0, s1, in_f1 * sizeof(float));
+                        }
+
+                        // apply gain
+                        for (size_t i = 0; i < static_cast<size_t>(samples_per_grain); ++i)
+                            temp[i] *= gain_linear;
+
+                        // scatter to output channel c (may also wrap)
+                        auto* d0 = reinterpret_cast<float*>(
+                            static_cast<uint8_t*>(out_slice.base.fragments[0].pointer)
+                            + c * out_slice.stride);
+                        std::memcpy(d0, temp.data(), out_f0 * sizeof(float));
+                        if (out_f1 > 0) {
+                            auto* d1 = reinterpret_cast<float*>(
+                                static_cast<uint8_t*>(out_slice.base.fragments[1].pointer)
+                                + c * out_slice.stride);
+                            std::memcpy(d1, temp.data() + out_f0, out_f1 * sizeof(float));
+                        }
+                    }
+                    mxlFlowWriterCommitSamples(out_writer);
+                    grain_count++;
+                } else {
+                    log("OpenSamples failed (%s) at index=%llu",
+                        dmf::mxl_status_str(ost), audio_index);
+                }
+
+                audio_index += static_cast<uint64_t>(samples_per_grain);
+                const uint64_t ns = mxlGetNsUntilIndex(audio_index, &audio_rate);
+                if (ns > 0 && ns < 200'000'000ULL) mxlSleepForNs(ns);
+
+            } else if (rst == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
+                mxlSleepForNs(1'000'000);
+
+            } else if (rst == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
+                late_count++;
+                mxlFlowRuntimeInfo ri{};
+                mxlFlowReaderGetRuntimeInfo(in_reader, &ri);
+                audio_index = ri.headIndex;
+
+            } else {
+                log("read error (%s) at index=%llu", dmf::mxl_status_str(rst), audio_index);
+                break;
+            }
+        }
+
+        log("stopped  grains=%llu  late=%llu", grain_count, late_count);
+        mxlReleaseFlowReader(instance(), in_reader);
+        mxlReleaseFlowWriter(instance(), out_writer);
+    }
+};
+
+int main() {
+    GainDbNode node;
+    return node.execute();
+}

nodes/pip/CMakeLists.txt

@@ -0,0 +1,4 @@
+add_executable(dmf-node-pip main.cpp)
+target_compile_features(dmf-node-pip PRIVATE cxx_std_20)
+target_link_libraries(dmf-node-pip PRIVATE dmf-shared)
+install(TARGETS dmf-node-pip RUNTIME DESTINATION bin)

nodes/pip/main.cpp

@@ -0,0 +1,201 @@
+#include <algorithm>
+#include <cstring>
+#include <string>
+#include <vector>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+#include "NodeBase.hpp"
+#include "FlowDef.hpp"
+#include "V210.hpp"
+
+// Round down to nearest V210-aligned pixel count (multiple of 6).
+static int v210_align(int pixels) { return (pixels / 6) * 6; }
+
+class PiPNode : public dmf::NodeBase {
+    void run() override {
+        if (!config().contains("background_flow_id")) { log("no background connected"); return; }
+        if (!config().contains("inset_flow_id"))      { log("no inset connected");      return; }
+        if (!config().contains("video_flow_id"))      { log("no output connected");      return; }
+
+        const auto bg_id    = config().at("background_flow_id").at("id").get<std::string>();
+        const auto inset_id = config().at("inset_flow_id").at("id").get<std::string>();
+        const auto out_id   = config().at("video_flow_id").at("id").get<std::string>();
+
+        // Position and size of the inset in the output frame.
+        // x and width are snapped to 6-pixel V210 boundaries.
+        const int pip_x = v210_align(config().value("x", 0));
+        const int pip_y = config().value("y", 0);
+        const int pip_w = v210_align(config().value("width",  480));
+        const int pip_h = config().value("height", 270);
+
+        // --- Wait for both input flows ---
+        for (const auto* fid : {&bg_id, &inset_id}) {
+            log("waiting for flow %s...", fid->c_str());
+            bool active = false;
+            while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
+                mxlIsFlowActive(instance(), fid->c_str(), &active);
+                if (!active) mxlSleepForNs(100'000'000);
+            }
+            if (!dmf::g_running) return;
+        }
+
+        // --- Create readers ---
+        mxlFlowReader     bg_reader{}, inset_reader{};
+        mxlFlowConfigInfo bg_cfg{}, inset_cfg{};
+
+        if (mxlCreateFlowReader(instance(), bg_id.c_str(), "", &bg_reader) != MXL_STATUS_OK) {
+            log("background mxlCreateFlowReader failed"); return;
+        }
+        if (mxlCreateFlowReader(instance(), inset_id.c_str(), "", &inset_reader) != MXL_STATUS_OK) {
+            log("inset mxlCreateFlowReader failed");
+            mxlReleaseFlowReader(instance(), bg_reader);
+            return;
+        }
+        mxlFlowReaderGetConfigInfo(bg_reader,    &bg_cfg);
+        mxlFlowReaderGetConfigInfo(inset_reader, &inset_cfg);
+
+        const uint32_t bg_stride    = bg_cfg.discrete.sliceSizes[0];
+        const uint32_t inset_stride = inset_cfg.discrete.sliceSizes[0];
+
+        // --- Read formats from flow_def.json ---
+        const auto bg_fi    = dmf::read_video_flow_info(domain(), bg_id);
+        const auto inset_fi = dmf::read_video_flow_info(domain(), inset_id);
+
+        const int bg_w    = bg_fi.width;
+        const int bg_h    = bg_fi.height;
+        const int fps_num = bg_fi.fps_num;
+        const int fps_den = bg_fi.fps_den;
+        const int inset_w = inset_fi.width;
+        const int inset_h = inset_fi.height;
+
+        log("background: %dx%d @ %d/%d fps  stride=%u",
+            bg_w, bg_h, fps_num, fps_den, bg_stride);
+        log("inset src:  %dx%d  stride=%u", inset_w, inset_h, inset_stride);
+        log("pip region: %dx%d at (%d,%d)", pip_w, pip_h, pip_x, pip_y);
+
+        // Clamp pip region to background bounds
+        const int clamped_w = v210_align(std::min(pip_w, bg_w - pip_x));
+        const int clamped_h = std::min(pip_h, bg_h - pip_y);
+        if (clamped_w <= 0 || clamped_h <= 0) {
+            log("pip region is outside background bounds — exiting");
+            mxlReleaseFlowReader(instance(), bg_reader);
+            mxlReleaseFlowReader(instance(), inset_reader);
+            return;
+        }
+
+        // --- Create output writer (same format as background) ---
+        mxlFlowWriter     out_writer{};
+        mxlFlowConfigInfo out_cfg{};
+        bool created = false;
+        mxlStatus vst = mxlCreateFlowWriter(
+            instance(),
+            dmf::make_video_flow_def(out_id, node_id(), bg_w, bg_h, fps_num, fps_den).c_str(),
+            "", &out_writer, &out_cfg, &created);
+        if (vst != MXL_STATUS_OK) {
+            log("mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(vst));
+            mxlReleaseFlowReader(instance(), bg_reader);
+            mxlReleaseFlowReader(instance(), inset_reader);
+            return;
+        }
+        const uint32_t out_stride = out_cfg.discrete.sliceSizes[0];
+        log("output: stride=%u  grain=%u B  ring=%u grains",
+            out_stride, out_stride * static_cast<uint32_t>(bg_h), out_cfg.discrete.grainCount);
+
+        // --- Pre-allocate bilinear scaling workspace (reused every frame) ---
+        std::vector<uint16_t> Y0(inset_w), Y1(inset_w);
+        std::vector<uint16_t> Cb0(inset_w / 2), Cb1(inset_w / 2);
+        std::vector<uint16_t> Cr0(inset_w / 2), Cr1(inset_w / 2);
+
+        // --- 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);
+
+        uint64_t frame_count = 0, stall_count = 0;
+        bool fatal = false;
+
+        while (dmf::g_running.load(std::memory_order_relaxed)) {
+            // 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 (bg_st == MXL_STATUS_OK && bg_buf) {
+                // Try inset at same index (short timeout — hardware often lands one frame behind).
+                // On any miss, fall back to headIndex: the ring buffer is the cache.
+                mxlGrainInfo inset_grain{};
+                uint8_t* inset_buf = nullptr;
+                mxlStatus in_st = mxlFlowReaderGetGrain(
+                    inset_reader, index, 8'000'000, &inset_grain, &inset_buf);
+
+                if (in_st != MXL_STATUS_OK) {
+                    mxlFlowRuntimeInfo ri{};
+                    mxlFlowReaderGetRuntimeInfo(inset_reader, &ri);
+                    in_st = mxlFlowReaderGetGrain(
+                        inset_reader, ri.headIndex, 8'000'000, &inset_grain, &inset_buf);
+                    stall_count++;
+                }
+
+                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 (in_st == MXL_STATUS_OK && inset_buf) {
+                        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 & 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++;
+                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("bg fatal (%s) at index=%llu", dmf::mxl_status_str(bg_st), index);
+                fatal = true;
+                break;
+            }
+        }
+
+        log("stopped: %s  frames=%llu  stalls=%llu  index=%llu",
+            fatal ? "fatal error" : "shutdown signal",
+            frame_count, stall_count, index);
+
+        mxlReleaseFlowReader(instance(), bg_reader);
+        mxlReleaseFlowReader(instance(), inset_reader);
+        mxlReleaseFlowWriter(instance(), out_writer);
+    }
+};
+
+int main() {
+    PiPNode node;
+    return node.execute();
+}

nodes/testpattern/main.cpp

@@ -69,8 +69,9 @@ class TestPatternNode : public dmf::NodeBase {
         const size_t      samples_per_frame =
             static_cast<size_t>(sample_rate) * static_cast<size_t>(fps_den) / static_cast<size_t>(fps_num);
 
-        // -18 dBFS broadcast reference level
+        // default -18 dBFS broadcast reference level
-        const float amplitude = static_cast<float>(std::pow(10.0, -18.0 / 20.0));
+        const float amplitude_db = config().value("amplitude_db", -18.0f);
+        const float amplitude    = std::pow(10.0f, amplitude_db / 20.0f);
 
         uint64_t video_index = mxlGetCurrentIndex(&video_rate);
         uint64_t audio_index = 0;

shared/V210.hpp

@@ -1,8 +1,10 @@
 #pragma once
+#include <algorithm>
 #include <array>
 #include <cstddef>
 #include <cstdint>
 #include <cstring>
+#include <vector>
 
 namespace dmf::v210 {
 
@@ -173,4 +175,124 @@ inline void YUV422P10toV210(
     }
 }
 
+// Unpack one V210 row into planar uint16_t Y (width values),
+// Cb and Cr (width/2 values each). Width must be a multiple of 6.
+inline void unpack_row(const uint8_t* src, int width,
+                       uint16_t* Y, uint16_t* Cb, uint16_t* Cr)
+{
+    const auto* w = reinterpret_cast<const uint32_t*>(src);
+    const int blocks = width / 6;
+    for (int b = 0; b < blocks; ++b, w += 4) {
+        const int x = b * 6;
+        Cb[x/2]   = (w[0] >>  0) & 0x3FF;
+        Y[x]      = (w[0] >> 10) & 0x3FF;
+        Cr[x/2]   = (w[0] >> 20) & 0x3FF;
+        Y[x+1]    = (w[1] >>  0) & 0x3FF;
+        Cb[x/2+1] = (w[1] >> 10) & 0x3FF;
+        Y[x+2]    = (w[1] >> 20) & 0x3FF;
+        Cr[x/2+1] = (w[2] >>  0) & 0x3FF;
+        Y[x+3]    = (w[2] >> 10) & 0x3FF;
+        Cb[x/2+2] = (w[2] >> 20) & 0x3FF;
+        Y[x+4]    = (w[3] >>  0) & 0x3FF;
+        Cr[x/2+2] = (w[3] >> 10) & 0x3FF;
+        Y[x+5]    = (w[3] >> 20) & 0x3FF;
+    }
+}
+
+// Scale the inset V210 frame into a rectangular region of dst using bilinear
+// interpolation. pip_x and pip_w must be multiples of 6 (V210 alignment).
+// Workspace vectors are passed in to avoid per-call heap allocation.
+inline void scale_and_overlay(
+    const uint8_t* inset,  uint32_t inset_stride, int inset_w, int inset_h,
+    uint8_t*       dst,    uint32_t dst_stride,
+    int pip_x, int pip_y, int pip_w, int pip_h,
+    std::vector<uint16_t>& Y0_buf,  std::vector<uint16_t>& Y1_buf,
+    std::vector<uint16_t>& Cb0_buf, std::vector<uint16_t>& Cb1_buf,
+    std::vector<uint16_t>& Cr0_buf, std::vector<uint16_t>& Cr1_buf)
+{
+    Y0_buf.resize(inset_w);  Y1_buf.resize(inset_w);
+    Cb0_buf.resize(inset_w / 2); Cb1_buf.resize(inset_w / 2);
+    Cr0_buf.resize(inset_w / 2); Cr1_buf.resize(inset_w / 2);
+
+    const int out_blocks   = pip_w / 6;
+    const int dst_x_bytes  = (pip_x / 6) * 16;
+    const float inv_pip_h  = static_cast<float>(inset_h) / pip_h;
+    const float inv_pip_w  = static_cast<float>(inset_w) / pip_w;
+    const float inv_pip_cw = static_cast<float>(inset_w / 2) / (pip_w / 2);
+
+    // Precompute horizontal source positions once — they are the same for every row.
+    // thread_local avoids heap allocation on repeated calls with the same dimensions.
+    struct XS { int x0, x1; float fx, ifx; };
+    static thread_local std::vector<XS> y_xs, c_xs;
+    static thread_local int cached_pip_w = 0, cached_inset_w = 0;
+    if (pip_w != cached_pip_w || inset_w != cached_inset_w) {
+        y_xs.resize(pip_w);
+        for (int dx = 0; dx < pip_w; ++dx) {
+            const float sx = (dx + 0.5f) * inv_pip_w - 0.5f;
+            const int   x0 = std::max(0, static_cast<int>(sx));
+            const float fx = sx - static_cast<float>(x0);
+            y_xs[dx] = { x0, std::min(inset_w - 1, x0 + 1), fx, 1.0f - fx };
+        }
+        c_xs.resize(pip_w / 2);
+        for (int cx = 0; cx < pip_w / 2; ++cx) {
+            const float sx = (cx + 0.5f) * inv_pip_cw - 0.5f;
+            const int   x0 = std::max(0, static_cast<int>(sx));
+            const float fx = sx - static_cast<float>(x0);
+            c_xs[cx] = { x0, std::min(inset_w / 2 - 1, x0 + 1), fx, 1.0f - fx };
+        }
+        cached_pip_w  = pip_w;
+        cached_inset_w = inset_w;
+    }
+
+    int cur_row0 = -1, cur_row1 = -1;
+
+    for (int dy = 0; dy < pip_h; ++dy) {
+        const float sy  = (dy + 0.5f) * inv_pip_h - 0.5f;
+        const int   sy0 = std::max(0, static_cast<int>(sy));
+        const int   sy1 = std::min(inset_h - 1, sy0 + 1);
+        const float fy  = sy - static_cast<float>(sy0);
+        const float w0  = 1.0f - fy;
+        const float w1  = fy;
+
+        if (sy0 != cur_row0) {
+            unpack_row(inset + static_cast<size_t>(sy0) * inset_stride, inset_w,
+                       Y0_buf.data(), Cb0_buf.data(), Cr0_buf.data());
+            cur_row0 = sy0;
+        }
+        if (sy1 != cur_row1) {
+            unpack_row(inset + static_cast<size_t>(sy1) * inset_stride, inset_w,
+                       Y1_buf.data(), Cb1_buf.data(), Cr1_buf.data());
+            cur_row1 = sy1;
+        }
+
+        uint8_t* dst_row = dst + static_cast<size_t>(pip_y + dy) * dst_stride + dst_x_bytes;
+
+        for (int b = 0; b < out_blocks; ++b) {
+            const int bx = b * 6;
+            uint16_t Y[6], Cb[3], Cr[3];
+
+            for (int i = 0; i < 6; ++i) {
+                const XS& xs = y_xs[bx + i];
+                Y[i] = static_cast<uint16_t>(
+                    (Y0_buf[xs.x0] * xs.ifx + Y0_buf[xs.x1] * xs.fx) * w0 +
+                    (Y1_buf[xs.x0] * xs.ifx + Y1_buf[xs.x1] * xs.fx) * w1 + 0.5f);
+            }
+            for (int i = 0; i < 3; ++i) {
+                const XS& cs = c_xs[b * 3 + i];
+                Cb[i] = static_cast<uint16_t>(
+                    (Cb0_buf[cs.x0] * cs.ifx + Cb0_buf[cs.x1] * cs.fx) * w0 +
+                    (Cb1_buf[cs.x0] * cs.ifx + Cb1_buf[cs.x1] * cs.fx) * w1 + 0.5f);
+                Cr[i] = static_cast<uint16_t>(
+                    (Cr0_buf[cs.x0] * cs.ifx + Cr0_buf[cs.x1] * cs.fx) * w0 +
+                    (Cr1_buf[cs.x0] * cs.ifx + Cr1_buf[cs.x1] * cs.fx) * w1 + 0.5f);
+            }
+
+            pack_block(dst_row + b * 16,
+                {0, Cb[0], Cr[0]}, Y[0], Y[1],
+                {0, Cb[1], Cr[1]}, Y[2], Y[3],
+                {0, Cb[2], Cr[2]}, Y[4], Y[5]);
+        }
+    }
+}
+
 } // namespace dmf::v210

tg-post.md

@@ -0,0 +1,78 @@
+# DMF Studio — dev update
+
+**DMF Studio** is an open-architecture broadcast signal processor built around a node graph model. The idea is simple: every processing element — signal generator, capture input, compositor, gain stage, output — is an independent node. You wire them together visually in a browser, hit run, and the system routes live video and audio between them in real time.
+
+Unlike traditional broadcast routers or vision mixers that bundle routing and processing into a closed appliance, DMF Studio runs on commodity hardware (a Linux server with DeckLink cards) and exposes the entire signal graph as software. Every node is a separate C++ process. Inter-node transport is MXL shared memory — a low-latency ring buffer system that timestamps every grain against a TAI clock. The browser frontend builds a JSON graph, sends it over WebSocket to a studio-manager daemon, which forks and execs the node processes and wires them together via UUID-addressed memory flows.
+
+The long-term goal: a fully modular, vendor-neutral broadcast backbone — live production, ingest, playout — where the signal graph is code you can version-control and deploy like any other infrastructure.
+
+---
+
+Building a node-based broadcast signal processor where every node is an isolated C++ process exchanging video and audio through MXL shared memory over a TAI clock grid. Think of it as a modular patchbay you assemble visually and run on bare metal — no GPU, no frameworks, just shared memory and tight timing.
+
+This stage we got a live SDI → PiP → SDI chain running from a browser UI. Here's what it took.
+
+---
+
+**What shipped**
+
+- **PiP node** — composites two live video flows into one. Background drives the clock, inset overlays with V210 10-bit bilinear scaling. Supports any combination of sources: testpattern, NDI In, DeckLink In.
+
+- **gaindb node** — audio gain/attenuation node. Reads a continuous MXL audio flow, applies dB gain via `pow(10, gain_db/20)` per sample, writes back out. Chain it anywhere between a source and a sink.
+
+- **DeckLink In / NDI In** — both now write to the correct TAI grain index. Turns out there was a double-pacing bug: the capture loop was blocking on hardware frame delivery AND sleeping to the MXL clock — two pacers fighting each other, causing silent frame drops when they drifted apart. Fixed by resolving `mxlGetCurrentIndex()` after the hardware frame lands, not before.
+
+- **Frontend** — Vue Flow canvas for building pipelines. Fixed a type coercion bug where select inputs (device port, sample rate) were storing strings, which made nlohmann/json throw on the C++ side. Also cleaned up auto-detected params: DeckLink and NDI sources don't expose resolution/fps in the UI anymore since they probe from the signal.
+
+---
+
+**The hard parts**
+
+**MXL sync groups and TAI time.**
+MXL uses TAI (not UTC, not wall clock — TAI, currently offset by 37 seconds from UTC). Grain indices are absolute TAI nanosecond timestamps divided by frame period. When we first wired up the PiP sync group, it was silently passing grain index (~44 billion) instead of TAI nanoseconds (~1.784 × 10¹⁸). The node exited in under a millisecond with status 0 and no log. We had to read MXL internals — `flow.cpp`, `Timing.hpp`, `FlowSynchronizationGroup.cpp` — to figure out that `mxlFlowSynchronizationGroupWaitForDataAt` expects nanoseconds since TAI epoch, not a frame counter.
+
+**Sync groups don't work with hardware inputs.**
+Even after fixing the timestamp, sync groups kept failing with TOO_LATE cascades when mixing testpattern with DeckLink. The reason: sync groups require both flows to have a grain at the *exact same TAI index*. A software source (testpattern) writes precisely on the TAI grid. A hardware source (DeckLink) writes to whatever index is current when the frame arrives from the SDI callback — which has hardware jitter of a few milliseconds. Even one miss causes a timeout, the node jumps forward, and you get a cascade.
+
+The fix: drop sync groups for hardware-mixed inputs. Background is the master clock (blocking grain read). Inset is best-effort — try the exact index, fall back to `ri.headIndex` from the MXL ring buffer. One frame stale on the inset is visually invisible in a PiP.
+
+Sync groups are still the right tool when both sources are locked to the same TAI reference via PTP — for example two genlocked DeckLink inputs with `ptp4l` + `phc2sys` on Linux syncing `CLOCK_TAI` to a PTP grandmaster (SMPTE ST 2059-2). On macOS dev machines we just live with best-effort.
+
+**Bilinear scaling at 1278×720 was too slow.**
+First test: testpattern background + DeckLink inset at configured 1280×720 PiP size → PiP running at 17fps instead of 25. The culprit was inside `scale_and_overlay`: for every output pixel on every row it was recomputing the source X coordinate — `sx = (dx + 0.5) * scale - 0.5`, floor, clamp, fractional weight. That's ~1.6 million float multiplies per frame that produce the same result on every row. Fixed by precomputing an X sample map (x0, x1, fx, 1-fx) once per call. Scale time dropped from ~32ms to ~8ms, PiP runs at 25fps with headroom.
+
+---
+
+**Architecture in one picture**
+
+```
+[testpattern]──video──┐
+                      ├──[pip]──video──[decklinkout]
+[decklinkin]──video───┘
+     │
+     └──audio──[gaindb]──audio──[decklinkout]
+```
+
+Every box is a separate process. Arrows are MXL shared memory flows — ring buffers in `/dev/shm`, addressed by UUID, timestamped in TAI. The studio-manager forks and execs nodes, injects `NODE_CONFIG` as a JSON env var, and monitors for crashes. The browser UI builds the graph and sends it over WebSocket.
+
+---
+
+**What's next**
+
+- More processing nodes: audio mixer, video switcher/mixer (A/B cut, dissolve)
+- PTP integration for production use — genlocked multi-source sync via `ptp4l` + `phc2sys` → `CLOCK_TAI`, enabling sync groups across hardware sources
+- Graph persistence and live reconfiguration without full restart
+- NDI discovery UI — pick sources by name, not by index number
+- Proper crash recovery in studio-manager — restart crashed nodes and reconnect flows
+
+**The bigger step: Kubernetes.**
+
+Right now every DMF Studio node is a process on the same machine, exchanging video through `/dev/shm`. That works — and it's fast, zero-copy, nanosecond-timestamped. But it doesn't scale beyond one box.
+
+The cluster layer for this is [`mxl-k8s`](https://github.com/qvest-digital/mxl-k8s) — a Kubernetes control plane for MXL built around the EBU Dynamic Media Facility Reference Architecture (V2.0, April 2026). The idea: each DMF Studio node becomes a pod. Flows that cross machine boundaries are handled transparently by a per-node gateway DaemonSet that owns the `libmxl-fabrics` handles (RDMA/RoCEv2/EFA/TCP), drives the cross-node grain transfer loop, and recovers on restarts. The media function itself never touches `libmxl-fabrics` — it still calls `mxlCreateFlowReader` against its local domain. An LD_PRELOAD shim intercepts the first access to a not-yet-materialised remote flow and blocks until the gateway has mirrored it locally.
+
+The control plane: an agent DaemonSet watches each node's MXL domain via `fanotify` and publishes flows to the Kubernetes API. A cluster-scoped operator reconciles `MxlReceiver` intent ("this pod wants to consume that flow") into `MxlFlowMirror` objects, with ref-counted sharing when multiple consumers on the same node want the same flow.
+
+What this means for DMF Studio: a testpattern pod on node A, a PiP pod on node B, a DeckLink Out pod on node C — wired together in the browser exactly the same way as today, with the cluster handling the fabric underneath. Horizontal scaling, failure isolation, and fabric rollout (from TCP in dev to RDMA in production) become Kubernetes operational concerns, not application code.
+
+Demo video attached.