spsc-fat: backpressure-preserving pool-exhaustion policy (default) — keep the hardware-ARQ contract under consumer stalls

CLAUDE.md

@@ -425,11 +425,14 @@ a hardware-ARQ loop (autonomous MAC retransmission until ACK). The ACK's
 horizon is chip-FIFO **admission** (bench: 8812EU responder,
 `tests/arq_e2e_delivery.sh` per-frame ledgers): receiver-side congestion
 upstream of admission declines the ACK, so the loop sees and retries it — but
-a host stage that drops on a full queue while keeping URBs armed (the
-spsc-fat ring at pool exhaustion — counted as `pool_dropped` in `rx.ring`)
-turns the same loss into ACKed-but-undelivered that the TX peer logs as
-delivered and never retries. Under ARQ or delivery accounting, prefer
-backpressure to the chip over drop-on-full host buffering. Per-frame TX
+a host stage that drops on a full queue while keeping URBs armed turns the
+same loss into ACKed-but-undelivered that the TX peer logs as delivered and
+never retries. The spsc-fat ring's `DEVOURER_RX_POOL_EXHAUST` policy is that
+choice (`src/DeviceConfig.h`): `backpressure` (default) parks exhausted URBs
+so the chip declines further ACKs and the loss stays ARQ-visible
+(`pool_stalls` in `rx.ring`; bench: 0 ACKed-but-undelivered under stalls that
+lose 5.5k+ frames on the `drop` policy, which remains opt-in and counted as
+`pool_dropped`). Per-frame TX
 outcomes surface as `tx.report` events (CCX via C2H) — the TX-side link
 sensor; C2H rides the RX path, so J1/J2 TX-only sessions see none (run
 `DEVOURER_TX_WITH_RX=thread`; J3's coex thread drains C2H regardless).

docs/logging.md

@@ -91,7 +91,7 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets];
 | `rx.corrupt` | RX (`DEVOURER_RX_DUMP_ALL`) | len, crc, icv, rate, bw, stbc, ldpc, sgi, rssi[2], evm[2], snr[2] |
 | `rx.txhit` | RX, TX | hits, total_rx, len, seq, paggr, ppdu, rate, bw, stbc, ldpc, ppdu_type — canonical-SA (57:42:75:05:d6:00) matcher; rate/ldpc prove what encoding was decoded (8814A reports ldpc=0 always — no HW indicator); ppdu_type is the AX RXD format nibble (7=HE_SU, 8=HE_ERSU; 255 on pre-AX chips) |
 | `rx.seq` | RX, duplex (`DEVOURER_RX_PCTR`) | pctr, tsfl, seq, crc, paggr, ppdu — the ground-truth per-frame delivery sequence for the RX-ring loss study: pctr is the u32 the txdemo QoS-Data path stamps at MPDU offset 26, so gaps in it are per-frame loss; paggr/ppdu carry the aggregate structure the host-vs-RF discriminator keys on. Lean by design (no body hex) so the emit can't perturb the pump thread. SA gate follows `DEVOURER_RX_AGG_SA` (required in duplex, whose canonical-SA `rx.frame` stream is a different transmitter), else canonical SA |
-| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"/"reorder-pool"/"spsc-fat"), n_urbs, armed (URBs posted to the HCD and awaiting a frame — the depth that starves under a slow inline consumer), min_armed (low-water mark since the last emit), cb_max_us (worst inline-consume latency in the window), resubmit_fail, completions (cumulative URB callbacks), empties (cumulative callbacks that left the ring with zero posted URBs), pool_free (−1 = no host pool), qdepth (spsc-fat consumer-queue depth; 0 in the other ring modes), pool_dropped (cumulative received frames discarded on the spsc-fat drop path — pool exhausted, or a failed re-arm, the latter also ticking resubmit_fail; each was already chip-ACKed, so a hardware-ARQ peer counts it delivered; `tests/arq_e2e_delivery.sh` measures it). Sync mode emits a reduced line (pool_free pinned at −1; no qdepth/pool_dropped/completions/empties). The mechanism-proof telemetry: empties/completions is the host-starvation rate — near-0 under RF loss (the ring stays armed because frames don't arrive), high under host starvation (frames out-race resubmit and drain the ring). Counted in the callback, so robust to the pump-thread starvation that makes the periodic emit sparse — but blind while the pump itself is frozen: a stalled consumer drops frames these counters never see, which the per-frame `rx.seq` ledger exists to catch |
+| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"/"reorder-pool"/"spsc-fat"), n_urbs, armed (URBs posted to the HCD and awaiting a frame — the depth that starves under a slow inline consumer), min_armed (low-water mark since the last emit), cb_max_us (worst inline-consume latency in the window), resubmit_fail, completions (cumulative URB callbacks), empties (cumulative callbacks that left the ring with zero posted URBs), pool_free (−1 = no host pool), qdepth (spsc-fat consumer-queue depth; 0 in the other ring modes), pool_dropped (cumulative received frames discarded at spsc-fat pool exhaustion under the `drop` policy — pool exhausted, or a failed re-arm, the latter also ticking resubmit_fail; each was already chip-ACKed, so a hardware-ARQ peer counts it delivered), pool_stalls (the `backpressure`-policy counterpart, cumulative URB park events: the payload still reaches the consumer, the ring shrinks and the chip declines further ACKs, so overload loss stays ARQ-visible — `DEVOURER_RX_POOL_EXHAUST`, default backpressure; `tests/arq_e2e_delivery.sh` measures both). Sync mode emits a reduced line (pool_free pinned at −1; no qdepth/pool_dropped/pool_stalls/completions/empties). The mechanism-proof telemetry: empties/completions is the host-starvation rate — near-0 under RF loss (the ring stays armed because frames don't arrive), high under host starvation (frames out-race resubmit and drain the ring). Counted in the callback, so robust to the pump-thread starvation that makes the periodic emit sparse — but blind while the pump itself is frozen: a stalled consumer drops frames these counters never see, which the per-frame `rx.seq` ledger exists to catch |
 | `rx.count` | TX (its RX thread) | total, len |
 | `rx.path` | RX (`DEVOURER_RX_ALLPATHS`) | seq, rssi[4], snr[4], evm[4] |
 | `rx.path_mask` | L (toggle spec) | t, mask "0xNN" |

examples/common/env_config.cpp

