diff --git a/docs/stream-timing.md b/docs/stream-timing.md index bdefcc95..f55b57f4 100644 --- a/docs/stream-timing.md +++ b/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 diff --git a/examples/common/stream_timing_tx.h b/examples/common/stream_timing_tx.h index d5545a49..caeece79 100644 --- a/examples/common/stream_timing_tx.h +++ b/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: diff --git a/examples/streamtx/main.cpp b/examples/streamtx/main.cpp index fc32a776..0812eb59 100644 --- a/examples/streamtx/main.cpp +++ b/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) diff --git a/examples/svctx/main.cpp b/examples/svctx/main.cpp index 32087b76..a40f5f74 100644 --- a/examples/svctx/main.cpp +++ b/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 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(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(channel)); timing.start(); - // Read all length-prefixed NALs up front. - std::vector> nals; - while (true) { - std::vector 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& 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& 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 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 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& 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 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> nals; + while (true) { + std::vector 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. */ diff --git a/tests/README.md b/tests/README.md index cc3d092a..59359802 100644 --- a/tests/README.md +++ b/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 diff --git a/tests/gen_svc_nals.py b/tests/gen_svc_nals.py index 3ae2ca39..d2cb1c3a 100644 --- a/tests/gen_svc_nals.py +++ b/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(" 0: + out.flush() + time.sleep(a.pace_us / 1e6) + out.flush() + except BrokenPipeError: + pass if __name__ == "__main__": diff --git a/tests/stream_timing_onair.sh b/tests/stream_timing_onair.sh index 7854b8d6..5b3c041f 100755 --- a/tests/stream_timing_onair.sh +++ b/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 =="