svctx: --live, each NAL injected as it arrives with the producer's capture stamp

docs/stream-timing.md

@@ -96,7 +96,8 @@ repeated, so a later difference has something to be judged against.
 | Kestrel 8832CU, ch6 | 2 | 90 | 0 | 126 | 30 | 0 | < 1 µs |
 | Jaguar1 8821AU, ch6 (beacon clock) | 5 | 358–1337 | 34 (first three) | 436–1373 | 30–40 | 2–19 | < 1 µs |
 | Jaguar3 8812EU, ch36, producer at 15 ms | 2 | 109–122 | 6 | 150 | 30 | 0 | < 1 µs |
-| Jaguar3 8812CU, `svctx` (no producer stamp) | 1 | 155 | — | — | — | 0 | < 1 µs |
+| Jaguar3 8812CU, `svctx` replay (no producer stamp) | 2 | 155–166 | — | — | — | 0 | < 1 µs |
+| Jaguar3 8812CU, `svctx --live` (stamped NALs, 20 ms delay on every 10th) | 1 | — | — | step 20.08 ms on 6.9% of stamped frames | 30 | 0 | < 1 µs |
 | Jaguar3 8812CU, `duplex` (TX+RX on one chip) | 1 | 124 | — | 150 | 30 | 0 | < 1 µs |
 
 The checks every cell passes or fails on:
@@ -146,8 +147,18 @@ The adversarial readings, in the same breath:
 - **The first seconds are the pipe, not the link.** The producer fills the
   stdin pipe while the chip is brought up, so the first ~1000 records arrive
   with stamps seconds old. The analyzer drops a 4 s warm-up for that reason.
-- **svctx has no producer stamp.** It pre-reads and replays a clip, so its
-  capture→send is each NAL's loop iteration; a live stdin mode is a follow-up.
+- **svctx's replay mode has no producer stamp.** It pre-reads and replays a
+  clip, so its capture→send is each NAL's loop iteration and `cap` stays 0.
+  `svctx --live` injects each NAL as it arrives and takes the producer's
+  `CAPTURE_TS`: one stamp per NAL, claimed by its first fragment (later
+  fragments measure from the read), so the per-NAL delay check holds through
+  fragmentation: the step comes back as 20.08 ms. The delayed share reads
+  6.9% rather than 10% because the producer's every-10th lands unevenly on
+  the clip's NAL classes and the fast-rate enhancement frames are lost more
+  often than the robust base frames, so the share is a delivery figure as
+  much as a producer one. `tests/gen_svc_nals.py --capture-ts --pace-us` is the
+  producer; the harness's `svctx-live` phase runs the same 20 ms every-10th
+  check as streamtx.
 
 ## Reading it
 

examples/common/stream_timing_tx.h

@@ -202,6 +202,12 @@ class StreamTimingTx {
     _pending_capture = false;
     return has_capture ? _pending_capture_ns : read_ns;
   }
+  // The input ended: a stamp still pending had no record to apply to, and
+  // counts as dropped like one that was overwritten.
+  void input_ended() {
+    if (_pending_capture) ++_capture_dropped;
+    _pending_capture = false;
+  }
   uint64_t capture_dropped() const { return _capture_dropped; }
 
  private:

examples/streamtx/main.cpp

@@ -515,6 +515,7 @@ int main(int argc, char **argv) {
     }
   }
 
+  timing.input_ended();
   timing.stop();
   devourer::Ev(logger->events(), "stream.done")
       .f("sent", tx_count)

examples/svctx/main.cpp

@@ -11,13 +11,19 @@
 //    produce. A production link would feed Annex-B or RTP instead — same NAL
 //    header parse, trivial framing adapter.)
 //