@@ -76,6 +76,12 @@ devourer::DeviceConfig devourer_config_from_env() {
     cfg.rx.rx_mode = parse_rx_mode(e);
   if (env_long("DEVOURER_RX_POOL_SPARE", &v))
     cfg.rx.pool_spare = static_cast<int>(v);
+  if (const char *e = env_str("DEVOURER_RX_POOL_EXHAUST")) {
+    if (str_ieq(e, "drop"))
+      cfg.rx.pool_exhaust = devourer::PoolExhaust::Drop;
+    else if (str_ieq(e, "backpressure") || str_ieq(e, "bp"))
+      cfg.rx.pool_exhaust = devourer::PoolExhaust::Backpressure;
+  }
   if (env_long("DEVOURER_RX_RING_MS", &v))
     cfg.rx.ring_ms = static_cast<int>(v);
   cfg.rx.phy_status_8821c = !env_flag("DEVOURER_8821C_NO_PHYST");

src/DeviceConfig.h

@@ -79,6 +79,22 @@ enum class RxMode : uint8_t {
   Decoupled,   /* naive worker-thread hand-off (known-bad A/B control) */
 };
 
+/* What the spsc-fat ring does when its buffer pool runs dry (consumer behind
+ * under sustained overload). The choice decides which side of the hardware-ARQ
+ * contract survives overload — the chip ACKs on FIFO admission, so anything
+ * the HOST discards afterwards is a loss the TX peer believes delivered. */
+enum class PoolExhaust : uint8_t {
+  Backpressure, /* park the URB unarmed: the chip FIFO fills and the chip
+                 * declines further ACKs — congestion loss stays ARQ-visible
+                 * (bench: 14,214/14,214 forced drops reported ok=0 and
+                 * retried). Ring re-arms as the consumer returns buffers. */
+  Drop,         /* keep the ring armed and discard the payload — the chip
+                 * already ACKed it, so the TX peer logs it delivered and
+                 * never retries (counted as rx.ring pool_dropped). Smoother
+                 * under overload, but only for consumers that accept silent
+                 * loss (no ARQ / delivery accounting on top). */
+};
+
 struct DeviceConfig {
   /* ---- RX ------------------------------------------------------------- */
   struct Rx {
@@ -125,6 +141,11 @@ struct DeviceConfig {
      * burst backlog host-side instead of overflowing the chip RX FIFO. 0 =
      * pool of exactly urbs buffers (no spare). Ignored by async/sync. */
     int pool_spare = 0;
+    /* env: DEVOURER_RX_POOL_EXHAUST — "backpressure" (default) | "drop":
+     * the SpscFat pool-exhaustion policy (see PoolExhaust above). Only read
+     * by SpscFat; the other modes never drop host-side (async/reorder degrade
+     * to inline consume, which backpressures the chip by construction). */
+    PoolExhaust pool_exhaust = PoolExhaust::Backpressure;
     /* env: DEVOURER_RX_RING_MS — cadence (ms) for the diagnostic rx.ring
      * telemetry event (armed-URB depth, min depth in the window, resubmit
      * failures, max inline-consume latency). Unset/0 = no telemetry (the

src/RtlAdapter.cpp

@@ -22,7 +22,8 @@ RtlAdapter::RtlAdapter(libusb_device_handle *dev_handle, Logger_t logger,
                        const devourer::DeviceConfig &cfg)
     : _transport{std::make_shared<devourer::UsbTransport>(
           dev_handle, logger, ctx, std::move(usb_lock), cfg.usb.rx_zerocopy,
-          cfg.rx.rx_mode, cfg.rx.pool_spare, cfg.rx.ring_ms.value_or(0))},
+          cfg.rx.rx_mode, cfg.rx.pool_spare, cfg.rx.ring_ms.value_or(0),
+          cfg.rx.pool_exhaust)},
       _logger{std::move(logger)} {
   init_from_transport(cfg);
 }

src/UsbTransport.cpp

@@ -89,15 +89,38 @@ struct AsyncRxShared {
    *
    * ARQ caveat, bench-measured (tests/arq_e2e_delivery.sh): keeping the ring
    * armed means the chip ADMITS AND ACKS every frame — so a frame dropped at
-   * pool exhaustion below is an ACKed-but-undelivered loss the hardware-ARQ
-   * peer will log as delivered and never retry. The default async ring loses
-   * the same frames chip-side instead, where the 8812EU declines the ACK and
-   * the ARQ loop recovers them (14,214/14,214 stall-window drops reported
-   * ok=0 there, vs 5,667 ok=1-but-lost here). Under an ARQ or
-   * delivery-accounting scheme, prefer backpressure over this mode, or watch
-   * pool_dropped. */
+   * pool exhaustion is an ACKed-but-undelivered loss the hardware-ARQ peer
+   * will log as delivered and never retry. The default async ring loses the
+   * same frames chip-side instead, where the 8812EU declines the ACK and the
+   * ARQ loop recovers them (14,214/14,214 stall-window drops reported ok=0
+   * there, vs 5,667 ok=1-but-lost under the drop policy here).
+   *
+   * PoolExhaust picks which contract survives overload: Backpressure (the
+   * default) PARKS the completed URB unarmed — the payload still reaches the
+   * consumer queue, the ring shrinks, the chip FIFO fills and the chip
+   * declines further ACKs, so overload loss stays ARQ-visible; parked URBs
+   * re-arm as the consumer returns buffers. Drop keeps the ring armed and
+   * discards the payload (counted in pool_dropped) — smoother, but only for
+   * consumers that accept silent loss. */
   bool spsc = false;
+  bool bp = true; /* PoolExhaust::Backpressure */
   std::atomic<unsigned long long> pool_dropped{0};
+  /* Backpressure-policy state, guarded by pool_mu with free_bufs: URBs parked
+   * bufferless at exhaustion, re-armed from the consumer's buffer-return path.
+   * parking_closed is teardown's gate — set (under pool_mu) before the cancel
+   * pass, so no late re-arm can launch an uncancellable infinite-timeout URB.
+   * pool_stalls counts park events (the backpressure counterpart of
+   * pool_dropped; nothing is lost host-side on this path). */
+  std::vector<libusb_transfer *> parked;
+  bool parking_closed = false;
+  std::atomic<unsigned long long> pool_stalls{0};
+  /* Device-gone latch. Plain async self-terminates on unplug (every URB
+   * error-retires, active hits 0, rx_loop returns) — parked URBs never
+   * complete, so backpressure mode needs an explicit signal or a dead device
+   * leaves the loop spinning until Stop(). Set from a NO_DEVICE transfer
+   * status or submit error; rx_loop exits on it and retires the parked set
+   * through the normal teardown path. */
+  std::atomic<bool> dead{false};
   std::mutex queue_mu;
   std::condition_variable queue_cv;
   std::deque<std::pair<uint8_t *, int>> queue;
@@ -122,6 +145,8 @@ inline void rx_consume(AsyncRxShared *s, const uint8_t *buf, int len) {
 
 extern "C" void LIBUSB_CALL devourer_rx_cb(libusb_transfer *t) {
   auto *s = static_cast<AsyncRxShared *>(t->user_data);
+  if (t->status == LIBUSB_TRANSFER_NO_DEVICE)
+    s->dead.store(true, std::memory_order_relaxed);
   /* This URB just completed — it has left the wire until resubmitted. */
   if (s->telemetry) {
     const int a = s->armed.fetch_sub(1, std::memory_order_relaxed) - 1;
@@ -219,13 +244,45 @@ extern "C" void LIBUSB_CALL devourer_rx_cb(libusb_transfer *t) {
         s->resubmit_fail.fetch_add(1, std::memory_order_relaxed);
     }
     /* Pool exhausted (consumer hopelessly behind under sustained overload) or
-     * submit failed: preserve the pump's never-block invariant by re-arming
-     * with the received buffer and DROPPING this frame — a bounded loss, vs the
-     * cascade an inline consume would trigger. The chip already ACKed this
-     * frame (see the mode comment above): count every received frame dropped
-     * here, exhaustion and failed-re-arm alike — both are post-ACK host drops,
-     * and the re-arm failure is separable because it also ticks resubmit_fail.
-     * `resubmit` gates the count: a teardown-window frame (stop requested) is
+     * submit failed. Two policies (see the mode comment above):
+     *
+     * Backpressure (default): the payload is NOT lost — hand it to the
+     * consumer queue and PARK this URB bufferless; the ring shrinks, the chip
+     * FIFO fills and the chip declines further ACKs, so the overload loss
+     * happens chip-side where the ARQ peer can see and retry it. The
+     * consumer's buffer-return path re-arms parked URBs. `active` is NOT
+     * decremented: a parked URB is pending, not retired (teardown retires the
+     * parked set explicitly before its cancel pass). */
+    if (s->bp && resubmit && rlen > 0) {
+      /* Queue the payload FIRST, park LAST. Once this URB is on the parked
+       * list the teardown path may retire it, finish, and destroy the shared
+       * state — so parked.push_back must be this callback's final shared-
+       * state access (the parked-path analogue of the retire paths' trailing
+       * `active--`, which teardown likewise blocks on). */
+      t->buffer = nullptr; /* payload buffer now belongs to the queue */
+      {
+        std::lock_guard<std::mutex> lk(s->queue_mu);
+        s->queue.emplace_back(received, rlen);
+      }
+      s->queue_cv.notify_one();
+      {
+        std::lock_guard<std::mutex> lk(s->pool_mu);
+        if (!s->parking_closed) {
+          s->pool_stalls.fetch_add(1, std::memory_order_relaxed);
+          s->parked.push_back(t);
+          return; /* lock releases on return; nothing may follow the park */
+        }
+      }
+      s->active--; /* parking closed = teardown: retire without resubmit */
+      return;
+    }
+    /* Drop policy: preserve the pump's never-block invariant by re-arming
+     * with the received buffer and DROPPING this frame — a bounded loss, vs
+     * the cascade an inline consume would trigger. The chip already ACKed
+     * this frame: count every received frame dropped here, exhaustion and
+     * failed-re-arm alike — both are post-ACK host drops, and the re-arm
+     * failure is separable because it also ticks resubmit_fail. `resubmit`
+     * gates the count: a teardown-window frame (stop requested) is
      * intentional loss, not an overload signal. */
     if (resubmit && rlen > 0)
       s->pool_dropped.fetch_add(1, std::memory_order_relaxed);
@@ -257,10 +314,11 @@ UsbTransport::UsbTransport(libusb_device_handle *dev_handle, Logger_t logger,
                            libusb_context *ctx,
                            std::shared_ptr<devourer::UsbDeviceLock> usb_lock,
                            bool rx_zerocopy, RxMode rx_mode, int pool_spare,
-                           int ring_ms)
+                           int ring_ms, PoolExhaust pool_exhaust)
     : _dev_handle{dev_handle}, _ctx{ctx}, _logger{std::move(logger)},
       _rx_zerocopy{rx_zerocopy}, _rx_mode{rx_mode}, _pool_spare{pool_spare},
-      _ring_ms{ring_ms}, _usb_lock{std::move(usb_lock)} {
+      _ring_ms{ring_ms}, _pool_exhaust{pool_exhaust},
+      _usb_lock{std::move(usb_lock)} {
   libusb_device_descriptor desc{};
   if (libusb_get_device_descriptor(libusb_get_device(_dev_handle), &desc) ==
       LIBUSB_SUCCESS) {
@@ -317,6 +375,7 @@ void UsbTransport::rx_loop(
   sh.telemetry = _ring_ms > 0;
   sh.reorder = reorder;
   sh.spsc = spsc;
+  sh.bp = _pool_exhaust == PoolExhaust::Backpressure;
   sh.buf_size = buf_size;
   /* reorder-pool and spsc-fat post n_urbs URBs but allocate pool_spare extra
    * buffers so a burst / stall backlog is absorbed in the host pool instead of
@@ -399,7 +458,37 @@ void UsbTransport::rx_loop(
           sh.queue.pop_front();
         }
         rx_consume(&sh, item.first, item.second);
+        /* Buffer return. Backpressure policy: a parked (bufferless) URB gets
+         * this buffer and goes straight back on the wire — the ring re-arms
+         * exactly as fast as the consumer frees capacity. The submit stays
+         * under pool_mu so teardown's parking_closed flip (also under
+         * pool_mu) strictly precedes-or-follows any re-arm: no late launch
+         * after the cancel pass. (libusb_submit_transfer is thread-safe and
+         * the event pump never holds pool_mu while inside libusb, so there is
+         * no lock-order inversion.) */
         std::lock_guard<std::mutex> lk(sh.pool_mu);
+        if (sh.bp && !sh.parking_closed && !sh.parked.empty()) {
+          libusb_transfer *t = sh.parked.back();
+          sh.parked.pop_back();
+          t->buffer = item.first;
+          t->length = sh.buf_size;
+          const int rc = libusb_submit_transfer(t);
+          if (rc == 0) {
+            if (sh.telemetry)
+              sh.armed.fetch_add(1, std::memory_order_relaxed);
+            continue;
+          }
+          /* Submit failure: re-park the URB, pool the buffer. Device gone
+           * latches `dead` so rx_loop exits and retires the parked set —
+           * re-park-forever would otherwise pin `active` above zero with no
+           * completion ever coming. */
+          t->buffer = nullptr;
+          sh.parked.push_back(t);
+          if (rc == LIBUSB_ERROR_NO_DEVICE)
+            sh.dead.store(true, std::memory_order_relaxed);
+          if (sh.telemetry)
+            sh.resubmit_fail.fetch_add(1, std::memory_order_relaxed);
+        }
         sh.free_bufs.push_back(item.first);
       }
     });
@@ -415,7 +504,8 @@ void UsbTransport::rx_loop(
     sh.min_armed.store(a0, std::memory_order_relaxed);
   }
   auto last_ring = std::chrono::steady_clock::now();
-  while (!should_stop() && sh.active > 0) {
+  while (!should_stop() && sh.active > 0 &&
+         !sh.dead.load(std::memory_order_relaxed)) {
     struct timeval tv {0, 100000};
     libusb_handle_events_timeout_completed(_ctx, &tv, nullptr);
     if (sh.telemetry) {
@@ -452,10 +542,23 @@ void UsbTransport::rx_loop(
             .f("pool_free", pool_free)
             .f("qdepth", qdepth)
             .f("pool_dropped", (unsigned long long)sh.pool_dropped.load(
-                                   std::memory_order_relaxed));
+                                   std::memory_order_relaxed))
+            .f("pool_stalls", (unsigned long long)sh.pool_stalls.load(
+                                  std::memory_order_relaxed));
       }
     }
   }
+  /* Close parking BEFORE the cancel pass (under pool_mu, so no consumer
+   * re-arm is mid-submit), then retire the parked URBs: they are not in
+   * flight, cancel would be a no-op on them, and the active-drain below would
+   * otherwise wait forever for completions that can never come. The transfer
+   * objects stay in `xfers` and are freed centrally. */
+  {
+    std::lock_guard<std::mutex> lk(sh.pool_mu);
+    sh.parking_closed = true;
+    sh.active -= static_cast<int>(sh.parked.size());
+    sh.parked.clear();
+  }
   for (auto *t : xfers)
     libusb_cancel_transfer(t);
   while (sh.active > 0) {