From 0727de6f7fb59e27b8ca2b0f58e36a1a4690aba0 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 20:09:30 +0300 Subject: [PATCH 1/3] =?UTF-8?q?spsc-fat:=20backpressure-preserving=20pool-?= =?UTF-8?q?exhaustion=20policy=20(default)=20=E2=80=94=20keep=20the=20hard?= =?UTF-8?q?ware-ARQ=20contract=20under=20consumer=20stalls?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The spsc-fat ring's drop-at-exhaustion converts receiver congestion into ACKed-but-undelivered loss the TX peer logs as delivered (bench: 5,468-5,739 per escalation run). The new PoolExhaust policy decides which contract survives overload: - backpressure (default): at exhaustion the completed URB PARKS bufferless — its payload still reaches the consumer queue, the ring shrinks, the chip FIFO fills and the chip declines further ACKs, so overload loss happens chip-side where the ARQ loop sees and retries it. The consumer's buffer-return path re-arms parked URBs, with the submit held under pool_mu so teardown's parking_closed flip strictly excludes a late launch of an uncancellable infinite-timeout URB; teardown retires the parked set before its cancel pass (they are not in flight and would hang the active-drain). Parked URBs stay counted in `active` — the pump loop must keep running while the whole ring is parked. - drop: the previous behaviour, kept for latency-over-integrity consumers, still counted as pool_dropped. Config: DeviceConfig::Rx::pool_exhaust / DEVOURER_RX_POOL_EXHAUST (backpressure|drop). Telemetry: rx.ring gains pool_stalls (cumulative park events — nothing is lost host-side on that path); docs/logging.md schema row updated. Harness: DUT_POOL_EXHAUST passthrough in arq_e2e_delivery.sh. Co-Authored-By: Claude Opus 4.8 --- docs/logging.md | 2 +- examples/common/env_config.cpp | 6 ++ src/DeviceConfig.h | 21 ++++++ src/RtlAdapter.cpp | 3 +- src/UsbTransport.cpp | 120 ++++++++++++++++++++++++++++----- src/UsbTransport.h | 4 +- tests/arq_e2e_delivery.sh | 2 + 7 files changed, 138 insertions(+), 20 deletions(-) diff --git a/docs/logging.md b/docs/logging.md index a6d9bf8..eef6df2 100644 --- a/docs/logging.md +++ b/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" | diff --git a/examples/common/env_config.cpp b/examples/common/env_config.cpp index 1331bc7..048741a 100644 --- a/examples/common/env_config.cpp +++ b/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(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(v); cfg.rx.phy_status_8821c = !env_flag("DEVOURER_8821C_NO_PHYST"); diff --git a/src/DeviceConfig.h b/src/DeviceConfig.h index b96b779..caa46c8 100644 --- a/src/DeviceConfig.h +++ b/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 diff --git a/src/RtlAdapter.cpp b/src/RtlAdapter.cpp index 88b40fc..4640a98 100644 --- a/src/RtlAdapter.cpp +++ b/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( 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); } diff --git a/src/UsbTransport.cpp b/src/UsbTransport.cpp index 4111664..4dca08a 100644 --- a/src/UsbTransport.cpp +++ b/src/UsbTransport.cpp @@ -89,15 +89,31 @@ 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 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 parked; + bool parking_closed = false; + std::atomic pool_stalls{0}; std::mutex queue_mu; std::condition_variable queue_cv; std::deque> queue; @@ -219,13 +235,44 @@ 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) { + bool parked_ok = false; + { + std::lock_guard lk(s->pool_mu); + if (!s->parking_closed) { + t->buffer = nullptr; /* payload buffer now belongs to the queue */ + s->parked.push_back(t); + parked_ok = true; + } + } + { + std::lock_guard lk(s->queue_mu); + s->queue.emplace_back(received, rlen); + } + s->queue_cv.notify_one(); + if (parked_ok) { + s->pool_stalls.fetch_add(1, std::memory_order_relaxed); + return; + } + 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 +304,11 @@ UsbTransport::UsbTransport(libusb_device_handle *dev_handle, Logger_t logger, libusb_context *ctx, std::shared_ptr 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 +365,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 +448,31 @@ 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 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; + if (libusb_submit_transfer(t) == 0) { + if (sh.telemetry) + sh.armed.fetch_add(1, std::memory_order_relaxed); + continue; + } + /* Rare submit failure: re-park the URB, pool the buffer. */ + t->buffer = nullptr; + sh.parked.push_back(t); + if (sh.telemetry) + sh.resubmit_fail.fetch_add(1, std::memory_order_relaxed); + } sh.free_bufs.push_back(item.first); } }); @@ -452,10 +525,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 lk(sh.pool_mu); + sh.parking_closed = true; + sh.active -= static_cast(sh.parked.size()); + sh.parked.clear(); + } for (auto *t : xfers) libusb_cancel_transfer(t); while (sh.active > 0) { diff --git a/src/UsbTransport.h b/src/UsbTransport.h index ec7676b..bbf62ce 100644 --- a/src/UsbTransport.h +++ b/src/UsbTransport.h @@ -35,7 +35,8 @@ class UsbTransport final : public IRtlTransport { libusb_context *ctx = nullptr, std::shared_ptr usb_lock = nullptr, bool rx_zerocopy = true, RxMode rx_mode = RxMode::Async, - int pool_spare = 0, int ring_ms = 0); + int pool_spare = 0, int ring_ms = 0, + PoolExhaust pool_exhaust = PoolExhaust::Backpressure); ~UsbTransport() override; bool is_usb() const override { return true; } @@ -143,6 +144,7 @@ class UsbTransport final : public IRtlTransport { RxMode _rx_mode = RxMode::Async; int _pool_spare = 0; int _ring_ms = 0; + PoolExhaust _pool_exhaust = PoolExhaust::Backpressure; /* rx_loop helpers for the servicing strategies dispatched off _rx_mode. */ void rx_loop_sync(int buf_size, diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh index 00e2917..068d65b 100644 --- a/tests/arq_e2e_delivery.sh +++ b/tests/arq_e2e_delivery.sh @@ -58,6 +58,7 @@ WARMUP_S=${WARMUP_S:-12} # feeder idle lead-in (duplex bring-up) PREFLIGHT_S=${PREFLIGHT_S:-14} # liveness check delay after drone start DUT_RX_MODE=${DUT_RX_MODE:-} # empty = default async ring DUT_POOL_SPARE=${DUT_POOL_SPARE:-16} # only read by the pool modes +DUT_POOL_EXHAUST=${DUT_POOL_EXHAUST:-} # spsc-fat: backpressure (default) | drop DUT_SPIN_US=${DUT_SPIN_US:-0} # per-frame inline consumer cost model DUT_STALL_MS=${DUT_STALL_MS:-0} # periodic consumer hiccup (GC pause) DUT_STALL_EVERY=${DUT_STALL_EVERY:-1500} @@ -139,6 +140,7 @@ env DEVOURER_VID="$DUT_VID" DEVOURER_PID="$DUT_PID" DEVOURER_CHANNEL="$CH" \ DEVOURER_RX_PCTR=1 DEVOURER_RX_AGG_SA="$TX_SA" DEVOURER_RX_RING_MS=200 \ ${DUT_RX_MODE:+DEVOURER_RX_MODE="$DUT_RX_MODE"} \ ${DUT_RX_MODE:+DEVOURER_RX_POOL_SPARE="$DUT_POOL_SPARE"} \ + ${DUT_POOL_EXHAUST:+DEVOURER_RX_POOL_EXHAUST="$DUT_POOL_EXHAUST"} \ DEVOURER_RX_SINK_SPIN_US="$DUT_SPIN_US" \ DEVOURER_RX_SINK_STALL_MS="$DUT_STALL_MS" \ DEVOURER_RX_SINK_STALL_EVERY="$DUT_STALL_EVERY" \ From 066d3aa17296f0112e7370983dc5161546ed53ff Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 20:26:55 +0300 Subject: [PATCH 2/3] CLAUDE.md: ACK-horizon paragraph reflects the pool-exhaustion policy Co-Authored-By: Claude Opus 4.8 --- CLAUDE.md | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7575a2e..17944c4 100644 --- a/CLAUDE.md +++ b/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). From 630ed492ed9f28c3cb309a1bc15484e0c018d492 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 20:44:42 +0300 Subject: [PATCH 3/3] spsc-fat backpressure: park last in the callback; latch device-gone for parked URBs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round on #366, both real: - The park path queued the payload and ticked counters AFTER inserting the URB into the parked list — but the moment it is parked, teardown may retire it, finish, and destroy the stack-owned shared state, leaving the callback epilogue (queue mutex, cv, counters) racing a destroyed object when it runs on a co-running TX event pump. The park insertion is now the callback's final shared-state access, mirroring the retire paths' trailing active-- that teardown already blocks on. - Plain async self-terminates on unplug (every URB error-retires, active hits 0, rx_loop returns); parked URBs never complete, so a dead device left the loop spinning until Stop(). A `dead` latch — set from a NO_DEVICE transfer status or submit error — now exits the loop, and the parked set retires through the normal teardown path. Validated: 48/48 ctest; fourth consecutive backpressure escalation run with acked_undelivered=0 (pool_stalls 1,445, pool_dropped 0); and an AddressSanitizer-instrumented escalation run (DEVOURER_SANITIZE=address, same stall/park/teardown load) with zero sanitizer reports across DUT, drone-sim and witness. Co-Authored-By: Claude Opus 4.8 --- src/UsbTransport.cpp | 47 ++++++++++++++++++++++++++++++-------------- 1 file changed, 32 insertions(+), 15 deletions(-) diff --git a/src/UsbTransport.cpp b/src/UsbTransport.cpp index 4dca08a..f9bc6ca 100644 --- a/src/UsbTransport.cpp +++ b/src/UsbTransport.cpp @@ -114,6 +114,13 @@ struct AsyncRxShared { std::vector parked; bool parking_closed = false; std::atomic 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 dead{false}; std::mutex queue_mu; std::condition_variable queue_cv; std::deque> queue; @@ -138,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(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; @@ -245,23 +254,24 @@ extern "C" void LIBUSB_CALL devourer_rx_cb(libusb_transfer *t) { * decremented: a parked URB is pending, not retired (teardown retires the * parked set explicitly before its cancel pass). */ if (s->bp && resubmit && rlen > 0) { - bool parked_ok = false; - { - std::lock_guard lk(s->pool_mu); - if (!s->parking_closed) { - t->buffer = nullptr; /* payload buffer now belongs to the queue */ - s->parked.push_back(t); - parked_ok = true; - } - } + /* 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 lk(s->queue_mu); s->queue.emplace_back(received, rlen); } s->queue_cv.notify_one(); - if (parked_ok) { - s->pool_stalls.fetch_add(1, std::memory_order_relaxed); - return; + { + std::lock_guard 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; @@ -462,14 +472,20 @@ void UsbTransport::rx_loop( sh.parked.pop_back(); t->buffer = item.first; t->length = sh.buf_size; - if (libusb_submit_transfer(t) == 0) { + const int rc = libusb_submit_transfer(t); + if (rc == 0) { if (sh.telemetry) sh.armed.fetch_add(1, std::memory_order_relaxed); continue; } - /* Rare submit failure: re-park the URB, pool the buffer. */ + /* 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); } @@ -488,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) {