diff --git a/Cargo.lock b/Cargo.lock index 83555451d05..e3231c03ebe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7048,6 +7048,7 @@ name = "qmux" version = "0.7.0" dependencies = [ "anyhow", + "bench-harness", "bytes", "futures", "kio", diff --git a/bench/README.md b/bench/README.md index 3df19a59dbf..6566fa19ad9 100644 --- a/bench/README.md +++ b/bench/README.md @@ -172,6 +172,27 @@ Both ends share one current-thread runtime, so its CPU covers sending and receiving alike. It is a local profiling tool rather than a nightly series: the `moq-net` session bench owns qmux's series. +Measure qmux receive ownership at the relay client defaults (4 MiB connection, +1 MiB stream, 4 MiB send): + +```bash +nix develop --command cargo bench --locked --manifest-path rs/qmux/Cargo.toml --bench receive +``` + +`qmux_receive` sweeps TCP and WebSocket framing, every qmux version, 1/8/64 +sessions, and 1/8 streams per session. Both endpoints run on one thread over +bounded in-memory byte pipes; this isolates framing and receive ownership from +TLS, kernel sockets, and congestion. Each stream holds an unread 512 KiB lease, +so eight together fill the connection window. The untimed output reports live +requested heap bytes, including both endpoints, their tasks, the pipes, stream +tables and receive backings, per session. The shared source payload and runtime +are excluded. This measures retained allocations, not RSS; allocator counters +also run during timing. Every timed round reads and checks all payload bytes, +returns lease credit, and fills each stream again. Throughput counts delivered +payload once. Nightly's supplemental Rust benchmark leg records the series and +keeps the memory output with its raw results. The broader MoQ session transport +and version sweep remains in [its quest](/quest/m2/bench-transport-sweep.md). + Compare one multi-threaded Tokio runtime with the same number of independent Tokio/epoll and io\_uring workers: diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 0cb327cabef..ca6ad8db3bd 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -395,8 +395,10 @@ Under light congestion video drops the tail of a group; under heavy congestion video stops and audio lags by at most 500 ms. No protocol change, just knobs. A long group is paced by its fastest reader instead. Each hop receives at most -16 MB of a group past its fastest reader, then stops reading the group's stream, -so stream flow control pushes back hop by hop to the publisher, and a reader +its stream receive window of a group past its fastest reader, capped at 16 MB +(half the group's tail), then stops reading the group's stream, so the hop +holds about what its upstream peer can have in flight on that stream and +stream flow control pushes back hop by hop to the publisher. A reader that keeps up loses nothing. A slower reader is never waited on: one that trails the fastest by more than the group's 32 MB tail reaches a shed frame, and asks for a new group on tracks that opt into [refresh](#refresh) or ends with diff --git a/quest/m1/release-perf/README.md b/quest/m1/release-perf/README.md index ffabf4712d9..9cdddf53cd5 100644 --- a/quest/m1/release-perf/README.md +++ b/quest/m1/release-perf/README.md @@ -315,6 +315,7 @@ subscribers. ## Required - [Relay CPU per frame since #5926](/quest/m1/release-perf/relay-cpu-bisect.md) - the 18-56% rise in relay instructions per delivered frame is bisected, attributed, and its avoidable cost fixed +- [Pacing's per-frame report](/quest/m1/release-perf/pacing-report.md) - relay fan-out reads with pacing are within noise of the pre-pacing parent ## Related diff --git a/quest/m1/release-perf/pacing-report.md b/quest/m1/release-perf/pacing-report.md new file mode 100644 index 00000000000..86fafc27d59 --- /dev/null +++ b/quest/m1/release-perf/pacing-report.md @@ -0,0 +1,22 @@ +# [S] Pacing's per-frame report costs relay fan-out nothing measurable + +## Goal + +Relay fan-out reads (`group_relay_frames`) with fastest-reader pacing are +within noise of the pre-pacing parent for groups of 512 or more frames, +where today they are 4.6-8.1% slower, with pacing still bounding relay +memory exactly as before. + +## Plan + +Planned 2026-11-04 (maintainer): pacing's cost was accepted to finish +[fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664), with this +trim as its follow-up. Measured in that PR, six rounds on a loaded host: +[moq-dev/moq#7388](https://github.com/moq-dev/moq/pull/7388), which added +pacing, against its parent `b51eb8c` is 4.6-8.1% slower on +`group_relay_frames` in all 12 cases with 512+ frames (64-byte payloads, the +worst case for a per-frame cost); 4-frame groups and `group_read_frames` are +unchanged. The likely cause is the per-read progress report #7388 added to +the batch read path; profile it first (`just bench-profile instructions`, as +[the relay CPU bisect](/quest/m1/release-perf/relay-cpu-bisect.md) does), +then report progress per batch or only when the pacing window can change. diff --git a/quest/m1/release-perf/relay-cpu-bisect.md b/quest/m1/release-perf/relay-cpu-bisect.md index e4f6313f667..c83840acfca 100644 --- a/quest/m1/release-perf/relay-cpu-bisect.md +++ b/quest/m1/release-perf/relay-cpu-bisect.md @@ -27,6 +27,8 @@ Decided 2026-11-03 (maintainer, on [moq-dev/moq#7605](https://github.com/moq-dev `Cursor::probe` is the relay's check for groups a reader has missed. Each time a subscriber asks for its next group, the probe looks at what arrived on the track since its last look and decides whether any group the reader still wants has left the cache without the live feed delivering it (evicted, skipped, or from before the feed started), and asks upstream to fill those. A live reader that takes each group as it arrives never has such a gap, yet it paid for the whole check on every group: locking the subscription and the track's shared state, cloning the track's fills, walking the arrival ledger, checking each new group for holes, and running the ask-and-answer loop. On top of that, each time the cache aged out an old group, the eviction landed with every reader, all of which had already taken it, and forced each one to re-run the full check. +Added 2026-11-04 (maintainer) from [fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664)'s comparison on a loaded host: `main` at `f37d964` reads `group_relay_frames` about 15% slower than `e7a5c10`. Pacing ([moq-dev/moq#7388](https://github.com/moq-dev/moq/pull/7388)) explains about 4% of that, and [its own quest](/quest/m1/release-perf/pacing-report.md) trims it; attribute the remaining ~10% here, on a quiet host. + Follow-up planning on 2026-11-03 (maintainer) added a load-independent mode to the driver: `just bench-fanout` counts the chat fan-out per delivered group on one pinned thread, through the model alone and through a relay over the in-memory transport, for any list of revisions. [The report](/.github/relay-cpu-bisect.md) attributes each step with it, and the fixes so far take `main`'s model path from 13,300 to 10,126 instructions per group (-23.9%) and the in-process session from 120,948 to 113,133 (-6.5%), at 512 subscribers. Decided 2026-11-03 (maintainer, on [moq-dev/moq#7605](https://github.com/moq-dev/moq/pull/7605)), after the attribution: diff --git a/quest/m1/socket/README.md b/quest/m1/socket/README.md index b6bdb4f0136..fe85508243a 100644 --- a/quest/m1/socket/README.md +++ b/quest/m1/socket/README.md @@ -76,7 +76,7 @@ pipe below; the second round above adds the acceptor to the session name: - Long-lived `tcp` sessions rely on unbounded groups (since [moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411) a group has no size cap, and each copy keeps a ~32 MiB tail). With one reader per copy, - [fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md) acts + [fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664) acts as TCP backpressure: a writer awaiting `group::Producer::ready` runs at most 16 MiB past its reader on each hop, and a reader that falls off the tail fails with `Evicted` rather than skipping data. Where pacing doesn't reach diff --git a/quest/m1/unbounded-groups/README.md b/quest/m1/unbounded-groups/README.md index 312bcb3a9af..526fe629144 100644 --- a/quest/m1/unbounded-groups/README.md +++ b/quest/m1/unbounded-groups/README.md @@ -32,7 +32,7 @@ Decided 2026-10-28 (maintainer, after an audit of the plan and shedding on pool debt alone let one live group grow without bound once the group caps went. The 32 MB counts a group copy's live tail (its `kio::log` run), and nothing is kept below it: no head, no refilled runs. -- With [fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md), +- With [fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664), the 32 MB is one budget split between "ahead" (bytes the fastest reader has not read, bounded by pacing) and "behind" (bytes it has read, kept for slower readers). Shedding never passes the fastest reader, and the ahead @@ -76,7 +76,7 @@ superseded 2026-10-28 by flights.) When nothing serves a hole, the reader gets `Evicted` and a joiner falls back to `Info::refresh`. The live edge is paced by its fastest reader (decided 2026-10-25, superseding 2026-10-24's "never paced by readers"; see -[Fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md)), so a +[Fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664)), so a byte stream over MoQ loses nothing to a reader that keeps up. Groups are bounded by bytes, with each frame's slot charged to the pool and the 32 MB tail. Front eviction landed in [moq-dev/moq#6993](https://github.com/moq-dev/moq/pull/6993) @@ -155,7 +155,6 @@ covered only at the model layer (maintainer, 2026-10-30). ## Required -- [Fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md) - every hop buffers at most a window past its fastest reader and pushes back through QUIC flow control - [Flights](/quest/m1/unbounded-groups/flights.md) - a reader at a shed frame replays through a separate paced copy that hands over to the copy ahead - [Byte holes](/quest/m1/unbounded-groups/byte-holes.md) - segments inside a frame shed like front frames and flights start at `(frame, byte)` on the current lite wip; the frame cap goes - [JS byte holes](/quest/m1/unbounded-groups/js-byte-holes.md) - `@moq/net` mirrors the byte-offset wire and continues a frame cut off mid-way diff --git a/quest/m1/unbounded-groups/fastest-reader.md b/quest/m1/unbounded-groups/fastest-reader.md deleted file mode 100644 index 7182f21d293..00000000000 --- a/quest/m1/unbounded-groups/fastest-reader.md +++ /dev/null @@ -1,217 +0,0 @@ -# [M] Fastest-reader pacing - -## Goal - -Every hop buffers at most a window of a group past its fastest reader, inside -the group's 32 MB tail, then stops reading the upstream QUIC stream, so stream -flow control pushes back hop by hop to the publisher. A hop with no reader -keeps reading and evicting, so pacing never blocks an unwatched track. A reader -that falls off the tail boards a -[flight](/quest/m1/unbounded-groups/flights.md) when a fill handler exists, and -otherwise refreshes or ends with `Error::Evicted`. A byte stream over MoQ then -loses nothing to a reader that keeps up, and each hop keeps its own congestion -control and retransmissions. - -The paced hops include qmux sessions (WebSocket, `tcp://`, `tls://`, -`unix://`, and WebRTC): they lease receive credit natively, so incomplete -messages are bounded by the session's receive window and the lowest-priority -holder is evicted when it is stuck, as on native QUIC. The per-session memory -a peer can make us hold is one deliberate, documented number across native -QUIC and qmux. The browser and JS keep reading unpaced; their pacing is -[JS fastest-reader pacing](/quest/m2/js-fastest-reader.md), in m2. - -## Plan - -Decided 2026-10-25 (maintainer, mesh planning interview), reopening the -2026-10-24 "the live edge is never paced by a reader" decision for all tracks: - -- Pace by the fastest reader, never the slowest: one slow reader must not stall - the publisher or its other readers. Rejected: pacing by the slowest reader - (TCP fanout semantics), and an opt-in flow-controlled track flag. -- Propagate with QUIC stream flow control per hop. No new wire messages, so it - lands in the m1 cut beside front eviction - ([moq-dev/moq#6993](https://github.com/moq-dev/moq/pull/6993)). - Rejected: explicit MoQ-level credit messages. -- A reader past the window gets no new mechanism: it is an ordinary front - eviction victim. Tracks without a fill handler, such as - [byte streams](/quest/m2/stream/README.md), end that reader with - `Evicted`, which a TCP proxy surfaces as a reset. - -Decided 2026-10-28 (maintainer, after an audit of the plan): - -- One 32 MB budget per group: the live tail that front eviction - (`GroupState::trim` in `rs/moq-net/src/model/group.rs`) holds is - "ahead" (bytes the fastest reader has not read, bounded by pacing) plus - "behind" (bytes it has read, kept for slower readers). Shedding never - passes the fastest reader, and the ahead window is at most 16 MB, half - the tail, so a writer more than the window ahead waits instead of - shedding unread bytes, and at least 16 MB always stays behind. - The behind window then slides at the fastest reader's pace with no pacing - of its own: a reader falls off once it trails the fastest by more than - 32 MB minus the ahead window. The ahead window is each hop's QUIC stream - receive window, capped at 16 MB. The cap keeps - [flight](/quest/m1/unbounded-groups/flights.md) admission sound: a reader - admitted within the behind half is not shed when outstanding credit fills - the ahead half. -- Reader progress is measured in bytes, not whole frames. Otherwise a frame - larger than the ahead window deadlocks a writer against a reader that - advances only on a finished frame. Until - [byte holes](/quest/m1/unbounded-groups/byte-holes.md) gives Rust a - chunked read, a whole-frame reader must not stall a paced writer. -- An origin producer that can wait (a byte stream) awaits a readiness future - that stays pending until a reader exists and the ahead window has room. - Live encoders ignore it and keep writing. Without the reader condition, a - dialer writing before its listener subscribes would shed its first bytes - once they passed 32 MB, and the stream would start with a hole. Relays - keep the "no reader keeps reading and evicting" rule above, since they only - hold groups some reader asked for. -- Cap the receive credit paused streams can hold. A paused stream keeps up to - its stream window counted against the connection window, so enough paused - groups (one per track with a slow reader) stall unrelated tracks on the - same connection and trigger moq-quic's receive eviction - (`CloseKind::Evicted`), which could reset the very byte stream pacing keeps - lossless. Bring the mechanism to the maintainer before coding, and align it - with the [receive pool](/quest/m2/receive-pool.md). -- Folded in 2026-11-01 (maintainer, follow-up of - [moq-dev/moq#7328](https://github.com/moq-dev/moq/pull/7328)): the same cap - covers streams parked waiting for TRACK_INFO. Group data that a SUBSCRIBE - sent alongside TRACK, or a FETCH, leaves unread until TRACK_INFO arrives - must not use up the connection receive window, or TRACK_INFO itself can't - arrive. TRACK_INFO and other control messages always get connection - credit, in Rust and JS. Test it: enough parked groups to fill the - connection window, with TRACK_INFO still arriving and the groups then read. - -Decided 2026-10-30 (maintainer, in the -[moq-dev/moq#7388](https://github.com/moq-dev/moq/pull/7388) iteration): - -- moq-net caps the receive credit paused groups hold per session at about half - the connection window, read live from the transport - (`poll::Session::receive_window`). A group that would push the total past - the cap keeps reading and shedding instead of pausing. This stands until the - [receive pool](/quest/m2/receive-pool.md) folds paused credit into its - per-worker budget. -- The readiness future is `group::Producer::poll_ready` and `async fn ready()`, - pending until a reader exists and the 16 MB window has room. - -Prior art splits the same way (2026-10-28 survey): Varnish streaming, ATS -read-while-writer, and WHATWG `tee()` fan out at the writer's pace and buffer -everything; Varnish `transit_buffer`, nginx unbuffered proxying, and ATS flow -control pace a single client; Node `pipe()` to several destinations paces the -slowest. This is `tee()` with a bounded buffer that drops the branch that -falls off it ([whatwg/streams#1235](https://github.com/whatwg/streams/issues/1235)). - -Landed in [moq-dev/moq#7388](https://github.com/moq-dev/moq/pull/7388) (Rust, -session ingest only): readers report the byte end of each frame they take -(`Pace` in `rs/moq-net/src/model/pace.rs`), and the shared transport reader's -payload loop (`coding::Reader::poll_read_frame`) waits on -`group::Producer::poll_paced` before taking more, and admits a frame only once -its declared size fits the window (`coding::Reader::poll_admit`), so moq-lite -and moq-transport ingest both pause past the 16 MB window. A reader that took every -committed frame counts as waiting on the writer, so a frame larger than the -window never stalls it. A paused stream is not read, so it is watched for a -reset instead (`poll::RecvStream::poll_reset`, which moq-quic answers through -moq-tokio and moq-uring without taking bytes). Tests cover one hop on lite-06, -lite-07-wip, and moq-transport-22: an idle reader holds the relay copy near the -window and then receives every frame, a slow reader ends `Evicted` while the -fast one keeps up, a session past its credit cap reads on, and a publisher's -reset ends a paused relay copy. The same PR landed the credit -cap (each paused stream reserves its stream window from a per-session budget -in `rs/moq-net/src/transport/boxed.rs`; moq-quic reports its windows through -moq-tokio and moq-uring, and a transport that reports none never pauses) and -the public readiness future. - -Decided 2026-11-02 (maintainer, /quest-grill interview on -[moq-dev/moq#7624](https://github.com/moq-dev/moq/pull/7624)): - -- Native only: finish the moq-quic path (tokio, io_uring, and qmux on its - mux). Browser parity waits on a real receive-credit contract for - WebTransport, since `Reader.closed` watches resets but nothing says how much - credit a paused stream holds; guessed window constants would move memory - outside flow control. Rejected: a browser read-ahead cap, and dropping the - browser from scope. -- qmux is handled here, absorbing the m2 qmux receive-lease quest: qmux's - `RecvStream` leases natively on moq-quic's mux instead of through - `poll::Buffered`, evicts by priority, and so closes the 2026-10-27 audit's - NET-3 gap (`Buffered` bounded each stream's incomplete message by the largest - message, and nothing bounded them in aggregate). Rejected: deferring qmux to - that quest, and adding hooks here while keeping leases in m2. -- The lease floor is one 1 MiB segment for every transport and version: the - lite-01/02 `ANNOUNCE_INIT` cap drops from 64 MiB to a segment less its 16 - bytes of prefixes, so `moq_net::transport::MIN_RECEIVE_WINDOW` is 1 MiB. A - legacy origin whose initial set exceeds it fails its announce loudly, on - both ends. Rejected: a 4 MiB cap, and keeping 64 MiB with a per-version - floor. -- Window defaults are one per-session number across native QUIC and qmux: - clients get a 4 MiB connection and a 1 MiB stream receive window, and qmux - takes the same client/peer split as native QUIC. The cluster peer windows - (`[limits.peer.quic]`, 640 MiB) and the client `send_window` (16 MiB) are - re-derived from measured bandwidth-delay needs, with the maintainer - suggesting about 16 MiB for peers. Rejected for clients: 16, 32, or 128 MiB - connection windows, and a 4 MiB stream window. -- Local readiness takes the next frame's `frame::Info` - (`group::Producer::poll_ready(waiter, &info)` and `ready(&info)`), reusing - ingest's admission semantics, including its whole-frame deadlock exception. - Rejected: a bare size on the same methods. -- Whole-batch fills move to [flights](/quest/m1/unbounded-groups/flights.md), - since the failure crosses groups while this mechanism paces one group copy. - -The adapter inventory this plan once listed is stale: moq-tokio has no Quinn -backend, and moq-ffi drives sessions through moq-tokio rather than a -transport of its own, so neither needs an adapter. - -Decided 2026-11-02 (maintainer, settling the open items of -[moq-dev/moq#7624](https://github.com/moq-dev/moq/pull/7624)): - -- Peer windows stay at 16 MiB receive and send in `[limits.peer.quic]`, with - the 1 MiB stream window from `[quic]`. Rejected: 32 MiB, and a separate - per-stream peer window. -- The client `send_window` is 4 MiB, matching its receive window. Rejected: - keeping 16 MiB. -- A GROUP stream waiting unread for its TRACK_INFO reserves nothing from the - paused budget: it waits at the lowest receive priority, so a full window - evicts it before control streams, and TRACK_INFO still gets credit. This - replaces the 2026-11-01 wording that the paused cap covers such streams. - Rejected: stopping it once the budget is full. -- Every qmux session a server accepts takes the configured `[quic]` windows, - including the relay's own WebSocket listener. Rejected: leaving them at the - defaults until admission. -- Split: this PR lands the windows, the floor, qmux leases, sized readiness, - TRACK_INFO priority, and the accept paths, and the rest stays here. - Rejected: finishing everything in one PR. - -Landed in [moq-dev/moq#7624](https://github.com/moq-dev/moq/pull/7624): -sized local readiness; the one-segment floor; 4 MiB, 1 MiB, and 4 MiB client -receive, stream, and send windows in moq-tokio, moq-uring, and qmux; 16 MiB -cluster peer windows; native qmux leases with eviction, window reporting, and -`poll_reset`; qmux sessions starting at the server's `[quic]` windows on every -accept path and widening to `[limits.peer.quic]` for peers; and a Bun sender -exhausting a qmux receiver over WebSocket (`just test receive-credit`). At the -client defaults a session pauses two group streams (half the window, a stream -window each), and a peer link eight; past that, groups read and shed as -before pacing. The same PR ranks streams waiting on TRACK_INFO (lite GROUP -streams and FETCH answers, and ietf data streams waiting on their alias) -below every other stream, and they watch `poll_reset`, which now reports an -eviction, so an evicted stream is stopped and eviction resumes. - -What remains: - -- A two-relay chain test with memory measured near the window. -- Per-session retained bytes and throughput across many qmux sessions at the - chosen window, in the - [transport x version benches](/quest/m2/bench-transport-sweep.md). -- A quiet-host comparison against the base of `group_read_frames`, - `group_contended`, and `group_relay_frames`, plus `group_paced` (a writer - parked at the window, over 1, 8, and 64 readers), which the first slice - added. The cursor's report is one relaxed load per frame or batch, and a - `fetch_max` only for the fastest reader. On the busy host that landed it, - runs swung by up to 50% either way between repeats. -- A relay-level test that the operator's `[quic]` windows reach WebSocket - sessions end to end through the relay's own `serve_ws` path. Today it is - covered only at `websocket::accept` level, because an accepted session's - window is not visible outside moq-tokio. - -Decided 2026-11-02 (maintainer, follow-up planning after the 2026-11-02 -/quest-complete run): JS parity moves to -[JS fastest-reader pacing](/quest/m2/js-fastest-reader.md) in m2, which waits -on [browser receive credit](/quest/m2/browser-receive-credit.md), so the m1 -cut does not wait on browsers. diff --git a/quest/m1/unbounded-groups/flights.md b/quest/m1/unbounded-groups/flights.md index 50981b7af3e..a0d9d0dd7ad 100644 --- a/quest/m1/unbounded-groups/flights.md +++ b/quest/m1/unbounded-groups/flights.md @@ -26,7 +26,7 @@ never does. - A flight is a separate copy of a group: its own upstream fill from its first rider's frame into its own group copy, with its own 32 MB tail, paced by its fastest rider like the live edge - ([Fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md)). + ([Fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664)). The live subscription is the first flight. Copies are never stitched together: two caches sharing one group handle caused the expiry and boundary bugs [Resume redesign](/quest/m2/resume-redesign.md) records. @@ -40,7 +40,7 @@ never does. frame within 16 MB, half the tail, of the flight's fastest rider, less one shedding unit (a log segment). Judge it against the worst-case front: the ahead window is capped at 16 MB - ([Fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md)), + ([Fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664)), so filling all outstanding credit still leaves an admitted reader held. Otherwise it starts its own flight, so it never boards at a front about to be shed. @@ -72,13 +72,9 @@ never does. - Fills that write a whole batch of groups at once evict their own earlier groups before the reader takes them (the follow-up from [moq-dev/moq#7339](https://github.com/moq-dev/moq/pull/7339)). Moved here - from [fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md) + from [fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664) on 2026-11-02 (maintainer): the failure crosses groups, while pacing holds back one group copy, so a flight's fill paces the batch by its riders instead. Public API: none. Wire: none, beyond the existing frame-offset fills. - -## Required - -- [Fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md) - flights pace on their fastest rider the same way the live edge does diff --git a/quest/m2/js-fastest-reader.md b/quest/m2/js-fastest-reader.md index edbe92ff7a1..9bf95ffa795 100644 --- a/quest/m2/js-fastest-reader.md +++ b/quest/m2/js-fastest-reader.md @@ -10,7 +10,7 @@ and on the JS qmux receiver. ## Plan Planned 2026-11-02 (maintainer, follow-up planning after the 2026-11-02 /quest-complete run): JS parity moves out of the m1 -[fastest-reader quest](/quest/m1/unbounded-groups/fastest-reader.md), so the m1 +[fastest-reader quest](https://github.com/moq-dev/moq/pull/7664), so the m1 cut does not wait on browsers. The JS qmux receiver could pace today but follows the browser, so one design covers both. diff --git a/quest/m2/receive-pool.md b/quest/m2/receive-pool.md index 4f70965cb64..633aa1aefa3 100644 --- a/quest/m2/receive-pool.md +++ b/quest/m2/receive-pool.md @@ -47,12 +47,12 @@ Open, for the implementer to bring to the maintainer: all. Idea for later (maintainer, 2026-10-30, from the -[fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md) iteration +[fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664) iteration in [moq-dev/moq#7388](https://github.com/moq-dev/moq/pull/7388)): instead of moq-quic granting `MAX_STREAM_DATA` automatically as a stream is read, the application could pause a stream, or grant it an explicit -`allow(fastest + 16 MB)`, so flow control never grants credit past the pacing -window. Paused streams would then hold no credit beyond that window, replacing +`allow(fastest + window)`, so flow control never grants credit past the pacing +window (the stream receive window, capped at 16 MB). Paused streams would then hold no credit beyond that window, replacing moq-net's per-session cap on paused credit. Until then, the pool must keep that cap honest (noted 2026-10-31 from diff --git a/quest/m2/stream/README.md b/quest/m2/stream/README.md index 62c725bcd44..8665a92df9b 100644 --- a/quest/m2/stream/README.md +++ b/quest/m2/stream/README.md @@ -30,7 +30,7 @@ Decided 2026-10-25 with the maintainer (mesh planning interview): Group finish is a FIN (half-close); a group abort with a code is a reset. The implementer measures and records a frame-size cap. - Lossless delivery comes from - [fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md), not + [fastest-reader pacing](https://github.com/moq-dev/moq/pull/7664), not from refill: a stream registers no fill handler, because most TCP traffic is not worth retaining for replay. A reader that falls behind the window ends `Evicted`, surfaced as a reset. @@ -58,7 +58,6 @@ the window. ## Required -- [Fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md) - relays push back instead of evicting a reader that keeps up - [Rust byte streams](/quest/m2/stream/rust.md) - a duplex `AsyncRead`/`AsyncWrite` stream, a listener and a dialer, and `moq` CLI forwarding - [JS byte streams](/quest/m2/stream/js.md) - the same surface in `@moq/net` - [Stream benchmark](/quest/m2/stream/bench.md) - throughput and latency against raw TCP through a relay chain, with and without loss diff --git a/quest/m2/stream/rust.md b/quest/m2/stream/rust.md index 7f001cb5dfe..1ddeda66fce 100644 --- a/quest/m2/stream/rust.md +++ b/quest/m2/stream/rust.md @@ -30,7 +30,3 @@ frame per TCP read, no fill handler. before the frame ends, round-trip bytes, half-close each way, reset each way, a reader ending `Evicted` surfaced as a reset, and a relay restart ending the stream with a typed error. - -## Required - -- [Fastest-reader pacing](/quest/m1/unbounded-groups/fastest-reader.md) - lossless delivery to a reader that keeps up diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index b0a5a2b7397..f14253f0e04 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -43,6 +43,13 @@ const WAKE_BUDGET: usize = 64 * 1024; /// A stream ingest may stop reading to pace a group's writer, which holds its receive /// credit meanwhile. pub(crate) trait Pause { + /// How far the group's writer reads this stream ahead of its fastest reader before + /// pausing: the stream's receive credit, capped at the live tail's ahead half (see + /// [`crate::model::pace::window`]). A stream outside any session takes the cap. + fn window(&self) -> u64 { + crate::model::pace::AHEAD + } + /// Reserve the credit this stream may hold while paused, returning whether it may pause. /// A stream outside any session always may. fn pause(&mut self) -> bool { @@ -54,6 +61,10 @@ pub(crate) trait Pause { } impl Pause for crate::transport::boxed::RecvStream { + fn window(&self) -> u64 { + crate::transport::boxed::RecvStream::window(self) + } + fn pause(&mut self) -> bool { crate::transport::boxed::RecvStream::pause(self) } @@ -429,7 +440,7 @@ impl Reader { let result = loop { // A writer too far ahead of the group's fastest reader leaves the rest in the // stream, so flow control pushes back upstream. - let lease = match self.poll_pace(waiter, frame.poll_ready(waiter)) { + let lease = match self.poll_pace(waiter, frame.poll_ready(waiter, self.stream.window())) { Poll::Ready(Ok(())) => self.poll_lease(waiter, frame.remaining().min(1)), Poll::Ready(Err(err)) => break Err(err), Poll::Pending => Poll::Pending, @@ -509,7 +520,7 @@ impl Reader { where S: Pause, { - self.poll_pace(waiter, group.poll_admit(waiter, header)) + self.poll_pace(waiter, group.poll_admit(waiter, header, self.stream.window())) } /// Wait out `paced`, the group's pace gate, if the session has room for the credit this diff --git a/rs/moq-net/src/model/frame.rs b/rs/moq-net/src/model/frame.rs index e5a2c09a10d..a5183c75f19 100644 --- a/rs/moq-net/src/model/frame.rs +++ b/rs/moq-net/src/model/frame.rs @@ -1376,12 +1376,12 @@ impl Producer { self.buf.write_chunk(chunk) } - /// Poll until the group's writer may take more of this frame from upstream; see - /// [`group::Producer::poll_paced`]. + /// Poll until the group's writer may take more of this frame from upstream, at most + /// `window` bytes past its fastest reader; see [`group::Producer::poll_paced`]. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] - pub(crate) fn poll_ready(&self, waiter: &kio::Waiter) -> Poll<()> { + pub(crate) fn poll_ready(&self, waiter: &kio::Waiter, window: u64) -> Poll<()> { self.group() - .poll_paced(waiter, self.buf.written(Ordering::Acquire) as u64) + .poll_paced(waiter, self.buf.written(Ordering::Acquire) as u64, window) } /// Publish what has been written so far, waking consumers parked on the group. diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index f5bfff14221..db4aca9b9a9 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -2222,7 +2222,7 @@ impl Producer { /// the group and is not waited on. Writing without waiting never fails: the group sheds /// its front instead. pub fn poll_ready(&self, waiter: &kio::Waiter, frame: &frame::Info) -> Poll<()> { - ready!(self.poll_paced(waiter, frame.size)); + ready!(self.poll_paced(waiter, frame.size, super::pace::AHEAD)); // A closed group reports `Closed`, which is ready too. self.handle.state.poll_wanted_changed(waiter, false).map(|_| ()) } @@ -2233,8 +2233,9 @@ impl Producer { } /// Poll until the writer may take more from upstream, `partial` bytes into a frame - /// it has not committed: until it is at most [`super::pace::AHEAD`] past the end of the last - /// frame its fastest reader took. + /// it has not committed: until it is at most `window` bytes past the end of the last + /// frame its fastest reader took. Ingest passes its stream's receive credit (see + /// [`super::pace::window`]); a local writer passes [`super::pace::AHEAD`]. /// /// Ready at once while no reader holds the group, so a group nobody watches keeps /// writing and sheds its front, and once it is closed, so the next write surfaces why. @@ -2243,16 +2244,16 @@ impl Producer { /// fallen off the group and is not waited on. /// /// A session's ingest waits on this; a local writer waits on [`Self::poll_ready`]. - pub(crate) fn poll_paced(&self, waiter: &kio::Waiter, partial: u64) -> Poll<()> { + pub(crate) fn poll_paced(&self, waiter: &kio::Waiter, partial: u64, window: u64) -> Poll<()> { let pace = &self.handle.alive.pace; - if !pace.ahead(partial) { + if !pace.ahead(partial, window) { return Poll::Ready(()); } // Either wakes the writer: the last reader leaving, or the group closing. if self.handle.poll_unwanted(waiter).is_ready() || self.handle.state.poll_closed(waiter).is_ready() { return Poll::Ready(()); } - match pace.paced(pace.park(waiter), partial) { + match pace.paced(pace.park(waiter), partial, window) { true => Poll::Pending, false => Poll::Ready(()), } diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index 3f46801d4d9..33f0ff63770 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -27,7 +27,7 @@ mod expiry; mod idle; mod skipped; // How far a group's writer reads ahead of its fastest reader. -mod pace; +pub(crate) mod pace; // The DEFLATE codec behind compressed tracks. mod flate; pub(crate) mod object; diff --git a/rs/moq-net/src/model/pace.rs b/rs/moq-net/src/model/pace.rs index 3118ca44fc6..2b0d9f6fcbd 100644 --- a/rs/moq-net/src/model/pace.rs +++ b/rs/moq-net/src/model/pace.rs @@ -1,11 +1,15 @@ //! Fastest-reader pacing: how far a group's writer reads ahead of its furthest reader. //! //! Readers report the frames they take; a paced writer (a session's ingest) stops reading -//! its upstream stream while it is more than [`AHEAD`] past the end of the last frame the +//! its upstream stream while it is more than its window past the end of the last frame the //! fastest reader took, so stream flow control pushes back to the publisher. Only the //! fastest reader counts: a slower one is never waited on, and falls off the group's //! front as the live tail sheds. //! +//! Ingest's window is the stream receive credit its transport grants, capped at [`AHEAD`] +//! (see [`window`]), so a hop holds about what its upstream peer can have in flight on +//! that stream. A local writer, which has no transport, uses [`AHEAD`] itself. +//! //! Progress is in bytes on one scale per group copy: each frame entering the copy's log //! ends where the frames before it, payload and slot, add up to. A frame larger than the //! window never stalls its writer, since a reader that took every committed frame is @@ -18,10 +22,19 @@ use std::sync::{ use std::task::Waker; use super::group::MAX_CACHE_BYTES; - -/// How far a paced writer runs ahead of its group's fastest reader: half the live tail, -/// so the other half stays behind the fastest reader for slower ones. -pub(super) const AHEAD: u64 = MAX_CACHE_BYTES / 2; +use crate::transport::ReceiveWindow; + +/// The furthest a paced writer runs ahead of its group's fastest reader: half the live +/// tail, so the other half stays behind the fastest reader for slower ones. +pub(crate) const AHEAD: u64 = MAX_CACHE_BYTES / 2; + +/// How far ingest reading a stream runs ahead of its group's fastest reader: the credit one +/// stream may hold under `receive`, the windows its transport grants, capped at [`AHEAD`]. +/// +/// A transport that reports no window never pauses a stream, so its window is moot. +pub(crate) fn window(receive: Option) -> u64 { + receive.map_or(AHEAD, |receive| receive.stream.min(receive.connection).min(AHEAD)) +} /// One group copy's reader progress, shared by its writer and every cursor. #[derive(Default)] @@ -88,19 +101,19 @@ impl Pace { self.written.load(Ordering::Acquire) } - /// Whether a writer `partial` bytes into an uncommitted frame is more than [`AHEAD`] - /// past the fastest reader, given it read up to `read`. + /// Whether a writer `partial` bytes into an uncommitted frame is more than `window` + /// bytes past the fastest reader, given it read up to `read`. /// /// Never while that reader took every committed frame, since it is waiting on the /// writer, nor once the frames it has yet to take were shed, since it fell off. - pub fn paced(&self, read: u64, partial: u64) -> bool { + pub fn paced(&self, read: u64, partial: u64, window: u64) -> bool { let written = self.written(); - read < written && read >= self.front.load(Ordering::SeqCst) && written.saturating_add(partial) - read > AHEAD + read < written && read >= self.front.load(Ordering::SeqCst) && written.saturating_add(partial) - read > window } /// Whether the writer is paced, as [`Self::paced`] judges it now. - pub fn ahead(&self, partial: u64) -> bool { - self.paced(self.read.load(Ordering::SeqCst), partial) + pub fn ahead(&self, partial: u64, window: u64) -> bool { + self.paced(self.read.load(Ordering::SeqCst), partial, window) } /// Register `waiter` for the next read or shed, returning how far readers got, read @@ -181,16 +194,44 @@ mod tests { let mut group = written(15); let mut fast = group.consume(); let _slow = group.consume(); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Ready(()), "within the window"); + assert_eq!( + group.poll_paced(&waiter, 0, AHEAD), + Poll::Ready(()), + "within the window" + ); group.write_frame(None, Bytes::from(vec![0u8; MB])).unwrap(); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Pending, "past the window"); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Pending, "past the window"); assert!(!flag.woken()); // A batch read takes several frames at once, which the writer counts as read. read(&mut fast); assert!(flag.woken(), "the fastest reader's read wakes the writer"); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Ready(())); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Ready(())); + } + + /// Ingest paces at its stream's receive credit, so a writer stops once that much is + /// past its fastest reader rather than half the live tail. + #[test] + fn a_smaller_window_paces_sooner() { + let (_flag, waiter) = flag(); + let group = written(3); + let _reader = group.consume(); + let credit = window(Some(ReceiveWindow::new(4 * MB as u64, MB as u64))); + assert_eq!(credit, MB as u64); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Ready(()), "within the cap"); + assert_eq!(group.poll_paced(&waiter, 0, credit), Poll::Pending, "past the credit"); + } + + /// The window is a stream's credit, no more than the connection's and never past the + /// ahead half of the live tail, which keeps the behind half for slower readers. + #[test] + fn the_window_is_the_stream_credit_capped() { + let mb = MB as u64; + assert_eq!(window(None), AHEAD, "a transport without windows never pauses"); + assert_eq!(window(Some(ReceiveWindow::new(4 * mb, mb))), mb); + assert_eq!(window(Some(ReceiveWindow::new(mb / 2, mb))), mb / 2); + assert_eq!(window(Some(ReceiveWindow::new(u64::MAX, 64 * mb))), AHEAD); } /// A one-byte frame whose properties hold `metadata` bytes. @@ -231,9 +272,9 @@ mod tests { group = frame.finish().unwrap(); } let group = crate::recv::Group::new(group); - assert_eq!(group.poll_admit(&waiter, &fielded(15, 0)), Poll::Ready(())); + assert_eq!(group.poll_admit(&waiter, &fielded(15, 0), AHEAD), Poll::Ready(())); assert_eq!( - group.poll_admit(&waiter, &fielded(15, 2 * MB)), + group.poll_admit(&waiter, &fielded(15, 2 * MB), AHEAD), Poll::Pending, "its metadata would pass the window" ); @@ -251,9 +292,9 @@ mod tests { timestamp: None, }) .unwrap(); - assert_eq!(frame.poll_ready(&waiter), Poll::Ready(())); + assert_eq!(frame.poll_ready(&waiter, AHEAD), Poll::Ready(())); frame.write(Bytes::from(vec![0u8; MB])).unwrap(); - assert_eq!(frame.poll_ready(&waiter), Poll::Pending); + assert_eq!(frame.poll_ready(&waiter, AHEAD), Poll::Pending); } /// A frame is admitted only once its declared size fits the window, since admitting @@ -263,11 +304,11 @@ mod tests { let (flag, waiter) = flag(); let group = written(15); let mut reader = group.consume(); - assert_eq!(group.poll_paced(&waiter, 20 * MB as u64), Poll::Pending); + assert_eq!(group.poll_paced(&waiter, 20 * MB as u64, AHEAD), Poll::Pending); while reader.poll_read_frame(&kio::Waiter::noop()).is_ready() {} assert!(flag.woken()); assert_eq!( - group.poll_paced(&waiter, 20 * MB as u64), + group.poll_paced(&waiter, 20 * MB as u64, AHEAD), Poll::Ready(()), "the reader took everything, so the frame sheds nothing it wants" ); @@ -279,17 +320,17 @@ mod tests { fn a_group_without_readers_never_waits() { let (flag, waiter) = flag(); let group = written(16); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Ready(())); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Ready(())); let reader = group.consume(); assert_eq!( - group.poll_paced(&waiter, 0), + group.poll_paced(&waiter, 0, AHEAD), Poll::Pending, "a reader that took nothing" ); drop(reader); assert!(flag.woken(), "the last reader leaving wakes the writer"); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Ready(())); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Ready(())); } /// A local writer waits for a reader before it writes, then for room in the window, and @@ -373,7 +414,7 @@ mod tests { }) .unwrap(); frame.write(Bytes::from(vec![0u8; 18 * MB])).unwrap(); - assert_eq!(frame.poll_ready(&waiter), Poll::Ready(())); + assert_eq!(frame.poll_ready(&waiter, AHEAD), Poll::Ready(())); } /// A reader whose next frame the tail shed fell off the group, so the writer stops @@ -387,7 +428,7 @@ mod tests { for _ in 0..40 { group.write_frame(None, Bytes::from(vec![0u8; MB])).unwrap(); } - assert_eq!(group.poll_paced(&waiter, 0), Poll::Ready(())); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Ready(())); } /// A reader that skipped past what the writer has written is waiting on it, so it @@ -403,7 +444,7 @@ mod tests { } // Each commit wakes the parked reader, which polls again and reports the edge. assert!(reader.poll_read_frame(&kio::Waiter::noop()).is_pending()); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Ready(())); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Ready(())); } /// Shedding past the fastest reader under cache pressure wakes a parked writer: that @@ -413,10 +454,10 @@ mod tests { let (flag, waiter) = flag(); let group = written(17); let _reader = group.consume(); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Pending); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Pending); group.handle().shed(u64::MAX, u64::MAX); assert!(flag.woken()); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Ready(())); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Ready(())); } /// An abort wakes a parked writer, so its next write surfaces why. @@ -425,9 +466,9 @@ mod tests { let (flag, waiter) = flag(); let group = written(17); let _reader = group.consume(); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Pending); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Pending); group.handle().abort(Error::Cancel).unwrap(); assert!(flag.woken()); - assert_eq!(group.poll_paced(&waiter, 0), Poll::Ready(())); + assert_eq!(group.poll_paced(&waiter, 0, AHEAD), Poll::Ready(())); } } diff --git a/rs/moq-net/src/recv.rs b/rs/moq-net/src/recv.rs index 8e87b288d58..8a130f6e217 100644 --- a/rs/moq-net/src/recv.rs +++ b/rs/moq-net/src/recv.rs @@ -31,14 +31,14 @@ impl Group { } /// Poll until the writer may admit the frame `header` opens, its payload and metadata, - /// without shedding a byte its fastest reader has yet to take; see - /// [`group::Producer::poll_paced`]. - pub fn poll_admit(&self, waiter: &kio::Waiter, header: &frame::Header) -> std::task::Poll<()> { + /// within `window` bytes past its fastest reader, so admitting it sheds no byte that + /// reader has yet to take; see [`group::Producer::poll_paced`]. + pub fn poll_admit(&self, waiter: &kio::Waiter, header: &frame::Header, window: u64) -> std::task::Poll<()> { let metadata = header.fields.as_ref().map_or(0, |fields| fields.heap()); self.writer .as_ref() .expect("group writer is available between frames") - .poll_paced(waiter, header.info.size.saturating_add(metadata)) + .poll_paced(waiter, header.info.size.saturating_add(metadata), window) } /// Restore the writer returned by a completed frame. diff --git a/rs/moq-net/src/transport/boxed.rs b/rs/moq-net/src/transport/boxed.rs index f541e489990..0772b4858ed 100644 --- a/rs/moq-net/src/transport/boxed.rs +++ b/rs/moq-net/src/transport/boxed.rs @@ -156,17 +156,27 @@ impl RecvStream { /// Box a stream outside any session, which therefore never pauses. #[cfg(test)] pub(crate) fn new(stream: impl poll::RecvStream + Send + Sync) -> Self { - Self::budgeted(stream, Weak::>::new()) + Self::budgeted(stream, Weak::>::new(), None) } - fn budgeted(stream: impl poll::RecvStream + Send + Sync, session: Budget) -> Self { + /// Box a stream of a session granting `receive`, read when the stream is accepted or + /// opened, which sizes its pacing [`window`](Self::window). + fn budgeted(stream: impl poll::RecvStream + Send + Sync, session: Budget, receive: Option) -> Self { Self(Box::new(Budgeted { stream, session, held: 0, + window: crate::model::pace::window(receive), })) } + /// How far ingest reads this stream ahead of its group's fastest reader before pausing: + /// the credit the stream may hold, so a paced hop buffers about what its peer can have + /// in flight on it. + pub(crate) fn window(&self) -> u64 { + self.0.window() + } + /// Reserve the receive credit this stream may hold while ingest stops reading it, returning /// whether it may stop: not once the session's paused streams hold half its connection /// window, nor on a transport that reports no window. @@ -186,6 +196,8 @@ struct Budgeted { session: Budget, /// The credit reserved while paused, zero while read. held: u64, + /// The pacing window, from the session's receive windows when the stream arrived. + window: u64, } impl Budgeted { @@ -311,6 +323,7 @@ trait ErasedRecv: Send + Sync { fn set_priority(&mut self, priority: i64); fn stop(&mut self, code: u32); fn poll_reset(&mut self, waiter: &kio::Waiter) -> Poll; + fn window(&self) -> u64; fn pause(&mut self) -> bool; fn resume(&mut self); } @@ -338,18 +351,22 @@ fn unbox<'a, G: 'static>(options: SendOptions<'a, Group>) -> SendOptions<'a, G> } } -fn bi((send, recv): poll::BiStreams, budget: Budget) -> (SendStream, RecvStream) { - (SendStream(Box::new(send)), RecvStream::budgeted(recv, budget)) +fn bi(session: &S, (send, recv): poll::BiStreams, budget: Budget) -> (SendStream, RecvStream) { + let receive = poll::Session::receive_window(session); + (SendStream(Box::new(send)), RecvStream::budgeted(recv, budget, receive)) } impl ErasedSession for S { fn poll_accept_uni(&self, waiter: &kio::Waiter, budget: Budget) -> Poll> { - poll::Session::poll_accept_uni(self, waiter) - .map(|res| res.map(|recv| RecvStream::budgeted(recv, budget)).map_err(Error::new)) + poll::Session::poll_accept_uni(self, waiter).map(|res| { + res.map(|recv| RecvStream::budgeted(recv, budget, poll::Session::receive_window(self))) + .map_err(Error::new) + }) } fn poll_accept_bi(&self, waiter: &kio::Waiter, budget: Budget) -> Poll> { - poll::Session::poll_accept_bi(self, waiter).map(|res| res.map(|bis| bi::(bis, budget)).map_err(Error::new)) + poll::Session::poll_accept_bi(self, waiter) + .map(|res| res.map(|bis| bi::(self, bis, budget)).map_err(Error::new)) } fn poll_open_uni(&self, waiter: &kio::Waiter, options: SendOptions<'_, Group>) -> Poll> { @@ -364,7 +381,7 @@ impl ErasedSession for S { budget: Budget, ) -> Poll> { poll::Session::poll_open_bi(self, waiter, unbox(options)) - .map(|res| res.map(|bis| bi::(bis, budget)).map_err(Error::new)) + .map(|res| res.map(|bis| bi::(self, bis, budget)).map_err(Error::new)) } fn poll_open_request( @@ -374,7 +391,7 @@ impl ErasedSession for S { budget: Budget, ) -> Poll, Error>> { poll::Session::poll_open_request(self, waiter, unbox(options)).map(|res| { - res.map(|streams| streams.map(|bis| bi::(bis, budget))) + res.map(|streams| streams.map(|bis| bi::(self, bis, budget))) .map_err(Error::new) }) } @@ -501,6 +518,10 @@ impl ErasedRecv for Budgeted { poll::RecvStream::poll_reset(&mut self.stream, waiter).map(Error::new) } + fn window(&self) -> u64 { + self.window + } + fn pause(&mut self) -> bool { if self.held > 0 { return true; @@ -736,6 +757,26 @@ mod test { assert!(accept(&session).pause(), "so did a dropped one"); } + /// A stream paces at the credit its session grants it, so a hop holds about what its + /// peer can have in flight; a transport without windows takes the cap, never pausing. + #[test] + fn a_stream_paces_at_its_credit() { + use crate::model::pace::AHEAD; + + let gate = kio::Producer::new(true); + let session = + Session::new(SinkSession::accepted_bi(gate.consume()).with_receive_window(ReceiveWindow::new(100, 20))); + assert_eq!(accept(&session).window(), 20); + + let session = Session::new( + SinkSession::accepted_bi(gate.consume()).with_receive_window(ReceiveWindow::new(u64::MAX, u64::MAX)), + ); + assert_eq!(accept(&session).window(), AHEAD, "capped at half the live tail"); + + let session = Session::new(SinkSession::accepted_bi(gate.consume())); + assert_eq!(accept(&session).window(), AHEAD); + } + /// A transport that reports no window never pauses a stream, since nothing bounds what /// it would hold; nor does a stream that outlived its session. #[test] diff --git a/rs/moq-net/tests/integration/fastest_reader.rs b/rs/moq-net/tests/integration/fastest_reader.rs index df0dce3b030..83cb18d0f01 100644 --- a/rs/moq-net/tests/integration/fastest_reader.rs +++ b/rs/moq-net/tests/integration/fastest_reader.rs @@ -1,13 +1,14 @@ #![cfg(feature = "async")] //! A relay's ingest buffers at most a window past the group's fastest reader, then leaves -//! the rest in the stream, so flow control pushes back to the publisher. A slower reader -//! is never waited on: it falls off the group's front. +//! the rest in the stream, so flow control pushes back to the publisher. The window is the +//! stream receive credit the relay's transport grants, capped at 16 MiB, half the group's +//! live tail. A slower reader is never waited on: it falls off the group's front. use std::time::Duration; use bytes::Bytes; -use moq_net::{Hop, Timestamp, Version, origin, track}; +use moq_net::{Hop, Timestamp, Version, origin, track, transport::ReceiveWindow}; use crate::support::harness::{MockPair, peer, spawn}; @@ -17,6 +18,11 @@ const MB: usize = 1 << 20; const VERSIONS: &[&str] = &["moq-lite-06", "moq-lite-07-wip", "moq-transport-22"]; +/// Receive windows whose stream credit, 64 MiB, is past the 16 MiB pacing cap. +fn uncapped() -> ReceiveWindow { + ReceiveWindow::new(u64::MAX, 64 * MB as u64) +} + fn origin(hop: u64) -> origin::Producer { let (origin, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); spawn(driver); @@ -66,46 +72,60 @@ async fn relay(version: &str) -> Relay { } } +/// An idle reader holds the relay to about its stream credit, or the 16 MiB cap past it, and +/// still receives every frame. #[kio::test] async fn an_idle_reader_paces_the_relay_and_loses_nothing() { - let payload = Bytes::from(vec![7u8; MB]); + // The stream credit the relay grants, and how many 1 MiB frames it then holds. + let cases = [ + (ReceiveWindow::new(16 * MB as u64, 4 * MB as u64), 3..=5), + (uncapped(), 15..=18), + ]; for version in VERSIONS { - let mut relay = relay(version).await; + for (window, expected) in cases.clone() { + idle_reader(version, window, expected).await; + } + } +} - let mut group = relay.track.append_group().unwrap(); +async fn idle_reader(version: &str, window: ReceiveWindow, expected: std::ops::RangeInclusive) { + let payload = Bytes::from(vec![7u8; MB]); + let mut relay = relay(version).await; + relay.pair.server_transport.set_receive_window(Some(window)); + + let mut group = relay.track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + let mut reader = kio::clock::timeout(TIMEOUT, relay.subscriber.recv_group()) + .await + .expect("group timeout") + .unwrap() + .expect("group 0"); + // Within the publisher's own 32 MB tail, so only the relay's pacing holds it back. + for _ in 1..30 { group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); - let mut reader = kio::clock::timeout(TIMEOUT, relay.subscriber.recv_group()) - .await - .expect("group timeout") - .unwrap() - .expect("group 0"); - // Within the publisher's own 32 MB tail, so only the relay's pacing holds it back. - for _ in 1..30 { - group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); - } - group.finish().unwrap(); + } + group.finish().unwrap(); - // Simulated time moves only once every task is parked: the relay's ingest stopped. - kio::clock::sleep(Duration::from_secs(1)).await; - let held = reader.frame_count(); - assert!( - (15..=18).contains(&held), - "{version}: the relay read {held} MB ahead of a reader that took nothing" - ); + // Simulated time moves only once every task is parked: the relay's ingest stopped. + kio::clock::sleep(Duration::from_secs(1)).await; + let held = reader.frame_count(); + assert!( + expected.contains(&held), + "{version}: the relay read {held} MB ahead of a reader that took nothing under {window:?}" + ); - for index in 0..30 { - let frame = kio::clock::timeout(TIMEOUT, reader.read_frame()) - .await - .unwrap_or_else(|_| panic!("{version}: frame {index} never arrived")) - .unwrap_or_else(|err| panic!("{version}: frame {index}: {err}")) - .expect("a frame"); - assert_eq!(frame.payload.len(), MB, "{version}: frame {index}"); - } - let end = kio::clock::timeout(TIMEOUT, reader.read_frame()) + for index in 0..30 { + let frame = kio::clock::timeout(TIMEOUT, reader.read_frame()) .await - .expect("end timeout"); - assert!(matches!(end, Ok(None)), "{version}: {end:?}"); + .unwrap_or_else(|_| panic!("{version}: frame {index} never arrived")) + .unwrap_or_else(|err| panic!("{version}: frame {index}: {err}")) + .expect("a frame"); + assert_eq!(frame.payload.len(), MB, "{version}: frame {index}"); } + let end = kio::clock::timeout(TIMEOUT, reader.read_frame()) + .await + .expect("end timeout"); + assert!(matches!(end, Ok(None)), "{version}: {end:?}"); } /// A paused stream is not read, so the relay watches it for a reset instead: a publisher @@ -139,11 +159,15 @@ async fn a_reset_reaches_a_paused_relay() { /// A frame larger than the room left waits to be admitted, since admitting it trims the /// tail to fit: an idle reader loses nothing to a large frame either. +/// +/// The relay grants past the 16 MiB cap, so it takes the first 15 frames and the publisher +/// sheds only what it already sent. #[kio::test] async fn a_large_frame_sheds_nothing_an_idle_reader_wants() { let payload = Bytes::from(vec![7u8; MB]); for version in VERSIONS { let mut relay = relay(version).await; + relay.pair.server_transport.set_receive_window(Some(uncapped())); let mut group = relay.track.append_group().unwrap(); group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); @@ -248,3 +272,100 @@ async fn a_slow_reader_falls_off_without_slowing_the_fast_one() { ); } } + +/// Backpressure crosses both relays, with one pacing window of unread payload per hop: the +/// 1 MiB stream credit each relay grants, not the 16 MiB cap. The upstream copy also keeps +/// sent history within its 32 MiB tail. +/// Cache charges include frame slots and relay metadata, not transport buffers or RSS. +#[kio::test] +async fn two_relays_bound_their_cache_and_resume_without_a_hole() { + use moq_net::{frame, transport::ReceiveWindow}; + + const FRAME: usize = 64 * 1024; + const TOTAL: usize = 64 * MB / FRAME; + for version in VERSIONS { + let publisher = origin(1); + let first = origin(2); + let second = origin(3); + let version = version.parse::().unwrap(); + let upstream = peer(version, &publisher, &first).await; + let downstream = peer(version, &first, &second).await; + for pair in [&upstream, &downstream] { + pair.server_transport + .set_receive_window(Some(ReceiveWindow::new(4 * MB as u64, MB as u64))); + } + let broadcast = publisher.create_broadcast("chain").unwrap(); + let track = broadcast + .create_track("data", track::Info::default().with_timescale(moq_net::Timescale::MILLI)) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + let remote = second + .consume() + .follow("chain", None) + .unwrap() + .available() + .await + .unwrap(); + let mut subscriber = remote + .track("data") + .unwrap() + .subscribe( + track::Subscription::default().with_delay(track::Delay::default().with_max(Duration::from_secs(3600))), + ) + .await + .unwrap(); + crate::support::harness::subscribed(&track).await; + let mut group = track.append_group().unwrap(); + let written = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let progress = written.clone(); + let writer = kio::clock::sim::spawn(async move { + for index in 0..TOTAL { + let payload = Bytes::from(vec![(index % 251) as u8; FRAME]); + let info = frame::Info { + size: FRAME as u64, + timestamp: Some(Timestamp::ZERO), + }; + group.ready(&info).await; + group.write_frame(Timestamp::ZERO, payload).unwrap(); + progress.store(index + 1, std::sync::atomic::Ordering::Relaxed); + } + group.finish().unwrap(); + }); + let mut reader = subscriber.recv_group().await.unwrap().unwrap(); + // Simulated time advances only once all drivers and the paced writer park. + kio::clock::sleep(Duration::from_secs(1)).await; + let parked = written.load(std::sync::atomic::Ordering::Relaxed); + assert!(parked < TOTAL, "{version}: no backpressure reached the producer"); + // Frame slots count toward pacing too, so the last whole frame can fall short. The + // upstream copy also holds what it sent downstream, within its 32 MiB tail. + for (hop, relay, max) in [(1, &first, 32 * MB + FRAME), (2, &second, MB + FRAME)] { + let held = relay.config().pool.used(); + assert!( + held >= (MB - FRAME) as u64, + "{version}: hop {hop} never filled its window: {held}" + ); + assert!(held < max as u64, "{version}: hop {hop} retained {held} bytes"); + eprintln!("{version}: hop {hop} parked cache={held} B, origin wrote {parked} frames"); + } + for index in 0..TOTAL { + let frame = kio::clock::timeout(TIMEOUT, reader.read_frame()) + .await + .expect("chain stalled") + .unwrap() + .unwrap(); + assert_eq!( + frame.payload, + Bytes::from(vec![(index % 251) as u8; FRAME]), + "{version}: frame {index}" + ); + for relay in [&first, &second] { + assert!( + relay.config().pool.used() < 33 * MB as u64, + "{version}: live tail grew past its budget" + ); + } + } + assert!(reader.read_frame().await.unwrap().is_none()); + writer.await.unwrap(); + } +} diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 6d9f039f548..440da5af490 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -2552,6 +2552,88 @@ mod tests { serving.abort(); } + /// Read the windows advertised by the real relay upgrade, before MoQ admission can + /// widen them. Nondefault values distinguish operator configuration from defaults. + #[cfg(feature = "websocket")] + #[tokio::test] + async fn websocket_upgrade_advertises_the_configured_receive_windows() { + use bytes::Buf; + use futures::StreamExt; + use qmux::ws::tungstenite::{Message, client::IntoClientRequest}; + + // QMux's frame type, lengths, and QUIC transport parameter values use QUIC varints. + fn decode_quic(bytes: &mut bytes::Bytes) -> Result { + let first = *bytes.first().ok_or("missing varint")?; + let len = 1 << (first >> 6); + if bytes.len() < len { + return Err("truncated varint"); + } + let mut value = (bytes.get_u8() & 0x3f) as u64; + for _ in 1..len { + value = (value << 8) | bytes.get_u8() as u64; + } + Ok(value) + } + + let mut config = crate::Config::default(); + config.listen.tcp.bind = Some("127.0.0.1:0".parse().unwrap()); + config.web.http.listen = Some("127.0.0.1:0".parse().unwrap()); + config.web.ws = true; + config.auth.public = vec![moq_auth::Pattern::all()]; + config.quic.receive_window = Some(3 << 20); + config.quic.stream_receive_window = Some(384 << 10); + config.quic.max_streams = Some(17); + let relay = crate::Relay::load(config).await.unwrap().with_signals(false); + let http = relay.web_addrs().http.unwrap(); + let serving = tokio::spawn(relay.run()); + + for version in qmux::Version::ALL { + let mut request = format!("ws://{http}/").into_client_request().unwrap(); + request + .headers_mut() + .insert("sec-websocket-protocol", version.alpn().parse().unwrap()); + let (mut socket, _) = qmux::ws::tokio_tungstenite::connect_async(request).await.unwrap(); + let Message::Binary(mut record) = kio::clock::timeout(std::time::Duration::from_secs(5), socket.next()) + .await + .unwrap() + .unwrap() + .unwrap() + else { + panic!("expected QMux transport parameters"); + }; + assert_eq!(decode_quic(&mut record).unwrap(), 0x3f5153300d0a0d0a); + let len = decode_quic(&mut record).unwrap() as usize; + let mut params = record.split_to(len); + let mut found = std::collections::BTreeMap::new(); + while params.has_remaining() { + let id = decode_quic(&mut params).unwrap(); + let len = decode_quic(&mut params).unwrap() as usize; + let mut value = params.split_to(len); + if (0x04..=0x09).contains(&id) { + found.insert(id, decode_quic(&mut value).unwrap()); + assert!(value.is_empty()); + } + } + assert_eq!( + found, + [ + (4, 3 << 20), + (5, 384 << 10), + (6, 384 << 10), + (7, 384 << 10), + (8, 17), + (9, 17) + ] + .into_iter() + .collect(), + "{version:?}" + ); + socket.close(None).await.unwrap(); + } + serving.abort(); + let _ = serving.await; + } + /// An upgraded WebSocket reads no more than one qmux record at a time: a frame /// declaring more is refused at its header, before SETUP and without buffering it, /// even with no timeout to end the wait. The socket is the listener's own, reclaimed from diff --git a/rs/qmux/Cargo.toml b/rs/qmux/Cargo.toml index a963848c868..f9728b36f17 100644 --- a/rs/qmux/Cargo.toml +++ b/rs/qmux/Cargo.toml @@ -41,6 +41,7 @@ libc = { workspace = true } [dev-dependencies] anyhow = { workspace = true } +criterion = { package = "bench-harness", path = "../bench-harness" } # The last release before this crate moved here, for wire interop tests only. qmux06 = { package = "qmux", version = "0.6.3", default-features = false, features = ["tcp", "ws"] } # A self-signed identity and the relay's crypto provider, for the bench example's TLS transports. @@ -63,5 +64,10 @@ required-features = ["tcp", "ws"] name = "bench" required-features = ["tls", "wss"] +[[bench]] +name = "receive" +harness = false +required-features = ["tcp", "ws"] + [lints] workspace = true diff --git a/rs/qmux/benches/receive.rs b/rs/qmux/benches/receive.rs new file mode 100644 index 00000000000..a412ed1cc0a --- /dev/null +++ b/rs/qmux/benches/receive.rs @@ -0,0 +1,224 @@ +//! Retained receive heap and delivery cost at the relay's 4 MiB / 1 MiB windows. +//! +//! Each session holds one or eight unread messages, each half a stream window, +//! under native leases. Eight together fill the connection window. Memory includes +//! both endpoints, the bounded byte pipes, tasks, stream tables, and payload backings; +//! it excludes the runtime and the shared source payload. It counts live requested +//! allocator bytes, not RSS or a sum of allocations. Every timed round reads, verifies, +//! releases, and refills every message. Transport, session count, stream count, and +//! QMux version are independent axes. TCP and WebSocket framing run over bounded +//! in-memory pipes to isolate receive ownership from socket scheduling and TLS. + +use kio::clock::Instant; +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicIsize, Ordering}; +use std::time::Duration; + +use bytes::{Buf, Bytes}; +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_net::transport::poll::{Lease as _, RecvStream as _, SendStream as _, Session as _}; +use qmux::{Config, RecvStream, SendStream, Session, Version, transport::Stream}; +use tokio::task::JoinSet; + +const CONNECTION: usize = 4 << 20; +const STREAM: usize = 1 << 20; +const MESSAGE: usize = STREAM / 2; + +struct Counter; +static LIVE: AtomicIsize = AtomicIsize::new(0); +#[global_allocator] +static ALLOCATOR: Counter = Counter; + +// Counts requested live bytes once per backing, including reallocations in place. +unsafe impl GlobalAlloc for Counter { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc(layout) }; + if !ptr.is_null() { + LIVE.fetch_add(layout.size() as isize, Ordering::Relaxed); + } + ptr + } + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc_zeroed(layout) }; + if !ptr.is_null() { + LIVE.fetch_add(layout.size() as isize, Ordering::Relaxed); + } + ptr + } + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 { + let ptr = unsafe { System.realloc(ptr, layout, size) }; + if !ptr.is_null() { + LIVE.fetch_add(size as isize - layout.size() as isize, Ordering::Relaxed); + } + ptr + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + LIVE.fetch_sub(layout.size() as isize, Ordering::Relaxed); + unsafe { System.dealloc(ptr, layout) } + } +} + +#[derive(Clone, Copy)] +struct Shape { + websocket: bool, + version: Version, + sessions: usize, + streams: usize, +} + +struct Receivers { + sessions: Vec<(Session, Session)>, + streams: Vec<(SendStream, RecvStream)>, + drivers: JoinSet<()>, + payload: Bytes, +} + +impl Receivers { + async fn new(shape: Shape, payload: Bytes) -> Self { + let Shape { + websocket, + version, + sessions, + streams, + } = shape; + let mut config = Config::new(version); + config.limits.receive_window = std::num::NonZeroU64::new(CONNECTION as u64); + config.limits.stream_receive_window = std::num::NonZeroU64::new(STREAM as u64); + config.limits.send_window = std::num::NonZeroU64::new(CONNECTION as u64); + let mut rig = Self { + sessions: Vec::with_capacity(sessions), + streams: Vec::with_capacity(sessions * streams), + drivers: JoinSet::new(), + payload, + }; + for _ in 0..sessions { + let (client, server) = tokio::io::duplex(64 << 10); + let (client, server) = if websocket { + use qmux::ws::{Upgraded, tokio_tungstenite::WebSocketStream, tungstenite::protocol::Role}; + let client = WebSocketStream::from_raw_socket(client, Role::Client, None).await; + let server = WebSocketStream::from_raw_socket(server, Role::Server, None).await; + let (client, driver) = Upgraded::new(client, version.alpn()) + .with_session_config(config.clone()) + .connect() + .unwrap(); + rig.drivers.spawn(driver); + let (server, driver) = Upgraded::new(server, version.alpn()) + .with_session_config(config.clone()) + .accept() + .unwrap(); + rig.drivers.spawn(driver); + (client, server) + } else { + let (client, server) = tokio::join!( + Session::connect(Stream::new(client), config.clone()), + Session::accept(Stream::new(server), config.clone()), + ); + let (client, driver) = client.unwrap(); + rig.drivers.spawn(driver); + let (server, driver) = server.unwrap(); + rig.drivers.spawn(driver); + (client, server) + }; + for _ in 0..streams { + let mut send = client.open_uni(Default::default()).await.unwrap(); + send.write_chunk(rig.payload.clone()).await.unwrap(); + let mut recv = server.accept_uni().await.unwrap(); + // Dropping an unread lease leaves the bytes charged and unread in the stream. + kio::wait(|waiter| recv.poll_lease(waiter, MESSAGE).map_ok(drop)) + .await + .unwrap(); + rig.streams.push((send, recv)); + } + rig.sessions.push((client, server)); + } + rig + } + + async fn round(&mut self) { + for (send, recv) in &mut self.streams { + kio::wait(|waiter| { + recv.poll_lease(waiter, MESSAGE).map_ok(|mut lease| { + let received = lease.copy_to_bytes(MESSAGE); + assert_eq!(received, self.payload); + lease.release(); + }) + }) + .await + .unwrap(); + send.write_chunk(self.payload.clone()).await.unwrap(); + kio::wait(|waiter| recv.poll_lease(waiter, MESSAGE).map_ok(drop)) + .await + .unwrap(); + } + assert!(self.drivers.try_join_next().is_none(), "a session driver ended"); + } + + async fn close(mut self) { + self.streams.clear(); + self.sessions.clear(); + self.drivers.shutdown().await; + } +} + +fn receive(c: &mut Criterion) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let payload = Bytes::from((0..MESSAGE).map(|i| (i % 251) as u8).collect::>()); + let mut group = c.benchmark_group("qmux_receive"); + for (transport, websocket) in [("pipe", false), ("ws", true)] { + for &version in Version::ALL { + for sessions in [1, 8, 64] { + for streams in [1, 8] { + let id = format!("{transport}/{}/{sessions}s/{streams}streams", version.alpn()); + group.throughput(Throughput::Bytes((sessions * streams * MESSAGE) as u64)); + let mut rig = None; + group.bench_function(BenchmarkId::from_parameter(&id), |b| { + let rig = rig.get_or_insert_with(|| { + let before = LIVE.load(Ordering::Relaxed); + let mut rig = runtime.block_on(Receivers::new( + Shape { + websocket, + version, + sessions, + streams, + }, + payload.clone(), + )); + runtime.block_on(rig.round()); + runtime.block_on(rig.round()); + let held = LIVE.load(Ordering::Relaxed) - before; + eprintln!( + "{id}: retained={held} B, per-session={} B, unread-per-session={} B, connection-window={CONNECTION} B", + held / sessions as isize, + streams * MESSAGE + ); + rig + }); + b.iter_custom(|iters| { + runtime.block_on(async { + let start = Instant::now(); + for _ in 0..iters { + rig.round().await; + } + start.elapsed() + }) + }); + }); + if let Some(rig) = rig { + runtime.block_on(rig.close()); + } + } + } + } + } + group.finish(); +} + +criterion_group! { + name = benches; + config = Criterion::default().warm_up_time(Duration::from_secs(1)).measurement_time(Duration::from_secs(2)).sample_size(10); + targets = receive +} +criterion_main!(benches); diff --git a/sh/gh/nightly.sh b/sh/gh/nightly.sh index 93538f69017..2725e2815a8 100755 --- a/sh/gh/nightly.sh +++ b/sh/gh/nightly.sh @@ -139,6 +139,7 @@ case "$1" in ;; # Keep added coverage out of the rust-bench and rust-shared legs. rust-extra) + criterion qmux --manifest-path rs/qmux/Cargo.toml --bench receive # The allocation/window/play-buffer/batch/receive-bursts tables keep their stdout in the run; # only Criterion estimates enter the trend report. criterion extra --no-fail-fast -p moq-crypto -p moq-json -p moq-archive -p moq-cli -p moq-mux -p moq-audio \