diff --git a/CMakeLists.txt b/CMakeLists.txt index 636d896..851859b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -193,6 +193,12 @@ endif() # Link devourer with libusb as found via pkg-config. target_link_libraries(devourer PUBLIC PkgConfig::libusb) +# Android (Termux / NDK): the diagnostics plane routes to logcat via +# __android_log_write (src/logger.h), which lives in the system `log` library. +if(ANDROID) + target_link_libraries(devourer PUBLIC log) +endif() + target_include_directories(devourer PUBLIC hal) target_include_directories(devourer PUBLIC hal/phydm/rtl8814a) target_include_directories(devourer PUBLIC hal/phydm/rtl8822c) diff --git a/docs/experiments/issue-330-rx-ring-starvation.md b/docs/experiments/issue-330-rx-ring-starvation.md new file mode 100644 index 0000000..9d43bc5 --- /dev/null +++ b/docs/experiments/issue-330-rx-ring-starvation.md @@ -0,0 +1,159 @@ +# Async RX-ring residual loss on a constrained host (issue #330) + +Instrumentation, reproduction, and candidate fixes for the ~1–2 % residual RX +loss a MediaTek phone (RTL8812AU @ USB2) shows on the async multi-URB RX ring — +the PixelPilot#6 follow-up. The prior investigation established the mechanism was +*possible* but could not reproduce it at realistic load on a fast x86 rig. This +campaign built proper mechanism telemetry, reproduced the starvation on a host we +drive, implemented three ring-servicing fixes, and root-caused why the bench +cannot resolve the end-to-end delivery benefit. + +## The mechanism + +`UsbTransport::rx_loop` posts N bulk-IN URBs; `devourer_rx_cb` runs the consumer +(`on_data` → `FrameParser` → the caller's sink) **inline on the libusb pump +thread, before resubmitting the URB**. If the consumer stalls or the pump thread +is preempted, no URB is re-armed, the posted-URB depth collapses, and the chip +RX FIFO overflows — dropping frames. `FrameParser` returns `Packet.Data` as a +*view into the URB's DMA buffer* across every aggregated subframe, so any fix +must not reuse that buffer until the consumer is done with it. + +## Instrumentation (all off by default; the async path is unchanged when unset) + +| Knob / event | What | +|---|---| +| `rx.ring` (`DEVOURER_RX_RING_MS=N`) | periodic ring telemetry: `armed` (posted-URB depth), `min_armed`, `cb_max_us` (worst inline consume), `resubmit_fail`, `completions`/`empties`, `pool_free`, `qdepth` | +| `DEVOURER_RX_MODE` | `async` (default) / `sync` (blocking single-buffer reference) / `reorder-pool` / `spsc-fat` / `decoupled` (unimpl.) | +| `DEVOURER_RX_POOL_SPARE=K` | spare buffers for the pool modes | +| `DEVOURER_RX_SINK_SPIN_US` | busy-spin per delivered frame (models the wfb consumer cost) | +| `DEVOURER_RX_SINK_STALL_MS` / `_EVERY` | a periodic consumer hiccup (GC/scheduler pause) | +| `rx.seq` (`DEVOURER_RX_PCTR`) | lean per-frame event: the txdemo payload counter + `tsfl` + crc — ground-truth delivery | +| txdemo `DEVOURER_TX_BURST_ON_MS` / `_OFF_MS` | keyframe-burst emulation (reuses the QoS-Data u32 counter stamp) | + +Harness `tests/rxq_starve.sh` (one run per invocation: TX flood + constrained RX +capture, with `taskset`/`nice`/`chrt` clamps, continuous `BUSY_THREADS`, and an +intermittent RT-priority preemptor `tests/rxq_preempt.c` modelling a VR +compositor). Analyzer `tests/rxq_analyze.py` computes delivery from the `pctr` +range (independent of `tx.stats`), a gap-width histogram, the ring telemetry, and +a differential across runs. + +## Discriminating host-starvation from RF loss + +- The `armed` counter decrements at callback *entry*, not physical URB + completion, so with a single pump thread it rarely reaches 0 — a weak depth + proxy. `cb_max_us` (ms-scale under a stalled consumer vs ~100 µs under RF loss) + is the robust single-run signal. +- Catastrophic RF loss *also* produces wide multi-frame gaps, so gap width alone + false-positives. **The definitive method is differential**: hold the RF link + constant, vary the host lever, and attribute the delivery drop to the host. + +## Reproduction (host we drive) + +Desktop RTL8812AU (`0bda:8812`, USB2 — the reporter's exact chip), 2.4 GHz, RF +held constant: + +| condition | delivery | cb_max | +|---|---|---| +| urbs=2, spin=500 µs, no clamp | 70 % | 1.5 ms | +| urbs=2, spin=500 µs, `taskset -c 0` + 4 busy threads | **24 %** | **23 ms** | + +**+46 pts host-induced loss** at a modest consumer cost, purely by removing CPU +headroom — the lever the prior x86 attempts lacked. The pump thread is +descheduled ~23 ms behind the busy threads (≈ a VR compositor frame). Shrinking +the ring alone (urbs 1→8 at modest spin) did **not** reproduce it — the x86 host +re-arms even a depth-1 ring fast enough; you must also remove the headroom. + +## Candidate fixes + +- **reorder-pool** — swap a spare buffer onto the wire and resubmit *before* + consuming the received one (zerocopy-safe). Helps only when the pump is + **scheduled** but the consume blocks resubmit; the re-arm still runs on the + pump, so it does not help pump preemption or a throughput deficit. +- **spsc-fat** — the pump only reaps + re-arms + enqueues; a separate consumer + thread runs `on_data`. Keeps the ring armed through a **consumer** stall, + converting chip-FIFO overflow (dropped) into bounded host-queue backlog + (delayed) up to the pool depth. Under a periodic 100 ms stall its `qdepth` grew + to the pool cap and `pool_free` cycled fully — the queue demonstrably absorbed + the backlog the async ring drops. +- Neither fixes a genuine **throughput deficit** (consumer permanently too slow): + that needs more CPU or a faster consumer, not buffering. + +## Why the bench cannot resolve the ~1–2 % delivery benefit + +1. **Noise floor.** SNR pins at ~4–5 dB across every 2.4 GHz channel and TX-power + offset (RSSI swings −30…−46 dBm but SNR does not), USB3/switching + self-interference between the two bench adapters. 6M delivery ceilings ~65–75 % + — small host effects sit under the RF variance. 5 GHz does not close here. +2. **Large ring buffering.** 16 KB URBs (8 KB device-side aggregation on USB2) + hold tens of 256-byte frames each, so even a shallow urbs=2 ring buffers tens + of ms and absorbs typical transient stalls in *all* modes. Only extreme + continuous conditions exceed the floor, and those are throughput deficits no + ring change fixes. + +This confirms and root-causes the prior "rig unsuitable" finding. The fixes are +implemented and their mechanisms proven to engage (telemetry), but the end-to-end +delivery gain is measurable only on a clean-RF link (conducted / attenuators — +absent on this rig) or the actual constrained device. + +## The Meta Quest 3 (Snapdragon XR2 Gen 2, Android) + +A genuinely constrained mobile host, driven **fully headlessly from the desktop** +— no in-headset interaction. The vehicle is a minimal auto-grant APK +(`tests/quest_apk/`, hand-built with `aapt`/`d8`/`apksigner`) modelled on how +PixelPilot avoids the USB permission dialog: a `USB_DEVICE_ATTACHED` +intent-filter + `device_filter` (VID 0x0bda / PID 0x8812) auto-grants USB +permission on attach, and a tiny JNI `fork()`+`execve()` helper hands the +`UsbDeviceConnection` fd to `rxdemo` (Android's `ProcessBuilder` closes fds ≥3 +before exec, so the fd must be inherited across an in-process fork, not a +spawned child). This also required porting the Android fd path +(`libusb_wrap_sys_device` + `LIBUSB_OPTION_NO_DEVICE_DISCOVERY`) into +`examples/rx/main.cpp` — it previously existed only in the TX demo; without it +libusb's device enumeration is SELinux-denied under the `untrusted_app` domain. +Harness: `tests/quest_rxq_run.sh` (desktop TX flood + per-mode capture + pull) +and `tests/quest_rxq_analyze.py` (delivery from the 802.11 `seq_num` sequence + +ring telemetry). + +**Bench traps found and pinned:** the Quest's own adb-over-WiFi rides 5180 MHz +(**channel 36**), so any 5 GHz experiment self-desenses the USB adapter inches +away in the same chassis — the experiment must run on 2.4 GHz (ch 6). The 8821AU +1T1R nano airs but couples poorly to the Quest adapter; the 2T2R flooder reaches +it. 1M CCK from the flooder airs but the Quest decodes ~0; **6M OFDM** works. +Off-head standby drops adb-WiFi (keep-awake: `prox_close` broadcast + +`svc power stayon`). + +**What the constrained host confirmed.** The slow-consumer-on-pump-thread signal +is real and *large*: `cb_max_us` reaches **1–1.8 ms at idle and ~15 ms under a +1.5 ms/frame consumer spin**, vs `< 10 µs` on the desktop. Android also denies +`dev_mem_alloc` to the app sandbox, so the zerocopy DMA ring falls back to heap +buffers — the exact constrained condition the phone reports run under. + +**What it did not show: a delivery benefit from the ring-servicing mode.** Across +three load regimes, `async` vs `reorder-pool` vs `spsc-fat` are within +run-to-run noise (Δ swings ±1.5 pts with no consistent ordering over repeats): + +| regime | delivery (all modes) | telemetry | +|---|---|---| +| idle (`spin=0`, ~667 fps) | ~66 % | `cb_max` 1–1.8 ms, `min_armed` 3, `resubmit_fail` 0 | +| sustained overload (`spin=1500 µs`) | ~38 % | inline modes' pump `completions` collapse to ~1.2 k (blocked ~12 ms/frame); `spsc-fat`'s pump keeps draining (~16.7 k) — **same delivery** | +| transient burst (`3–4 ms` on / `30 ms` off + moderate spin) | ~63–66 % | Δdeliv across reorder-pool/spsc-fat swings ±1.5 pts over repeats = noise | + +The reasons mirror the desktop: at low load the ~34 % loss is the marginal-RF +baseline (near-field 2.4 GHz here) and the 4-URB(+spare) ring absorbs the +transient stalls (`min_armed` never below 3); at high load the consumer is +throughput-deficient (`spin × rate > 1 CPU`) and no buffering strategy recovers +a sustained deficit — `spsc-fat` provably keeps the pump draining but delivery is +unchanged because the *consumer* is the wall. The narrow window where a deeper +pool should win (bursts that overflow 4 URBs yet fit the pool, with average +headroom to drain) did not produce a signal above the RF-baseline noise. + +**Conclusion.** The three fixes are correct and their mechanisms provably engage +(telemetry), and they ship **off by default** (`DEVOURER_RX_MODE=async` +unchanged). But neither the desktop nor the Quest attributes the reporter's +~1–2 % residual to ring-servicing strategy: at realistic load the ring is not the +bottleneck (RF or consumer-throughput is), and the deep 16 KB-URB ring (#317) +already absorbs the transient stalls a shallower ring would drop. The shippable +product is therefore the **`rx.ring` telemetry as a field diagnostic**: if a +reporter's own phone shows `cb_max_us` / `qdepth` spiking coincident with +keyframe loss, the loss is consumer/pump-limited and `spsc-fat` is the remedy; +if not (the likely case), the residual is a CPU-provisioning limit that no ring +change fixes. Enablement + autonomous-driving details: `tests/quest_rxq.md`. diff --git a/docs/logging.md b/docs/logging.md index d439f39..9928652 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -90,6 +90,8 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets]; | `rx.body` | RX (`DEVOURER_DUMP_BODY`) | rate, rssi[2], evm[2], snr[2], crc, len, body hex | | `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 (`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`, else canonical SA | +| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"), 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). 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 | | `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 d22af58..fa43d42 100644 --- a/examples/common/env_config.cpp +++ b/examples/common/env_config.cpp @@ -38,6 +38,21 @@ bool env_long(const char *name, long *out) { return true; } +/* DEVOURER_RX_MODE spelling -> RxMode (UsbTransport RX-ring strategy). Accepts + * hyphen or underscore; unrecognised falls back to the default async ring. */ +devourer::RxMode parse_rx_mode(const char *s) { + if (str_ieq(s, "sync")) + return devourer::RxMode::Sync; + if (str_ieq(s, "reorder-pool") || str_ieq(s, "reorder_pool") || + str_ieq(s, "pool")) + return devourer::RxMode::ReorderPool; + if (str_ieq(s, "spsc-fat") || str_ieq(s, "spsc_fat") || str_ieq(s, "spsc")) + return devourer::RxMode::SpscFat; + if (str_ieq(s, "decoupled")) + return devourer::RxMode::Decoupled; + return devourer::RxMode::Async; +} + } // namespace devourer::DeviceConfig devourer_config_from_env() { @@ -57,6 +72,12 @@ devourer::DeviceConfig devourer_config_from_env() { cfg.rx.urbs = static_cast(v); if (env_long("DEVOURER_RX_URB_BYTES", &v)) cfg.rx.urb_bytes = static_cast(v); + if (const char *e = env_str("DEVOURER_RX_MODE")) + cfg.rx.rx_mode = parse_rx_mode(e); + if (env_long("DEVOURER_RX_POOL_SPARE", &v)) + cfg.rx.pool_spare = static_cast(v); + 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"); cfg.rx.abs_noise_floor = env_flag("DEVOURER_RX_NOISE_FLOOR"); if (env_long("DEVOURER_IGI", &v)) diff --git a/examples/rx/main.cpp b/examples/rx/main.cpp index a8f5931..073da67 100644 --- a/examples/rx/main.cpp +++ b/examples/rx/main.cpp @@ -484,6 +484,42 @@ static bool agg_sa_match(const Packet &packet) { std::memcmp(packet.Data.data() + 10, g_agg_sa, 6) == 0; } +/* DEVOURER_RX_SINK_SPIN_US: busy-spin this many microseconds at the top of + * every delivered frame in packetProcessor, modelling the inline consumer cost + * (in PixelPilot: wfb-ng FEC+AES+UDP, ~20-50 us/frame) that in the async ring + * delays the URB resubmit. packetProcessor runs once per aggregated MPDU, so + * this is a per-subframe cost — the same granularity the real consumer pays. + * 0/unset = off. */ +static const long g_rx_sink_spin_us = []() { + const char *e = std::getenv("DEVOURER_RX_SINK_SPIN_US"); + return e ? std::strtol(e, nullptr, 0) : 0L; +}(); + +/* DEVOURER_RX_SINK_STALL_MS / _EVERY: a PERIODIC consumer stall — every _EVERY + * frames, busy-spin _MS milliseconds — modelling an occasional hiccup (GC pause, + * scheduler preemption of the consumer) on a consumer that otherwise keeps up. + * This is the regime where moving the consumer off the pump thread (spsc-fat) + * matters: async/reorder consume on the pump, so a stall drains the ring and + * drops frames; spsc-fat keeps the pump re-arming and queues the backlog. */ +static const long g_rx_stall_ms = []() { + const char *e = std::getenv("DEVOURER_RX_SINK_STALL_MS"); + return e ? std::strtol(e, nullptr, 0) : 0L; +}(); +static const long g_rx_stall_every = []() { + const char *e = std::getenv("DEVOURER_RX_SINK_STALL_EVERY"); + return e ? std::strtol(e, nullptr, 0) : 100L; +}(); + +/* DEVOURER_RX_PCTR: emit a lean rx.seq event (payload counter + tsfl + crc + + * aggregate markers) per SA-matched frame — the ground-truth per-frame delivery + * sequence for the RX-ring loss study. Deliberately lean (no body hex) so the + * emit cost does not itself perturb the pump thread it measures. The counter is + * the u32 txdemo stamps at the QoS-Data body start (MPDU offset 26). */ +static const bool g_rx_pctr = []() { + const char *e = std::getenv("DEVOURER_RX_PCTR"); + return e != nullptr && std::strcmp(e, "0") != 0; +}(); + /* Emit the frame-free NHM power histogram (IRtlDevice::GetRxEnergy fills it) as * a distinct rx.nhm event so it never disturbs the rx.energy * fields its consumers key on. `peak` = the fullest bucket (0 = quiet @@ -548,6 +584,26 @@ static void packetProcessor(const Packet &packet) { ++g_rx_count; + /* Model the inline consumer cost that delays URB resubmit in the async ring + * (DEVOURER_RX_SINK_SPIN_US). This runs on the libusb pump thread — the very + * thread whose resubmit latency the rx.ring telemetry measures — so it + * reproduces the burst-starvation mechanism on a host with headroom to + * spare. */ + if (g_rx_sink_spin_us > 0) { + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::microseconds(g_rx_sink_spin_us); + while (std::chrono::steady_clock::now() < deadline) { + /* busy-wait: a sleep would yield the pump thread and defeat the model */ + } + } + if (g_rx_stall_ms > 0 && (g_rx_count % g_rx_stall_every) == 0) { + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(g_rx_stall_ms); + while (std::chrono::steady_clock::now() < deadline) { + /* periodic consumer hiccup */ + } + } + /* HE Trigger frame (802.11 control, FC=0x24) — aired in a legacy PPDU, so * even an 11ac witness captures the bytes. Decode + surface it as rx.trigger * so a monitor validates what an AX AP's F2P / UL_FIXINFO scheduler airs @@ -616,6 +672,32 @@ static void packetProcessor(const Packet &packet) { packet.RxAtrib.icv_err); } + /* rx.seq — the ground-truth per-frame delivery sequence for the RX-ring loss + * study. pctr is the u32 txdemo stamps at the QoS-Data body start (MPDU + * offset 26); tsfl is the chip RX timestamp; paggr/ppdu are the aggregate + * structure the host-vs-RF loss discriminator keys on (a host FIFO overflow + * drops a whole aggregate; an RF loss drops individual frames). The SA gate + * follows DEVOURER_RX_AGG_SA when set, else the canonical txdemo SA. */ + if (g_rx_pctr && packet.Data.size() >= 30) { + const bool seq_sa = + g_agg_sa_env + ? agg_sa_match(packet) + : (packet.Data.size() >= 16 && + std::memcmp(packet.Data.data() + 10, kTxSa, 6) == 0); + if (seq_sa) { + uint32_t pctr; + std::memcpy(&pctr, packet.Data.data() + 26, 4); + devourer::Ev(*g_ev, "rx.seq") + .t() /* host monotonic ms — correlates a pctr gap with an rx.ring dip */ + .f("pctr", (unsigned long long)pctr) + .f("tsfl", packet.RxAtrib.tsfl) + .f("seq", packet.RxAtrib.seq_num) + .f("crc", packet.RxAtrib.crc_err ? 1 : 0) + .f("paggr", packet.RxAtrib.paggr ? 1 : 0) + .f("ppdu", packet.RxAtrib.ppdu_cnt); + } + } + if (g_rx_count == 1) { devourer::Ev(*g_ev, "init.timing") .f("stage", "demo.first_rx_frame") @@ -857,10 +939,17 @@ static void packetProcessor(const Packet &packet) { } } -int main() { +int main(int argc, char **argv) { libusb_context *ctx; int rc; + /* Termux/Android: argv[1] = numeric USB fd handed to us by the app that holds + * the USB-host permission (libusb can't enumerate /dev/bus/usb under an + * untrusted_app SELinux domain, so we wrap the pre-opened fd instead). Mirrors + * the TX demo. fd==0 -> normal VID/PID open on Linux/macOS. */ + const long termux_fd = (argc >= 2) ? std::strtol(argv[1], nullptr, 0) : 0; + const bool termux_mode = (termux_fd > 0); + auto logger = std::make_shared(); apply_logging_env(*logger); /* DEVOURER_LOG_LEVEL / DEVOURER_EVENTS / ... */ g_ev = &logger->events(); @@ -964,6 +1053,15 @@ int main() { } #endif /* DEVOURER_HAVE_PCIE */ + if (termux_mode) { + logger->info("Termux/Android mode: wrapping USB fd {}", termux_fd); + /* Both options are global (ctx==NULL) and must precede libusb_init: skip the + * device-discovery scan entirely, and take the weak-authority path that + * tolerates the app-sandboxed usbfs fd. */ + libusb_set_option(NULL, LIBUSB_OPTION_NO_DEVICE_DISCOVERY); + libusb_set_option(NULL, LIBUSB_OPTION_WEAK_AUTHORITY); + } + rc = libusb_init(&ctx); if (rc < 0) { return rc; @@ -980,11 +1078,22 @@ int main() { ? LIBUSB_LOG_LEVEL_DEBUG : LIBUSB_LOG_LEVEL_WARNING); - /* DEVOURER_PID / DEVOURER_VID / DEVOURER_USB_BUS / DEVOURER_USB_PORT device - * selection — shared with the other multi-adapter demos (usb_select.h). */ - libusb_device_handle *dev_handle = open_selected_usb( - ctx, logger, kRealtekProductIds, - sizeof(kRealtekProductIds) / sizeof(kRealtekProductIds[0])); + libusb_device_handle *dev_handle = nullptr; + if (termux_mode) { + rc = libusb_wrap_sys_device(ctx, (intptr_t)termux_fd, &dev_handle); + if (rc != 0 || dev_handle == nullptr) { + logger->error("libusb_wrap_sys_device(fd={}) failed: {} ({})", termux_fd, + libusb_error_name(rc), rc); + libusb_exit(ctx); + return 1; + } + } else { + /* DEVOURER_PID / DEVOURER_VID / DEVOURER_USB_BUS / DEVOURER_USB_PORT device + * selection — shared with the other multi-adapter demos (usb_select.h). */ + dev_handle = open_selected_usb( + ctx, logger, kRealtekProductIds, + sizeof(kRealtekProductIds) / sizeof(kRealtekProductIds[0])); + } if (dev_handle == NULL) { libusb_exit(ctx); return 1; @@ -1003,8 +1112,10 @@ int main() { /* Reopen variant: recovers in place when the reset re-enumerates the device * (a warm Kestrel drops its firmware back to ROM on reset — the handle goes * stale and the dongle may pass through its ZeroCD id before returning). */ + /* Reset skipped in termux_mode: a wrapped app-owned fd can't be re-enumerated + * (a USB reset would orphan the handle the device-list scan can't re-find). */ rc = devourer::claim_interface_reset_reopen(ctx, dev_handle, logger, - std::getenv("DEVOURER_SKIP_RESET") == nullptr, usb_lock); + !termux_mode && std::getenv("DEVOURER_SKIP_RESET") == nullptr, usb_lock); devourer::Ev(*g_ev, "init.timing") .f("stage", "demo.usb_reset") .f("ms", ms_since_start()); diff --git a/examples/tx/main.cpp b/examples/tx/main.cpp index 947c983..eaf1e79 100644 --- a/examples/tx/main.cpp +++ b/examples/tx/main.cpp @@ -900,6 +900,26 @@ int main(int argc, char **argv) { tx_gap_set = true; } + /* Keyframe-burst emulation for the RX-ring loss study: send flat-out for + * DEVOURER_TX_BURST_ON_MS, then idle for DEVOURER_TX_BURST_OFF_MS, repeat. + * This reproduces the bursty arrival a video keyframe imposes on the receiver + * — the load pattern that starves the async RX ring — instead of the steady + * cadence tx_gap/tx_interval produce. When on_ms > 0 it overrides the gap. */ + long burst_on_ms = 0, burst_off_ms = 0; + if (const char *e = std::getenv("DEVOURER_TX_BURST_ON_MS")) { + burst_on_ms = std::strtol(e, nullptr, 0); + if (burst_on_ms < 0) burst_on_ms = 0; + } + if (const char *e = std::getenv("DEVOURER_TX_BURST_OFF_MS")) { + burst_off_ms = std::strtol(e, nullptr, 0); + if (burst_off_ms < 0) burst_off_ms = 0; + } + const bool burst_mode = burst_on_ms > 0 && burst_off_ms > 0; + auto burst_cycle_start = std::chrono::steady_clock::now(); + if (burst_mode) + logger->info("DEVOURER_TX_BURST — keyframe emulation {} ms on / {} ms off", + burst_on_ms, burst_off_ms); + /* DEVOURER_TX_BATCH=N: drive the send_packets batch API in groups of N * frames per call — the USB TX aggregation bench mode. Pair with * DEVOURER_TX_USB_AGG (library knob) to actually pack the group into shared @@ -1405,7 +1425,31 @@ int main(int argc, char **argv) { consec_fail); break; } - if (tx_gap_set) { + bool burst_off_window = false; + if (burst_mode) { + /* Keyframe-burst pacing: send at the normal inter-frame rate during the + * ON window, idle during OFF. A single sleep spans the remainder of the + * OFF window, then the cycle restarts — reproducing the bursty arrival + * that starves the RX ring. (ON keeps the tx_gap rate rather than going + * gap-0: a flat-out flood wedges the Jaguar1 async TX path, and a high + * finite rate already over-subscribes a slow RX consumer.) */ + const auto now = std::chrono::steady_clock::now(); + const long long in_cycle_ms = + std::chrono::duration_cast( + now - burst_cycle_start) + .count(); + if (in_cycle_ms >= burst_on_ms) { + burst_off_window = true; + const long long total = burst_on_ms + burst_off_ms; + if (in_cycle_ms < total) + std::this_thread::sleep_for( + std::chrono::milliseconds(total - in_cycle_ms)); + burst_cycle_start = std::chrono::steady_clock::now(); + } + } + if (burst_off_window) { + /* idled above for the OFF window — no additional inter-frame gap. */ + } else if (tx_gap_set) { /* explicit gap wins: >0 sleeps, 0 = no inter-frame sleep (max flood) */ if (tx_gap_us > 0) std::this_thread::sleep_for(std::chrono::microseconds(tx_gap_us)); diff --git a/src/DeviceConfig.h b/src/DeviceConfig.h index ac65301..d6c4131 100644 --- a/src/DeviceConfig.h +++ b/src/DeviceConfig.h @@ -65,6 +65,20 @@ enum class Dpdt8822eMode : uint8_t { Skip, /* leave the DPDT/pin-mux untouched */ }; +/* USB RX-ring servicing strategy (UsbTransport::rx_loop). The default async + * ring runs the consumer inline on the libusb pump thread before resubmitting + * each URB, so a slow consumer under a burst can stall resubmit and collapse + * the armed-URB depth. The other modes trade that coupling off for A/B study: + * a blocking single-buffer reference, and two buffer-pool variants that re-arm + * the ring with a spare buffer before consuming the received one. */ +enum class RxMode : uint8_t { + Async, /* current: consume inline, then resubmit the same URB */ + Sync, /* blocking single-buffer bulk-IN read loop (legacy reference) */ + ReorderPool, /* swap a spare buffer onto the wire, resubmit, then consume */ + SpscFat, /* pool + SPSC hand-off to a consumer thread (deep absorption) */ + Decoupled, /* naive worker-thread hand-off (known-bad A/B control) */ +}; + struct DeviceConfig { /* ---- RX ------------------------------------------------------------- */ struct Rx { @@ -102,6 +116,20 @@ struct DeviceConfig { * fills past 16 KB anyway. Kestrel ignores this knob: the 8852C RXAGG * LEN_TH is ~20 KB and its ring must hold a full aggregate (32 KB). */ std::optional urb_bytes; + /* env: DEVOURER_RX_MODE — USB RX-ring servicing strategy (RxMode above). + * Default async = the current inline-consume-then-resubmit ring. The other + * modes are A/B / fix vehicles for the burst-starvation study. */ + RxMode rx_mode = RxMode::Async; + /* env: DEVOURER_RX_POOL_SPARE — extra RX buffers beyond the URB count for + * the buffer-pool modes (ReorderPool/SpscFat): a deeper pool absorbs a + * 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_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 + * default RX path is then byte-for-byte unchanged). */ + std::optional ring_ms; /* env: DEVOURER_8821C_NO_PHYST (inverted) — 8821C: prepend the 32-byte * PHY-status to RX frames (per-frame RSSI/SNR/EVM). Disable only for the * leanest possible RX path. */ diff --git a/src/RtlAdapter.cpp b/src/RtlAdapter.cpp index f23fdf9..88b40fc 100644 --- a/src/RtlAdapter.cpp +++ b/src/RtlAdapter.cpp @@ -21,7 +21,8 @@ RtlAdapter::RtlAdapter(libusb_device_handle *dev_handle, Logger_t logger, std::shared_ptr usb_lock, const devourer::DeviceConfig &cfg) : _transport{std::make_shared( - dev_handle, logger, ctx, std::move(usb_lock), cfg.usb.rx_zerocopy)}, + 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))}, _logger{std::move(logger)} { init_from_transport(cfg); } diff --git a/src/UsbTransport.cpp b/src/UsbTransport.cpp index 2f978d0..d561fca 100644 --- a/src/UsbTransport.cpp +++ b/src/UsbTransport.cpp @@ -1,9 +1,14 @@ #include "UsbTransport.h" #include +#include #include #include +#include +#include #include +#include +#include #include #include "Event.h" @@ -13,6 +18,19 @@ namespace devourer { namespace { +/* Saturating atomic min/max for the RX-ring telemetry — relaxed and + * best-effort (a rare lost update is acceptable for a diagnostic counter). */ +inline void atomic_min(std::atomic &m, int v) { + int cur = m.load(std::memory_order_relaxed); + while (v < cur && !m.compare_exchange_weak(cur, v, std::memory_order_relaxed)) + ; +} +inline void atomic_max(std::atomic &m, long long v) { + long long cur = m.load(std::memory_order_relaxed); + while (v > cur && !m.compare_exchange_weak(cur, v, std::memory_order_relaxed)) + ; +} + /* Shared state for the async RX URB queue. */ struct AsyncRxShared { const std::function *cb; @@ -21,15 +39,196 @@ struct AsyncRxShared { * event pump and the TX event loop may run this callback, so `active` is * written from the pump thread while the loop below reads it. */ std::atomic active{0}; + + /* Diagnostic telemetry, populated only when `telemetry` (DEVOURER_RX_RING_MS + * > 0) — the default path pays nothing. `armed` is the count of URBs + * currently posted to the host controller and awaiting a frame: THIS is the + * depth that collapses when a slow inline consumer delays resubmit, and it is + * distinct from `active` (which counts URBs not yet permanently retired and + * so sits at ~n_urbs regardless of starvation). `min_armed` is the low-water + * mark since the last rx.ring emit; `cb_max_us` the worst inline-consume + * latency in the window; `resubmit_fail` cumulative wanted-but-failed + * resubmits. */ + bool telemetry = false; + std::atomic armed{0}; + std::atomic min_armed{0}; + std::atomic cb_max_us{0}; + std::atomic resubmit_fail{0}; + /* completions = URB callbacks; empties = those that left the ring with zero + * URBs posted. empties/completions is the robust host-starvation rate: under + * RF loss frames rarely arrive so the ring stays armed (empties ~0); under + * host starvation frames arrive faster than resubmit so the ring drains to + * empty (empties high) — and unlike the poll-loop rx.ring emit, this is + * counted in the callback, so it survives the pump-thread starvation that + * makes the periodic telemetry sparse exactly when it matters. */ + std::atomic completions{0}; + std::atomic empties{0}; + + /* reorder-pool mode (RxMode::ReorderPool): a free-list of spare RX buffers. + * On completion the callback swaps a fresh buffer onto the wire and resubmits + * the URB BEFORE consuming the received one, so the ring stays armed through + * the slow inline consume — the fix for the burst-starvation of the default + * async ring. Zerocopy-safe by construction: the buffer being DMA'd is never + * the one being read. The mutex guards the free-list because the callback can + * run from the RX pump and, in co-running TX+RX, the TX pump. */ + bool reorder = false; + int buf_size = 0; + std::mutex pool_mu; + std::vector free_bufs; + + /* spsc-fat mode (RxMode::SpscFat): the pump callback only reaps + re-arms the + * URB (microseconds, never blocked) and hands the received buffer to a + * SEPARATE consumer thread over this queue; the consumer runs on_data and + * returns the buffer to the pool. This keeps the ring fully armed even while + * the consumer stalls or is preempted — converting a chip-FIFO overflow + * (dropped frames) into bounded host-queue backlog (delayed frames). The fix + * for a stalling/preempted CONSUMER, where reorder-pool can't help because it + * still consumes on the pump thread. */ + bool spsc = false; + std::mutex queue_mu; + std::condition_variable queue_cv; + std::deque> queue; + bool consumer_stop = false; }; + +/* Run the caller's consumer, timing it for the rx.ring cb_max_us telemetry. */ +inline void rx_consume(AsyncRxShared *s, const uint8_t *buf, int len) { + if (len <= 0) + return; + if (s->telemetry) { + const auto t0 = std::chrono::steady_clock::now(); + (*s->cb)(buf, len); + atomic_max(s->cb_max_us, + std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0) + .count()); + } else { + (*s->cb)(buf, len); + } +} + extern "C" void LIBUSB_CALL devourer_rx_cb(libusb_transfer *t) { auto *s = static_cast(t->user_data); - if (t->status == LIBUSB_TRANSFER_COMPLETED && t->actual_length > 0) - (*s->cb)(t->buffer, t->actual_length); - bool resubmit = !(*s->stop)() && (t->status == LIBUSB_TRANSFER_COMPLETED || - t->status == LIBUSB_TRANSFER_TIMED_OUT); - if (resubmit && libusb_submit_transfer(t) == 0) + /* 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; + atomic_min(s->min_armed, a); + s->completions.fetch_add(1, std::memory_order_relaxed); + if (a <= 0) + s->empties.fetch_add(1, std::memory_order_relaxed); + } + const bool resubmit = !(*s->stop)() && + (t->status == LIBUSB_TRANSFER_COMPLETED || + t->status == LIBUSB_TRANSFER_TIMED_OUT); + const int rlen = t->status == LIBUSB_TRANSFER_COMPLETED ? t->actual_length : 0; + + if (s->reorder) { + uint8_t *received = t->buffer; + uint8_t *fresh = nullptr; + if (resubmit) { + std::lock_guard lk(s->pool_mu); + if (!s->free_bufs.empty()) { + fresh = s->free_bufs.back(); + s->free_bufs.pop_back(); + } + } + if (fresh) { + /* Re-arm the ring with a DIFFERENT buffer, then consume the received one + * — the ring is never idle during the slow consume. */ + t->buffer = fresh; + t->length = s->buf_size; + if (libusb_submit_transfer(t) == 0) { + if (s->telemetry) + s->armed.fetch_add(1, std::memory_order_relaxed); + rx_consume(s, received, rlen); + std::lock_guard lk(s->pool_mu); + s->free_bufs.push_back(received); + return; + } + /* submit failed — return the spare, restore, fall to the inline path. */ + { + std::lock_guard lk(s->pool_mu); + s->free_bufs.push_back(fresh); + } + t->buffer = received; + if (s->telemetry) + s->resubmit_fail.fetch_add(1, std::memory_order_relaxed); + } + /* Pool exhausted (consumer hopelessly behind), submit failed, or teardown: + * degrade to the inline consume-then-resubmit-same behaviour. */ + rx_consume(s, received, rlen); + if (resubmit && libusb_submit_transfer(t) == 0) { + if (s->telemetry) + s->armed.fetch_add(1, std::memory_order_relaxed); + return; + } + if (resubmit && s->telemetry) + s->resubmit_fail.fetch_add(1, std::memory_order_relaxed); + s->active--; + return; + } + + if (s->spsc) { + uint8_t *received = t->buffer; + uint8_t *fresh = nullptr; + if (resubmit) { + std::lock_guard lk(s->pool_mu); + if (!s->free_bufs.empty()) { + fresh = s->free_bufs.back(); + s->free_bufs.pop_back(); + } + } + if (fresh) { + /* Re-arm the ring with a fresh buffer, then hand the received one to the + * consumer thread — the pump never runs on_data, so it can re-arm in + * microseconds and the ring stays armed through any consumer stall. */ + t->buffer = fresh; + t->length = s->buf_size; + if (libusb_submit_transfer(t) == 0) { + if (s->telemetry) + s->armed.fetch_add(1, std::memory_order_relaxed); + if (rlen > 0) { + std::lock_guard lk(s->queue_mu); + s->queue.emplace_back(received, rlen); + s->queue_cv.notify_one(); + } else { /* empty completion — nothing to consume, recycle the buffer */ + std::lock_guard lk(s->pool_mu); + s->free_bufs.push_back(received); + } + return; + } + { + std::lock_guard lk(s->pool_mu); + s->free_bufs.push_back(fresh); + } + t->buffer = received; + if (s->telemetry) + 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. */ + if (resubmit && libusb_submit_transfer(t) == 0) { + if (s->telemetry) + s->armed.fetch_add(1, std::memory_order_relaxed); + return; + } + if (resubmit && s->telemetry) + s->resubmit_fail.fetch_add(1, std::memory_order_relaxed); + s->active--; return; + } + + /* Default async ring: consume inline, then resubmit the same URB. */ + rx_consume(s, t->buffer, rlen); + if (resubmit && libusb_submit_transfer(t) == 0) { + if (s->telemetry) + s->armed.fetch_add(1, std::memory_order_relaxed); /* back on the wire */ + return; + } + if (resubmit && s->telemetry) + s->resubmit_fail.fetch_add(1, std::memory_order_relaxed); s->active--; /* not resubmitted -> this URB is done */ } } // namespace @@ -37,9 +236,11 @@ extern "C" void LIBUSB_CALL devourer_rx_cb(libusb_transfer *t) { UsbTransport::UsbTransport(libusb_device_handle *dev_handle, Logger_t logger, libusb_context *ctx, std::shared_ptr usb_lock, - bool rx_zerocopy) + bool rx_zerocopy, RxMode rx_mode, int pool_spare, + int ring_ms) : _dev_handle{dev_handle}, _ctx{ctx}, _logger{std::move(logger)}, - _rx_zerocopy{rx_zerocopy}, _usb_lock{std::move(usb_lock)} { + _rx_zerocopy{rx_zerocopy}, _rx_mode{rx_mode}, _pool_spare{pool_spare}, + _ring_ms{ring_ms}, _usb_lock{std::move(usb_lock)} { libusb_device_descriptor desc{}; if (libusb_get_device_descriptor(libusb_get_device(_dev_handle), &desc) == LIBUSB_SUCCESS) { @@ -79,9 +280,27 @@ void UsbTransport::rx_loop( int buf_size, int n_urbs, const std::function &on_data, const std::function &should_stop) { + if (_rx_mode == RxMode::Sync) { + rx_loop_sync(buf_size, on_data, should_stop); + return; + } + if (_rx_mode == RxMode::Decoupled) + _logger->warn("RX: mode {} not yet implemented — falling back to async ring", + static_cast(_rx_mode)); + const bool reorder = _rx_mode == RxMode::ReorderPool; + const bool spsc = _rx_mode == RxMode::SpscFat; AsyncRxShared sh{&on_data, &should_stop}; + sh.telemetry = _ring_ms > 0; + sh.reorder = reorder; + sh.spsc = spsc; + 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 + * overflowing the chip RX FIFO. Plain async has no spares. */ + const int spare = (reorder || spsc) && _pool_spare > 0 ? _pool_spare : 0; + const int n_pool = n_urbs + spare; std::vector xfers; - /* Zerocopy RX ring: allocate each URB buffer from kernel DMA memory + /* Zerocopy RX ring: allocate each buffer from kernel DMA memory * (libusb_dev_mem_alloc = USBDEVFS_ALLOC on Linux) so the bulk-IN DMAs a * frame straight into this mmap'd buffer and usbfs skips the copy-to-user on * reap. dev_mem_alloc returns NULL on backends/HCDs that don't support it (or @@ -89,10 +308,10 @@ void UsbTransport::rx_loop( * copy-on-reap path. Buffers outlive every in-flight transfer (freed only * after the drain below), so the mmap'd memory is never released under a live * URB. */ - std::vector bufs(n_urbs, nullptr); - std::vector is_devmem(n_urbs, false); + std::vector bufs(n_pool, nullptr); + std::vector is_devmem(n_pool, false); int zc = 0; - for (int i = 0; i < n_urbs; i++) { + for (int i = 0; i < n_pool; i++) { if (_rx_zerocopy) { bufs[i] = libusb_dev_mem_alloc(_dev_handle, buf_size); if (bufs[i]) { @@ -102,8 +321,16 @@ void UsbTransport::rx_loop( } if (!bufs[i]) bufs[i] = static_cast(malloc(buf_size)); + } + /* Hand the reorder spares (the pool tail) to the callback's free-list BEFORE + * submitting any URB, so an early completion (e.g. a co-running TX pump + * draining events) never races an unpopulated list. */ + for (int i = n_urbs; i < n_pool; i++) + if (bufs[i]) + sh.free_bufs.push_back(bufs[i]); + for (int i = 0; i < n_urbs; i++) { if (!bufs[i]) - continue; /* both allocs failed — skip this URB */ + continue; /* alloc failed — skip this URB (buffer freed in the sweep) */ libusb_transfer *t = libusb_alloc_transfer(0); if (!t) continue; /* buffer freed in the by-kind sweep below */ @@ -125,11 +352,83 @@ void UsbTransport::rx_loop( libusb_free_transfer(t); } } - _logger->info("RX: async queue of {} URBs submitted ({} zerocopy DMA, {} heap)", - sh.active.load(), zc, n_urbs - zc); + _logger->info("RX: {} ring of {} URBs submitted (+{} spare, {} zerocopy DMA, " + "{} heap)", + spsc ? "spsc-fat" : reorder ? "reorder-pool" : "async", + sh.active.load(), static_cast(sh.free_bufs.size()), zc, + n_pool - zc); + /* spsc-fat: the consumer thread. The pump callback only reaps + re-arms + + * enqueues; this thread drains the queue and runs on_data, returning each + * buffer to the pool. Joined at teardown after the URBs are drained. */ + std::thread consumer; + if (spsc) { + consumer = std::thread([&sh]() { + for (;;) { + std::pair item; + { + std::unique_lock lk(sh.queue_mu); + sh.queue_cv.wait( + lk, [&sh] { return !sh.queue.empty() || sh.consumer_stop; }); + if (sh.queue.empty()) /* stop requested and drained */ + return; + item = sh.queue.front(); + sh.queue.pop_front(); + } + rx_consume(&sh, item.first, item.second); + std::lock_guard lk(sh.pool_mu); + sh.free_bufs.push_back(item.first); + } + }); + } + if (sh.telemetry) { + /* Seed both the live depth and its low-water mark to the just-submitted URB + * count. min_armed{0} would otherwise pin the first window's low-water at 0 + * (atomic_min only lowers), reporting a starvation that never happened; the + * store also wipes any transient recorded by a callback that fired between + * URB submission above and this seed. */ + const int a0 = sh.active.load(std::memory_order_relaxed); + sh.armed.store(a0, std::memory_order_relaxed); + sh.min_armed.store(a0, std::memory_order_relaxed); + } + auto last_ring = std::chrono::steady_clock::now(); while (!should_stop() && sh.active > 0) { struct timeval tv {0, 100000}; libusb_handle_events_timeout_completed(_ctx, &tv, nullptr); + if (sh.telemetry) { + const auto now = std::chrono::steady_clock::now(); + if (std::chrono::duration_cast(now - last_ring) + .count() >= _ring_ms) { + last_ring = now; + const int a = sh.armed.load(std::memory_order_relaxed); + long long pool_free = -1; /* -1 = no host-side pool (plain async) */ + long long qdepth = 0; + if (reorder || spsc) { + std::lock_guard lk(sh.pool_mu); + pool_free = static_cast(sh.free_bufs.size()); + } + if (spsc) { + std::lock_guard lk(sh.queue_mu); + qdepth = static_cast(sh.queue.size()); + } + Ev(_logger->events(), "rx.ring") + .t() + .f("mode", + spsc ? "spsc-fat" : reorder ? "reorder-pool" : "async") + .f("n_urbs", n_urbs) + .f("armed", a) + .f("min_armed", sh.min_armed.exchange(a, std::memory_order_relaxed)) + .f("cb_max_us", + sh.cb_max_us.exchange(0, std::memory_order_relaxed)) + .f("resubmit_fail", (unsigned long long)sh.resubmit_fail.load( + std::memory_order_relaxed)) + .f("completions", (unsigned long long)sh.completions.load( + std::memory_order_relaxed)) + .f("empties", (unsigned long long)sh.empties.load( + std::memory_order_relaxed)) + .f("pool_free", pool_free) + .f("qdepth", qdepth); + } + } } for (auto *t : xfers) libusb_cancel_transfer(t); @@ -139,9 +438,22 @@ void UsbTransport::rx_loop( } for (auto *t : xfers) libusb_free_transfer(t); - /* All URBs are drained (sh.active == 0) — no transfer references a buffer, so - * releasing the ring is safe. Free each by the way it was allocated. */ - for (int i = 0; i < n_urbs; i++) { + /* spsc-fat: stop and join the consumer before releasing the pool, so no + * on_data is in flight over a buffer we are about to free. */ + if (spsc) { + { + std::lock_guard lk(sh.queue_mu); + sh.consumer_stop = true; + } + sh.queue_cv.notify_all(); + if (consumer.joinable()) + consumer.join(); + } + /* All URBs are drained (sh.active == 0) — no transfer references a buffer and + * the free-list is quiescent, so releasing the whole pool is safe. Ownership + * is `bufs` (every buffer, including the ones reorder swapped onto transfers + * or parked on the free-list); free each once, by the way it was allocated. */ + for (int i = 0; i < n_pool; i++) { if (!bufs[i]) continue; if (is_devmem[i]) @@ -151,6 +463,59 @@ void UsbTransport::rx_loop( } } +void UsbTransport::rx_loop_sync( + int buf_size, const std::function &on_data, + const std::function &should_stop) { + /* Legacy single-buffer blocking bulk-IN read — the pre-async-ring reference + * (floppyhammer-era model). There is no posted-URB ring: while on_data runs, + * nothing is armed on the wire, so this starves on consume latency rather + * than resubmit latency. Kept as an A/B baseline for the burst-starvation + * study; the 200 ms read timeout bounds should_stop() responsiveness on an + * idle channel. */ + std::vector buf(buf_size); + _logger->info("RX: synchronous blocking-read loop (buf {} B)", buf_size); + const bool telemetry = _ring_ms > 0; + auto last_ring = std::chrono::steady_clock::now(); + long long cb_max_us = 0; + while (!should_stop()) { + int n = rx_raw(buf.data(), buf_size, 200); + if (n > 0) { + if (telemetry) { + const auto t0 = std::chrono::steady_clock::now(); + on_data(buf.data(), n); + const long long us = std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0) + .count(); + if (us > cb_max_us) + cb_max_us = us; + } else { + on_data(buf.data(), n); + } + } else if (n == LIBUSB_ERROR_NO_DEVICE) { + _logger->error("RX: device gone (sync read) — stopping"); + break; + } + /* n == LIBUSB_ERROR_TIMEOUT (idle channel) or a transient error: loop. */ + if (telemetry) { + const auto now = std::chrono::steady_clock::now(); + if (std::chrono::duration_cast(now - last_ring) + .count() >= _ring_ms) { + last_ring = now; + Ev(_logger->events(), "rx.ring") + .t() + .f("mode", "sync") + .f("n_urbs", 1) + .f("armed", 0) /* no ring: nothing armed while consuming */ + .f("min_armed", 0) + .f("cb_max_us", cb_max_us) + .f("resubmit_fail", (unsigned long long)0) + .f("pool_free", -1); + cb_max_us = 0; + } + } + } +} + int UsbTransport::rx_raw(uint8_t *buf, int len, int timeout_ms) { int actual = 0; int rc = libusb_bulk_transfer(_dev_handle, _info.bulk_in_ep, buf, len, diff --git a/src/UsbTransport.h b/src/UsbTransport.h index 9a02027..d235614 100644 --- a/src/UsbTransport.h +++ b/src/UsbTransport.h @@ -32,7 +32,8 @@ class UsbTransport final : public IRtlTransport { UsbTransport(libusb_device_handle *dev_handle, Logger_t logger, libusb_context *ctx = nullptr, std::shared_ptr usb_lock = nullptr, - bool rx_zerocopy = true); + bool rx_zerocopy = true, RxMode rx_mode = RxMode::Async, + int pool_spare = 0, int ring_ms = 0); ~UsbTransport() override; bool is_usb() const override { return true; } @@ -120,6 +121,18 @@ class UsbTransport final : public IRtlTransport { * unsupported. See rx_loop and DeviceConfig::Usb::rx_zerocopy. */ bool _rx_zerocopy = true; + /* RX-ring servicing strategy + buffer-pool depth + diagnostic telemetry + * cadence, from DeviceConfig::Rx. rx_loop reads these; the defaults preserve + * the historic inline async ring with no extra buffers and no telemetry. */ + RxMode _rx_mode = RxMode::Async; + int _pool_spare = 0; + int _ring_ms = 0; + + /* rx_loop helpers for the servicing strategies dispatched off _rx_mode. */ + void rx_loop_sync(int buf_size, + const std::function &on_data, + const std::function &should_stop); + /* Exclusive per-adapter USB lock (UsbDeviceLock.h), held for the transport * lifetime; released when the device (and thus the transport) dies. */ std::shared_ptr _usb_lock; diff --git a/tests/build_android.sh b/tests/build_android.sh new file mode 100755 index 0000000..c1721c2 --- /dev/null +++ b/tests/build_android.sh @@ -0,0 +1,72 @@ +#!/usr/bin/env bash +# +# build_android.sh — cross-compile rxdemo for arm64-v8a Android (e.g. Meta Quest 3, +# a phone under Termux) for the issue #330 RX-ring test. Produces a stripped, +# self-contained bundle (rxdemo + libusb1.0.so) you `adb push` and run in Termux +# via `termux-usb -e`. See tests/quest_rxq.md. +# +# Requirements: an Android NDK (r26+), curl, cmake. Point NDK at it: +# ANDROID_NDK=/path/to/android-ndk-r26d tests/build_android.sh +# +# Build knobs (env): ABI (default arm64-v8a), API (default 28), OUT (bundle dir). +# +set -euo pipefail + +NDK="${ANDROID_NDK:?set ANDROID_NDK to your NDK root (r26+)}" +ABI="${ABI:-arm64-v8a}" +API="${API:-28}" +SRC="$(cd "$(dirname "$0")/.." && pwd)" +WORK="${WORK:-/tmp/devourer-android}" +OUT="${OUT:-$WORK/bundle}" +LIBUSB_VER="${LIBUSB_VER:-1.0.27}" + +mkdir -p "$WORK"; cd "$WORK" + +# --- 1. libusb for the target ABI via ndk-build ------------------------------ +if [ ! -d "libusb-$LIBUSB_VER" ]; then + curl -sL -o "libusb-$LIBUSB_VER.tar.bz2" \ + "https://github.com/libusb/libusb/releases/download/v$LIBUSB_VER/libusb-$LIBUSB_VER.tar.bz2" + tar xjf "libusb-$LIBUSB_VER.tar.bz2" +fi +( cd "libusb-$LIBUSB_VER/android/jni" && "$NDK/ndk-build" APP_ABI="$ABI" APP_PLATFORM="android-$API" -j"$(nproc)" ) +SO=$(find "$WORK/libusb-$LIBUSB_VER/android/libs/$ABI" -name 'libusb1.0.so' | head -1) + +# --- 2. stage a prefix + a pkg-config .pc the devourer CMake consumes --------- +PFX="$WORK/prefix-$ABI" +rm -rf "$PFX"; mkdir -p "$PFX/lib/pkgconfig" "$PFX/include/libusb-1.0" +cp "$SO" "$PFX/lib/libusb-1.0.so" # -lusb-1.0 wants this name +cp "$WORK/libusb-$LIBUSB_VER/libusb/libusb.h" "$PFX/include/libusb-1.0/" +cat > "$PFX/lib/pkgconfig/libusb-1.0.pc" </dev/null || true + +echo "=== bundle ($ABI) ==="; file "$OUT/rxdemo"; ls -lh "$OUT" +echo "push: adb push $OUT/rxdemo $OUT/libusb1.0.so /data/local/tmp/" diff --git a/tests/quest_apk/.gitignore b/tests/quest_apk/.gitignore new file mode 100644 index 0000000..f79605c --- /dev/null +++ b/tests/quest_apk/.gitignore @@ -0,0 +1,5 @@ +# Build artifacts placed here by build_apk.sh (native payloads copied/compiled +# from the arm64 rxdemo bundle + NDK). Never checked in. +lib/ +build/ +*.apk diff --git a/tests/quest_apk/AndroidManifest.xml b/tests/quest_apk/AndroidManifest.xml new file mode 100644 index 0000000..c33cd85 --- /dev/null +++ b/tests/quest_apk/AndroidManifest.xml @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/quest_apk/build_apk.sh b/tests/quest_apk/build_apk.sh new file mode 100755 index 0000000..46c3dda --- /dev/null +++ b/tests/quest_apk/build_apk.sh @@ -0,0 +1,68 @@ +#!/usr/bin/env bash +# Hand-build the RXQ USB-host APK (no gradle): javac -> d8 -> aapt package -> +# add dex + native libs -> zipalign -> apksigner. Autonomous vehicle for the +# issue-330 Quest run: the APK auto-grants USB permission via its +# USB_DEVICE_ATTACHED intent-filter (the PixelPilot mechanism) and execs the +# bundled devourer rxdemo, so nothing needs doing in the headset. See +# tests/quest_rxq.md. +# +# Prereqs: an Android SDK (build-tools 30.0.3 + android-29 platform), an NDK +# (r23+), and an arm64 rxdemo bundle from tests/build_android.sh. Point at them: +# ANDROID_SDK=/opt/android-sdk ANDROID_NDK=.../ndk/23.1.7779620 \ +# RXQ_BUNDLE=/tmp/devourer-android/bundle tests/quest_apk/build_apk.sh +set -euo pipefail + +SDK="${ANDROID_SDK:-${ANDROID_SDK_ROOT:-/opt/android-sdk}}" +BT="$SDK/build-tools/${BUILD_TOOLS:-30.0.3}" +AJAR="$SDK/platforms/${PLATFORM:-android-29}/android.jar" +NDK="${ANDROID_NDK:-$SDK/ndk/23.1.7779620}" +API="${API:-28}" +HERE="$(cd "$(dirname "$0")" && pwd)" +# RXQ_BUNDLE holds the arm64 rxdemo + libusb1.0.so (output of build_android.sh). +BUNDLE="${RXQ_BUNDLE:-/tmp/devourer-android/bundle}" +WORK="${RXQ_WORK:-/tmp/rxq-apk}" # build artifacts + keystore, out of tree +OUT="$WORK/build" +KS="$WORK/debug.keystore" + +rm -rf "$OUT"; mkdir -p "$OUT/obj" "$HERE/lib/arm64-v8a" + +# 0. Native payload: rxdemo must be named lib*.so so Android extracts it +# executable into nativeLibraryDir; its DT_NEEDED "libusb1.0.so" resolves +# via LD_LIBRARY_PATH=nativeLibraryDir set by the launcher. +cp "$BUNDLE/rxdemo" "$HERE/lib/arm64-v8a/librxdemo.so" +cp "$BUNDLE/libusb1.0.so" "$HERE/lib/arm64-v8a/libusb1.0.so" + +# 0b. JNI fork()+exec() helper — ProcessBuilder can't pass a USB fd to a child, +# so the app forks in-process via this .so. +CC=$(ls "$NDK"/toolchains/llvm/prebuilt/*/bin/aarch64-linux-android${API}-clang | head -1) +"$CC" -shared -fPIC -O2 -o "$HERE/lib/arm64-v8a/libspawn.so" "$HERE/jni/spawn.c" -llog + +# 1. debug keystore (once) +if [ ! -f "$KS" ]; then + keytool -genkeypair -keystore "$KS" -storepass android -keypass android \ + -alias rxq -keyalg RSA -keysize 2048 -validity 10000 \ + -dname "CN=RXQ,O=OpenIPC,C=US" +fi + +# 2. compile + dex +javac -source 1.8 -target 1.8 -bootclasspath "$AJAR" \ + -d "$OUT/obj" "$HERE"/src/org/openipc/rxq/*.java +"$BT/d8" --min-api 28 --output "$OUT" "$OUT"/obj/org/openipc/rxq/*.class + +# 3. package manifest + resources +"$BT/aapt" package -f -M "$HERE/AndroidManifest.xml" -S "$HERE/res" \ + -I "$AJAR" -F "$OUT/rxq.unsigned.apk" + +# 4. add classes.dex + native libs (paths inside the apk preserved) +( cd "$OUT" && "$BT/aapt" add rxq.unsigned.apk classes.dex >/dev/null ) +( cd "$HERE" && "$BT/aapt" add "$OUT/rxq.unsigned.apk" \ + lib/arm64-v8a/librxdemo.so lib/arm64-v8a/libusb1.0.so \ + lib/arm64-v8a/libspawn.so >/dev/null ) + +# 5. align + sign +"$BT/zipalign" -f 4 "$OUT/rxq.unsigned.apk" "$OUT/rxq.aligned.apk" +"$BT/apksigner" sign --ks "$KS" --ks-pass pass:android --key-pass pass:android \ + --out "$OUT/rxq.apk" "$OUT/rxq.aligned.apk" + +echo "BUILT: $OUT/rxq.apk" +"$BT/apksigner" verify --print-certs "$OUT/rxq.apk" >/dev/null && echo "signature OK" diff --git a/tests/quest_apk/jni/spawn.c b/tests/quest_apk/jni/spawn.c new file mode 100644 index 0000000..8052b09 --- /dev/null +++ b/tests/quest_apk/jni/spawn.c @@ -0,0 +1,75 @@ +/* spawn.c — a minimal JNI fork()+execve() helper. + * + * Android's ProcessBuilder closes every fd >= 3 in the child before exec, so a + * USB fd from UsbDeviceConnection can't reach an exec'd binary that way. Here we + * fork from inside the app process (where the fd is valid) and execve rxdemo + * ourselves without closing the fd — the child inherits it. Only + * async-signal-safe calls run between fork and exec; the env array is built in + * the parent beforehand. This is the same in-process-fd trick termux-usb and + * PixelPilot rely on, minus the socket dance. + */ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +JNIEXPORT jint JNICALL +Java_org_openipc_rxq_MainActivity_nativeSpawn(JNIEnv *env, jclass cls, jint fd, + jstring jexe, jstring jout, + jobjectArray jenv) { + const char *exe = (*env)->GetStringUTFChars(env, jexe, 0); + const char *out = (*env)->GetStringUTFChars(env, jout, 0); + char exebuf[1024], outbuf[1024]; + strncpy(exebuf, exe, sizeof(exebuf) - 1); exebuf[sizeof(exebuf) - 1] = 0; + strncpy(outbuf, out, sizeof(outbuf) - 1); outbuf[sizeof(outbuf) - 1] = 0; + (*env)->ReleaseStringUTFChars(env, jexe, exe); + (*env)->ReleaseStringUTFChars(env, jout, out); + + int nenv = jenv ? (*env)->GetArrayLength(env, jenv) : 0; + char **envp = calloc(nenv + 1, sizeof(char *)); + for (int i = 0; i < nenv; i++) { + jstring s = (jstring)(*env)->GetObjectArrayElement(env, jenv, i); + const char *cs = (*env)->GetStringUTFChars(env, s, 0); + envp[i] = strdup(cs); + (*env)->ReleaseStringUTFChars(env, s, cs); + } + envp[nenv] = NULL; + + char fdstr[16]; + snprintf(fdstr, sizeof(fdstr), "%d", fd); + + pid_t pid = fork(); + if (pid == 0) { + int o = open(outbuf, O_WRONLY | O_CREAT | O_TRUNC, 0644); + if (o >= 0) { dup2(o, 1); dup2(o, 2); if (o > 2) close(o); } + fcntl(fd, F_SETFD, 0); /* ensure the USB fd is not CLOEXEC */ + char *argv[] = { exebuf, fdstr, NULL }; + execve(exebuf, argv, envp); + _exit(127); + } + + for (int i = 0; i < nenv; i++) free(envp[i]); + free(envp); + return pid; +} + +JNIEXPORT jint JNICALL +Java_org_openipc_rxq_MainActivity_nativeWait(JNIEnv *env, jclass cls, jint pid) { + int st = 0; + if (waitpid(pid, &st, 0) < 0) return -1; + return WIFEXITED(st) ? WEXITSTATUS(st) : 128 + WTERMSIG(st); +} + +JNIEXPORT void JNICALL +Java_org_openipc_rxq_MainActivity_nativeKill(JNIEnv *env, jclass cls, jint pid) { + /* SIGKILL, not SIGTERM: rxdemo's SIGTERM handler runs a slow clean chip + * de-init during which the USB device stays claimed, so the next capture's + * openDevice/claim fails. For RX there's no TX state to unwind — kill hard + * and reap so the device frees immediately. */ + if (pid > 0) { kill(pid, SIGKILL); waitpid(pid, NULL, 0); } +} diff --git a/tests/quest_apk/res/xml/device_filter.xml b/tests/quest_apk/res/xml/device_filter.xml new file mode 100644 index 0000000..741fa7d --- /dev/null +++ b/tests/quest_apk/res/xml/device_filter.xml @@ -0,0 +1,6 @@ + + + + + diff --git a/tests/quest_apk/src/org/openipc/rxq/MainActivity.java b/tests/quest_apk/src/org/openipc/rxq/MainActivity.java new file mode 100644 index 0000000..c0db206 --- /dev/null +++ b/tests/quest_apk/src/org/openipc/rxq/MainActivity.java @@ -0,0 +1,195 @@ +package org.openipc.rxq; + +import android.app.Activity; +import android.app.PendingIntent; +import android.content.BroadcastReceiver; +import android.content.Context; +import android.content.Intent; +import android.content.IntentFilter; +import android.hardware.usb.UsbDevice; +import android.hardware.usb.UsbDeviceConnection; +import android.hardware.usb.UsbManager; +import android.os.Bundle; +import android.util.Log; + +import java.io.File; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; + +/* + * Minimal USB-host launcher for the devourer rxdemo, modelled on how + * PixelPilot avoids the permission dialog: a USB_DEVICE_ATTACHED intent-filter + * with a device_filter auto-grants this app permission to the matching adapter + * and launches us with EXTRA_DEVICE set. We open the device, hand its file + * descriptor to the native rxdemo (packaged as librxdemo.so so Android extracts + * it executable into nativeLibraryDir), and redirect its JSONL to + * getExternalFilesDir() where adb can pull it. Everything is driven headlessly + * via `am start` extras (rxmode/channel/tag/...); no in-headset interaction. + */ +public class MainActivity extends Activity { + static final String TAG = "RXQ"; + static final String ACTION_PERM = "org.openipc.rxq.PERM"; + static final int VID = 0x0bda, PID = 0x8812; + + // The running rxdemo child pid + reaper thread, kept across launches so a new + // run (or action=stop) can reap the previous one. sConn is held open for the + // child's lifetime so the USB fd stays valid. + static int sPid; + static Thread sWaiter; + static UsbDeviceConnection sConn; + + static { System.loadLibrary("spawn"); } + + // JNI fork()+execve() helper (spawn.c) — ProcessBuilder can't hand a USB fd + // to a child (it closes all fds >= 3 before exec), so we fork in-process + // where the fd is valid and exec rxdemo ourselves without closing it. + static native int nativeSpawn(int fd, String exe, String out, String[] env); + static native int nativeWait(int pid); + static native void nativeKill(int pid); + + @Override protected void onCreate(Bundle b) { + super.onCreate(b); + handle(getIntent()); + } + + @Override protected void onNewIntent(Intent i) { + super.onNewIntent(i); + setIntent(i); + handle(i); + } + + void handle(Intent intent) { + if (intent != null && "stop".equals(intent.getStringExtra("action"))) { + killChild(); + writeStatus("stopped"); + Log.i(TAG, "stopped on request"); + return; + } + UsbManager um = (UsbManager) getSystemService(Context.USB_SERVICE); + UsbDevice dev = null; + if (intent != null) + dev = intent.getParcelableExtra(UsbManager.EXTRA_DEVICE); + if (dev == null) dev = findDevice(um); + if (dev == null) { + Log.e(TAG, "no matching USB device (" + Integer.toHexString(VID) + + ":" + Integer.toHexString(PID) + ")"); + writeStatus("no-device"); + return; + } + if (um.hasPermission(dev)) { + launch(um, dev, intent); + } else { + // Fallback for the already-attached case: request permission. With + // the intent-filter present this is normally auto-granted on + // attach, so this path is a safety net. + Log.w(TAG, "no permission yet, requesting"); + // targetSdk 28 is exempt from the API-31 mutable-PendingIntent + // requirement, so FLAG_UPDATE_CURRENT compiles against older jars. + PendingIntent pi = PendingIntent.getBroadcast(this, 0, + new Intent(ACTION_PERM), PendingIntent.FLAG_UPDATE_CURRENT); + registerReceiver(new BroadcastReceiver() { + @Override public void onReceive(Context c, Intent i) { + UsbDevice d = i.getParcelableExtra(UsbManager.EXTRA_DEVICE); + boolean ok = i.getBooleanExtra( + UsbManager.EXTRA_PERMISSION_GRANTED, false); + Log.i(TAG, "perm result granted=" + ok); + if (ok && d != null) launch(um, d, getIntent()); + try { c.unregisterReceiver(this); } catch (Exception e) {} + } + }, new IntentFilter(ACTION_PERM)); + um.requestPermission(dev, pi); + } + } + + UsbDevice findDevice(UsbManager um) { + HashMap m = um.getDeviceList(); + for (UsbDevice d : m.values()) + if (d.getVendorId() == VID && d.getProductId() == PID) return d; + return null; + } + + void launch(UsbManager um, UsbDevice dev, Intent intent) { + killChild(); + UsbDeviceConnection conn = um.openDevice(dev); + if (conn == null) { + Log.e(TAG, "openDevice returned null"); + writeStatus("open-failed"); + return; + } + // fork()+execve() in-process (spawn.c): the fd from getFileDescriptor() + // is valid here and the native helper does not close it, so the child + // inherits it directly. Keep sConn open for the child's lifetime. + int fd = conn.getFileDescriptor(); + sConn = conn; + String tag = ext(intent, "tag", "run"); + File out = new File(getExternalFilesDir(null), "rxq_" + tag + ".jsonl"); + String nlib = getApplicationInfo().nativeLibraryDir; + + Map e = new HashMap<>(); + e.put("LD_LIBRARY_PATH", nlib); + e.put("DEVOURER_EVENTS", "stdout"); + e.put("DEVOURER_LOG_LEVEL", ext(intent, "loglevel", "info")); + e.put("DEVOURER_CHANNEL", ext(intent, "channel", "36")); + e.put("DEVOURER_RX_MODE", ext(intent, "rxmode", "async")); + e.put("DEVOURER_RX_URBS", ext(intent, "urbs", "4")); + // rx.seq is the ground-truth per-frame delivery sequence (pctr counter + // the txdemo QoS-Data flood stamps); always on for this study. + e.put("DEVOURER_RX_PCTR", ext(intent, "pctr", "1")); + putIf(e, "DEVOURER_RX_POOL_SPARE", ext(intent, "poolspare", null)); + putIf(e, "DEVOURER_RX_RING_MS", ext(intent, "ringms", "100")); + putIf(e, "DEVOURER_RX_SINK_SPIN_US", ext(intent, "sinkspin", null)); + putIf(e, "DEVOURER_RX_KEEP_CORRUPTED", ext(intent, "keepcorrupt", null)); + putIf(e, "DEVOURER_BW", ext(intent, "bw", null)); + ArrayList envList = new ArrayList<>(); + for (Map.Entry en : e.entrySet()) + envList.add(en.getKey() + "=" + en.getValue()); + String[] envArr = envList.toArray(new String[0]); + + Log.i(TAG, "spawn fd=" + fd + " mode=" + e.get("DEVOURER_RX_MODE") + + " ch=" + e.get("DEVOURER_CHANNEL") + " -> " + out); + final int pid = nativeSpawn(fd, nlib + "/librxdemo.so", + out.getAbsolutePath(), envArr); + sPid = pid; + if (pid <= 0) { + writeStatus("spawn-failed pid=" + pid); + return; + } + writeStatus("running tag=" + tag + " pid=" + pid + " fd=" + fd + + " mode=" + e.get("DEVOURER_RX_MODE")); + sWaiter = new Thread(new Runnable() { + @Override public void run() { + int rc = nativeWait(pid); + Log.i(TAG, "rxdemo pid=" + pid + " exit " + rc); + writeStatus("exited rc=" + rc + " pid=" + pid); + } + }); + sWaiter.start(); + } + + static void killChild() { + if (sPid > 0) { nativeKill(sPid); } + sPid = 0; + // Release the USB device so the next capture's openDevice/claim is clean. + if (sConn != null) { try { sConn.close(); } catch (Exception e) {} sConn = null; } + } + + void putIf(Map m, String k, String v) { + if (v != null && !v.isEmpty()) m.put(k, v); + } + + static String ext(Intent i, String k, String def) { + if (i == null) return def; + String v = i.getStringExtra(k); + return v != null ? v : def; + } + + void writeStatus(String s) { + try { + File f = new File(getExternalFilesDir(null), "status.txt"); + java.io.FileWriter w = new java.io.FileWriter(f, false); + w.write(s + "\n"); + w.close(); + } catch (Exception e) { Log.e(TAG, "status write", e); } + } +} diff --git a/tests/quest_rxq.md b/tests/quest_rxq.md new file mode 100644 index 0000000..eb11f16 --- /dev/null +++ b/tests/quest_rxq.md @@ -0,0 +1,83 @@ +# Quest 3 RX-ring runbook (issue #330) + +Confirmatory on-device test for the async-RX-ring residual loss. The Meta Quest 3 +(Snapdragon XR2 Gen 2, Android) is a constrained mobile host — the most faithful +stand-in for the reporter's MediaTek phone available. The whole run is driven +**headlessly from the desktop over adb**; nothing happens in the headset. + +Vehicle: a minimal auto-grant APK (`tests/quest_apk/`) modelled on how PixelPilot +avoids the USB permission dialog — a `USB_DEVICE_ATTACHED` intent-filter + +`device_filter` auto-grants USB permission when the adapter attaches, and a JNI +`fork()`+`execve()` helper hands the `UsbDeviceConnection` fd to the bundled +`rxdemo`. Results (`docs/experiments/issue-330-rx-ring-starvation.md`): the +ring-servicing mode shows no delivery benefit above the RF-baseline noise on the +real device either; the shippable product is the `rx.ring` telemetry as a field +diagnostic. + +## Enablement (once) + +1. **Developer mode** on the headset (Meta Quest mobile app / a developer org), + then `Settings → System → Developer → USB debugging`. +2. **adb over Wi-Fi** — frees the single USB-C port for the adapter: + ```sh + adb tcpip 5555 # once over USB, then unplug + adb connect :5555 + ``` + The Quest's adb rides its **internal Wi-Fi**; check the band with + `adb shell dumpsys wifi | grep frequencyMhz`. If it is 5 GHz (5180 = ch 36), + the experiment must run on 2.4 GHz — a 5 GHz flood self-desenses the USB + adapter inches away in the same chassis. +3. **Keep-awake** (off-head standby drops adb-Wi-Fi): + ```sh + adb shell am broadcast -a com.oculus.vrpowermanager.prox_close + adb shell svc power stayon true + ``` +4. Plug the **RTL8812AU** into the Quest's USB-C (via OTG). Confirm the kernel + enumerates it: `adb shell dumpsys usb | grep -A2 host_manager` should show + `manufacturer=3034 product=34834` (0x0bda:0x8812). + +## Build + install the APK + +```sh +# 1. arm64 rxdemo bundle (rxdemo + libusb1.0.so) +ANDROID_NDK=/opt/android-sdk/ndk/23.1.7779620 tests/build_android.sh +# 2. hand-build + sign the APK (no gradle) +ANDROID_SDK=/opt/android-sdk ANDROID_NDK=/opt/android-sdk/ndk/23.1.7779620 \ + RXQ_BUNDLE=/tmp/devourer-android/bundle tests/quest_apk/build_apk.sh +adb install -r /tmp/rxq-apk/build/rxq.apk +``` + +First launch requests USB permission once; tap **"Always open …"** + **OK** (or +drive it headlessly: `adb shell uiautomator dump`, then `adb shell input tap` the +`alwaysUse` checkbox and `button1`). Thereafter every attach auto-grants. + +## Run the matrix + +The RX (Quest) side is launched per-mode via `am start` extras; the TX flood + +analysis run on the desktop. One command does the lot: + +```sh +MODES="async reorder-pool spsc-fat" SINK_SPIN_US=500 DUR=20 \ + TX_VID=0x2357 TX_PID=0x0120 TX_GAP_US=200 \ + BURST_ON_MS=4 BURST_OFF_MS=30 POOL_SPARE=16 \ + tests/quest_rxq_run.sh +``` + +It keeps the headset awake, floods counter-stamped 6M QoS-Data, captures each +mode, pulls the JSONL, and prints per-mode delivery (from the 802.11 `seq_num` +sequence) + ring telemetry via `tests/quest_rxq_analyze.py`. + +**Link tips** (2.4 GHz ch 6): use a 2T2R flooder — the 8821AU 1T1R nano couples +poorly to the Quest adapter; use **6M OFDM** — 1M CCK from the flooder airs but +the Quest decodes ~0. The jaguar3 dies wedge on repeated soft-kill (need a +replug); the 8821AU (jaguar1) is kill-robust if it reaches the Quest. + +## Direct app control + +```sh +adb shell am start -n org.openipc.rxq/.MainActivity \ + --es tag run --es rxmode reorder-pool --es channel 6 \ + --es urbs 4 --es poolspare 16 --es sinkspin 500 +adb shell am start -n org.openipc.rxq/.MainActivity --es action stop +# output: /sdcard/Android/data/org.openipc.rxq/files/rxq_.jsonl ; status.txt +``` diff --git a/tests/quest_rxq_analyze.py b/tests/quest_rxq_analyze.py new file mode 100644 index 0000000..fbe95a6 --- /dev/null +++ b/tests/quest_rxq_analyze.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +# /// script +# requires-python = ">=3.9" +# /// +"""Analyse issue-330 Quest RX-ring captures: per-mode delivery (from the 802.11 +seq_num sequence, which increments per aired frame) plus ring-health telemetry +(cb_max_us consumer stalls, min_armed depth, resubmit_fail, empties). Compares +RX-ring servicing modes under the same TX flood — the differential that isolates +host-side ring loss from the (mode-invariant) RF baseline.""" +import glob +import json +import sys + + +def load(path): + seq, ring = [], [] + for line in open(path): + if '"ev":"rx.seq"' in line: + try: + seq.append(json.loads(line)) + except Exception: + pass + elif '"ev":"rx.ring"' in line: + try: + ring.append(json.loads(line)) + except Exception: + pass + return seq, ring + + +def delivery_from_seqnum(seq): + """802.11 seq_num counts every aired frame and wraps at 4096. Unfold the + wraps into a monotonic stream, then delivery = received / span.""" + sn = [e["seq"] for e in seq if "seq" in e] + if len(sn) < 2: + return None + unfolded, base, prev = [], 0, sn[0] + for v in sn: + if v < prev - 2048: # forward wrap + base += 4096 + elif v > prev + 2048: # backward blip (reorder) — ignore the jump + base -= 4096 + unfolded.append(base + v) + prev = v + lo, hi = min(unfolded), max(unfolded) + span = hi - lo + 1 + got = len(set(unfolded)) + return got, span, 100.0 * got / span if span else 0.0 + + +def ring_stats(ring): + if not ring: + return {} + cb = [r.get("cb_max_us", 0) for r in ring] + ma = [r.get("min_armed", 0) for r in ring] + comp = [r.get("completions", 0) for r in ring] + return { + "cb_max_us": max(cb), + "cb_p50_us": sorted(cb)[len(cb) // 2], + "min_armed": min(ma), + "resubmit_fail": max(r.get("resubmit_fail", 0) for r in ring), + "empties": max(r.get("empties", 0) for r in ring), + "completions": (max(comp) - min(comp)) if comp else 0, + } + + +def main(paths): + rows = [] + for p in sorted(paths): + seq, ring = load(p) + mode = p.split("rxq_m_")[-1].replace(".jsonl", "").replace("_", "-") \ + if "rxq_m_" in p else p.split("/")[-1] + d = delivery_from_seqnum(seq) + rs = ring_stats(ring) + rows.append((mode, len(seq), d, rs)) + + print(f"{'mode':<14}{'rx.seq':>8}{'got/span':>14}{'deliv%':>8}" + f"{'cbmax_us':>9}{'cbp50':>7}{'minArm':>7}{'resubF':>7}{'compl':>7}") + for mode, nseq, d, rs in rows: + gs = f"{d[0]}/{d[1]}" if d else "-" + dv = f"{d[2]:.1f}" if d else "-" + print(f"{mode:<14}{nseq:>8}{gs:>14}{dv:>8}" + f"{rs.get('cb_max_us','-'):>9}{rs.get('cb_p50_us','-'):>7}" + f"{rs.get('min_armed','-'):>7}{rs.get('resubmit_fail','-'):>7}" + f"{rs.get('completions','-'):>7}") + if len(rows) > 1 and all(r[2] for r in rows): + base = rows[0] + print(f"\nbaseline={base[0]} deliv={base[2][2]:.1f}%") + for mode, _, d, _ in rows[1:]: + print(f" {mode:<14} Δdeliv = {d[2] - base[2][2]:+.1f} pts") + + +if __name__ == "__main__": + args = sys.argv[1:] + paths = [x for a in args for x in glob.glob(a)] if args else [] + if not paths: + print("usage: quest_rxq_analyze.py ") + sys.exit(1) + main(paths) diff --git a/tests/quest_rxq_run.sh b/tests/quest_rxq_run.sh new file mode 100755 index 0000000..9eb4e90 --- /dev/null +++ b/tests/quest_rxq_run.sh @@ -0,0 +1,74 @@ +#!/usr/bin/env bash +# +# quest_rxq_run.sh — issue #330 confirmatory run on a Meta Quest 3 (constrained +# XR2 host) driven fully headlessly. The Quest hosts the RTL8812AU RX via the +# org.openipc.rxq auto-grant APK (fork()+exec of the devourer rxdemo with the +# USB fd); this desktop side floods a counter-stamped 6M QoS-Data stream from +# the 8812EU and, for each RX-ring servicing MODE, launches the Quest capture, +# pulls the JSONL, and computes seq_num delivery + ring telemetry. +# +# The link only closes with: TX=8812EU (0bda:a81a, not the 8821AU nano), 6M +# OFDM (not 1M CCK), on 2.4GHz ch6 (the Quest's adb-WiFi lives on 5GHz ch36, so +# 5GHz self-desenses). See kaeru issue330-quest-link-WORKS-6M-adb-band-collision. +set -euo pipefail + +MODES=${MODES:-"async reorder-pool spsc-fat"} +SINK_SPIN_US=${SINK_SPIN_US:-0} +DUR=${DUR:-20} +CH=${CH:-6} +TX_RATE=${TX_RATE:-6M} +TX_GAP_US=${TX_GAP_US:-800} +TX_PAYLOAD=${TX_PAYLOAD:-200} +BURST_ON_MS=${BURST_ON_MS:-0} # keyframe-burst model: flat-out for ON ms ... +BURST_OFF_MS=${BURST_OFF_MS:-0} # ... then idle for OFF ms +URBS=${URBS:-4} +POOL_SPARE=${POOL_SPARE:-8} +TX_VID=${TX_VID:-0x0bda}; TX_PID=${TX_PID:-0xa81a} # 8812EU flooder +BUILD=${BUILD:-./build} +PKG=org.openipc.rxq +FDIR=/storage/emulated/0/Android/data/$PKG/files +OUT=${OUT:-/tmp/quest-rxq/$(printf '%(%s)T' -1)} +mkdir -p "$OUT" + +TXLOG="$OUT/tx.log" +cleanup() { + sudo -n pkill -x -f "$BUILD/txdemo" 2>/dev/null || true + adb shell am start -n $PKG/.MainActivity --es action stop >/dev/null 2>&1 || true +} +trap cleanup EXIT INT TERM + +echo "[quest-rxq] modes=[$MODES] spin=${SINK_SPIN_US}us dur=${DUR}s ch=$CH rate=$TX_RATE gap=${TX_GAP_US}us out=$OUT" >&2 + +# keep the headset awake (off-head standby drops adb-WiFi) +adb shell am broadcast -a com.oculus.vrpowermanager.prox_close >/dev/null 2>&1 || true +adb shell svc power stayon true >/dev/null 2>&1 || true + +# start the steady TX flood (counter-stamped QoS-Data) +sudo -n env DEVOURER_VID=$TX_VID DEVOURER_PID=$TX_PID DEVOURER_CHANNEL=$CH \ + DEVOURER_TX_RATE=$TX_RATE DEVOURER_TX_QOS_DATA=1 DEVOURER_TX_QOS_NOACK=1 \ + DEVOURER_TX_PAYLOAD_BYTES=$TX_PAYLOAD DEVOURER_TX_GAP_US=$TX_GAP_US \ + DEVOURER_TX_BURST_ON_MS=$BURST_ON_MS DEVOURER_TX_BURST_OFF_MS=$BURST_OFF_MS \ + DEVOURER_EVENTS=off DEVOURER_LOG_LEVEL=warn "$BUILD/txdemo" >"$TXLOG" 2>&1 & +sleep 3 + +for mode in $MODES; do + tag="m_${mode//-/_}" + adb shell am start -n $PKG/.MainActivity --es action stop >/dev/null 2>&1 || true + sleep 3 + adb shell am start -n $PKG/.MainActivity \ + --es tag "$tag" --es rxmode "$mode" --es channel "$CH" \ + --es urbs "$URBS" --es poolspare "$POOL_SPARE" \ + --es sinkspin "$SINK_SPIN_US" >/dev/null 2>&1 + echo "[quest-rxq] mode=$mode capturing ${DUR}s ..." >&2 + sleep "$DUR" + adb shell am start -n $PKG/.MainActivity --es action stop >/dev/null 2>&1 || true + sleep 3 + adb pull "$FDIR/rxq_${tag}.jsonl" "$OUT/rxq_${tag}.jsonl" >/dev/null 2>&1 || \ + echo "[quest-rxq] WARN pull failed for $tag" >&2 +done + +sudo -n pkill -x -f "$BUILD/txdemo" 2>/dev/null || true +trap - EXIT INT TERM + +echo "[quest-rxq] analysis:" >&2 +python3 tests/quest_rxq_analyze.py "$OUT"/rxq_*.jsonl \ No newline at end of file diff --git a/tests/rxq_analyze.py b/tests/rxq_analyze.py new file mode 100644 index 0000000..400a4ab --- /dev/null +++ b/tests/rxq_analyze.py @@ -0,0 +1,222 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# dependencies = [] +# /// +"""Analyse an rxq_starve.sh capture for issue #330. + +Reads one or more capture directories (each holding rx.jsonl + meta.json) and +reports, per run: + + * delivery = distinct rx.seq counters received / the TX counter range they + span (ground truth from the payload counter — independent of tx.stats), + * the gap-width histogram (a host FIFO overflow drops a whole aggregate, so + its gaps cluster at k*pkt_cnt; an RF loss drops single frames, width ~1), + * the rx.ring armed-depth telemetry (min depth, fraction of windows the depth + collapsed to 0, worst inline-consume latency), + * the discriminator verdict: a loss cluster is HOST-STARVATION when the + armed depth collapsed to 0 in the same host-time window as the gap; it is + RF/other when the depth held up. + +Usage: tests/rxq_analyze.py DIR [DIR2 ...] + (multiple dirs print a comparison table — e.g. async vs a fix mode) +""" +import json +import os +import sys +from collections import Counter + + +def load(path): + rows = [] + if not os.path.exists(path): + return rows + with open(path) as f: + for line in f: + line = line.strip() + if not line or not line.startswith("{"): + continue + try: + rows.append(json.loads(line)) + except json.JSONDecodeError: + pass # a torn final line under FLUSH=0; skip it + return rows + + +def analyse(dirpath): + rx = load(os.path.join(dirpath, "rx.jsonl")) + meta = {} + mp = os.path.join(dirpath, "meta.json") + if os.path.exists(mp): + try: + meta = json.loads(open(mp).read().replace("\n", "")) + except json.JSONDecodeError: + pass + + seq = [e for e in rx if e.get("ev") == "rx.seq"] + ring = [e for e in rx if e.get("ev") == "rx.ring"] + + out = {"dir": dirpath, "meta": meta, "n_seq": len(seq), "n_ring": len(ring)} + if len(seq) < 2: + out["error"] = "too few rx.seq frames (link down / wrong SA / TX dead)" + return out + + # --- delivery from the payload counter -------------------------------- + ctrs = sorted({e["pctr"] for e in seq}) + lo, hi = ctrs[0], ctrs[-1] + span = hi - lo + 1 + got = len(ctrs) + out["pctr_lo"], out["pctr_hi"], out["span"], out["received"] = lo, hi, span, got + out["delivery"] = got / span if span else 0.0 + out["lost"] = span - got + + # --- gaps (runs of consecutive missing counters) ---------------------- + # keyed by t (host ms) of the frame that FOLLOWS the gap, so we can align + # each gap to the rx.ring depth telemetry. + t_by_ctr = {} + for e in seq: + t_by_ctr.setdefault(e["pctr"], e.get("t")) + gaps = [] # (width, t_after) + for i in range(1, len(ctrs)): + d = ctrs[i] - ctrs[i - 1] + if d > 1: + gaps.append((d - 1, t_by_ctr.get(ctrs[i]))) + out["n_gaps"] = len(gaps) + widths = Counter(w for w, _ in gaps) + out["gap_width_hist"] = dict(sorted(widths.items())) + out["gap_w1"] = widths.get(1, 0) + out["gap_wmulti"] = sum(c for w, c in widths.items() if w > 1) + + # --- ring depth telemetry -------------------------------------------- + # empties/completions is the primary, poll-loop-independent signal: the + # fraction of URB completions that drained the ring to zero posted URBs. + # RF loss leaves the ring armed (frames don't arrive) -> ~0; host + # starvation drains it (frames out-race resubmit) -> high. Catastrophic RF + # ALSO makes wide gaps, so gap width alone can't be the trigger — the empty + # rate is what separates "consumer/pump was the bottleneck" from "the link + # was". + if ring: + out["cb_max_us"] = max(r.get("cb_max_us", 0) for r in ring) + out["resubmit_fail"] = max(r.get("resubmit_fail", 0) for r in ring) + comp = max((r.get("completions", 0) for r in ring), default=0) + emp = max((r.get("empties", 0) for r in ring), default=0) + out["completions"], out["empties"] = comp, emp + out["empty_rate"] = emp / comp if comp else 0.0 + out["ring_min"] = min((r.get("min_armed", -1) for r in ring), default=-1) + # collapse intervals for the secondary gap-coincidence corroboration + collapse_iv = [(ring[i - 1].get("t", 0) if i else 0, r.get("t", 0)) + for i, r in enumerate(ring) if r.get("min_armed", -1) == 0] + else: + collapse_iv = [] + out["cb_max_us"] = out["resubmit_fail"] = out["ring_min"] = None + out["completions"] = out["empties"] = 0 + out["empty_rate"] = None + + def in_collapse(t): + return t is not None and any(a <= t <= b for a, b in collapse_iv) + + out["gaps_in_collapse"] = sum(1 for _, t in gaps if in_collapse(t)) + + # --- stall-window attribution (RF-robust) ----------------------------- + # An injected consumer stall shows as a >=STALL_MS gap in the rx.seq host-t + # stream; the pctr jump ACROSS that gap is the loss that stall caused. RF + # loss is time-uniform, so it lands in the non-stall remainder. This + # separates the host effect from a high RF floor without needing a clean + # link: spsc-fat keeps the pump armed during a consumer stall (frames queued, + # no pctr jump) while async/reorder drop them (big jump). + STALL_T_MS = 12 + seq_t = [(e.get("t"), e["pctr"]) for e in seq if e.get("t") is not None] + seq_t.sort() + stall_loss = stalls = 0 + for i in range(1, len(seq_t)): + dt = seq_t[i][0] - seq_t[i - 1][0] + if dt >= STALL_T_MS: + stalls += 1 + stall_loss += max(0, seq_t[i][1] - seq_t[i - 1][1] - 1) + out["stalls"] = stalls + out["stall_loss"] = stall_loss + out["nonstall_loss"] = max(0, out["lost"] - stall_loss) + + # Single-run verdict. cb_max_us is the robust consumer-stall signal: a + # healthy parser consumes in ~100-200us, so a cb_max of milliseconds means + # the pump thread stalled in on_data and could not re-arm the ring — the + # host was the bottleneck, not the link. (RF loss leaves cb_max small.) + # This catches the injected-spin model; pump PREEMPTION with a fast consumer + # (e.g. the Quest compositor) shows small cb_max and is proven instead by + # the DIFFERENTIAL below: same RF, delivery responds to the host lever. + CB_STALL_US = 5000 + cbmax = out.get("cb_max_us") or 0 + if out["lost"] == 0: + out["verdict"] = "CLEAN" + elif cbmax >= CB_STALL_US: + out["verdict"] = "HOST_STARVATION" + else: + out["verdict"] = "RF_OR_OTHER" + return out + + +def fmt(o): + if "error" in o: + return f" {os.path.basename(o['dir'])}: {o['error']} (n_seq={o['n_seq']})" + m = o.get("meta", {}) + cfg = (f"mode={m.get('mode','?')} urbs={m.get('urbs','?')} " + f"spin={m.get('sink_spin_us','?')}us " + f"burst={m.get('burst_on_ms','?')}/{m.get('burst_off_ms','?')}ms " + f"busy={m.get('busy_threads','?')} taskset={m.get('taskset','') or '-'}") + lines = [ + f" {os.path.basename(o['dir'])} [{cfg}]", + f" delivery {o['delivery']*100:6.2f}% " + f"received {o['received']}/{o['span']} lost {o['lost']}", + f" gaps {o['n_gaps']} (width1={o['gap_w1']} multi={o['gap_wmulti']}) " + f"hist={o['gap_width_hist']}", + ] + if o.get("empty_rate") is not None: + lines.append( + f" ring empty_rate={o['empty_rate']*100:5.1f}% " + f"(empties {o['empties']}/{o['completions']}) " + f"min_armed={o['ring_min']} cb_max={o['cb_max_us']}us " + f"resubmit_fail={o['resubmit_fail']}") + if o.get("stalls"): + lines.append( + f" stalls {o['stalls']} detected " + f"stall_loss={o['stall_loss']} nonstall(RF)_loss={o['nonstall_loss']}") + lines.append(f" VERDICT {o['verdict']}") + return "\n".join(lines) + + +def main(argv): + dirs = argv[1:] + if not dirs: + print(__doc__) + return 2 + results = [analyse(d) for d in dirs] + for o in results: + print(fmt(o)) + # machine-readable one-liner for scripting + slim = {k: o.get(k) for k in ( + "delivery", "received", "span", "lost", "n_gaps", "gap_w1", + "gap_wmulti", "ring_min", "empty_rate", "empties", "completions", + "cb_max_us", "resubmit_fail", "gaps_in_collapse", "verdict")} + slim["ev"] = "rxq.result" + slim["dir"] = o["dir"] + print(json.dumps(slim)) + + # --- differential: host-induced loss vs the best-delivery run ---------- + # With the RF link held constant across runs, the drop in delivery from the + # best run is loss the HOST caused (ring depth / consumer cost / mode). This + # is the definitive proof — cleaner than any single-run classifier. + ok = [o for o in results if "delivery" in o] + if len(ok) >= 2: + base = max(ok, key=lambda o: o["delivery"]) + print(f"\n DIFFERENTIAL (baseline = {os.path.basename(base['dir'])}, " + f"{base['delivery']*100:.2f}% delivery):") + for o in ok: + d = (base["delivery"] - o["delivery"]) * 100 + tag = "" if o is base else f" host-induced loss +{d:.2f} pts" + print(f" {os.path.basename(o['dir']):28s} {o['delivery']*100:6.2f}%" + f" cb_max={o.get('cb_max_us')}us{tag}") + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv)) diff --git a/tests/rxq_preempt.c b/tests/rxq_preempt.c new file mode 100644 index 0000000..853b6cd --- /dev/null +++ b/tests/rxq_preempt.c @@ -0,0 +1,53 @@ +/* rxq_preempt — an intermittent, real-time-priority CPU preemptor that models a + * VR compositor (or any high-priority periodic task) starving a background USB + * pump thread. Pins itself to CORE, raises to SCHED_FIFO, then repeats: + * busy-hog for ON_US, sleep for OFF_US. During each hog the FIFO priority fully + * preempts the normal-priority devourer RX pump on that core — reproducing the + * transient ring starvation of issue #330 on a host that otherwise has CPU to + * spare. SCHED_FIFO needs privilege (run under sudo); it falls back to a plain + * busy-hog (weaker, ~50%% share) if setscheduler fails. + * + * Build: cc -O2 -o build/rxq_preempt tests/rxq_preempt.c -lpthread + * Usage: rxq_preempt CORE ON_US OFF_US + */ +#define _GNU_SOURCE +#include +#include +#include +#include + +static long now_us(void) { + struct timespec t; + clock_gettime(CLOCK_MONOTONIC, &t); + return t.tv_sec * 1000000L + t.tv_nsec / 1000; +} + +int main(int argc, char **argv) { + if (argc < 4) { + fprintf(stderr, "usage: rxq_preempt CORE ON_US OFF_US\n"); + return 2; + } + int core = atoi(argv[1]); + long on_us = atol(argv[2]); + long off_us = atol(argv[3]); + + cpu_set_t set; + CPU_ZERO(&set); + CPU_SET(core, &set); + if (sched_setaffinity(0, sizeof(set), &set) != 0) + perror("sched_setaffinity"); + + struct sched_param sp = {.sched_priority = 50}; + if (sched_setscheduler(0, SCHED_FIFO, &sp) != 0) + perror("sched_setscheduler(FIFO) — falling back to normal priority"); + + volatile unsigned long sink = 0; + for (;;) { + long deadline = now_us() + on_us; + while (now_us() < deadline) + sink += 1; /* burn the core at RT priority */ + struct timespec ts = {off_us / 1000000, (off_us % 1000000) * 1000}; + nanosleep(&ts, NULL); + } + return (int)sink; /* unreachable */ +} diff --git a/tests/rxq_starve.sh b/tests/rxq_starve.sh new file mode 100755 index 0000000..994c966 --- /dev/null +++ b/tests/rxq_starve.sh @@ -0,0 +1,163 @@ +#!/usr/bin/env bash +# +# rxq_starve.sh — one RX-ring starvation experiment run for issue #330. +# +# Floods a TX adapter with counter-stamped QoS-Data frames (optionally bursty to +# emulate video keyframes) and captures the RTL8812AU RX under a chosen RX-ring +# servicing mode + injected inline-consumer cost + host CPU constraint. Emits a +# JSONL capture (rx.ring depth telemetry + rx.seq per-frame counters + tx.stats) +# for tests/rxq_analyze.py. +# +# The point the desktop rig missed before: at realistic consumer cost a fast +# xhci host has too much slack to starve. The faithful self-driven lever is to +# SHRINK THE RING (RX_URBS=1..2) so a slow inline consume collapses the armed +# depth immediately — mimicking the phone's shallow dwc3 controller — combined +# with a CPU clamp. See docs/experiments/issue-330-rx-ring-starvation.md. +# +# All config is env-overridable; run one config per invocation. Example: +# SINK_SPIN_US=1500 RX_URBS=2 RX_MODE=async BURST_ON_MS=8 BURST_OFF_MS=40 \ +# RX_TASKSET=0 tests/rxq_starve.sh +# +set -euo pipefail + +BUILD=${BUILD:-./build} +OUTDIR=${OUTDIR:-/tmp/rxq/$(printf '%(%s)T' -1)} +SUDO=${SUDO:-} # set to "sudo" if libusb needs privilege + +# --- adapters (VID/PID pick the device; different VIDs need no topology hint) -- +TX_VID=${TX_VID:-0x2357}; TX_PID=${TX_PID:-0x0120} # 8821AU flooder (Jaguar1) +RX_VID=${RX_VID:-0x0bda}; RX_PID=${RX_PID:-0x8812} # 8812AU DUT (the reporter chip) + +# --- RF / link --- +CH=${CH:-6} +TX_RATE=${TX_RATE:-MCS4} +TX_PWR_OFFSET_QDB=${TX_PWR_OFFSET_QDB:--40} # -10 dB: de-saturate near field +TX_PAYLOAD_BYTES=${TX_PAYLOAD_BYTES:-256} + +# --- stimulus --- +# NB: gap 0 (max flood) wedges the Jaguar1 async TX path — keep a modest floor; +# a high finite rate already over-subscribes a slow RX consumer. +TX_GAP_US=${TX_GAP_US:-200} # ~5000 fps ceiling per sender +BURST_ON_MS=${BURST_ON_MS:-0} +BURST_OFF_MS=${BURST_OFF_MS:-0} + +# --- RX ring / consumer / telemetry --- +RX_MODE=${RX_MODE:-async} +RX_URBS=${RX_URBS:-8} +RX_URB_BYTES=${RX_URB_BYTES:-16384} +SINK_SPIN_US=${SINK_SPIN_US:-0} +RING_MS=${RING_MS:-50} + +# --- host constraint on the RX process --- +RX_TASKSET=${RX_TASKSET:-} # e.g. "0" pins the pump to cpu0 +RX_NICE=${RX_NICE:-} # e.g. "19" +RX_CHRT=${RX_CHRT:-} # e.g. "idle" (SCHED_IDLE) — deepest deprio +BUSY_THREADS=${BUSY_THREADS:-0} # continuous CPU hogs (permanently-loaded SoC) +# Intermittent RT-priority preemptor (faithful VR-compositor model): hog the RX +# core for PREEMPT_ON_MS every PREEMPT_ON_MS+OFF_MS. This is the TRANSIENT +# starvation the reporter sees on keyframes — the ring recovers between hogs. +PREEMPT_ON_MS=${PREEMPT_ON_MS:-0} +PREEMPT_OFF_MS=${PREEMPT_OFF_MS:-0} +PREEMPT_CORE=${PREEMPT_CORE:-${RX_TASKSET:-0}} + +DUR=${DUR:-15} + +mkdir -p "$OUTDIR" +RX_JSONL="$OUTDIR/rx.jsonl"; RX_ERR="$OUTDIR/rx.err" +TX_ERR="$OUTDIR/tx.err"; META="$OUTDIR/meta.json" + +TX_PIDF=""; RX_PIDF=""; PREEMPT_PIDF=""; BUSY_PIDS=() +cleanup() { + set +e + [ -n "$RX_PIDF" ] && kill "$RX_PIDF" 2>/dev/null + [ -n "$TX_PIDF" ] && kill "$TX_PIDF" 2>/dev/null + [ -n "$PREEMPT_PIDF" ] && $SUDO kill "$PREEMPT_PIDF" 2>/dev/null + for p in "${BUSY_PIDS[@]:-}"; do [ -n "$p" ] && kill "$p" 2>/dev/null; done + # exact-comm backstops (never a broad pkill) + pkill -x -f "$BUILD/txdemo" 2>/dev/null + pkill -x rxq-busy 2>/dev/null + $SUDO pkill -x rxq_preempt 2>/dev/null + wait 2>/dev/null +} +trap cleanup EXIT INT TERM + +echo "[rxq] out=$OUTDIR mode=$RX_MODE urbs=$RX_URBS spin=${SINK_SPIN_US}us burst=${BURST_ON_MS}/${BURST_OFF_MS}ms taskset=${RX_TASKSET:-none}" >&2 + +cat > "$META" <&2; exit 1; } + fi + $SUDO "$BUILD/rxq_preempt" "$PREEMPT_CORE" "$((PREEMPT_ON_MS*1000))" \ + "$((PREEMPT_OFF_MS*1000))" >/dev/null 2>&1 & + PREEMPT_PIDF=$! + echo "[rxq] preemptor: core $PREEMPT_CORE hog ${PREEMPT_ON_MS}ms / idle ${PREEMPT_OFF_MS}ms (FIFO)" >&2 +fi + +# --- TX flood: counter-stamped QoS-Data, optional keyframe bursts ------------- +TXENV=( DEVOURER_VID="$TX_VID" DEVOURER_PID="$TX_PID" + DEVOURER_CHANNEL="$CH" DEVOURER_TX_RATE="$TX_RATE" + DEVOURER_TX_QOS_DATA=1 DEVOURER_TX_QOS_NOACK=1 + DEVOURER_TX_PWR_OFFSET_QDB="$TX_PWR_OFFSET_QDB" + DEVOURER_TX_PAYLOAD_BYTES="$TX_PAYLOAD_BYTES" + DEVOURER_TX_GAP_US="$TX_GAP_US" + DEVOURER_TX_BURST_ON_MS="$BURST_ON_MS" DEVOURER_TX_BURST_OFF_MS="$BURST_OFF_MS" + DEVOURER_LOG_LEVEL=warn DEVOURER_EVENTS=stdout DEVOURER_EVENT_FLUSH=0 ) +TX_JSONL="$OUTDIR/tx.jsonl" +$SUDO env "${TXENV[@]}" "$BUILD/txdemo" >"$TX_JSONL" 2>"$TX_ERR" & +TX_PIDF=$! +sleep 2 # let the flood come up before the RX starts counting + +# --- RX capture: ring telemetry + per-frame counters + injected consumer cost - +RXENV=( DEVOURER_VID="$RX_VID" DEVOURER_PID="$RX_PID" DEVOURER_CHANNEL="$CH" + DEVOURER_RX_MODE="$RX_MODE" DEVOURER_RX_URBS="$RX_URBS" + DEVOURER_RX_URB_BYTES="$RX_URB_BYTES" DEVOURER_RX_RING_MS="$RING_MS" + DEVOURER_RX_PCTR=1 DEVOURER_RX_AGG_SA=canon + DEVOURER_RX_SINK_SPIN_US="$SINK_SPIN_US" + DEVOURER_RX_SINK_STALL_MS="${STALL_MS:-0}" + DEVOURER_RX_SINK_STALL_EVERY="${STALL_EVERY:-100}" + DEVOURER_RX_POOL_SPARE="${RX_POOL_SPARE:-0}" + DEVOURER_RX_ENERGY_MS="${RX_ENERGY_MS:-500}" DEVOURER_LINKHEALTH=1 + DEVOURER_EVENT_FLUSH=0 DEVOURER_LOG_LEVEL=info ) +PREFIX=() +[ -n "$RX_CHRT" ] && PREFIX=(chrt --"$RX_CHRT" 0) +[ -n "$RX_NICE" ] && PREFIX=(nice -n "$RX_NICE" "${PREFIX[@]}") +[ -n "$RX_TASKSET" ] && PREFIX=(taskset -c "$RX_TASKSET" "${PREFIX[@]}") + +$SUDO env "${RXENV[@]}" "${PREFIX[@]}" "$BUILD/rxdemo" >"$RX_JSONL" 2>"$RX_ERR" & +RX_PIDF=$! + +sleep "$DUR" +kill "$RX_PIDF" 2>/dev/null; RX_PIDF="" +kill "$TX_PIDF" 2>/dev/null; TX_PIDF="" +sleep 0.5 + +# --- quick summary (full analysis in tests/rxq_analyze.py) -------------------- +ring=$(grep -c '"ev":"rx.ring"' "$RX_JSONL" 2>/dev/null || echo 0) +seq=$(grep -c '"ev":"rx.seq"' "$RX_JSONL" 2>/dev/null || echo 0) +hit=$(grep -c '"ev":"rx.seq"' "$RX_JSONL" 2>/dev/null || echo 0) +txsub=$(grep '"ev":"tx.stats"' "$TX_JSONL" 2>/dev/null | tail -1 | grep -oE '"submitted":[0-9]+' | cut -d: -f2 || echo "?") +lh=$(grep '"ev":"link.health"' "$RX_JSONL" 2>/dev/null | tail -1 | grep -oE '"verdict":"[A-Z_]+"' | cut -d'"' -f4 || echo "?") +echo "[rxq] captured: rx.ring=$ring rx.seq=$seq tx.submitted=$txsub link=$lh -> $RX_JSONL" >&2 +grep -m1 '"ev":"rx.ring"' "$RX_JSONL" 2>/dev/null >&2 || true +grep '"ev":"link.health"' "$RX_JSONL" 2>/dev/null | tail -1 >&2 || true +echo "$OUTDIR"