stream timing: join the CCX TX report to its frame on HalMAC (queue time, retries)

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_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 |

docs/stream-timing.md

@@ -191,6 +191,64 @@ 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 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.
+
+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.
+
 ## Reading it
 
 `rxdemo` with `DEVOURER_STREAM_OUT=1`: every `rx.frame` carries `fc0` (so a

examples/common/stream_timing_tx.h

@@ -18,6 +18,8 @@
 // gets no absolute latency and says so.
 #pragma once
 
+#include <algorithm>
+#include <atomic>
 #include <chrono>
 #include <cstdint>
 #include <cstdlib>
@@ -28,6 +30,10 @@
 #include "AdapterCaps.h"
 #include "Event.h"
 #include "IRadio.h"
+#include "IRtlRadio.h"
+#include "TxReport.h"
+#include <deque>
+#include <mutex>
 #include "StreamTelemetry.h"
 #include "TxStats.h"
 #include "host_tsf_fit.h"
@@ -83,8 +89,29 @@ 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<IRtlRadio *>(&_dev);
+    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<std::mutex> lk(_rpt_mu);
+        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)");
+    }
   }
   void stop() {
+    if (_join_on) {
+      _dev.SetTxReportSink({});
+      _join_on = false;
+    }
     if (_beacon) {
       _dev.StopBeacon();
       _beacon = false;
@@ -128,6 +155,16 @@ class StreamTimingTx {
       t.tsf10_lo = static_cast<uint16_t>((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<uint32_t>(_t0 > read_ns ? (_t0 - read_ns) / 1000 : 0);
+        rec.c2s_us = static_cast<uint32_t>(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 +179,43 @@ 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<QueuedReport> q;
+    {
+      std::lock_guard<std::mutex> lk(_rpt_mu);
+      q.swap(_rpt_q);
+    }
+    for (const auto &qr : q) {
+      const devourer::TxReport &r = qr.r;
+      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)
+            /* 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);
+    }
   }
 
   // Before each data frame: air the marker when due. `radiotap` is the stream
@@ -174,7 +248,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<std::ptrdiff_t>(_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,10 +282,35 @@ 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_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) {
@@ -230,6 +343,15 @@ 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;
+  struct QueuedReport { devourer::TxReport r; uint64_t rx_ns; };
+  std::mutex _rpt_mu;
+  std::deque<QueuedReport> _rpt_q;
+  std::atomic<uint64_t> _rpt_overflow{0};
+  devourer::stream_timing::TxReportJoin _join;
+  std::vector<uint32_t> _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<HostTsfFit> _fit;
   devourer::stream_timing::TimingWindow _window;

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());
       }
     }

src/IRadio.h

@@ -13,6 +13,9 @@
 #include "RxQuality.h"
 #include "SelectedChannel.h"
 #include "ThermalStatus.h"
+#include "TxReport.h"
+#include <condition_variable>
+#include <mutex>
 #include "Sounding.h"
 #include "TriggerTwt.h"
 #include "TxCaps.h"
@@ -660,6 +663,22 @@ 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<void(const devourer::TxReport &)> sink) {
+    std::unique_lock<std::mutex> 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 /
    * 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 +780,33 @@ 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<void(const devourer::TxReport &)> sink;
+    {
+      std::lock_guard<std::mutex> lk(_tx_report_sink_mu);
+      if (!_tx_report_sink) return;
+      sink = _tx_report_sink;
+      ++_tx_report_inflight;
+    }
+    sink(r);
+    {
+      std::lock_guard<std::mutex> 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<void(const devourer::TxReport &)> _tx_report_sink;
+
 };
 
 #endif /* IRADIO_H */

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<uint8_t> 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