From 68bd58ee06d56a9e29e48b0747a7134cf4b5efe0 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Sat, 10 Oct 2026 17:44:18 +0300 Subject: [PATCH 1/2] stream timing: join the CCX TX report to its frame on HalMAC (queue time, retries) The one stage the host cannot time is the chip's own queue. With DEVOURER_TX_REPORT on, the HalMAC firmware's CCX report echoes the descriptor's SW_DEFINE tag; two small library seams make that joinable: IRtlRadio::NextTxReportTag() (the tag the next send_packet will carry, Jaguar2/3) and IRadio::SetTxReportSink() (each decoded report handed to the caller beside the tx.report event; every generation's emit site goes through one DeliverTxReport). The TX timing helper reads the tag before every send into a 256-slot ring (src/StreamTelemetry.h TxReportJoin, selftested), joins the reports on the TX thread, and folds the on-chip queue time (raw firmware units) and retry count into stream.timing (rpt_n, rpt_fail, q_p50_raw, q_max_raw, retries_max, rpt_unmatched) plus a sampled per-frame ledger, stream.txrpt, that also carries the report's host age. Nothing on the air changes. Measured (report per frame, 20 s): steady state joins every frame on the 8812CU (streamtx, ~460 fps) and the 8812BU/T3U (duplex, ~680 fps) -- 1.02 reports per data frame (the markers), 0 unmatched, report age p50 2.2 ms, queue time p50 1 raw, max 41-607. The startup burst (the stdin backlog aired at full rate) is where reports go missing: latency outruns the tag ring and the 8812CU firmware drops reports in gaps of 7-11; the harness's `report` phase judges the second half of the run and says so. Closes #476. Co-Authored-By: Claude Fable 5.1 --- docs/logging.md | 3 +- docs/stream-timing.md | 40 +++++++++++ examples/common/stream_timing_tx.h | 106 +++++++++++++++++++++++++++- src/IRadio.h | 32 +++++++++ src/IRtlRadio.h | 8 +++ src/StreamTelemetry.h | 40 +++++++++++ src/jaguar1/RtlJaguarDevice.cpp | 2 +- src/jaguar2/RtlJaguar2Device.cpp | 6 +- src/jaguar2/RtlJaguar2Device.h | 4 ++ src/jaguar3/RtlJaguar3Device.cpp | 12 ++-- src/jaguar3/RtlJaguar3Device.h | 4 ++ tests/README.md | 5 +- tests/stream_telemetry_selftest.cpp | 16 +++++ tests/stream_timing_onair.sh | 54 +++++++++++++- 14 files changed, 318 insertions(+), 14 deletions(-) diff --git a/docs/logging.md b/docs/logging.md index 85ab03c7..541eb1f3 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -188,7 +188,8 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets]; |---|---|---| | `stream.rx` / `stream.ctl` / `stream.eof` / `stream.tx` | duplex (stdout), streamtx (stderr) | hits / op, len, [applied 0 — streamtx reports the live-knob opcodes it does not apply; op 4 CAPTURE_TS is consumed silently by both] / tx_count, [bytes] / n, ok, psdu, [total] | | `stream.ready` | duplex (stdout) | channel — the chip's bring-up (InitWrite) succeeded and the TX thread exists: from here a record on stdin is sent, not lost. A feeder waits on this, not on a lead time; a refused bring-up emits `stream.eof {tx_count 0, bringup_failed 1}` instead and exits 1 | -| `stream.timing` | streamtx, svctx, duplex (`DEVOURER_STREAM_TIMING=N`, every N data frames) | ok (the marker frame's send), frames, tq_p50_us, tq_max_us, tw_p50_us, tw_max_us, c2s_p50_us, c2s_max_us, depth_max, captured, tsf_pred, fit_ppm, fit_n, fit_resid_us, fit_resets, fit_unsupported (the host↔TSF fit, one ReadTsf per 100 ms; a reset = the chip's TSF jumped under it; unsupported = ReadTsf returns 0 on this part), presp_stamped, beacon, tx_async — the same window the marker carries on air (`examples/common/stream_timing_tx.h`) | +| `stream.timing` | streamtx, svctx, duplex (`DEVOURER_STREAM_TIMING=N`, every N data frames) | ok (the marker frame's send), frames, tq_p50_us, tq_max_us, tw_p50_us, tw_max_us, c2s_p50_us, c2s_max_us, depth_max, captured, tsf_pred, fit_ppm, fit_n, fit_resid_us, fit_resets, fit_unsupported (the host↔TSF fit, one ReadTsf per 100 ms; a reset = the chip's TSF jumped under it; unsupported = ReadTsf returns 0 on this part), presp_stamped, beacon, tx_async — the same window the marker carries on air (`examples/common/stream_timing_tx.h`); plus the CCX join when `DEVOURER_TX_REPORT` is on and the die echoes a tag (HalMAC): rpt_join, rpt_n (reports joined to frames in this window), rpt_fail (state ≠ 0), q_p50_raw, q_max_raw (on-chip queue time, raw firmware units), retries_max, rpt_unmatched (running), rpt_overflow | +| `stream.txrpt` | streamtx, svctx, duplex (`DEVOURER_TX_REPORT`, HalMAC) | a CCX report joined to its frame by the SW_DEFINE tag (first 5, then every 100th joined): frame, tag, q_raw, retries, state, final_rate, tq_us, c2s_us (the frame's host stages), age_us (send_packet → report arrival on the host), marker (0/1) | | `stream.done` | streamtx (stderr) | sent, capture_dropped (CAPTURE_TS stamps that no record followed) | | `svc.stats` | svctx | frames, crit, t0, t1, t2, t3plus | | `doctor.verdict` | doctor | verdict, reasons "0x…", efuse_reads, efuse_mismatch, efuse_bad_id, efuse_id, fw_attempted, fw_ready, rx_ok, rx_crc, init | diff --git a/docs/stream-timing.md b/docs/stream-timing.md index e9ff6978..73af6f61 100644 --- a/docs/stream-timing.md +++ b/docs/stream-timing.md @@ -191,6 +191,46 @@ The adversarial readings, in the same breath: producer; the harness's `svctx-live` phase runs the same 20 ms every-10th check as streamtx. +## The chip's own queue time: joining the CCX report + +The one stage the host cannot time is how long the chip held the frame. On +the HalMAC dies with `DEVOURER_TX_REPORT` on, the firmware answers each +transmission with a CCX report that echoes the descriptor's 8-bit SW_DEFINE +tag (`src/TxReport.h`). `IRtlRadio::NextTxReportTag()` is the tag the next +`send_packet` will carry; the TX helper reads it before every send, keeps a +256-slot ring of frame records keyed by tag, and `IRadio::SetTxReportSink` +hands it each report as it decodes (on the C2H-draining thread; the TX +thread joins them). The on-chip queue time (raw firmware units — the HalMAC +unit is not documented; Jaguar1's 256 µs is) and the retry count then enter +the `stream.timing` window and a sampled per-frame ledger, `stream.txrpt`, +which also carries the report's age: send to report on the host. + +Measured with a report requested on every frame, 20 s runs: + +| | 8812CU, streamtx (~460 fps) | 8812BU (T3U), duplex (~680 fps) | +|---|---|---| +| steady state, reports joined per data frame | 1.02 (every frame, plus the markers) | 1.02 | +| whole run, joined / frames | 8198 / 9250 | 12940 / 13550 | +| reports matching no frame | 13 | 870, all in the first seconds | +| `tx.report` tag deltas | 1 × 8113, then gaps of 7–11 | 1 × 13860 | +| report age, send → host | p50 2.2 ms steady, 7–10 ms in the burst | — | +| queue time p50 / max (raw) | 1 / 508 | 1 / 596 | + +The whole-run shortfall is the start, not the link: the first ~1000 records +are the stdin backlog aired at full rate while the chip came up, and there +the report latency outruns the 256-slot tag ring (a report arriving after +256 further sends finds its slot overwritten — the unmatched count) and the +firmware's emission ceiling drops reports outright on the 8812CU (the tag +gaps; `docs/scheduled-mac.md`). From the first paced window on, every frame +has its report. The join is exact where a report exists: the tag advances +once per send, markers included, which is why the ledger's `tag` runs ahead +of `frame` by the number of markers aired. Nothing is added to the air: this +is a transmit-side instrument. C2H must flow for it — Jaguar3 +drains it on its coex thread, a Jaguar2 transmitter needs an RX loop +(duplex, or `DEVOURER_TX_WITH_RX=thread`); Jaguar1 reports carry no tag, so +there is no join there; Kestrel, the RTL8733B and the MT7612U have no CCX +report in this form. + ## Reading it `rxdemo` with `DEVOURER_STREAM_OUT=1`: every `rx.frame` carries `fc0` (so a diff --git a/examples/common/stream_timing_tx.h b/examples/common/stream_timing_tx.h index 227fee12..97f72735 100644 --- a/examples/common/stream_timing_tx.h +++ b/examples/common/stream_timing_tx.h @@ -18,6 +18,7 @@ // gets no absolute latency and says so. #pragma once +#include #include #include #include @@ -28,6 +29,10 @@ #include "AdapterCaps.h" #include "Event.h" #include "IRadio.h" +#include "IRtlRadio.h" +#include "TxReport.h" +#include +#include #include "StreamTelemetry.h" #include "TxStats.h" #include "host_tsf_fit.h" @@ -83,8 +88,28 @@ class StreamTimingTx { _fit->start(); _log.info("stream timing: marker every {} frames; host<->TSF fit started " "(one ReadTsf per 100 ms)", _marker_every); + /* CCX report join (HalMAC dies with DeviceConfig tx.report on): every + * send records the tag its descriptor will carry; the device's sink + * queues each report, the TX thread drains and joins them. The on-chip + * queue time and retry count then enter the window (stream.timing) and + * a sampled per-frame ledger (stream.txrpt). Needs C2H to flow: an RX + * loop on Jaguar2, Jaguar3's coex thread regardless. */ + _rtl = dynamic_cast(&_dev); + if (_rtl && _rtl->NextTxReportTag()) { + _join_on = true; + _dev.SetTxReportSink([this](const devourer::TxReport &r) { + std::lock_guard lk(_rpt_mu); + if (_rpt_q.size() < 4096) _rpt_q.push_back(r); + else ++_rpt_overflow; + }); + _log.info("stream timing: CCX report join armed (tag echo)"); + } } void stop() { + if (_join_on) { + _dev.SetTxReportSink({}); + _join_on = false; + } if (_beacon) { _dev.StopBeacon(); _beacon = false; @@ -128,6 +153,16 @@ class StreamTimingTx { t.tsf10_lo = static_cast((tsf / 10) & 0xffff); } t.encode(addr3); + if (_join_on) { + if (auto tag = _rtl->NextTxReportTag()) { + devourer::stream_timing::TxFrameRec rec; + rec.frame = _frames; + rec.send_ns = _t0; + rec.t_queue_us = static_cast(_t0 > read_ns ? (_t0 - read_ns) / 1000 : 0); + rec.c2s_us = static_cast(c2s); + _join.sent(*tag, rec); + } + } } // Right after send_packet returned (`ok` = its result). No marker is aired // before the first data frame went out: the duplex demo's TX thread can run @@ -142,6 +177,41 @@ class StreamTimingTx { _last_write_us = tw; _window.add(tq, tw, c2s, _has_capture, _depth); ++_frames; + drain_reports(); + } + + // Join every queued CCX report to its frame; fold queue time and retries + // into the window and the sampled stream.txrpt ledger. + void drain_reports() { + if (!_join_on) return; + std::deque q; + { + std::lock_guard lk(_rpt_mu); + q.swap(_rpt_q); + } + const uint64_t now = now_ns(); + for (const auto &r : q) { + auto rec = _join.match(r.sw_define); + if (!rec) continue; + ++_w_rpt; + if (r.queue_time_raw > _w_q_max) _w_q_max = r.queue_time_raw; + if (_w_q.size() < 4096) _w_q.push_back(r.queue_time_raw); + if (r.data_retries > _w_retries_max) _w_retries_max = r.data_retries; + if (r.state != 0) ++_w_rpt_fail; + const uint64_t n = _join.joined(); + if (n <= 5 || n % 100 == 0) + devourer::Ev(_ev, "stream.txrpt") + .f("frame", (unsigned long long)rec->frame) + .f("tag", r.sw_define) + .f("q_raw", r.queue_time_raw) + .f("retries", r.data_retries) + .f("state", r.state) + .f("final_rate", r.final_rate) + .f("tq_us", rec->t_queue_us) + .f("c2s_us", rec->c2s_us) + .f("age_us", (unsigned long long)((now - rec->send_ns) / 1000)) + .f("marker", rec->marker ? 1 : 0); + } } // Before each data frame: air the marker when due. `radiotap` is the stream @@ -174,7 +244,21 @@ class StreamTimingTx { _buf.clear(); _buf.insert(_buf.end(), radiotap.begin(), radiotap.end()); devourer::stream_timing::append_marker_mpdu(_buf, _sa, _channel, m); + if (_join_on) { + if (auto tag = _rtl->NextTxReportTag()) { + devourer::stream_timing::TxFrameRec rec; + rec.frame = _frames; rec.send_ns = hn; rec.marker = true; + _join.sent(*tag, rec); + } + } const bool ok = _dev.send_packet(_buf.data(), _buf.size()); + drain_reports(); + uint32_t q_p50 = 0; + if (!_w_q.empty()) { + auto mid = _w_q.begin() + static_cast(_w_q.size() / 2); + std::nth_element(_w_q.begin(), mid, _w_q.end()); + q_p50 = *mid; + } devourer::Ev(_ev, "stream.timing") .f("ok", ok ? 1 : 0) .f("frames", m.frames) @@ -194,7 +278,19 @@ class StreamTimingTx { .f("fit_unsupported", _fit && _fit->unsupported() ? 1 : 0) .f("presp_stamped", _presp_stamped ? 1 : 0) .f("beacon", _beacon ? 1 : 0) - .f("tx_async", _tx_async ? 1 : 0); + .f("tx_async", _tx_async ? 1 : 0) + /* The CCX join (HalMAC + tx.report): reports joined in this window, + * the on-chip queue time (raw firmware units) p50/max, the worst + * retry count, failed deliveries, and the running unmatched total. */ + .f("rpt_join", _join_on ? 1 : 0) + .f("rpt_n", _w_rpt) + .f("rpt_fail", _w_rpt_fail) + .f("q_p50_raw", q_p50) + .f("q_max_raw", _w_q_max) + .f("retries_max", _w_retries_max) + .f("rpt_unmatched", (unsigned long long)_join.unmatched()) + .f("rpt_overflow", (unsigned long long)_rpt_overflow); + _w_rpt = _w_rpt_fail = 0; _w_q_max = 0; _w_retries_max = 0; _w_q.clear(); return true; } @@ -230,6 +326,14 @@ class StreamTimingTx { uint8_t _channel = 0; long _marker_every = 0; bool _started = false, _data_sent_ok = false, _warned_unsupported = false; + IRtlRadio *_rtl = nullptr; + bool _join_on = false; + std::mutex _rpt_mu; + std::deque _rpt_q; + uint64_t _rpt_overflow = 0; + devourer::stream_timing::TxReportJoin _join; + std::vector _w_q; + uint32_t _w_rpt = 0, _w_rpt_fail = 0, _w_q_max = 0, _w_retries_max = 0; bool _tx_async = false, _presp_stamped = false, _beacon = false; std::unique_ptr _fit; devourer::stream_timing::TimingWindow _window; diff --git a/src/IRadio.h b/src/IRadio.h index 5b54b7dd..f386e6a2 100644 --- a/src/IRadio.h +++ b/src/IRadio.h @@ -13,6 +13,8 @@ #include "RxQuality.h" #include "SelectedChannel.h" #include "ThermalStatus.h" +#include "TxReport.h" +#include #include "Sounding.h" #include "TriggerTwt.h" #include "TxCaps.h" @@ -660,6 +662,18 @@ class IRadio { virtual void SetTxMode(const devourer::TxMode & /*mode*/) {} virtual void ClearTxMode() {} + /* Per-frame CCX TX reports (TxReport.h, DeviceConfig tx.report) handed to + * the caller as they decode, in addition to the `tx.report` event. The sink + * runs on whichever thread drains C2H for this generation (the RX worker, + * or Jaguar3's coex thread), so it must be cheap and thread-safe; set it + * before the RX/coex path starts. On the HalMAC dies a report's sw_define + * is the tag IRtlRadio::NextTxReportTag() gave the frame, which is how a + * caller joins a report to the frame it sent. An empty function clears it. */ + void SetTxReportSink(std::function sink) { + std::lock_guard lk(_tx_report_sink_mu); + _tx_report_sink = std::move(sink); + } + /* TX submission health snapshot (see TxStats.h) — the driver-side drop / * congestion signal an adaptive-link controller uses to detect a full TX FIFO * (a bulk-OUT TIMEOUT = recoverable back-pressure) vs a hard error. Counted at @@ -761,6 +775,24 @@ class IRadio { * this is the only place the failure is visible to a caller. */ virtual devourer::FwBootStatus GetFwBootStatus() { return {}; } + +protected: + /* Emit the `tx.report` event and hand the report to the sink, if any. */ + void DeliverTxReport(devourer::EventSink &events, const devourer::TxReport &r, + const char *fmt) { + devourer::emit_tx_report(events, r, fmt); + std::function sink; + { + std::lock_guard lk(_tx_report_sink_mu); + sink = _tx_report_sink; + } + if (sink) sink(r); + } + +private: + std::mutex _tx_report_sink_mu; + std::function _tx_report_sink; + }; #endif /* IRADIO_H */ diff --git a/src/IRtlRadio.h b/src/IRtlRadio.h index 90ba4ee9..944b33d0 100644 --- a/src/IRtlRadio.h +++ b/src/IRtlRadio.h @@ -35,6 +35,14 @@ * it, and it saves five identical per-backend overrides. */ class IRtlRadio : public IRadio { public: + /* The SW_DEFINE tag the NEXT send_packet's descriptor will carry, on the + * HalMAC dies with DeviceConfig tx.report on (the report echoes it, so a + * caller that reads this right before each send can join every TxReport + * to its frame). nullopt where there is no tag echo (Jaguar1, Kestrel, + * RTL8733B, MT7612U) or reports are off. Single-sender semantics: the tag + * advances once per frame the device builds a data descriptor for. */ + virtual std::optional NextTxReportTag() const { return std::nullopt; } + /* Crystal (XTAL) load-capacitance trim — the CFO lever. Writes the AFE * crystal-cap field (a per-chip register), pulling the chip's reference * oscillator a few ppm to align a marginal TX/RX crystal pair; the payoff diff --git a/src/StreamTelemetry.h b/src/StreamTelemetry.h index bf563487..d1408113 100644 --- a/src/StreamTelemetry.h +++ b/src/StreamTelemetry.h @@ -45,6 +45,7 @@ #include #include #include +#include #include namespace devourer { @@ -359,6 +360,45 @@ class TimingWindow { uint64_t _q_max = 0, _w_max = 0, _c_max = 0; }; +/* ---- CCX report join ----------------------------------------------------- */ + +/* Joins a HalMAC CCX TX report to the frame it describes by the 8-bit + * SW_DEFINE tag the descriptor carried (IRtlRadio::NextTxReportTag read right + * before the send). A 256-slot ring keyed by tag: a slot is written per + * send, consumed by the report that echoes its tag, and a report whose slot + * is empty or already consumed (a tag that wrapped past 256 unreported + * frames, or a report for a frame sent before the join was armed) counts as + * unmatched. Pure; the TX thread owns it. */ +struct TxFrameRec { + uint64_t frame = 0; /* the transmitter's frame counter */ + uint64_t send_ns = 0; /* host clock at send_packet */ + uint32_t t_queue_us = 0, c2s_us = 0; + bool marker = false; /* a marker frame, not a data frame */ +}; +class TxReportJoin { + public: + void sent(uint8_t tag, const TxFrameRec &rec) { + _slot[tag] = rec; + _live[tag] = true; + ++_sent; + } + /* The record for a report's tag, or nullopt (counted as unmatched). */ + std::optional match(uint8_t tag) { + if (!_live[tag]) { ++_unmatched; return std::nullopt; } + _live[tag] = false; + ++_joined; + return _slot[tag]; + } + uint64_t sent_count() const { return _sent; } + uint64_t joined() const { return _joined; } + uint64_t unmatched() const { return _unmatched; } + + private: + std::array _slot{}; + std::array _live{}; + uint64_t _sent = 0, _joined = 0, _unmatched = 0; +}; + } // namespace stream_timing } // namespace devourer diff --git a/src/jaguar1/RtlJaguarDevice.cpp b/src/jaguar1/RtlJaguarDevice.cpp index 271df27d..51c0c4dd 100644 --- a/src/jaguar1/RtlJaguarDevice.cpp +++ b/src/jaguar1/RtlJaguarDevice.cpp @@ -1963,7 +1963,7 @@ void RtlJaguarDevice::StartRxLoop(Action_ParsedRadioPacket packetProcessor) { const devourer::TxReport r = devourer::parse_ccx_8812(p.Data.data() + 2, p.Data.size() - 2); if (r.valid) - devourer::emit_tx_report(_logger->events(), r, "8812"); + DeliverTxReport(_logger->events(), r, "8812"); } _packetProcessor(p); } diff --git a/src/jaguar2/RtlJaguar2Device.cpp b/src/jaguar2/RtlJaguar2Device.cpp index b73061c7..1cb08fd4 100644 --- a/src/jaguar2/RtlJaguar2Device.cpp +++ b/src/jaguar2/RtlJaguar2Device.cpp @@ -711,9 +711,9 @@ void RtlJaguar2Device::StartRxLoop(Action_ParsedRadioPacket packetProcessor) { * C2H pkt 0xFF/0x0F carries per-frame delivery + retry count — * decode + emit tx.report (src/TxReport.h). */ if (is_c2h && devourer::is_ccx_halmac(f.frame, f.frame_len)) - devourer::emit_tx_report( - _logger->events(), - devourer::parse_ccx_halmac(f.frame, f.frame_len), "halmac"); + DeliverTxReport(_logger->events(), + devourer::parse_ccx_halmac(f.frame, f.frame_len), + "halmac"); /* Per-frame RSSI/SNR/EVM from the jgr2 PHY-status (present when * APP_PHYSTS is on, i.e. drvinfo carries the 32-byte report). CCK rates * (DESC_RATE1M..11M = 0..3) use type0, everything else type1. C2H has no diff --git a/src/jaguar2/RtlJaguar2Device.h b/src/jaguar2/RtlJaguar2Device.h index f47ef04d..c61273d2 100644 --- a/src/jaguar2/RtlJaguar2Device.h +++ b/src/jaguar2/RtlJaguar2Device.h @@ -53,6 +53,10 @@ class RtlJaguar2Device : public IRtlRadio { * runs this on its own std::thread next to the TX loop. Starts (and on exit * stops) the DIG thread — TX-only sessions stay DIG-free. */ void StartRxLoop(Action_ParsedRadioPacket packetProcessor) override; + std::optional NextTxReportTag() const override { + if (!_cfg.tx.report) return std::nullopt; + return static_cast(_tx_rpt_tag.load(std::memory_order_relaxed) & 0xff); + } void StopRxLoop() override { _rx_stop = true; } void SetMonitorChannel(SelectedChannel channel) override; /* Lean frequency-hop retune (the Jaguar2 port of the Jaguar1 FastRetune — diff --git a/src/jaguar3/RtlJaguar3Device.cpp b/src/jaguar3/RtlJaguar3Device.cpp index 81307b93..a6372ba8 100644 --- a/src/jaguar3/RtlJaguar3Device.cpp +++ b/src/jaguar3/RtlJaguar3Device.cpp @@ -379,9 +379,9 @@ void RtlJaguar3Device::StartRxLoop(Action_ParsedRadioPacket packetProcessor) { * C2H pkt 0xFF/0x0F carries per-frame delivery + retry count — * decode + emit tx.report (src/TxReport.h). */ if (is_c2h && devourer::is_ccx_halmac(f.frame, f.frame_len)) - devourer::emit_tx_report( - _logger->events(), - devourer::parse_ccx_halmac(f.frame, f.frame_len), "halmac"); + DeliverTxReport(_logger->events(), + devourer::parse_ccx_halmac(f.frame, f.frame_len), + "halmac"); /* Decode the jgr3 PHY-status report (per-frame RSSI/SNR/EVM), which * sits immediately after the 24-byte descriptor inside the drvinfo * area (monitor_rx_cfg enables APP_PHYSTS + RX_DRVINFO_SZ=4). @@ -543,9 +543,9 @@ void RtlJaguar3Device::coex_runtime_loop() { static_cast(n) - off, f)) { if ((buf[off + 11] & 0x10) && devourer::is_ccx_halmac(f.frame, f.frame_len)) - devourer::emit_tx_report( - _logger->events(), - devourer::parse_ccx_halmac(f.frame, f.frame_len), "halmac"); + DeliverTxReport(_logger->events(), + devourer::parse_ccx_halmac(f.frame, f.frame_len), + "halmac"); if (f.next_offset == 0) break; off += f.next_offset; diff --git a/src/jaguar3/RtlJaguar3Device.h b/src/jaguar3/RtlJaguar3Device.h index 12d143ec..15de9c39 100644 --- a/src/jaguar3/RtlJaguar3Device.h +++ b/src/jaguar3/RtlJaguar3Device.h @@ -50,6 +50,10 @@ class RtlJaguar3Device : public IRtlRadio { * Takes over the bulk-IN endpoint from the coex thread's C2H drain for as * long as it runs. */ void StartRxLoop(Action_ParsedRadioPacket packetProcessor) override; + std::optional NextTxReportTag() const override { + if (!_cfg.tx.report) return std::nullopt; + return static_cast(_tx_rpt_tag.load(std::memory_order_relaxed) & 0xff); + } void StopRxLoop() override { _rx_stop = true; } void SetMonitorChannel(SelectedChannel channel) override; /* Lean frequency-hop retune (Jaguar3 port of the Jaguar1 FastRetune — see diff --git a/tests/README.md b/tests/README.md index 4eade682..a9b84bc3 100644 --- a/tests/README.md +++ b/tests/README.md @@ -600,7 +600,10 @@ 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` (replay), `svctx-live` (`svctx --live` fed by `gen_svc_nals.py --capture-ts ---pace-us`, the same producer-delay check per NAL), `duplex`. `tests/mcast_da_rx_check.sh` is the gate behind the +--pace-us`, the same producer-delay check per NAL), `duplex`, `report` +(streamtx with a CCX report per frame: the reports must join their frames by +tag; `TX_REPORT=1` adds the join verdict to the duplex phase too, which is how +a Jaguar2 transmitter is covered). `tests/mcast_da_rx_check.sh` is the gate behind the addr1 extension: an injector alternates the broadcast DA and the group DA the extension's encoder ships (`07:…`) and each receiver family (the MT7612U with `MT7612U_FW_DIR`) must deliver both at parity. `tests/stream_timing_analyze.py` is the per-capture summary and the diff --git a/tests/stream_telemetry_selftest.cpp b/tests/stream_telemetry_selftest.cpp index 07ebc1db..2013095c 100644 --- a/tests/stream_telemetry_selftest.cpp +++ b/tests/stream_telemetry_selftest.cpp @@ -162,6 +162,22 @@ int main() { CHECK(m.t_queue_max_10 == 4000, "long window max exact"); } + // ---- CCX join ring -------------------------------------------------------- + { + TxReportJoin j; + std::optional r; + CHECK(!j.match(5) && j.unmatched() == 1, "report before any send is unmatched"); + TxFrameRec a; a.frame = 7; a.t_queue_us = 30; + j.sent(5, a); + r = j.match(5); + CHECK(r && r->frame == 7 && r->t_queue_us == 30 && j.joined() == 1, "report joins its frame"); + CHECK(!j.match(5) && j.unmatched() == 2, "a second report for the same tag is unmatched"); + for (unsigned i = 0; i < 300; ++i) { TxFrameRec x; x.frame = 100 + i; j.sent(uint8_t(i), x); } + r = j.match(10); // tag 10 was written at i=10 and again at i=266 + CHECK(r && r->frame == 100 + 266, "a wrapped tag yields the latest frame with it"); + CHECK(j.sent_count() == 301, "sent count"); + } + if (fails == 0) std::printf("stream_telemetry selftest OK\n"); return fails == 0 ? 0 : 1; } diff --git a/tests/stream_timing_onair.sh b/tests/stream_timing_onair.sh index 5b3c041f..76d164c5 100755 --- a/tests/stream_timing_onair.sh +++ b/tests/stream_timing_onair.sh @@ -18,6 +18,12 @@ # 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. +# report streamtx with DEVOURER_TX_REPORT=1 (HalMAC CCX reports, tag echo): +# every report must join the frame it describes — rpt_n ≈ frames in +# each marker window, few unmatched — so the on-chip queue time and +# retry count enter stream.timing / stream.txrpt. Jaguar3 drains +# C2H on its coex thread; a Jaguar2 transmitter needs an RX loop +# (run the duplex phase with TX_REPORT=1 there). # 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). @@ -42,6 +48,7 @@ RX_VID="${RX_VID:-0x0bda}"; RX_PID="${RX_PID:-0xb812}" # CF-924AC (8822BU) CH="${CH:-6}"; SECS="${SECS:-20}"; REPS="${REPS:-3}" MARKER_EVERY="${MARKER_EVERY:-50}" PACE_US="${PACE_US:-2000}" +TX_REPORT="${TX_REPORT:-}" # set to 1 to request a CCX report per frame in every phase DELAY_N="${DELAY_N:-10}"; DELAY_MS="${DELAY_MS:-20}" HOP_CHANNELS="${HOP_CHANNELS:-1,6,11}"; HOP_SLOT_MS="${HOP_SLOT_MS:-50}" PHASES="${PHASES:-floor delay hop corrupt}" @@ -98,7 +105,8 @@ run_tx() { "${UVRUN[@]}" tools/precoder/stream_tx.py --input "$OUT/src.bin" --repeat 2000 \ --pace-us "$PACE_US" $prod 2>"$OUT/prod_$tag.log" | sudo env DEVOURER_VID="$TX_VID" DEVOURER_PID="$TX_PID" DEVOURER_CHANNEL="$CH" \ - DEVOURER_STREAM_TIMING="$MARKER_EVERY" DEVOURER_LOG_LEVEL=info "$@" \ + DEVOURER_STREAM_TIMING="$MARKER_EVERY" DEVOURER_LOG_LEVEL=info \ + ${TX_REPORT:+DEVOURER_TX_REPORT="$TX_REPORT"} "$@" \ timeout "$SECS" ./build/streamtx --interval-ms 0 >"$OUT/tx_$tag.out" 2>"$OUT/tx_$tag.log" sudo pkill -INT -x streamtx 2>/dev/null || true sleep 1 @@ -229,6 +237,48 @@ if [[ " $PHASES " == *" svctx "* ]]; then fi fi +# The CCX join verdict from a TX log's stream.timing windows: reports joined +# vs frames aired, unmatched reports, and that queue time actually arrived. +join_verdict() { # $1 = tx log, $2 = label + $PY - "$1" "$2" <<'PY' +import sys +sys.path.insert(0, "tests") +from devourer_events import iter_events +w = [e for e in iter_events(open(sys.argv[1]), ev="stream.timing")] +# The first ~1000 records are the stdin backlog aired at full rate while the +# chip came up; there the report latency outruns the tag ring and the +# firmware's CCX emission ceiling drops reports (docs/scheduled-mac.md). Judge +# the steady state: the second half of the run's windows. +w = [e for e in w if e.get("rpt_join") == 1] +w = w[len(w) // 2:] +frames = sum(e["frames"] for e in w); rpt = sum(e["rpt_n"] for e in w) +fail = sum(e["rpt_fail"] for e in w) +unm = (w[-1]["rpt_unmatched"] - w[0]["rpt_unmatched"]) if w else 0 +ovf = w[-1]["rpt_overflow"] if w else 0 +qmax = max((e["q_max_raw"] for e in w), default=0) +q50 = sorted(e["q_p50_raw"] for e in w)[len(w) // 2] if w else 0 +ratio = rpt / frames if frames else 0 +# The firmware drops some reports (tag gaps; docs/scheduled-mac.md): 0.8 is +# the floor a healthy join clears, the ratio itself is the number to read. +ok = frames >= 500 and ratio >= 0.8 and unm <= 0.02 * max(rpt, 1) and ovf == 0 +print(f"{'PASS' if ok else 'FAIL'}: {sys.argv[2]}: joined {rpt}/{frames} ({ratio:.3f}) unmatched={unm} " + f"overflow={ovf} fail={fail} q_p50_raw={q50} q_max_raw={qmax} windows={len(w)}") +sys.exit(0 if ok else 1) +PY +} + +# ---- report -------------------------------------------------------------- +if [[ " $PHASES " == *" report "* ]]; then + echo "== report: streamtx with a CCX report per frame, joined by tag ==" + start_witness report + off=$(mark report) + TX_REPORT=1 run_tx report "--capture-ts" + slice report "$off" report + stop_witness + if join_verdict "$OUT/tx_report.log" "report"; then pass "report: CCX reports join their frames (see line above)"; else fail "report: join"; fi + $PY tests/stream_timing_analyze.py "$OUT/run_report.jsonl" >"$OUT/sum_report.json" || fail "report: the stream itself degraded with reports on ($OUT/sum_report.json)" +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 ==" @@ -263,11 +313,13 @@ if [[ " $PHASES " == *" duplex "* ]]; then --pace-us "$PACE_US" --capture-ts 2>"$OUT/prod_duplex.log" ) | sudo env DEVOURER_VID="$TX_VID" DEVOURER_PID="$TX_PID" DEVOURER_CHANNEL="$CH" \ DEVOURER_STREAM_TIMING="$MARKER_EVERY" DEVOURER_LOG_LEVEL=info \ + ${TX_REPORT:+DEVOURER_TX_REPORT="$TX_REPORT"} \ timeout $((SECS + DUPLEX_LEAD_S + 10)) ./build/duplex --interval-ms 0 >"$OUT/tx_duplex.out" 2>"$OUT/tx_duplex.log" grep -q -F '"ev":"stream.ready"' "$OUT/tx_duplex.out" || fail "duplex: no stream.ready event" sudo pkill -INT -x duplex 2>/dev/null || true; sleep 1 slice duplex "$off" duplex stop_witness + if [ -n "$TX_REPORT" ]; then join_verdict "$OUT/tx_duplex.out" "duplex report" || fail "duplex: join"; fi if $PY tests/stream_timing_analyze.py "$OUT/run_duplex.jsonl" >"$OUT/sum_duplex.json"; then pass "duplex: $(grep -o '"lat_frames": [0-9]*\|"lat_p50_us": [0-9]*\|"c2a_p50_us": [0-9]*\|"markers": [0-9]*' "$OUT/sum_duplex.json" | tr '\n' ' ')" else From de5a162d81d86c7132176b24681ab961038cb430 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Sat, 10 Oct 2026 20:51:05 +0300 Subject: [PATCH 2/2] =?UTF-8?q?CCX=20join:=20review=20round=20=E2=80=94=20?= =?UTF-8?q?overwrite=20counter,=20in-flight-safe=20sink=20removal,=20hop-s?= =?UTF-8?q?ync=20tags?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - TxReportJoin counts a send that reuses a still-live tag (rpt_overwritten): an eight-bit tag cannot name its generation, so such a window's joins are suspect; selftest covers the late-report-after-reuse case and says what happens. The harness verdict requires zero overwrites in steady state. - IRadio::SetTxReportSink returns only once no earlier sink is still executing (in-flight count + condition variable), so a helper may destroy what its sink captured right after clearing it. - The report queue stamps each report when the host decoded it; age_us is send -> that instant, not the TX thread's drain time. The overflow counter is atomic. - streamtx's hop sync marker records its tag through the helper (note_external_send) so its report joins as a marker. - The duplex join verdict runs only for a positive report divisor; the report phase is documented as Jaguar3's (streamtx runs no RX loop, so a Jaguar2 transmitter is covered by the duplex phase), and the verdict says so when no join window exists. Measured after the changes: report phase 8812CU 1.02 joins/frame, 0 unmatched, 0 overwritten; duplex 8812BU/T3U ch36 1.02, 0, 0; hop + reports on the 8812CU joins 1.30/frame steady (markers included) while ~15% of sends get no report in bursts of 7-13 around retunes -- rpt_overwritten counts that backlog, documented as such. Co-Authored-By: Claude Fable 5.1 --- docs/logging.md | 4 ++-- docs/stream-timing.md | 34 ++++++++++++++++++++------- examples/common/stream_timing_tx.h | 36 +++++++++++++++++++++-------- examples/streamtx/main.cpp | 1 + src/IRadio.h | 18 +++++++++++++-- src/StreamTelemetry.h | 12 ++++++++-- tests/stream_telemetry_selftest.cpp | 9 ++++++-- tests/stream_timing_onair.sh | 15 ++++++++---- 8 files changed, 99 insertions(+), 30 deletions(-) diff --git a/docs/logging.md b/docs/logging.md index 541eb1f3..672c7406 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -188,8 +188,8 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets]; |---|---|---| | `stream.rx` / `stream.ctl` / `stream.eof` / `stream.tx` | duplex (stdout), streamtx (stderr) | hits / op, len, [applied 0 — streamtx reports the live-knob opcodes it does not apply; op 4 CAPTURE_TS is consumed silently by both] / tx_count, [bytes] / n, ok, psdu, [total] | | `stream.ready` | duplex (stdout) | channel — the chip's bring-up (InitWrite) succeeded and the TX thread exists: from here a record on stdin is sent, not lost. A feeder waits on this, not on a lead time; a refused bring-up emits `stream.eof {tx_count 0, bringup_failed 1}` instead and exits 1 | -| `stream.timing` | streamtx, svctx, duplex (`DEVOURER_STREAM_TIMING=N`, every N data frames) | ok (the marker frame's send), frames, tq_p50_us, tq_max_us, tw_p50_us, tw_max_us, c2s_p50_us, c2s_max_us, depth_max, captured, tsf_pred, fit_ppm, fit_n, fit_resid_us, fit_resets, fit_unsupported (the host↔TSF fit, one ReadTsf per 100 ms; a reset = the chip's TSF jumped under it; unsupported = ReadTsf returns 0 on this part), presp_stamped, beacon, tx_async — the same window the marker carries on air (`examples/common/stream_timing_tx.h`); plus the CCX join when `DEVOURER_TX_REPORT` is on and the die echoes a tag (HalMAC): rpt_join, rpt_n (reports joined to frames in this window), rpt_fail (state ≠ 0), q_p50_raw, q_max_raw (on-chip queue time, raw firmware units), retries_max, rpt_unmatched (running), rpt_overflow | -| `stream.txrpt` | streamtx, svctx, duplex (`DEVOURER_TX_REPORT`, HalMAC) | a CCX report joined to its frame by the SW_DEFINE tag (first 5, then every 100th joined): frame, tag, q_raw, retries, state, final_rate, tq_us, c2s_us (the frame's host stages), age_us (send_packet → report arrival on the host), marker (0/1) | +| `stream.timing` | streamtx, svctx, duplex (`DEVOURER_STREAM_TIMING=N`, every N data frames) | ok (the marker frame's send), frames, tq_p50_us, tq_max_us, tw_p50_us, tw_max_us, c2s_p50_us, c2s_max_us, depth_max, captured, tsf_pred, fit_ppm, fit_n, fit_resid_us, fit_resets, fit_unsupported (the host↔TSF fit, one ReadTsf per 100 ms; a reset = the chip's TSF jumped under it; unsupported = ReadTsf returns 0 on this part), presp_stamped, beacon, tx_async — the same window the marker carries on air (`examples/common/stream_timing_tx.h`); plus the CCX join when `DEVOURER_TX_REPORT` is on and the die echoes a tag (HalMAC): rpt_join, rpt_n (reports joined to frames in this window), rpt_fail (state ≠ 0), q_p50_raw, q_max_raw (on-chip queue time, raw firmware units), retries_max, rpt_unmatched (running), rpt_overwritten (running: a send reused a tag whose report had not returned — an unreported send, and that window's joins are suspect), rpt_overflow | +| `stream.txrpt` | streamtx, svctx, duplex (`DEVOURER_TX_REPORT`, HalMAC) | a CCX report joined to its frame by the SW_DEFINE tag (first 5, then every 100th joined): frame, tag, q_raw, retries, state, final_rate, tq_us, c2s_us (the frame's host stages), age_us (send_packet → the host decoding the report, stamped in the sink), marker (0/1) | | `stream.done` | streamtx (stderr) | sent, capture_dropped (CAPTURE_TS stamps that no record followed) | | `svc.stats` | svctx | frames, crit, t0, t1, t2, t3plus | | `doctor.verdict` | doctor | verdict, reasons "0x…", efuse_reads, efuse_mismatch, efuse_bad_id, efuse_id, fw_attempted, fw_ready, rx_ok, rx_crc, init | diff --git a/docs/stream-timing.md b/docs/stream-timing.md index 73af6f61..6cadfbf0 100644 --- a/docs/stream-timing.md +++ b/docs/stream-timing.md @@ -218,16 +218,34 @@ Measured with a report requested on every frame, 20 s runs: The whole-run shortfall is the start, not the link: the first ~1000 records are the stdin backlog aired at full rate while the chip came up, and there -the report latency outruns the 256-slot tag ring (a report arriving after -256 further sends finds its slot overwritten — the unmatched count) and the -firmware's emission ceiling drops reports outright on the 8812CU (the tag -gaps; `docs/scheduled-mac.md`). From the first paced window on, every frame -has its report. The join is exact where a report exists: the tag advances +the report latency outruns the 256-slot tag ring (a send reusing a tag whose +report has not returned overwrites the slot — `rpt_overwritten`; an +eight-bit tag cannot name its generation, so a late report for the old frame +would land on the new one and the new frame's own report go unmatched, which +is why a window with overwrites is suspect) and the firmware's emission +ceiling drops reports outright on the 8812CU (the tag gaps; +`docs/scheduled-mac.md`). From the first paced window on, every frame has +its report, and `rpt_overwritten` stays at zero. The join is exact where a +report exists: the tag advances once per send, markers included, which is why the ledger's `tag` runs ahead of `frame` by the number of markers aired. Nothing is added to the air: this -is a transmit-side instrument. C2H must flow for it — Jaguar3 -drains it on its coex thread, a Jaguar2 transmitter needs an RX loop -(duplex, or `DEVOURER_TX_WITH_RX=thread`); Jaguar1 reports carry no tag, so +is a transmit-side instrument. + +With slot hopping on as well (8812CU, 1/6/11 at 50 ms, a report per frame), +the steady state still joins 1.30 reports per data frame (the timing and +hop-sync markers included, both recorded through the helper), but about 15% +of all sends never get a report: the `tx.report` tag sequence shows ~190 +gaps of 7–13 over 340 dwells, so the firmware drops a burst of reports +around a retune. A dropped report leaves its slot live until the tag wraps, +and `rpt_overwritten` (1732 over that run) then counts the backlog — an +upper bound on suspect joins, since a report that never arrives cannot +mis-join. Read the counter as "this many sends went unreported", and judge +hop-mode queue times by their p50, not their maximum. + +C2H must flow for it — Jaguar3 +drains it on its coex thread; a Jaguar2 transmitter needs an RX loop on the +same handle, which streamtx never runs, so `duplex` is the Jaguar2 path (as +measured above); Jaguar1 reports carry no tag, so there is no join there; Kestrel, the RTL8733B and the MT7612U have no CCX report in this form. diff --git a/examples/common/stream_timing_tx.h b/examples/common/stream_timing_tx.h index 97f72735..a9ecb603 100644 --- a/examples/common/stream_timing_tx.h +++ b/examples/common/stream_timing_tx.h @@ -19,6 +19,7 @@ #pragma once #include +#include #include #include #include @@ -98,9 +99,10 @@ class StreamTimingTx { if (_rtl && _rtl->NextTxReportTag()) { _join_on = true; _dev.SetTxReportSink([this](const devourer::TxReport &r) { + const uint64_t rx_ns = now_ns(); /* when the host decoded it */ std::lock_guard lk(_rpt_mu); - if (_rpt_q.size() < 4096) _rpt_q.push_back(r); - else ++_rpt_overflow; + if (_rpt_q.size() < 4096) _rpt_q.push_back({r, rx_ns}); + else _rpt_overflow.fetch_add(1, std::memory_order_relaxed); }); _log.info("stream timing: CCX report join armed (tag echo)"); } @@ -184,13 +186,13 @@ class StreamTimingTx { // into the window and the sampled stream.txrpt ledger. void drain_reports() { if (!_join_on) return; - std::deque q; + std::deque q; { std::lock_guard lk(_rpt_mu); q.swap(_rpt_q); } - const uint64_t now = now_ns(); - for (const auto &r : q) { + for (const auto &qr : q) { + const devourer::TxReport &r = qr.r; auto rec = _join.match(r.sw_define); if (!rec) continue; ++_w_rpt; @@ -209,7 +211,9 @@ class StreamTimingTx { .f("final_rate", r.final_rate) .f("tq_us", rec->t_queue_us) .f("c2s_us", rec->c2s_us) - .f("age_us", (unsigned long long)((now - rec->send_ns) / 1000)) + /* send_packet -> the host decoding the report, stamped in the + * sink, not when this thread got round to draining it. */ + .f("age_us", (unsigned long long)((qr.rx_ns - rec->send_ns) / 1000)) .f("marker", rec->marker ? 1 : 0); } } @@ -289,11 +293,24 @@ class StreamTimingTx { .f("q_max_raw", _w_q_max) .f("retries_max", _w_retries_max) .f("rpt_unmatched", (unsigned long long)_join.unmatched()) - .f("rpt_overflow", (unsigned long long)_rpt_overflow); + .f("rpt_overwritten", (unsigned long long)_join.overwritten()) + .f("rpt_overflow", (unsigned long long)_rpt_overflow.load(std::memory_order_relaxed)); _w_rpt = _w_rpt_fail = 0; _w_q_max = 0; _w_retries_max = 0; _w_q.clear(); return true; } + // A frame this helper does not build is about to go out on the same device + // (streamtx's hop sync marker): record the tag it will carry so its report + // joins as a marker instead of counting as unmatched. + void note_external_send() { + if (!_join_on) return; + if (auto tag = _rtl->NextTxReportTag()) { + devourer::stream_timing::TxFrameRec rec; + rec.frame = _frames; rec.send_ns = now_ns(); rec.marker = true; + _join.sent(*tag, rec); + } + } + // A producer capture stamp arrived (kCtlCaptureTs): remembered for the next // data record. A second stamp before a record replaces the first (counted). void capture_stamp(uint64_t ns) { @@ -328,9 +345,10 @@ class StreamTimingTx { bool _started = false, _data_sent_ok = false, _warned_unsupported = false; IRtlRadio *_rtl = nullptr; bool _join_on = false; + struct QueuedReport { devourer::TxReport r; uint64_t rx_ns; }; std::mutex _rpt_mu; - std::deque _rpt_q; - uint64_t _rpt_overflow = 0; + std::deque _rpt_q; + std::atomic _rpt_overflow{0}; devourer::stream_timing::TxReportJoin _join; std::vector _w_q; uint32_t _w_rpt = 0, _w_rpt_fail = 0, _w_q_max = 0, _w_retries_max = 0; diff --git a/examples/streamtx/main.cpp b/examples/streamtx/main.cpp index 5d9112db..7a0f0673 100644 --- a/examples/streamtx/main.cpp +++ b/examples/streamtx/main.cpp @@ -485,6 +485,7 @@ int main(int argc, char **argv) { auto wire = devourer::HopSyncMarker::encode(marker); sync_buf.insert(sync_buf.end(), wire.begin(), wire.end()); } + timing.note_external_send(); rtlDevice->send_packet(sync_buf.data(), sync_buf.size()); } } diff --git a/src/IRadio.h b/src/IRadio.h index f386e6a2..acb9818f 100644 --- a/src/IRadio.h +++ b/src/IRadio.h @@ -14,6 +14,7 @@ #include "SelectedChannel.h" #include "ThermalStatus.h" #include "TxReport.h" +#include #include #include "Sounding.h" #include "TriggerTwt.h" @@ -670,8 +671,12 @@ class IRadio { * is the tag IRtlRadio::NextTxReportTag() gave the frame, which is how a * caller joins a report to the frame it sent. An empty function clears it. */ void SetTxReportSink(std::function sink) { - std::lock_guard lk(_tx_report_sink_mu); + std::unique_lock lk(_tx_report_sink_mu); _tx_report_sink = std::move(sink); + /* Returns only once no earlier sink is still executing, so a caller may + * destroy what its sink captured right after. Never call this from + * inside the sink itself. */ + _tx_report_sink_cv.wait(lk, [this] { return _tx_report_inflight == 0; }); } /* TX submission health snapshot (see TxStats.h) — the driver-side drop / @@ -784,13 +789,22 @@ class IRadio { std::function sink; { std::lock_guard lk(_tx_report_sink_mu); + if (!_tx_report_sink) return; sink = _tx_report_sink; + ++_tx_report_inflight; } - if (sink) sink(r); + sink(r); + { + std::lock_guard lk(_tx_report_sink_mu); + --_tx_report_inflight; + } + _tx_report_sink_cv.notify_all(); } private: std::mutex _tx_report_sink_mu; + std::condition_variable _tx_report_sink_cv; + int _tx_report_inflight = 0; std::function _tx_report_sink; }; diff --git a/src/StreamTelemetry.h b/src/StreamTelemetry.h index d1408113..6a7d8ceb 100644 --- a/src/StreamTelemetry.h +++ b/src/StreamTelemetry.h @@ -368,7 +368,13 @@ class TimingWindow { * send, consumed by the report that echoes its tag, and a report whose slot * is empty or already consumed (a tag that wrapped past 256 unreported * frames, or a report for a frame sent before the join was armed) counts as - * unmatched. Pure; the TX thread owns it. */ + * unmatched. An eight-bit tag cannot name its generation: when a send + * reuses a tag whose slot is still live (its report has not come back after + * 256 further sends), the slot is overwritten and counted in overwritten(); + * a late report for the old frame then joins the NEW frame — the ring + * cannot tell them apart. A non-zero overwritten() therefore means that + * window's joins are suspect (the startup burst does this; a paced steady + * state does not). Pure; the TX thread owns it. */ struct TxFrameRec { uint64_t frame = 0; /* the transmitter's frame counter */ uint64_t send_ns = 0; /* host clock at send_packet */ @@ -378,6 +384,7 @@ struct TxFrameRec { class TxReportJoin { public: void sent(uint8_t tag, const TxFrameRec &rec) { + if (_live[tag]) ++_overwritten; _slot[tag] = rec; _live[tag] = true; ++_sent; @@ -392,11 +399,12 @@ class TxReportJoin { uint64_t sent_count() const { return _sent; } uint64_t joined() const { return _joined; } uint64_t unmatched() const { return _unmatched; } + uint64_t overwritten() const { return _overwritten; } private: std::array _slot{}; std::array _live{}; - uint64_t _sent = 0, _joined = 0, _unmatched = 0; + uint64_t _sent = 0, _joined = 0, _unmatched = 0, _overwritten = 0; }; } // namespace stream_timing diff --git a/tests/stream_telemetry_selftest.cpp b/tests/stream_telemetry_selftest.cpp index 2013095c..b806ffaa 100644 --- a/tests/stream_telemetry_selftest.cpp +++ b/tests/stream_telemetry_selftest.cpp @@ -173,8 +173,13 @@ int main() { CHECK(r && r->frame == 7 && r->t_queue_us == 30 && j.joined() == 1, "report joins its frame"); CHECK(!j.match(5) && j.unmatched() == 2, "a second report for the same tag is unmatched"); for (unsigned i = 0; i < 300; ++i) { TxFrameRec x; x.frame = 100 + i; j.sent(uint8_t(i), x); } - r = j.match(10); // tag 10 was written at i=10 and again at i=266 - CHECK(r && r->frame == 100 + 266, "a wrapped tag yields the latest frame with it"); + // Tag 10 was written at i=10 and again at i=266 while still live: the + // ring cannot tell an old report from the new one — a late report for + // frame 110 joins frame 366, and the overwrite is counted, not hidden. + CHECK(j.overwritten() == 300 - 256, "live-slot reuse is counted"); + r = j.match(10); + CHECK(r && r->frame == 100 + 266, "a late report after reuse joins the NEW frame (documented ambiguity)"); + CHECK(!j.match(10), "and the new frame's own report is then unmatched"); CHECK(j.sent_count() == 301, "sent count"); } diff --git a/tests/stream_timing_onair.sh b/tests/stream_timing_onair.sh index 76d164c5..16800cb1 100755 --- a/tests/stream_timing_onair.sh +++ b/tests/stream_timing_onair.sh @@ -22,8 +22,9 @@ # every report must join the frame it describes — rpt_n ≈ frames in # each marker window, few unmatched — so the on-chip queue time and # retry count enter stream.timing / stream.txrpt. Jaguar3 drains -# C2H on its coex thread; a Jaguar2 transmitter needs an RX loop -# (run the duplex phase with TX_REPORT=1 there). +# C2H on its coex thread, so this phase is for a Jaguar3 TX; a +# Jaguar2 transmitter has no C2H without an RX loop (streamtx never +# runs one) — cover it with the duplex phase and TX_REPORT=1. # 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). @@ -250,6 +251,9 @@ w = [e for e in iter_events(open(sys.argv[1]), ev="stream.timing")] # firmware's CCX emission ceiling drops reports (docs/scheduled-mac.md). Judge # the steady state: the second half of the run's windows. w = [e for e in w if e.get("rpt_join") == 1] +if not w: + print(f"FAIL: {sys.argv[2]}: no join windows — reports need C2H to flow (Jaguar3 coex thread, " + f"or an RX loop: the duplex phase on a Jaguar2 transmitter)"); sys.exit(1) w = w[len(w) // 2:] frames = sum(e["frames"] for e in w); rpt = sum(e["rpt_n"] for e in w) fail = sum(e["rpt_fail"] for e in w) @@ -260,9 +264,10 @@ q50 = sorted(e["q_p50_raw"] for e in w)[len(w) // 2] if w else 0 ratio = rpt / frames if frames else 0 # The firmware drops some reports (tag gaps; docs/scheduled-mac.md): 0.8 is # the floor a healthy join clears, the ratio itself is the number to read. -ok = frames >= 500 and ratio >= 0.8 and unm <= 0.02 * max(rpt, 1) and ovf == 0 +ovw = (w[-1]["rpt_overwritten"] - w[0]["rpt_overwritten"]) if w else 0 +ok = frames >= 500 and ratio >= 0.8 and unm <= 0.02 * max(rpt, 1) and ovf == 0 and ovw == 0 print(f"{'PASS' if ok else 'FAIL'}: {sys.argv[2]}: joined {rpt}/{frames} ({ratio:.3f}) unmatched={unm} " - f"overflow={ovf} fail={fail} q_p50_raw={q50} q_max_raw={qmax} windows={len(w)}") + f"overwritten={ovw} overflow={ovf} fail={fail} q_p50_raw={q50} q_max_raw={qmax} windows={len(w)}") sys.exit(0 if ok else 1) PY } @@ -319,7 +324,7 @@ if [[ " $PHASES " == *" duplex "* ]]; then sudo pkill -INT -x duplex 2>/dev/null || true; sleep 1 slice duplex "$off" duplex stop_witness - if [ -n "$TX_REPORT" ]; then join_verdict "$OUT/tx_duplex.out" "duplex report" || fail "duplex: join"; fi + if [ "${TX_REPORT:-0}" -gt 0 ] 2>/dev/null; then join_verdict "$OUT/tx_duplex.out" "duplex report" || fail "duplex: join"; fi if $PY tests/stream_timing_analyze.py "$OUT/run_duplex.jsonl" >"$OUT/sum_duplex.json"; then pass "duplex: $(grep -o '"lat_frames": [0-9]*\|"lat_p50_us": [0-9]*\|"c2a_p50_us": [0-9]*\|"markers": [0-9]*' "$OUT/sum_duplex.json" | tr '\n' ' ')" else