-// All records are read up front, then injected in a continuous loop (so a
-// receiver can capture a stable rate histogram). A live deployment would
-// inject each NAL as it arrives, once.
+// Two modes. Default: all records are read up front, then injected in a
+// continuous loop (a receiver can capture a stable rate histogram). --live:
+// each NAL is injected as it arrives, once, and EOF ends the run — what a
+// deployment does. Live mode honours the stdin control escape
+// (stream_stdin.h): a producer's CAPTURE_TS stamp applies to the next NAL
+// and rides the per-frame timing field (capture->send, has_capture) on that
+// NAL's first fragment; later fragments of the same NAL carry the time since
+// the NAL was read. The replay mode has no producer, so its frames say so
+// (has_capture=0).
 //
 // Usage:
 //   DEVOURER_PID=0x8812 DEVOURER_CHANNEL=6 ./build/svctx \
-//       [--mtu N] [--gap-us US] < nals.bin
+//       [--live] [--mtu N] [--gap-us US] < nals.bin
 
 #ifndef NOMINMAX
 #define NOMINMAX  // keep <windows.h> from defining min()/max() macros (breaks std::min)
@@ -111,9 +117,12 @@ int main(int argc, char** argv) {
   size_t mtu = 1400;
   long gap_us = 2000;
   long termux_fd = 0;
+  bool live = false;
   for (int i = 1; i < argc; ++i) {
     std::string a = argv[i];
-    if (a == "--mtu" && i + 1 < argc)
+    if (a == "--live")
+      live = true;
+    else if (a == "--mtu" && i + 1 < argc)
       mtu = static_cast<size_t>(std::strtoul(argv[++i], nullptr, 0));
     else if (a == "--gap-us" && i + 1 < argc)
       gap_us = std::strtol(argv[++i], nullptr, 0);
@@ -217,61 +226,106 @@ int main(int argc, char** argv) {
   timing.set_channel(static_cast<uint8_t>(channel));
   timing.start();
 
-  // Read all length-prefixed NALs up front.
-  std::vector<std::vector<uint8_t>> nals;
-  while (true) {
-    std::vector<uint8_t> nal;
-    /* Any incomplete record ends the read-ahead: this loop drains a whole clip
-     * before transmitting anything, so a truncated tail is simply where the
-     * clip stops. */
-    if (stream_stdin::read_record(stdin, nal, 200000) !=
-        stream_stdin::RecordResult::Ok)
-      break;
-    nals.push_back(std::move(nal));
-  }
-  if (nals.empty()) { logger->error("no NALs on stdin"); return 2; }
-  logger->info("svctx: {} NALs, mtu={}, ch{} — looping", nals.size(), mtu,
-               channel);
-
   long sent[9] = {0};  // [0..7] per TID, [8] = critical
   long frames = 0;
-  while (!g_devourer_should_stop) {
-    for (const auto& nal : nals) {
+  auto stats = [&]() {
+    /* Per-layer injection counters. JSON keys are lowercase (t3plus stands
+     * in for "T3+": temporal layers 3..7 aggregated). */
+    devourer::Ev(logger->events(), "svc.stats")
+        .f("frames", frames)
+        .f("crit", sent[8])
+        .f("t0", sent[0])
+        .f("t1", sent[1])
+        .f("t2", sent[2])
+        .f("t3plus", sent[3] + sent[4] + sent[5] + sent[6] + sent[7]);
+  };
+  /* One NAL onto the air: classify, fragment to the MTU, every fragment at
+   * the layer's rate. `capture_ns` / `has_capture` describe the NAL (a
+   * producer stamp in live mode, the read instant otherwise); only the first
+   * fragment claims the stamp, the rest measure from the read. */
+  auto inject = [&](const std::vector<uint8_t>& nal, uint64_t read_ns,
+                    uint64_t capture_ns, bool has_capture) {
+    svc::NalInfo info = svc::parse_hevc_nal(nal.data(), nal.size());
+    const std::vector<uint8_t>& rt =
+        info.critical ? rt_crit
+        : rt_tid.empty() ? rt_crit
+        : rt_tid[info.tid < rt_tid.size() ? info.tid : rt_tid.size() - 1];
+    sent[info.critical ? 8 : (info.tid > 7 ? 7 : info.tid)]++;
+    bool first = true;
+    for (size_t off = 0; off < nal.size(); off += mtu) {
+      size_t n = std::min(mtu, nal.size() - off);
+      std::vector<uint8_t> frame;
+      frame.reserve(rt.size() + dot11.size() + n);
+      frame.insert(frame.end(), rt.begin(), rt.end());
+      frame.insert(frame.end(), dot11.begin(), dot11.end());
+      frame.insert(frame.end(), nal.begin() + off, nal.begin() + off + n);
+      timing.maybe_marker(rt);
+      timing.stamp(frame.data() + rt.size() + 16, read_ns,
+                   first ? capture_ns : read_ns, first && has_capture);
+      timing.sent(rtlDevice->send_packet(frame.data(), frame.size()));
+      first = false;
+      if (gap_us > 0)
+        std::this_thread::sleep_for(std::chrono::microseconds(gap_us));
+    }
+    if (++frames % 200 == 0) stats();
+  };
+
+  if (live) {
+    logger->info("svctx: live — each NAL injected as it arrives, mtu={}, ch{}",
+                 mtu, channel);
+    std::vector<uint8_t> nal, ctl;
+    while (!g_devourer_should_stop) {
+      uint32_t len = 0;
+      const auto r = stream_stdin::read_item(stdin, nal, ctl, 200000, &len);
       if (g_devourer_should_stop) break;
-      const uint64_t nal_ns = StreamTimingTx::now_ns();
-      svc::NalInfo info = svc::parse_hevc_nal(nal.data(), nal.size());
-      const std::vector<uint8_t>& rt =
-          info.critical ? rt_crit
-          : rt_tid.empty() ? rt_crit
-          : rt_tid[info.tid < rt_tid.size() ? info.tid : rt_tid.size() - 1];
-      sent[info.critical ? 8 : (info.tid > 7 ? 7 : info.tid)]++;
-      // Fragment the NAL to the radio MTU; every fragment carries the layer rate.
-      for (size_t off = 0; off < nal.size(); off += mtu) {
-        size_t n = std::min(mtu, nal.size() - off);
-        std::vector<uint8_t> frame;
-        frame.reserve(rt.size() + dot11.size() + n);
-        frame.insert(frame.end(), rt.begin(), rt.end());
-        frame.insert(frame.end(), dot11.begin(), dot11.end());
-        frame.insert(frame.end(), nal.begin() + off, nal.begin() + off + n);
-        timing.maybe_marker(rt);
-        timing.stamp(frame.data() + rt.size() + 16, nal_ns, nal_ns, false);
-        timing.sent(rtlDevice->send_packet(frame.data(), frame.size()));
-        if (gap_us > 0)
-          std::this_thread::sleep_for(std::chrono::microseconds(gap_us));
+      if (r == stream_stdin::RecordResult::Eof) break;
+      if (r == stream_stdin::RecordResult::Control) {
+        uint64_t ns = 0;
+        if (stream_stdin::parse_capture_ts(ctl, ns)) timing.capture_stamp(ns);
+        else
+          devourer::Ev(logger->events(), "stream.ctl")
+              .f("op", ctl[0]).f("len", ctl.size()).f("applied", 0);
+        continue;
       }
-      /* Per-layer injection counters. JSON keys are lowercase (t3plus stands
-       * in for "T3+": temporal layers 3..7 aggregated). */
-      if (++frames % 200 == 0) {
-        devourer::Ev(logger->events(), "svc.stats")
-            .f("frames", frames)
-            .f("crit", sent[8])
-            .f("t0", sent[0])
-            .f("t1", sent[1])
-            .f("t2", sent[2])
-            .f("t3plus", sent[3] + sent[4] + sent[5] + sent[6] + sent[7]);
+      if (r != stream_stdin::RecordResult::Ok) {
+        logger->warn("svctx: stdin ended mid-record (len {:#x}); stopping", len);
+        break;
+      }
+      const uint64_t read_ns = StreamTimingTx::now_ns();
+      bool has_capture = false;
+      const uint64_t capture_ns = timing.take_capture(read_ns, has_capture);
+      inject(nal, read_ns, capture_ns, has_capture);
+    }
+    stats();
+    timing.input_ended();
+    devourer::Ev(logger->events(), "stream.done")
+        .f("sent", frames)
+        .f("capture_dropped", (unsigned long long)timing.capture_dropped());
+  } else {
+    // Read all length-prefixed NALs up front.
+    std::vector<std::vector<uint8_t>> nals;
+    while (true) {
+      std::vector<uint8_t> nal;
+      /* Any incomplete record ends the read-ahead: this loop drains a whole
+       * clip before transmitting anything, so a truncated tail is simply
+       * where the clip stops. */
+      if (stream_stdin::read_record(stdin, nal, 200000) !=
+          stream_stdin::RecordResult::Ok)
+        break;
+      nals.push_back(std::move(nal));
+    }
+    if (nals.empty()) { logger->error("no NALs on stdin"); return 2; }
+    logger->info("svctx: {} NALs, mtu={}, ch{} — looping", nals.size(), mtu,
+                 channel);
+    while (!g_devourer_should_stop) {
+      for (const auto& nal : nals) {
+        if (g_devourer_should_stop) break;
+        const uint64_t nal_ns = StreamTimingTx::now_ns();
+        inject(nal, nal_ns, nal_ns, false);
       }
     }
   }
+  timing.stop();
   /* Device, then interface, handle and context (DeviceSession.h). Explicit
    * only because the process has nothing left to do here — the destructor
    * does exactly the same on every other exit path. */

tests/README.md

@@ -598,8 +598,9 @@ identical runs, the run-to-run sd of the median latency reported before any
 claim), `delay` (the producer sleeps 20 ms between its capture stamp and
 every 10th record; the receiver must see that step on ~10% of frames), `hop`
 (slot hopping; the clock survives retunes where the part stamps injected
-frames), `corrupt` (CRC-failed frames kept and never fed to the fit), `svctx`,
-`duplex`. `tests/stream_timing_analyze.py` is the per-capture summary and the
+frames), `corrupt` (CRC-failed frames kept and never fed to the fit), `svctx`
+(replay), `svctx-live` (`svctx --live` fed by `gen_svc_nals.py --capture-ts
+--pace-us`, the same producer-delay check per NAL), `duplex`. `tests/stream_timing_analyze.py` is the per-capture summary and the
 checks; `PACE_US` slows the producer for a part that cannot take 500 fps (the
 8812EU on 5 GHz). Which parts stamp an injected probe response is measured by
 `tests/probe_resp_egress_tsf_check.sh` (`HWBEACON=1` for the hardware

tests/gen_svc_nals.py

@@ -71,16 +71,41 @@ def gen_nals(gops: int = 6, idr_period: int = 2,
 
 
 def _main() -> None:
-    gops = int(sys.argv[1]) if len(sys.argv) > 1 else 6
-    # legacy positional PAYLOAD arg: flatten every class to that size
+    import argparse
+    import os
+    import time
+    ap = argparse.ArgumentParser(description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("gops", nargs="?", type=int, default=6)
+    ap.add_argument("payload", nargs="?", type=int, default=None,
+                    help="legacy: flatten every class to this many body bytes")
+    ap.add_argument("--repeat", type=int, default=1,
+                    help="emit the clip this many times (a live svctx run of N seconds)")
+    ap.add_argument("--pace-us", type=int, default=0,
+                    help="sleep between NALs and flush each: a live producer, not a dump")
+    # CAPTURE_TS for svctx --live: tools/precoder/stdin_ctl.py (stdlib only).
+    sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                                    "..", "tools", "precoder"))
+    import stdin_ctl  # noqa: E402
+    stdin_ctl.add_capture_args(ap)
+    a = ap.parse_args()
     sizes = None
-    if len(sys.argv) > 2:
-        p = int(sys.argv[2])
+    if a.payload is not None:
+        p = a.payload
         sizes = {"param": p, "idr": p, "t0": p, "t1": p, "t2": p}
+    stamper = stdin_ctl.capture_stamper_from_args(a)
     out = sys.stdout.buffer
-    for nal in gen_nals(gops, sizes=sizes):
-        out.write(struct.pack("<I", len(nal)) + nal)
-    out.flush()
+    clip = gen_nals(a.gops, sizes=sizes)
+    try:
+        for _ in range(max(1, a.repeat)):
+            for nal in clip:
+                out.write(stamper.prefix() + struct.pack("<I", len(nal)) + nal)
+                if a.pace_us > 0:
+                    out.flush()
+                    time.sleep(a.pace_us / 1e6)
+        out.flush()
+    except BrokenPipeError:
+        pass
 
 
 if __name__ == "__main__":

tests/stream_timing_onair.sh

@@ -18,6 +18,9 @@
 #           decoded for what they are and never feed the clock fit.
 #   svctx   the SVC demo as the transmitter (synthetic NALs, per-layer rates):
 #           the same addr3 field and marker from a second TX demo.
+#   svctx-live  svctx --live fed by a paced, CAPTURE_TS-stamping producer with
+#           the DELAY_MS/DELAY_N producer delay: the stamp survives the UEP
+#           path (one capture per NAL, on its first fragment).
 #   duplex  the duplex demo as the transmitter (TX+RX on one chip): same again,
 #           with the capture stamp through its control escape.
 #
@@ -226,6 +229,26 @@ if [[ " $PHASES " == *" svctx "* ]]; then
     fi
 fi
 
+# ---- svctx-live ----------------------------------------------------------
+if [[ " $PHASES " == *" svctx-live "* ]]; then
+    echo "== svctx --live: stamped, paced NALs with a ${DELAY_MS} ms delay on every ${DELAY_N}th =="
+    start_witness svclive
+    off=$(mark svclive)
+    $PY tests/gen_svc_nals.py 40 --repeat 50 --pace-us "$PACE_US" --capture-ts \
+        --capture-delay "$DELAY_N:$DELAY_MS" 2>"$OUT/prod_svclive.log" |
+    sudo env DEVOURER_VID="$TX_VID" DEVOURER_PID="$TX_PID" DEVOURER_CHANNEL="$CH" \
+        DEVOURER_STREAM_TIMING="$MARKER_EVERY" DEVOURER_LOG_LEVEL=info \
+        timeout "$SECS" ./build/svctx --live --gap-us 0 >"$OUT/tx_svclive.out" 2>"$OUT/tx_svclive.log"
+    sudo pkill -INT -x svctx 2>/dev/null || true; sleep 1
+    slice svclive "$off" svclive
+    stop_witness
+    if $PY tests/stream_timing_analyze.py "$OUT/run_svclive.jsonl" --expect-delay "$DELAY_N:$DELAY_MS" >"$OUT/sum_svclive.json"; then
+        pass "svctx-live: $(grep -o '"lat_frames": [0-9]*\|"cap_frames": [0-9]*\|"delay_step_ms": [0-9.]*\|"delay_frac": [0-9.]*' "$OUT/sum_svclive.json" | tr '\n' ' ')"
+    else
+        fail "svctx-live: $(cat "$OUT/sum_svclive.json")"
+    fi
+fi
+
 # ---- duplex --------------------------------------------------------------
 if [[ " $PHASES " == *" duplex "* ]]; then
     echo "== duplex: TX+RX on the transmitter chip, capture stamp via the control escape =="