af_xdp tryout

.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,6 +141,14 @@ 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)

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 }
+    }
+  ]
+}

mtl.md

@@ -1,81 +1,317 @@
-# ST 2110-20 receiver with Intel MTL
+# ST 2110-20 Receiver Runbook
 
-The MTL pattern is structurally identical to `DeckLinkReceiver` — callback-based, MTL manages frame buffers, you drain them into MXL.
+This project uses Intel Media Transport Library (MTL) for the `2110in` node.
+The current receiver path is:
 
-## MTL initialization (once per process)
+```text
-
+DeckLink ST 2110-20 -> MTL kernel backend -> MXL video/v210 flow
-```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
+The first validated stream was:
 
-```cpp
+```text
-st20_rx_ops ops{};
+1920x1080p50
-ops.port.num_port = 1;
+YCbCr 4:2:2 10-bit
-memcpy(ops.port.sip_addr[MTL_PORT_P], multicast_ip, 4); // join this group
+RTP payload type 96
-ops.port.udp_port[MTL_PORT_P] = 20000;
+multicast 239.255.197.181:16388
-ops.width  = 1280;
+source 192.168.0.2
-ops.height = 720;
+receiver interface eno1np0
-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()`
+## Runtime Setup
 
-```cpp
+Run these after boot before starting the receiver.
-static int on_frame_ready(void* priv, void* frame, st20_rx_frame_meta* meta) {
+
-    auto* self = static_cast<St2110InNode*>(priv);
+### Hugepages
-    // frame points to a complete assembled frame — MTL did the reassembly
+
-    const uint64_t index = mxlGetCurrentIndex(&self->video_rate);
+MTL initializes DPDK EAL even when using the kernel socket backend, so hugepages
-    mxlGrainInfo grain{};
+must exist.
-    uint8_t* buf = nullptr;
+
-    if (mxlFlowWriterOpenGrain(self->writer, index, &grain, &buf) == MXL_STATUS_OK) {
+```bash
-        memcpy(buf, frame, self->frame_size); // ← see format note
+sudo mkdir -p /mnt/huge
-        grain.validSlices = grain.totalSlices;
+sudo mount -t hugetlbfs nodev /mnt/huge
-        mxlFlowWriterCommitGrain(self->writer, &grain);
+echo 1024 | sudo tee /proc/sys/vm/nr_hugepages
-    }
+grep Huge /proc/meminfo
-    st20_rx_put_framebuff(self->rx, frame); // return buffer to MTL pool
+```
-    return 0;
+
+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..."
 }
 ```
 
-## Format conversion: RFC 4175 vs V210
+The SDP parser currently supports:
 
-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:
+```text
-- RFC 4175: 5 bytes per 2 pixels, big-endian packed
+m=video
-- V210: 4 bytes per 3 luma + 2 chroma, little-endian with padding bits
+c=IN IP4
-
+a=source-filter
-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`.
+a=fmtp width/height/depth/sampling/exactframerate
-
-Check `st_frame_fmt` enum in MTL headers for what's available.
-
-## Teardown
-
-```cpp
-st20_rx_free(rx);
-mtl_uninit(dev);
 ```
 
-## Main loop
+Supported video formats:
 
-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`:
+```text
-
+YCbCr-4:2:2 depth=8  -> MTL UYVY output -> local UYVY to v210 conversion
-```cpp
+YCbCr-4:2:2 depth=10 -> MTL V210 output -> direct copy to MXL
-while (dmf::g_running.load(std::memory_order_relaxed))
-    mxlSleepForNs(10'000'000);
 ```
 
-MTL drives the pacing, exactly like DeckLink hardware does.
+## 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/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();
+}

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

@@ -119,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;