af_xdp tryout

Reviewed-on: #6

.vscode/c_cpp_properties.json

@@ -6,7 +6,8 @@
                 "${workspaceFolder}/**",
                 "${workspaceFolder}/shared",
                 "${HOME}/SDK/NDI/include",
-                "${HOME}/SDK/decklink-sdk/Linux/include"
+                "${HOME}/SDK/decklink-sdk/Linux/include",
+                "${HOME}/SDK/mtl-26.01/include"
             ],
             "defines": [],
             "compilerPath": "/usr/bin/clang",

CMakeLists.txt

@@ -76,6 +76,35 @@ FetchContent_MakeAvailable(json)
 # ── FFmpeg ──────────────────────────────────────────────────────────────────
 find_package(PkgConfig REQUIRED)
 
+# ── Media Transport Library (SMPTE ST 2110) ─────────────────────────────────
+# MTL is Meson-based, so consume an installed Meson build instead of adding the
+# source tree as a CMake subdirectory.
+set(MTL_SDK_DIR "" CACHE PATH "Path to installed Media Transport Library prefix")
+if(MTL_SDK_DIR)
+    find_library(MTL_LIBRARY
+        NAMES mtl
+        PATHS "${MTL_SDK_DIR}/lib" "${MTL_SDK_DIR}/lib64"
+        NO_DEFAULT_PATH
+        REQUIRED
+    )
+
+    add_library(mtl::mtl SHARED IMPORTED GLOBAL)
+    set_target_properties(mtl::mtl PROPERTIES
+        IMPORTED_LOCATION "${MTL_LIBRARY}"
+        INTERFACE_INCLUDE_DIRECTORIES "${MTL_SDK_DIR}/include"
+    )
+else()
+    pkg_check_modules(MTL IMPORTED_TARGET mtl)
+    if(MTL_FOUND)
+        add_library(mtl::mtl INTERFACE IMPORTED GLOBAL)
+        target_link_libraries(mtl::mtl INTERFACE PkgConfig::MTL)
+        target_include_directories(mtl::mtl INTERFACE ${MTL_INCLUDE_DIRS})
+        target_link_directories(mtl::mtl INTERFACE ${MTL_LIBRARY_DIRS})
+        target_compile_options(mtl::mtl INTERFACE ${MTL_CFLAGS_OTHER})
+        target_link_options(mtl::mtl INTERFACE ${MTL_LDFLAGS_OTHER})
+    endif()
+endif()
+
 # Check for FFmpeg components
 pkg_check_modules(FFMPEG REQUIRED IMPORTED_TARGET
     libavformat
@@ -112,7 +141,17 @@ if(DECKLINK_SDK_DIR)
     add_subdirectory(nodes/decklinkout)
 endif()
 
+# ── SMPTE-2110 (Media Transport Library) nodes ──────────────────────────────
+if(TARGET mtl::mtl)
+    add_subdirectory(nodes/st2110in)
+else()
+    message(STATUS "Media Transport Library not found; skipping SMPTE ST 2110 nodes")
+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`

flow-graphs/2110-20.json

@@ -0,0 +1,23 @@
+{
+  "nodes": [
+    {
+      "id": "2110in",
+      "type": "2110",
+      "params": {
+        "interface": "eno1np0",
+        "backend": "af_xdp",
+        "af_xdp_zero_copy": false,
+        "local_ip": "192.168.0.3",
+        "sdp": "v=0\no=- 3150588975 5 IN IP4 192.168.0.2\ns=DeckLink IP/SDI HD (1): Video\nt=0 0\nm=video 16388 RTP/AVP 96\nc=IN IP4 239.255.197.181/255\na=source-filter:incl IN IP4 239.255.197.181 192.168.0.2\na=rtpmap:96 raw/90000\na=fmtp:96 sampling=YCbCr-4:2:2; depth=10; width=1920; height=1080; exactframerate=50; colorimetry=BT709; PM=2110GPM; SSN=ST2110-20:2017; TP=2110TPN;\na=ts-refclk:ptp=IEEE1588-2008:7C-2E-0D-FF-FE-A7-23-B6:127\na=mediaclk:direct=0\na=ssrc:1258156108 cname:EB02FE63F5DB433DA6739B90A9B67B02"
+      }
+    },
+    { "id": "fakesink",      "type": "fakesink",      "params": {} }
+  ],
+  "edges": [
+    {
+      "from": "2110in", "from_port": "video_flow_id",
+      "to":   "fakesink",      "to_port":   "video_flow_id",
+      "format": { "kind": "video", "width": 1920, "height": 1080, "fps_num": 50, "fps_den": 1 }
+    }
+  ]
+}

flow-graphs/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,317 @@
+# ST 2110-20 Receiver Runbook
+
+This project uses Intel Media Transport Library (MTL) for the `2110in` node.
+The current receiver path is:
+
+```text
+DeckLink ST 2110-20 -> MTL kernel backend -> MXL video/v210 flow
+```
+
+The first validated stream was:
+
+```text
+1920x1080p50
+YCbCr 4:2:2 10-bit
+RTP payload type 96
+multicast 239.255.197.181:16388
+source 192.168.0.2
+receiver interface eno1np0
+```
+
+## Runtime Setup
+
+Run these after boot before starting the receiver.
+
+### Hugepages
+
+MTL initializes DPDK EAL even when using the kernel socket backend, so hugepages
+must exist.
+
+```bash
+sudo mkdir -p /mnt/huge
+sudo mount -t hugetlbfs nodev /mnt/huge
+echo 1024 | sudo tee /proc/sys/vm/nr_hugepages
+grep Huge /proc/meminfo
+```
+
+What it does:
+
+- `hugetlbfs` provides the hugepage filesystem DPDK expects.
+- `nr_hugepages=1024` reserves about 2 GB with 2 MB pages.
+- `HugePages_Free` should be greater than zero before running the node.
+
+### RX Ring Size
+
+The Mellanox interface defaulted to RX ring `1024`, which caused
+`rx_out_of_buffer` increments and RTP timestamp gaps. Increase it to the card
+maximum.
+
+```bash
+sudo ethtool -g eno1np0
+sudo ethtool -G eno1np0 rx 8192
+sudo ethtool -g eno1np0
+```
+
+What it does:
+
+- Increases the NIC receive descriptor ring.
+- Gives the driver more buffers to absorb ST 2110 burstiness and scheduler jitter.
+- Prevents drops reported as `rx_out_of_buffer`.
+
+Expected result:
+
+```text
+Current hardware settings:
+RX: 8192
+```
+
+### Kernel Receive Buffers
+
+Increase kernel receive buffering for the kernel socket backend.
+
+```bash
+sudo sysctl -w net.core.rmem_max=268435456
+sudo sysctl -w net.core.rmem_default=268435456
+sudo sysctl -w net.core.netdev_max_backlog=250000
+```
+
+What each setting does:
+
+- `net.core.rmem_max`: maximum receive socket buffer size. Needed so high-rate
+  UDP receivers can request/use large buffers.
+- `net.core.rmem_default`: default receive socket buffer size for sockets that do
+  not explicitly set a larger one.
+- `net.core.netdev_max_backlog`: maximum packets queued in the kernel networking
+  backlog when the kernel cannot immediately process all received packets.
+
+These settings are especially relevant while using `kernel:<interface>` MTL
+ports. DPDK or AF_XDP paths reduce dependence on this kernel socket buffering.
+
+## Verification During A Run
+
+Start with a clean baseline:
+
+```bash
+ethtool -S eno1np0 | grep rx_out_of_buffer
+```
+
+Watch NIC drop-related counters while `2110in` is running:
+
+```bash
+watch -n1 "ethtool -S eno1np0 | grep -E 'rx_out_of_buffer|rx_discards_phy|rx_crc_errors_phy'"
+```
+
+Expected:
+
+```text
+rx_out_of_buffer does not increase
+rx_discards_phy remains 0
+rx_crc_errors_phy does not increase
+```
+
+Watch node stats:
+
+```text
+incomplete=0
+bad_fmt=0
+mxl_open_fail=0
+rtp_gap=0
+rtp_dup=0
+skipped=0
+```
+
+Meaning:
+
+- `incomplete`: MTL delivered incomplete frames. Should stay zero.
+- `bad_fmt`: MTL output format did not match the expected SDP-derived format.
+- `mxl_open_fail`: MXL writer could not open the target grain.
+- `rtp_gap`: RTP timestamp skipped one or more frame positions. Usually packet
+  loss, sender frame drops, or receiver drops.
+- `rtp_dup`: duplicate/backwards RTP timestamp.
+- `skipped`: MXL indices skipped by timestamp mapping. Should stay zero in a
+  clean run.
+
+## Persistent Setup
+
+### Persistent sysctl
+
+Create `/etc/sysctl.d/99-st2110.conf`:
+
+```bash
+sudo tee /etc/sysctl.d/99-st2110.conf >/dev/null <<'EOF'
+net.core.rmem_max=268435456
+net.core.rmem_default=268435456
+net.core.netdev_max_backlog=250000
+EOF
+```
+
+Apply without reboot:
+
+```bash
+sudo sysctl --system
+```
+
+### Persistent Hugepages
+
+Create `/etc/sysctl.d/98-hugepages.conf`:
+
+```bash
+sudo tee /etc/sysctl.d/98-hugepages.conf >/dev/null <<'EOF'
+vm.nr_hugepages=1024
+EOF
+```
+
+Ensure `hugetlbfs` is mounted at boot by adding this line to `/etc/fstab`:
+
+```text
+nodev /mnt/huge hugetlbfs defaults 0 0
+```
+
+Create the mount point and test:
+
+```bash
+sudo mkdir -p /mnt/huge
+sudo mount /mnt/huge
+mount | grep hugetlbfs
+```
+
+### Persistent RX Ring With systemd
+
+`ethtool -G` is not persistent by itself. Use a systemd oneshot service.
+
+Create `/etc/systemd/system/st2110-nic-tuning.service`:
+
+```ini
+[Unit]
+Description=ST 2110 NIC tuning
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+ExecStart=/usr/sbin/ethtool -G eno1np0 rx 8192
+RemainAfterExit=yes
+
+[Install]
+WantedBy=multi-user.target
+```
+
+Enable and start:
+
+```bash
+sudo systemctl daemon-reload
+sudo systemctl enable --now st2110-nic-tuning.service
+sudo systemctl status st2110-nic-tuning.service
+```
+
+Verify after reboot:
+
+```bash
+sudo ethtool -g eno1np0
+```
+
+Expected:
+
+```text
+Current hardware settings:
+RX: 8192
+```
+
+## Receiver Config
+
+The receiver config should carry local NIC settings plus SDP:
+
+```json
+{
+  "interface": "eno1np0",
+  "local_ip": "192.168.0.3",
+  "sdp": "v=0\nm=video 16388 RTP/AVP 96\nc=IN IP4 239.255.197.181/255\n..."
+}
+```
+
+The SDP parser currently supports:
+
+```text
+m=video
+c=IN IP4
+a=source-filter
+a=fmtp width/height/depth/sampling/exactframerate
+```
+
+Supported video formats:
+
+```text
+YCbCr-4:2:2 depth=8  -> MTL UYVY output -> local UYVY to v210 conversion
+YCbCr-4:2:2 depth=10 -> MTL V210 output -> direct copy to MXL
+```
+
+## Indexing Mode
+
+Default:
+
+```json
+"mxl_index_mode": "rtp"
+```
+
+RTP mode maps `frame->rtp_timestamp` to the MXL grain index. This preserves sender
+media cadence and exposes real RTP timestamp gaps.
+
+Alternative:
+
+```json
+"mxl_index_mode": "live"
+```
+
+Live mode publishes near `mxlGetCurrentIndex() + mxl_latency_frames`. It keeps
+sinks close to the local MXL clock but may skip indices if the source clock and
+local MXL clock drift.
+
+Keep RTP mode for normal ST 2110 ingest.
+
+## Known Failure Signatures
+
+### RTP gaps with `rx_out_of_buffer` increasing
+
+Cause:
+
+```text
+Receiver-side NIC/kernel buffering loss.
+```
+
+Fix:
+
+```text
+Increase RX ring and kernel receive buffers.
+```
+
+### `mxl-gst-sink` reports TOO_EARLY after long run
+
+Cause:
+
+```text
+Writer fell behind the MXL reader clock, usually from clock-domain drift or an
+indexing policy that does not follow source timestamps.
+```
+
+Fix:
+
+```text
+Use mxl_index_mode=rtp and verify rtp_gap=0.
+```
+
+### Clean NIC counters but `rtp_gap` increases
+
+Likely causes:
+
+```text
+Sender/source frame drops, sender media-clock discontinuity, or loss before the
+receiver NIC.
+```
+
+Next debug step:
+
+```bash
+sudo tcpdump -i eno1np0 -nn -s 128 udp port 16388 -w st2110-gap.pcap
+```
+
+Inspect RTP sequence numbers and timestamps around the gap.

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 ---
-        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));
+        // 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 audio_index = has_audio ? mxlGetCurrentIndex(&audio_rate) : 0;
+        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,18 +127,26 @@ 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(
                     audio_writer, audio_index, static_cast<size_t>(samples_written), &slice);
                 if (ast == MXL_STATUS_OK) {
                     for (int ch = 0; ch < channels; ++ch) {
-                        const uint8_t* src        = reinterpret_cast<const uint8_t*>(
+                        const uint8_t* src         = reinterpret_cast<const uint8_t*>(
                             audio_buf.data() + ch * max_audio_samples);
-                        uint8_t*       dst0       = static_cast<uint8_t*>(
+                        uint8_t*       dst0        = static_cast<uint8_t*>(
                             slice.base.fragments[0].pointer) + ch * slice.stride;
                         const size_t   frag0_bytes = slice.base.fragments[0].size;
                         const size_t   total_bytes = static_cast<size_t>(samples_written) * sizeof(float);
@@ -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/st2110in/CMakeLists.txt

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

nodes/st2110in/main.cpp

@@ -0,0 +1,267 @@
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <cstring>
+#include <exception>
+#include <limits>
+
+#include <mxl/flow.h>
+#include <mxl/time.h>
+
+#include "NodeBase.hpp"
+#include "ST2110Receiver.hpp"
+#include "V210.hpp"
+
+namespace dmf {
+
+namespace {
+
+class RtpIndexMapper {
+public:
+    RtpIndexMapper(const mxlRational& rate, int latency_frames)
+        : rate_(rate), latency_frames_(latency_frames) {}
+
+    uint64_t index_for(uint32_t rtp_timestamp, uint64_t current_mxl_index) {
+        const uint64_t rtp_ext = extend_rtp(rtp_timestamp);
+        if (!anchored_) {
+            anchored_ = true;
+            base_rtp_ext_ = rtp_ext;
+            base_mxl_index_ = current_mxl_index + static_cast<uint64_t>(latency_frames_);
+            last_offset_ = 0;
+            return base_mxl_index_;
+        }
+
+        const uint64_t rtp_delta = rtp_ext - base_rtp_ext_;
+        const uint64_t denominator = 90'000ULL * static_cast<uint64_t>(rate_.denominator);
+        const uint64_t numerator = rtp_delta * static_cast<uint64_t>(rate_.numerator);
+        const uint64_t offset = (numerator + denominator / 2) / denominator;
+        if (offset <= last_offset_) {
+            duplicate_or_backwards_++;
+            last_gap_frames_ = 0;
+        } else if (offset > last_offset_ + 1) {
+            last_gap_frames_ = offset - (last_offset_ + 1);
+            rtp_gap_frames_ += last_gap_frames_;
+        } else {
+            last_gap_frames_ = 0;
+        }
+        last_offset_ = offset;
+        return base_mxl_index_ + offset;
+    }
+
+    uint32_t last_rtp_delta() const { return last_rtp_delta_; }
+    uint32_t expected_rtp_delta() const {
+        return static_cast<uint32_t>(
+            (90'000ULL * static_cast<uint64_t>(rate_.denominator)) /
+            static_cast<uint64_t>(rate_.numerator));
+    }
+    uint64_t last_gap_frames() const { return last_gap_frames_; }
+    uint64_t rtp_gap_frames() const { return rtp_gap_frames_; }
+    uint64_t duplicate_or_backwards() const { return duplicate_or_backwards_; }
+    uint64_t base_mxl_index() const { return base_mxl_index_; }
+
+private:
+    uint64_t extend_rtp(uint32_t rtp_timestamp) {
+        if (!have_last_rtp_) {
+            have_last_rtp_ = true;
+            last_rtp_ = rtp_timestamp;
+            return rtp_timestamp;
+        }
+
+        if (rtp_timestamp < last_rtp_ &&
+            static_cast<uint32_t>(last_rtp_ - rtp_timestamp) > 0x80000000u) {
+            rtp_cycles_ += 0x1'0000'0000ULL;
+        }
+        last_rtp_delta_ = rtp_timestamp - last_rtp_;
+        last_rtp_ = rtp_timestamp;
+        return rtp_cycles_ + rtp_timestamp;
+    }
+
+    mxlRational rate_{};
+    int latency_frames_ = 2;
+    bool anchored_ = false;
+    bool have_last_rtp_ = false;
+    uint32_t last_rtp_ = 0;
+    uint64_t rtp_cycles_ = 0;
+    uint64_t base_rtp_ext_ = 0;
+    uint64_t base_mxl_index_ = 0;
+    uint64_t last_offset_ = 0;
+    uint64_t rtp_gap_frames_ = 0;
+    uint64_t duplicate_or_backwards_ = 0;
+    uint64_t last_gap_frames_ = 0;
+    uint32_t last_rtp_delta_ = 0;
+};
+
+} // namespace
+
+class ST2110In : public NodeBase {
+    void run() override {
+        ST2110ReceiverConfig cfg;
+        try {
+            cfg = parse_st2110_receiver_config(config());
+        } catch (const std::exception& e) {
+            log("config error: %s", e.what());
+            return;
+        }
+
+        log("SMPTE 2110-20 RX %s:%u from %s on %s backend=%s local=%s %dx%d depth=%d @ %d/%d latency=%d index_mode=%s",
+            cfg.mcast_ip.c_str(), cfg.udp_port, cfg.source_ip.c_str(), cfg.ifname.c_str(),
+            cfg.backend.c_str(), cfg.local_ip.c_str(), cfg.width, cfg.height, cfg.depth,
+            cfg.fps_num, cfg.fps_den, cfg.mxl_latency_frames, cfg.mxl_index_mode.c_str());
+
+        try {
+            MTLContext mtl(cfg);
+            ST20RxSession rx(mtl.get(), mtl.port_name(), cfg);
+            MXLVideoWriter writer(instance(), cfg, node_id());
+
+            const mxlRational video_rate = {cfg.fps_num, cfg.fps_den};
+            const uint32_t video_stride = writer.config().discrete.sliceSizes[0];
+            const uint32_t uyvy_stride = static_cast<uint32_t>(cfg.width * 2);
+            const auto expected_fmt = cfg.output_fmt();
+            RtpIndexMapper rtp_mapper(video_rate, cfg.mxl_latency_frames);
+            uint64_t last_published_index = std::numeric_limits<uint64_t>::max();
+            uint64_t frames_written = 0;
+            uint64_t incomplete_frames = 0;
+            uint64_t unexpected_format_frames = 0;
+            uint64_t mxl_open_failures = 0;
+            uint64_t index_resyncs = 0;
+            uint64_t skipped_indices = 0;
+            uint64_t backwards_indices = 0;
+            uint64_t last_report_frames = 0;
+            auto last_report = std::chrono::steady_clock::now();
+
+            log("MTL version=%s port=%s output_stride=%u input_stride=%u direct_v210=%d",
+                mtl_version(), mtl.port_name().c_str(), video_stride,
+                cfg.direct_v210() ? video_stride : uyvy_stride, cfg.direct_v210() ? 1 : 0);
+
+            while (g_running.load(std::memory_order_relaxed)) {
+                st_frame* frame = st20p_rx_get_frame(rx.get());
+                if (!frame) {
+                    continue;
+                }
+
+                if (!st_is_frame_complete(frame->status)) {
+                    incomplete_frames++;
+                    st20p_rx_put_frame(rx.get(), frame);
+                    continue;
+                }
+
+                if (frame->fmt != expected_fmt) {
+                    unexpected_format_frames++;
+                    log("unexpected MTL frame fmt=%d; expected %d", frame->fmt, expected_fmt);
+                    st20p_rx_put_frame(rx.get(), frame);
+                    continue;
+                }
+
+                mxlGrainInfo grain{};
+                uint8_t* video_buf = nullptr;
+                const uint64_t current_index = mxlGetCurrentIndex(&video_rate);
+                uint64_t video_index = 0;
+                if (cfg.mxl_index_mode == "rtp") {
+                    video_index = rtp_mapper.index_for(frame->rtp_timestamp, current_index);
+                    if (rtp_mapper.last_gap_frames() > 0) {
+                        log("RTP gap frame=%llu rtp=%u delta=%u expected=%u gap_frames=%llu",
+                            static_cast<unsigned long long>(frames_written),
+                            frame->rtp_timestamp,
+                            rtp_mapper.last_rtp_delta(),
+                            rtp_mapper.expected_rtp_delta(),
+                            static_cast<unsigned long long>(rtp_mapper.last_gap_frames()));
+                    }
+                } else {
+                    video_index = current_index + static_cast<uint64_t>(cfg.mxl_latency_frames);
+                }
+
+                if (last_published_index != std::numeric_limits<uint64_t>::max()) {
+                    if (video_index <= last_published_index) {
+                        backwards_indices++;
+                        video_index = last_published_index + 1;
+                    } else if (video_index > last_published_index + 1) {
+                        skipped_indices += video_index - (last_published_index + 1);
+                    }
+                }
+
+                mxlStatus st = mxlFlowWriterOpenGrain(writer.get(), video_index, &grain, &video_buf);
+                if (st != MXL_STATUS_OK) {
+                    mxl_open_failures++;
+                    log("mxlFlowWriterOpenGrain failed (%s) index=%llu current=%llu", mxl_status_str(st),
+                        static_cast<unsigned long long>(video_index),
+                        static_cast<unsigned long long>(current_index));
+                    st20p_rx_put_frame(rx.get(), frame);
+                    continue;
+                }
+
+                if (last_published_index != std::numeric_limits<uint64_t>::max() &&
+                    video_index > last_published_index + 1) {
+                    index_resyncs++;
+                    log("MXL index resync last=%llu next=%llu current=%llu",
+                        static_cast<unsigned long long>(last_published_index),
+                        static_cast<unsigned long long>(video_index),
+                        static_cast<unsigned long long>(current_index));
+                }
+
+                if (cfg.direct_v210()) {
+                    const auto* src = static_cast<const uint8_t*>(frame->addr[0]);
+                    const size_t src_stride = frame->linesize[0] ? frame->linesize[0]
+                                                                 : static_cast<size_t>(cfg.width / 6) * 16;
+                    const size_t row_bytes = static_cast<size_t>(cfg.width / 6) * 16;
+                    for (int y = 0; y < cfg.height; ++y) {
+                        std::memcpy(video_buf + static_cast<size_t>(y) * video_stride,
+                                    src + static_cast<size_t>(y) * src_stride, row_bytes);
+                    }
+                } else {
+                    const uint32_t src_stride = frame->linesize[0]
+                        ? static_cast<uint32_t>(frame->linesize[0])
+                        : uyvy_stride;
+                    v210::UYVYtoV210(static_cast<const uint8_t*>(frame->addr[0]), video_buf,
+                                     cfg.width, cfg.height, src_stride, video_stride);
+                }
+                grain.flags = 0;
+                grain.validSlices = grain.totalSlices;
+                mxlFlowWriterCommitGrain(writer.get(), &grain);
+                st20p_rx_put_frame(rx.get(), frame);
+
+                frames_written++;
+                last_published_index = video_index;
+                const auto now = std::chrono::steady_clock::now();
+                const auto elapsed = std::chrono::duration<double>(now - last_report).count();
+                if (elapsed >= 5.0) {
+                    const uint64_t delta = frames_written - last_report_frames;
+                    const double measured_fps = static_cast<double>(delta) / elapsed;
+                    log("stats frames=%llu fps=%.2f incomplete=%llu bad_fmt=%llu mxl_open_fail=%llu resync=%llu skipped=%llu backwards=%llu rtp_gap=%llu rtp_dup=%llu last_index=%llu current=%llu",
+                        static_cast<unsigned long long>(frames_written), measured_fps,
+                        static_cast<unsigned long long>(incomplete_frames),
+                        static_cast<unsigned long long>(unexpected_format_frames),
+                        static_cast<unsigned long long>(mxl_open_failures),
+                        static_cast<unsigned long long>(index_resyncs),
+                        static_cast<unsigned long long>(skipped_indices),
+                        static_cast<unsigned long long>(backwards_indices),
+                        static_cast<unsigned long long>(rtp_mapper.rtp_gap_frames()),
+                        static_cast<unsigned long long>(rtp_mapper.duplicate_or_backwards()),
+                        static_cast<unsigned long long>(last_published_index),
+                        static_cast<unsigned long long>(mxlGetCurrentIndex(&video_rate)));
+                    last_report = now;
+                    last_report_frames = frames_written;
+                }
+            }
+
+            log("stopped frames=%llu incomplete=%llu bad_fmt=%llu mxl_open_fail=%llu resync=%llu skipped=%llu backwards=%llu rtp_gap=%llu rtp_dup=%llu",
+                static_cast<unsigned long long>(frames_written),
+                static_cast<unsigned long long>(incomplete_frames),
+                static_cast<unsigned long long>(unexpected_format_frames),
+                static_cast<unsigned long long>(mxl_open_failures),
+                static_cast<unsigned long long>(index_resyncs),
+                static_cast<unsigned long long>(skipped_indices),
+                static_cast<unsigned long long>(backwards_indices),
+                static_cast<unsigned long long>(rtp_mapper.rtp_gap_frames()),
+                static_cast<unsigned long long>(rtp_mapper.duplicate_or_backwards()));
+        } catch (const std::exception& e) {
+            log("error: %s", e.what());
+        }
+    }
+};
+
+} // namespace dmf
+
+int main() {
+    dmf::ST2110In 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
-        const float amplitude = static_cast<float>(std::pow(10.0, -18.0 / 20.0));
+        // default -18 dBFS broadcast reference level
+        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/ST2110Receiver.hpp

@@ -0,0 +1,403 @@
+#pragma once
+
+#include <arpa/inet.h>
+
+#include <cstdint>
+#include <cstdio>
+#include <cstdlib>
+#include <sstream>
+#include <stdexcept>
+#include <string>
+
+#include <nlohmann/json.hpp>
+
+extern "C" {
+#include <mtl/mtl_api.h>
+#include <mtl/st20_api.h>
+#include <mtl/st_pipeline_api.h>
+}
+
+#include <mxl/flow.h>
+
+#include "FlowDef.hpp"
+#include "NodeBase.hpp"
+
+namespace dmf {
+
+struct ST2110ReceiverConfig {
+    std::string flow_id;
+    std::string ifname;
+    std::string local_ip;
+    std::string source_ip;
+    std::string mcast_ip;
+    uint16_t udp_port = 0;
+    uint8_t payload_type = 96;
+    int width = 1920;
+    int height = 1080;
+    int depth = 8;
+    int fps_num = 25;
+    int fps_den = 1;
+    int framebuff_cnt = 3;
+    int mxl_latency_frames = 2;
+    std::string mxl_index_mode = "rtp";
+    std::string backend = "kernel";
+    bool af_xdp_zero_copy = true;
+
+    bool direct_v210() const { return depth == 10; }
+    enum st20_fmt transport_fmt() const {
+        return direct_v210() ? ST20_FMT_YUV_422_10BIT : ST20_FMT_YUV_422_8BIT;
+    }
+    enum st_frame_fmt output_fmt() const {
+        return direct_v210() ? ST_FRAME_FMT_V210 : ST_FRAME_FMT_UYVY;
+    }
+};
+
+inline std::string st2110_trim(std::string s) {
+    const auto first = s.find_first_not_of(" \t\r\n");
+    if (first == std::string::npos) return {};
+    const auto last = s.find_last_not_of(" \t\r\n");
+    return s.substr(first, last - first + 1);
+}
+
+inline bool st2110_parse_int(const std::string& text, int* out) {
+    char* end = nullptr;
+    const long value = std::strtol(text.c_str(), &end, 10);
+    if (!end || *end != '\0') return false;
+    *out = static_cast<int>(value);
+    return true;
+}
+
+inline void st2110_parse_exactframerate(const std::string& value,
+                                        ST2110ReceiverConfig& cfg) {
+    const auto slash = value.find('/');
+    if (slash == std::string::npos) {
+        if (!st2110_parse_int(value, &cfg.fps_num)) {
+            throw std::runtime_error("invalid SDP exactframerate: " + value);
+        }
+        cfg.fps_den = 1;
+        return;
+    }
+
+    const std::string num = value.substr(0, slash);
+    const std::string den = value.substr(slash + 1);
+    if (!st2110_parse_int(num, &cfg.fps_num) || !st2110_parse_int(den, &cfg.fps_den)) {
+        throw std::runtime_error("invalid SDP exactframerate: " + value);
+    }
+}
+
+inline void st2110_apply_fmtp_param(const std::string& key, const std::string& value,
+                                    ST2110ReceiverConfig& cfg, std::string& sampling) {
+    if (key == "width") {
+        if (!st2110_parse_int(value, &cfg.width)) {
+            throw std::runtime_error("invalid SDP width: " + value);
+        }
+    } else if (key == "height") {
+        if (!st2110_parse_int(value, &cfg.height)) {
+            throw std::runtime_error("invalid SDP height: " + value);
+        }
+    } else if (key == "depth") {
+        if (!st2110_parse_int(value, &cfg.depth)) {
+            throw std::runtime_error("invalid SDP depth: " + value);
+        }
+    } else if (key == "sampling") {
+        sampling = value;
+    } else if (key == "exactframerate") {
+        st2110_parse_exactframerate(value, cfg);
+    }
+}
+
+inline void st2110_parse_fmtp(const std::string& line, ST2110ReceiverConfig& cfg,
+                              std::string& sampling) {
+    const auto space = line.find(' ');
+    if (space == std::string::npos) return;
+
+    std::stringstream params(line.substr(space + 1));
+    std::string item;
+    while (std::getline(params, item, ';')) {
+        item = st2110_trim(item);
+        if (item.empty()) continue;
+
+        const auto eq = item.find('=');
+        if (eq == std::string::npos) continue;
+        const std::string key = st2110_trim(item.substr(0, eq));
+        const std::string value = st2110_trim(item.substr(eq + 1));
+        st2110_apply_fmtp_param(key, value, cfg, sampling);
+    }
+}
+
+inline void st2110_apply_sdp(const std::string& sdp, ST2110ReceiverConfig& cfg) {
+    std::stringstream lines(sdp);
+    std::string line;
+    std::string sampling = "YCbCr-4:2:2";
+
+    while (std::getline(lines, line)) {
+        line = st2110_trim(line);
+        if (line.rfind("m=video ", 0) == 0) {
+            std::stringstream media(line.substr(8));
+            int port = 0;
+            std::string proto;
+            int payload = 0;
+            if (media >> port >> proto >> payload) {
+                if (port < 0 || port > 65535 || payload < 0 || payload > 255) {
+                    throw std::runtime_error("SDP m=video port or payload out of range");
+                }
+                cfg.udp_port = static_cast<uint16_t>(port);
+                cfg.payload_type = static_cast<uint8_t>(payload);
+            }
+        } else if (line.rfind("c=IN IP4 ", 0) == 0) {
+            std::string addr = line.substr(9);
+            const auto slash = addr.find('/');
+            if (slash != std::string::npos) addr.resize(slash);
+            cfg.mcast_ip = st2110_trim(addr);
+        } else if (line.rfind("a=source-filter:incl IN IP4 ", 0) == 0) {
+            std::stringstream filter(line.substr(28));
+            std::string group;
+            std::string source;
+            if (filter >> group >> source) {
+                cfg.source_ip = source;
+            }
+        } else if (line.rfind("a=fmtp:", 0) == 0) {
+            st2110_parse_fmtp(line.substr(7), cfg, sampling);
+        }
+    }
+
+    if (sampling != "YCbCr-4:2:2" || (cfg.depth != 8 && cfg.depth != 10)) {
+        throw std::runtime_error("only SDP YCbCr-4:2:2 depth=8 or depth=10 is supported currently");
+    }
+}
+
+inline void st2110_set_ip(uint8_t dst[MTL_IP_ADDR_LEN], const std::string& ip) {
+    if (inet_pton(AF_INET, ip.c_str(), dst) != 1) {
+        throw std::runtime_error("invalid IP address: " + ip);
+    }
+}
+
+inline uint16_t st2110_checked_u16(const nlohmann::json& j, const char* key) {
+    const int value = j.at(key).get<int>();
+    if (value < 0 || value > 65535) {
+        throw std::runtime_error(std::string(key) + " out of uint16 range");
+    }
+    return static_cast<uint16_t>(value);
+}
+
+inline uint8_t st2110_checked_u8(const nlohmann::json& j, const char* key, int fallback) {
+    const int value = j.value(key, fallback);
+    if (value < 0 || value > 255) {
+        throw std::runtime_error(std::string(key) + " out of uint8 range");
+    }
+    return static_cast<uint8_t>(value);
+}
+
+inline enum st_fps st2110_to_st_fps(int fps_num, int fps_den) {
+    if (fps_den == 1) {
+        switch (fps_num) {
+            case 24: return ST_FPS_P24;
+            case 25: return ST_FPS_P25;
+            case 30: return ST_FPS_P30;
+            case 50: return ST_FPS_P50;
+            case 60: return ST_FPS_P60;
+            case 100: return ST_FPS_P100;
+            case 120: return ST_FPS_P120;
+            default: break;
+        }
+    }
+    if (fps_num == 24000 && fps_den == 1001) return ST_FPS_P23_98;
+    if (fps_num == 30000 && fps_den == 1001) return ST_FPS_P29_97;
+    if (fps_num == 60000 && fps_den == 1001) return ST_FPS_P59_94;
+    if (fps_num == 120000 && fps_den == 1001) return ST_FPS_P119_88;
+    throw std::runtime_error("unsupported ST 2110 frame rate");
+}
+
+inline ST2110ReceiverConfig parse_st2110_receiver_config(const nlohmann::json& cfg) {
+    if (!cfg.contains("video_flow_id")) {
+        throw std::runtime_error("no video output connected");
+    }
+
+    ST2110ReceiverConfig out;
+    out.flow_id = cfg.at("video_flow_id").at("id").get<std::string>();
+    out.ifname = cfg.at("interface").get<std::string>();
+    out.local_ip = cfg.at("local_ip").get<std::string>();
+
+    if (cfg.contains("sdp")) {
+        st2110_apply_sdp(cfg.at("sdp").get<std::string>(), out);
+    }
+
+    if (cfg.contains("source_ip")) out.source_ip = cfg.at("source_ip").get<std::string>();
+    if (cfg.contains("mcast_ip")) out.mcast_ip = cfg.at("mcast_ip").get<std::string>();
+    if (cfg.contains("udp_port")) out.udp_port = st2110_checked_u16(cfg, "udp_port");
+    out.payload_type = st2110_checked_u8(cfg, "payload_type", out.payload_type);
+    out.width = cfg.value("width", out.width);
+    out.height = cfg.value("height", out.height);
+    out.depth = cfg.value("depth", out.depth);
+    out.fps_num = cfg.value("fps_num", out.fps_num);
+    out.fps_den = cfg.value("fps_den", out.fps_den);
+    out.framebuff_cnt = cfg.value("framebuff_cnt", out.framebuff_cnt);
+    out.mxl_latency_frames = cfg.value("mxl_latency_frames", out.mxl_latency_frames);
+    out.mxl_index_mode = cfg.value("mxl_index_mode", out.mxl_index_mode);
+    out.backend = cfg.value("backend", out.backend);
+    out.af_xdp_zero_copy = cfg.value("af_xdp_zero_copy", out.af_xdp_zero_copy);
+
+    if (out.width <= 0 || out.height <= 0) {
+        throw std::runtime_error("width and height must be positive");
+    }
+    if (out.depth != 8 && out.depth != 10) {
+        throw std::runtime_error("only depth=8 and depth=10 are supported currently");
+    }
+    if (out.source_ip.empty() || out.mcast_ip.empty() || out.udp_port == 0) {
+        throw std::runtime_error("source_ip, mcast_ip and udp_port are required unless provided by sdp");
+    }
+    if (out.width % 6 != 0) {
+        throw std::runtime_error("width must be divisible by 6 for v210 output");
+    }
+    if (out.fps_num <= 0 || out.fps_den <= 0) {
+        throw std::runtime_error("fps_num and fps_den must be positive");
+    }
+    if (out.framebuff_cnt < 2 || out.framebuff_cnt > ST20_FB_MAX_COUNT) {
+        throw std::runtime_error("framebuff_cnt must be in [2, ST20_FB_MAX_COUNT]");
+    }
+    if (out.mxl_latency_frames < 1 || out.mxl_latency_frames > 30) {
+        throw std::runtime_error("mxl_latency_frames must be in [1, 30]");
+    }
+    if (out.mxl_index_mode != "rtp" && out.mxl_index_mode != "live") {
+        throw std::runtime_error("mxl_index_mode must be 'rtp' or 'live'");
+    }
+    if (out.backend != "kernel" && out.backend != "af_xdp") {
+        throw std::runtime_error("backend must be 'kernel' or 'af_xdp'");
+    }
+
+    return out;
+}
+
+class MTLContext {
+public:
+    explicit MTLContext(const ST2110ReceiverConfig& cfg) {
+        mtl_init_params params{};
+        params.num_ports = 1;
+
+        if (cfg.backend == "kernel") {
+            port_name_ = "kernel:" + cfg.ifname;
+            params.pmd[MTL_PORT_P] = MTL_PMD_KERNEL_SOCKET;
+        } else if (cfg.backend == "af_xdp") {
+            port_name_ = "native_af_xdp:" + cfg.ifname;
+            params.pmd[MTL_PORT_P] = MTL_PMD_NATIVE_AF_XDP;
+            if (!cfg.af_xdp_zero_copy) {
+                params.flags |= MTL_FLAG_AF_XDP_ZC_DISABLE;
+            }
+        }
+
+        std::snprintf(params.port[MTL_PORT_P], sizeof(params.port[MTL_PORT_P]), "%s",
+                      port_name_.c_str());
+
+        params.net_proto[MTL_PORT_P] = MTL_PROTO_STATIC;
+        params.rx_queues_cnt[MTL_PORT_P] = 1;
+        params.tx_queues_cnt[MTL_PORT_P] = 0;
+        params.log_level = MTL_LOG_LEVEL_INFO;
+        params.flags |= MTL_FLAG_DEV_AUTO_START_STOP;
+        st2110_set_ip(params.sip_addr[MTL_PORT_P], cfg.local_ip);
+
+        handle_ = mtl_init(&params);
+        if (!handle_) {
+            throw std::runtime_error("mtl_init failed");
+        }
+    }
+
+    ~MTLContext() {
+        if (handle_) {
+            mtl_uninit(handle_);
+        }
+    }
+
+    MTLContext(const MTLContext&) = delete;
+    MTLContext& operator=(const MTLContext&) = delete;
+
+    mtl_handle get() const { return handle_; }
+    const std::string& port_name() const { return port_name_; }
+
+private:
+    mtl_handle handle_{nullptr};
+    std::string port_name_;
+};
+
+class ST20RxSession {
+public:
+    ST20RxSession(mtl_handle mt, const std::string& port_name,
+                  const ST2110ReceiverConfig& cfg) {
+        st20p_rx_ops ops{};
+        ops.name = "dmf-st2110in-video";
+        ops.port.num_port = 1;
+        ops.port.udp_port[MTL_SESSION_PORT_P] = cfg.udp_port;
+        ops.port.payload_type = cfg.payload_type;
+        std::snprintf(ops.port.port[MTL_SESSION_PORT_P],
+                      sizeof(ops.port.port[MTL_SESSION_PORT_P]), "%s", port_name.c_str());
+        st2110_set_ip(ops.port.ip_addr[MTL_SESSION_PORT_P], cfg.mcast_ip);
+        st2110_set_ip(ops.port.mcast_sip_addr[MTL_SESSION_PORT_P], cfg.source_ip);
+
+        ops.width = static_cast<uint32_t>(cfg.width);
+        ops.height = static_cast<uint32_t>(cfg.height);
+        ops.fps = st2110_to_st_fps(cfg.fps_num, cfg.fps_den);
+        ops.interlaced = false;
+        ops.transport_fmt = cfg.transport_fmt();
+        ops.output_fmt = cfg.output_fmt();
+        ops.device = ST_PLUGIN_DEVICE_AUTO;
+        ops.framebuff_cnt = static_cast<uint16_t>(cfg.framebuff_cnt);
+        ops.flags = ST20P_RX_FLAG_BLOCK_GET;
+
+        handle_ = st20p_rx_create(mt, &ops);
+        if (!handle_) {
+            throw std::runtime_error("st20p_rx_create failed");
+        }
+        st20p_rx_set_block_timeout(handle_, 100'000'000);
+    }
+
+    ~ST20RxSession() {
+        if (handle_) {
+            st20p_rx_wake_block(handle_);
+            st20p_rx_free(handle_);
+        }
+    }
+
+    ST20RxSession(const ST20RxSession&) = delete;
+    ST20RxSession& operator=(const ST20RxSession&) = delete;
+
+    st20p_rx_handle get() const { return handle_; }
+
+private:
+    st20p_rx_handle handle_{nullptr};
+};
+
+class MXLVideoWriter {
+public:
+    MXLVideoWriter(mxlInstance instance, const ST2110ReceiverConfig& cfg,
+                   const std::string& node_id)
+        : instance_(instance) {
+        bool created = false;
+        const std::string flow_def =
+            make_video_flow_def(cfg.flow_id, node_id, cfg.width, cfg.height, cfg.fps_num,
+                                cfg.fps_den);
+        const mxlStatus st =
+            mxlCreateFlowWriter(instance_, flow_def.c_str(), "", &writer_, &config_, &created);
+        if (st != MXL_STATUS_OK) {
+            throw std::runtime_error(std::string("mxlCreateFlowWriter failed: ") +
+                                     mxl_status_str(st));
+        }
+    }
+
+    ~MXLVideoWriter() {
+        if (writer_) {
+            mxlReleaseFlowWriter(instance_, writer_);
+        }
+    }
+
+    MXLVideoWriter(const MXLVideoWriter&) = delete;
+    MXLVideoWriter& operator=(const MXLVideoWriter&) = delete;
+
+    mxlFlowWriter get() const { return writer_; }
+    const mxlFlowConfigInfo& config() const { return config_; }
+
+private:
+    mxlInstance instance_{nullptr};
+    mxlFlowWriter writer_{nullptr};
+    mxlFlowConfigInfo config_{};
+};
+
+} // namespace dmf

shared/V210.hpp

@@ -1,8 +1,10 @@
 #pragma once
+#include <algorithm>
 #include <array>
 #include <cstddef>
 #include <cstdint>
 #include <cstring>
+#include <vector>
 
 namespace dmf::v210 {
 
@@ -117,7 +119,7 @@ inline void fill_white(uint8_t* buf, int width, int height, uint32_t stride)
     fill_solid(buf, width, height, stride, {940, 512, 512});
 }
 
-inline void UYVYtoV210(uint8_t* src_buf, uint8_t* dst_buf, int width, int height, uint32_t src_stride, uint32_t dst_stride)
+inline void UYVYtoV210(const uint8_t* src_buf, uint8_t* dst_buf, int width, int height, uint32_t src_stride, uint32_t dst_stride)
 {
     const uint8_t* src = src_buf;
     uint8_t* dst = dst_buf;
@@ -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.