diff --git a/.config/nextest.toml b/.config/nextest.toml index 5a0c97c0ed4..eb336ae8e40 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -37,6 +37,12 @@ test-group = "io-uring" filter = "package(moq-ffi) & test(=bench::benchmark_ffi)" slow-timeout = { period = "60s" } +# The receive-window benchmark sweeps a grid twice, once through a bottleneck +# whose rate bounds how fast a run can go, so it outlasts the hang budget. +[[profile.default.overrides]] +filter = "package(moq-tokio) & test(=receive_credit::benchmark_receive_windows)" +slow-timeout = { period = "60s", terminate-after = 5 } + [test-groups] io-uring = { max-threads = 4 } diff --git a/.github/workflows/interop.yml b/.github/workflows/interop.yml index c9c79b954f3..a036636c49b 100644 --- a/.github/workflows/interop.yml +++ b/.github/workflows/interop.yml @@ -65,6 +65,15 @@ jobs: run: nix develop --command just test max-age shell: bash -leo pipefail {0} + - name: Receive credit recovery + id: receive-credit + if: ${{ !cancelled() && steps.setup.outcome == 'success' }} + run: nix develop --command just test receive-credit + shell: bash -leo pipefail {0} + env: + # Fail rather than skip the io_uring arm if the runner's kernel lacks it. + MOQ_STRICT: 1 + - name: Harness regressions id: harness if: ${{ !cancelled() && steps.setup.outcome == 'success' }} @@ -196,6 +205,7 @@ jobs: "Rust cache": "${{ steps.rust-cache.outcome }}", "Playwright Chromium": "${{ steps.setup.outcome }}", "Optional publisher retention": "${{ steps.max-age.outcome }}", + "Receive credit recovery": "${{ steps.receive-credit.outcome }}", "Harness regressions": "${{ steps.harness.outcome }}", "Subscription termination interop": "${{ steps.bare-fin.outcome }}", "P2P data-channel transport": "${{ steps.p2p.outcome }}", diff --git a/.github/workflows/release-gate.yml b/.github/workflows/release-gate.yml index 9005351ec46..9b863fcbbd4 100644 --- a/.github/workflows/release-gate.yml +++ b/.github/workflows/release-gate.yml @@ -112,6 +112,9 @@ jobs: - "test/max-age/**" - "test/portrait/**" - "rs/moq-tokio/tests/integration/max_age.rs" + - "test/receive-credit/**" + - "rs/moq-tokio/tests/integration/receive_credit.rs" + - "rs/moq-uring/tests/integration/receive_credit.rs" # The interop job also depends on the recipe that invokes it and the # workspace membership that links the JS clients to source. - "test/justfile" diff --git a/Cargo.lock b/Cargo.lock index 4a94380c3ef..b8cb020c2fd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5330,6 +5330,7 @@ dependencies = [ "moq-ice", "moq-net", "moq-quic", + "moq-shaper", "moq-sock", "noq-proto", "notify", diff --git a/bun.lock b/bun.lock index ffdc58cf072..847bffa72d3 100644 --- a/bun.lock +++ b/bun.lock @@ -509,6 +509,13 @@ "typescript": "7.0.2", }, }, + "test/receive-credit": { + "name": "@moq/receive-credit-test", + "dependencies": { + "@moq/net": "workspace:*", + "@moq/web-transport": "^0.1.5", + }, + }, "test/video": { "name": "@moq/video-test", "dependencies": { @@ -846,6 +853,8 @@ "@moq/qmux-interop": ["@moq/qmux-interop@workspace:test/qmux"], + "@moq/receive-credit-test": ["@moq/receive-credit-test@workspace:test/receive-credit"], + "@moq/room": ["@moq/room@workspace:js/room"], "@moq/signals": ["@moq/signals@workspace:js/signals"], diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 4adaacf811b..b76ef9e601c 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -77,7 +77,7 @@ idle_timeout = "10s" # Drop a connection after this long with no keep_alive = "3s" # Ping interval; "0s" disables it. iroh still pings at least every 5s. Default. gso = true # UDP segmentation offload. iroh cannot turn it off. mtu_discovery = false # Path MTU discovery. Default. -receive_window = 67108864 # Whole-connection window, in bytes. Default (64 MiB). +receive_window = 134217728 # Whole-connection window, in bytes. Default (128 MiB). stream_receive_window = 4194304 # Per-stream window, in bytes. Default (4 MiB). send_window = 16777216 # Unacknowledged outgoing data, in bytes. Default (16 MiB). qlog = "/var/log/moq/qlog" # Existing directory. Needs the `qlog` build feature. @@ -96,10 +96,16 @@ QUIC uses the smaller of the two endpoints' values, so this also bounds the other end. Keep `keep_alive` under a third of it. Raise the receive windows when a fat, long path idles below the link rate. The -64 MiB `receive_window` default carries several Gbps at a 100 ms RTT and bounds -how much unread data one peer can make the relay buffer. Keep -`stream_receive_window` well under it so one slow group cannot starve the -connection. +128 MiB `receive_window` default carries several Gbps at a 100 ms RTT and bounds +how much unread data and incomplete messages one peer can make the relay +buffer. It has to fit the largest message, just over 64 MiB, so a smaller +value, here or in `[peer.quic]`, is refused at startup. When incomplete +messages fill the window, the relay cancels the least urgent stream: the +lowest subscription priority, then the group its subscription would deliver +last. Over iroh the window bounds only +unread data, and over WebSocket or TCP nothing bounds incomplete messages. +`stream_receive_window` only paces one stream: its credit returns as bytes are +read, whether or not their message is complete. ## \[session] and \[peer] diff --git a/package.json b/package.json index 0d4137b75db..a8542ae3c07 100644 --- a/package.json +++ b/package.json @@ -60,6 +60,7 @@ "test/interop/clients/js", "test/interop/clients/js-native", "test/drain", + "test/receive-credit", "test/video", "test/p2p", "test/qmux", diff --git a/quest/m1/quic/README.md b/quest/m1/quic/README.md index 474050188b0..ca0ad7fa446 100644 --- a/quest/m1/quic/README.md +++ b/quest/m1/quic/README.md @@ -8,8 +8,8 @@ MoQ never negotiates. One core serves the tokio backend, the thread-per-core `moq-uring` backend, and qmux; iroh stays on upstream noq. BBR correctness, reliable stream resets (#5690), hierarchical send groups (#5711, #5770), one `Limits` type for relay peers (#5871), and endpoint sharding (#5688) are on -`main`. What remains is bounded incomplete-message memory and clean exits -for connection-blocked writers; per-broadcast +`main`, and receive leases bound incomplete-message memory (#5878). What +remains is clean exits for connection-blocked writers; per-broadcast fairness gives the crate-private send domains (#5872) their consumer in [Broadcast fairness](/quest/m2/broadcast-fairness.md). qmux already runs on the shared stream state machine, outside this epic. @@ -71,7 +71,6 @@ cache limits remain separate; bitrate pacing composes with this memory cap. ## Required -- [Bound message assembly](/quest/m1/quic/receive-memory.md) - complete headers and frames return credit, and receiver priority cancellation restores progress - [Connection-blocked writers end cleanly](/quest/m1/quic/connection-blocked.md) - an ended stream leaves the blocked-writer list, and STOP_SENDING surfaces under exhausted connection credit ## Related diff --git a/quest/m1/quic/receive-memory.md b/quest/m1/quic/receive-memory.md deleted file mode 100644 index 1409f9fb367..00000000000 --- a/quest/m1/quic/receive-memory.md +++ /dev/null @@ -1,119 +0,0 @@ -# [XL] Bound incomplete messages with QUIC connection credit - -## Goal - -Native MoQ receivers bound incomplete frames and protocol messages by the -connection memory allowance, returning connection credit only when each -complete unit is parsed and handed off or discarded. When concurrent partial -messages prevent progress, the receiver cancels streams by MoQ priority and -continues. This works with Rust and JavaScript senders through tokio and -io_uring; completed media caching remains separately governed. - -## Plan - -Decided in the 2026-10-10 interview: cover headers and frame/group data alike. -The frame preallocation budget only bounds speculative allocation, and the -cache has soft limits; neither bounds aggregate incomplete messages. Use -peer-visible QUIC flow control instead of a separate header quota error. -The native transport primitive is a separate prerequisite; do not activate -whole-message retention without its recovery policy in this quest. - -In flight as [#5878](https://github.com/moq-dev/moq/pull/5878) (ready for -review) on branch `quest/m1/quic/receive-memory-lease`; the earlier drafts -#5735 and #5752 were closed in its favor. - -Decided 2026-10-20 (maintainer): three quests break moq-net's -`transport::poll` traits, so they land in order and each rebases its trait -change on the last: this quest first, then -[Rust request credit](/quest/m1/rs-request-credit.md), then -the send domain in [broadcast fairness](/quest/m2/broadcast-fairness.md). - -That primitive landed as a provisional seam to finalize here: both native -drivers' receive streams have a copying `poll_read_held` that yields a -`Credit` for exactly the bytes read. Dropping a `Credit` returns it, so a -reader buffer keeps one merged `Credit` and drops `split(n)` for each -complete unit; read-ahead stays in the remainder. io_uring's `poll_closed` -read-ahead keeps its credit until the application reads it. Both drivers' -`Connection::stats().credit` reports held and still-available credit. -`poll_read_held` stays separate until this quest folds it into `poll_read`. -moq-net's transport trait and the WebTransport wrappers do not carry it yet; -reshape any of it to fit the consumer. - -Hold connection credit through reader buffers, type/length prefixes, headers, -extensions, and frame assembly. Attribute read-ahead bytes to the next unit -rather than releasing them with the completed one. Release on complete-unit -handoff, not after application playback or cache eviction. Avoid retaining -an entire multi-frame group as one unit. Account for received-but-unread and -out-of-order transport bytes as well as parser-held bytes; they share the -same allowance. Speculative allocations and copies need separate accounting -or bounds: QUIC byte credit is not a claim about total process RSS. - -Keep the existing stream-window setting and honor explicit overrides. -Give the native path a very large default so it normally adds no round trips; -stream credit is replenished on reads, independently of message completion. -Do not remove or ignore the setting without benchmark evidence and a new -maintainer decision. The connection memory allowance is the shared cap. - -Decided 2026-10-16: once [Segmented frames](/quest/m2/unbounded-groups/segmented-frames.md) -lands, a frame payload releases credit per segment rather than per frame, -so frames of any size never need a matching allowance. Do not build on -whole-frame units beyond what that change replaces. - -Require that allowance to fit every supported encoded message, including -framing and extensions. Validate configuration and align native defaults -rather than discovering an impossible frame after receiving it. The current -32 MiB frame limit exceeds io_uring's 16 MiB connection default; a legacy -64 MiB announcement body also needs room for framing beyond 64 MiB. Derive -the minimum from the supported units, not just the payload limit. Lowering -a protocol/frame size limit is a separate maintainer decision, not an implicit -way around this requirement. Runtime memory-limit updates obey the same -minimum; already-advertised credit cannot be revoked. A rate limit can pace -a smaller amount of outstanding credit while the memory allowance remains -large enough for the complete unit. - -MoQ selects victims; the transport only reports pressure and sends -STOP_SENDING. Do not add sender-initiated shedding. This keeps recovery -available with browser senders and avoids treating intentional rate pacing -as receiver deadlock. The sender's required RESET_STREAM response is still -normal QUIC behavior, not a second recovery policy. - -Use the receiver's assembly and credit accounting to recover when partial -units exhaust progress capacity. Do not require peer DATA_BLOCKED, an -arbitrary timeout, loss, or a deadline. Demonstrate that useful work can -complete after cancellation, including when incoming streams keep arriving; -merely cycling through victims is not recovery. Distinguish bytes still able -to arrive under advertised credit from memory held by incomplete units. - -Within the existing broadcast fairness domains, cancel lower-priority media -and less useful groups first, respecting subscription priority and group -order. Prefer low-priority attributed media over unidentified headers so -headers have a chance to identify themselves; unidentified streams remain -eligible if they prevent progress. Control streams have the highest priority -but are eligible too. If canceling an essential control stream terminates the -session under its protocol, surface that outcome instead of leaving tasks -parked. There is no permanently protected class or separate control reserve. -Use deterministic tie-breaking without comparing unrelated broadcasts' raw -priority values. Do not invent another subscription-priority API here. - -Regression coverage must include many partial headers and frames filling the -shared allowance, a finite stream window smaller than a frame, the largest -supported encoded unit, read-ahead across unit boundaries, cancellation and -late reset/data races, unidentified and control-only pressure, and continued -progress after selecting victims. Exercise a JavaScript sender against both -native receive paths and run all-language interop. Measure retained bytes, -throughput, and completion latency across concurrent streams and frame sizes, -including finite versus very large stream windows. Wire checks and benchmarks -into CI; no wall-clock sleeps in unit tests. - -Public API: native receive-credit integration and validation of the existing -window configuration; no new header-specific quota. Wire: existing QUIC -flow-control and STOP_SENDING/RESET_STREAM behavior, no MoQ format change. -Browser/iroh receive-side guarantees and completed-cache limits are outside -this quest. Update existing configuration and transport docs inline; no -separate guide quest is needed. - -## Related - -- [Unbounded groups](/quest/m2/unbounded-groups/README.md) - removes the frame cap; frame payload credit becomes per segment -- [Broadcast fairness](/quest/m2/broadcast-fairness.md) - broadcast fairness domains on the same sessions -- [QUIC bitrate caps](/quest/m2/rate-quic.md) - pace credit grants inside the independently bounded memory allowance diff --git a/quest/m1/release-audit/README.md b/quest/m1/release-audit/README.md index 58c4f100770..d87107521af 100644 --- a/quest/m1/release-audit/README.md +++ b/quest/m1/release-audit/README.md @@ -113,7 +113,6 @@ child records whether its behavior is new on main or inherited from release. - [JS subscription ranges](/quest/m1/subscribe-ranges/js.md) - range lists and frame offsets build on the bounds fixed here - [OBS video source](/quest/m2/obs-moq-video/source.md) - reproduced inherited timing and EOF defects remain with the decoder replacement owner -- [Receive memory](/quest/m1/quic/receive-memory.md) - incomplete reliable input remains a separate connection-credit concern - [QPACK dependency fix](/quest/m0/web-transport-proto-0.6.3.md) - decoded header expansion is already tracked - [Compressed tracks](/quest/m1/compress/README.md) - mid-group widening and application migration have existing owners - [Archive consumer](/quest/m1/archive/consumer.md) - current producer and CLI reader still need the new-format roundtrip diff --git a/quest/m1/release-audit/datagram-memory.md b/quest/m1/release-audit/datagram-memory.md index e22eb461e90..0e3c6995d7a 100644 --- a/quest/m1/release-audit/datagram-memory.md +++ b/quest/m1/release-audit/datagram-memory.md @@ -45,5 +45,4 @@ with the maintainer before implementing it. ## Related -- [Reliable receive memory](/quest/m1/quic/receive-memory.md) - parser-held incomplete streams, not completed datagram queues - [Compressed tracks](/quest/m1/compress/README.md) - shared compression behavior and fallback diff --git a/quest/m1/release-audit/webtransport-streams.md b/quest/m1/release-audit/webtransport-streams.md index 08b8f81fe3d..b89015eddfc 100644 --- a/quest/m1/release-audit/webtransport-streams.md +++ b/quest/m1/release-audit/webtransport-streams.md @@ -31,5 +31,4 @@ if affected. ## Related -- [Receive memory](/quest/m1/quic/receive-memory.md) - classification must preserve connection-credit ownership - [QPACK dependency fix](/quest/m0/web-transport-proto-0.6.3.md) - separate decoded-header expansion problem diff --git a/quest/m1/rs-request-credit.md b/quest/m1/rs-request-credit.md index ebb544e7f5e..ec4bebd5aa1 100644 --- a/quest/m1/rs-request-credit.md +++ b/quest/m1/rs-request-credit.md @@ -36,10 +36,7 @@ Propose the error variant's name in the PR. Public API: one new `Error` variant, and possibly a `transport::poll` session method. Wire: none. Decided 2026-10-20 (maintainer): the `transport::poll` trait breaks land in -order, [receive memory](/quest/m1/quic/receive-memory.md) first, then this -quest, then the send domain in [broadcast fairness](/quest/m2/broadcast-fairness.md), so +order, this quest on the receive-lease shape from +[moq-dev/moq#5878](https://github.com/moq-dev/moq/pull/5878), then the send +domain in [broadcast fairness](/quest/m2/broadcast-fairness.md), so implementers rebase on one shape at a time. - -## Required - -- [Bound message assembly](/quest/m1/quic/receive-memory.md) - lands its `transport::poll` change first; this quest builds its non-waiting open on that shape diff --git a/quest/m2/README.md b/quest/m2/README.md index fb0581d4ad1..a87a29349a5 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -77,6 +77,7 @@ ranked just below deterministic TS legs, which it requires. - [Ordered owed wait](/quest/m2/ordered-owed-wait.md) - an ordered reader waits within max_delay for a missing group only while the track still owes it - [lite-05/06 gap judgement](/quest/m2/lite-legacy-gap.md) - a resumed copy on lite-05/06 never reads a descending track's newest group as its start - [Unbounded groups](/quest/m2/unbounded-groups/README.md) - groups and frames lose their size caps; the cache sheds parts of a group under pressure and refills the holes in place +- [Receive pool](/quest/m2/receive-pool.md) - one byte budget per worker covers every connection's receive credit and the cache, billing each frame once - [Rust-backed @moq/net](/quest/m2/rs2ts/README.md) - measure browser baselines, complete WASM bindings, and adopt Rust behind @moq/net before the later JS-only stage - [Front parking](/quest/m2/origin-front-parks.md) - an unroutable request waits on a front instead of re-asking on every route-table move - [Route wakes](/quest/m2/route-wakes.md) - a route change wakes only the fronts it can move, so pool churn stops scaling with served paths diff --git a/quest/m2/broadcast-fairness.md b/quest/m2/broadcast-fairness.md index ea2b07f2188..46515aa2a7f 100644 --- a/quest/m2/broadcast-fairness.md +++ b/quest/m2/broadcast-fairness.md @@ -41,7 +41,7 @@ Direction carried over from 2026-09-19, not re-confirmed: - The draft would say a relay MAY scope priority to the broadcast on a relay-to-relay connection; no carrier is needed on the wire. - Decided 2026-10-20 (maintainer): the `transport::poll` trait breaks land in - order, [receive memory](/quest/m1/quic/receive-memory.md), then + order, receive leases ([moq-dev/moq#5878](https://github.com/moq-dev/moq/pull/5878)), then [Rust request credit](/quest/m1/rs-request-credit.md), then the send domain. Receive ranks, folded in from @@ -103,4 +103,3 @@ update moves an already-ranked stream. ## Related - [One parked group per subscription](https://github.com/moq-dev/moq/pull/6140) - the parked group each subscription keeps must still yield the next stream by broadcast -- [Bound incomplete messages](/quest/m1/quic/receive-memory.md) - the receive-side counterpart on the same moq-quic sessions diff --git a/quest/m2/rate-quic.md b/quest/m2/rate-quic.md index 256eb569286..8f209bb163b 100644 --- a/quest/m2/rate-quic.md +++ b/quest/m2/rate-quic.md @@ -26,7 +26,7 @@ backpressure the peer cannot ignore, rather than metering and closing. flow-control limited (its congestion controller does not back off) and queues until [the grant](/quest/m2/rate-grant.md) clamps its encoder. Keep rate pacing separate from the connection memory allowance in - [message assembly](/quest/m1/quic/receive-memory.md). The memory allowance + message assembly (receive leases, [#5878](https://github.com/moq-dev/moq/pull/5878)). The memory allowance must fit the largest supported encoded unit even when it takes many rate intervals to receive. Limit each grant by both rate allowance and available memory backing; do not wait for a whole-message credit release to grant @@ -95,7 +95,3 @@ pacing alone does not trigger pressure cancellation. Decided in the 2026-10-10 receive-memory interview: memory allowance and rate pacing are independent bounds. This replaces the earlier assumption that a one-second rate window can also be the complete assembly allowance. - -## Required - -- [Bound message assembly](/quest/m1/quic/receive-memory.md) - separate retained memory from paced outstanding credit diff --git a/quest/m2/receive-pool.md b/quest/m2/receive-pool.md new file mode 100644 index 00000000000..957a31b0ad8 --- /dev/null +++ b/quest/m2/receive-pool.md @@ -0,0 +1,63 @@ +# [L] One receive budget per worker, shared with the cache + +## Goal + +Each worker has one byte budget that covers both the receive credit every +connection holds and the groups its cache keeps. A frame's bytes are billed once, +from the first leased byte until the cache drops them. An operator sets that one +number instead of sizing `quic.receive_window` against the cache capacity and the +expected peer count. + +Today each connection's window bounds only its own incomplete messages, and the +cache `Pool` is billed again for the same bytes once a frame is taken. So the worst +case is connections × 128 MiB plus the cache. A relay with many peers either +reserves memory it never uses or sizes the window below what one peer's largest +unit needs. + +Non-goals: revoking credit already granted, cross-peer eviction, and zero-copy +send buffers (see [registered TX buffers](/quest/m3/3204-moq-uring-register-tx-pool-buffers-for-zero-copy-sends.md)). + +## Plan + +Decided 2026-10-21 in the [moq-dev/moq#5878](https://github.com/moq-dev/moq/pull/5878) +iteration: + +- **Per worker.** One budget per io_uring worker thread, with no cross-thread + atomics; the tokio runtime gets one budget. The cache `Pool` splits the same way, + so each worker's cache and receive credit draw on one number. This reverses the + relay's single process-wide cache pool, which this quest must migrate (config, + `MIGRATE.md`, `doc/bin/relay/`). +- **Under pressure, stop granting.** Credit already advertised is never revoked. + New `MAX_DATA` comes from the budget by fair share, so a peer over its share + stalls rather than evicting anyone else. Eviction stays within one connection, as + in #5878. A peer still cannot cost another peer its streams. +- **Accounting first.** Received bytes stay `Bytes` slices, compacted when a reader + waits on them (#5878). The budget counts leased plus cached bytes once. Move to + QUICHE-style fixed blocks (8 KiB, allocated lazily and freed as consumed) only if a + benchmark shows allocator or fragmentation cost the accounting cannot see. + +Open, for the implementer to bring to the maintainer: + +- **The 64 MiB floor.** Each connection must be able to assemble its largest unit, a + legacy `AnnounceInit` (`MIN_RECEIVE_WINDOW`). Should a connection get its floor + only on a `want` that needs it, or should the announcement cap shrink first? +- **The handoff from lease to cache.** A frame taken from a lease keeps its credit + until complete. Once complete, the credit becomes a cache charge. The charge has + to move to the cache without a window where the bytes are billed twice or not at + all. + +Lessons from other stacks: QUICHE frees 8 KiB blocks as they are consumed, msquic +never advertises credit it could not back, and quinn and lsquic both let a slice +pin its packet buffer. #5878 already handles the last with compaction. + +Benchmark: sweep peers × window over a fixed budget on the `benchmark_receive_windows` +harness. Report peak resident bytes against the budget, evictions, and throughput. + +## Related + +- [Unbounded groups](/quest/m2/unbounded-groups/README.md) - drops the frame cap that + sets half of `MIN_RECEIVE_WINDOW` +- [Perf](/quest/m2/perf/README.md) - the per-worker hot paths this budget must not add + contention to +- [Broadcast fairness](/quest/m2/broadcast-fairness.md) - the receive rank's + fairness domains diff --git a/quest/m2/unbounded-groups/segmented-frames.md b/quest/m2/unbounded-groups/segmented-frames.md index 7ca5d55219b..d862b6b9ba8 100644 --- a/quest/m2/unbounded-groups/segmented-frames.md +++ b/quest/m2/unbounded-groups/segmented-frames.md @@ -22,8 +22,9 @@ See the [epic](/quest/m2/unbounded-groups/README.md) for the decisions. - Choose the segment size from benchmarks, and keep it a storage detail rather than a public knob. Coalescing the chunks the transport delivers is fine. -- Update [Receive memory](/quest/m1/quic/receive-memory.md) so a frame's - payload releases credit per segment. The frame header stays a whole unit. +- Release a frame payload's receive-lease credit per segment, rather than + per frame ([#5878](https://github.com/moq-dev/moq/pull/5878)). The frame + header stays a whole unit. - Measure `group.rs`, `frame.rs` (`frame_recv`), and `memory.rs` before and after. Add a large-frame case swept over frame size. diff --git a/rs/moq-ffi/src/transport/adapter.rs b/rs/moq-ffi/src/transport/adapter.rs index 7b6a08e2e10..1474709552a 100644 --- a/rs/moq-ffi/src/transport/adapter.rs +++ b/rs/moq-ffi/src/transport/adapter.rs @@ -1,4 +1,4 @@ -use bytes::{Buf, BufMut, Bytes}; +use bytes::{Buf, Bytes}; use moq_net::transport::{self as moq, poll}; use std::task::{Context, Poll}; use web_transport_trait::{self as wt, poll as backend}; @@ -55,17 +55,18 @@ impl Session { } impl + Clone + 'static> poll::Session for Session { type SendStream = SendStream; - type RecvStream = RecvStream; + // The backend cannot hold its receive window for moq-net, so reads are buffered here. + type RecvStream = poll::Buffered>; type Error = Error; fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { backend::Session::poll_accept_uni(&mut self.0, cx) .map_err(Error) - .map(|res| res.map(RecvStream)) + .map(|res| res.map(|recv| poll::Buffered::new(RecvStream(recv)))) } fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll, Self::Error>> { backend::Session::poll_accept_bi(&mut self.0, cx) .map_err(Error) - .map(|res| res.map(|(send, recv)| (SendStream(send), RecvStream(recv)))) + .map(|res| res.map(|(send, recv)| (SendStream(send), poll::Buffered::new(RecvStream(recv))))) } fn poll_open_uni( &mut self, @@ -83,7 +84,14 @@ impl + Clone + 'st ) -> Poll, Self::Error>> { backend::Session::poll_open_bi(&mut self.0, cx) .map_err(Error) - .map(|res| res.map(|(send, recv)| (SendStream(send).scheduled(options), RecvStream(recv)))) + .map(|res| { + res.map(|(send, recv)| { + ( + SendStream(send).scheduled(options), + poll::Buffered::new(RecvStream(recv)), + ) + }) + }) } fn send_group(&mut self, _: i64) -> Option { None @@ -156,27 +164,14 @@ impl poll::SendStream for SendStream { } /// An incoming WebTransport stream adapted to moq-net. pub struct RecvStream(S); -impl poll::RecvStream for RecvStream { +impl poll::Chunks for RecvStream { type Error = Error; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { - backend::RecvStream::poll_read(&mut self.0, cx, dst).map_err(Error) - } - fn poll_read_buf( - &mut self, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll, Self::Error>> { - backend::RecvStream::poll_read_buf(&mut self.0, cx, buf).map_err(Error) - } fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { backend::RecvStream::poll_read_chunk(&mut self.0, cx, max).map_err(Error) } fn stop(&mut self, code: u32) { backend::RecvStream::stop(&mut self.0, code) } - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { - backend::RecvStream::poll_closed(&mut self.0, cx).map_err(Error) - } } #[cfg(test)] diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index f60898f5f5a..11aa62bf313 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,28 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Receive streams are read through leases (#5878) + +`transport::poll::RecvStream` now has `poll_lease(cx, want)`, `set_priority`, and `stop`, plus the provided `poll_read_chunk` and `read_chunk`. +`poll_read`, `poll_read_buf`, `poll_closed`, `read`, `read_buf`, `closed`, `Read`, and `ReadBuf` are gone. +A lease borrows the stream's unreleased bytes as a `bytes::Buf`: reading moves past bytes for good, but they keep their flow-control credit until `Lease::release()`. +Bytes not read stay for the next lease, `Lease::chunks()` peeks without reading, and `is_fin` replaces `closed`. +A transport that can only hand out chunks implements `transport::poll::Chunks` (`poll_read_chunk` and `stop`) and wraps each stream in `transport::poll::Buffered`. + +```rust +// Before +let n = stream.read(&mut buf).await?; + +// After: copy what arrived, then return its credit +let mut lease = poll_fn(|cx| stream.poll_lease(cx, 1)).await?; +let n = lease.remaining().min(buf.len()); // 0 with `lease.is_fin()` is the end +lease.copy_to_slice(&mut buf[..n]); +lease.release(); +``` + +`transport::CloseKind` gained `Evicted`, which `Error::from_transport` maps to `Error::Evicted`. +`transport::Stats` gained `streams_evicted()`, and `session::Stats` the matching `streams_evicted` field. + ### Stage a requested track before subscribers see it (#6111) `track::Request::stage(info)` returns a `track::Staged` that derefs to the track's `Producer` while subscribers waiting on the request stay pending; `commit()` publishes it with everything written so far and returns the `Producer`. diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 889a4aa9598..5a2f9fa5231 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -593,9 +593,9 @@ mod tests { fn new(protocol: Option<&'static str>, server_control_bytes: Vec) -> Self { let writes = Arc::new(Mutex::new(Vec::new())); let send = FakeSendStream { writes: writes.clone() }; - let recv = FakeRecvStream { + let recv = crate::transport::poll::Buffered::new(FakeChunks { data: VecDeque::from(server_control_bytes), - }; + }); let state = FakeSessionState { protocol, control_stream: Mutex::new(Some((send, recv))), @@ -766,30 +766,25 @@ mod tests { } } - struct FakeRecvStream { + type FakeRecvStream = crate::transport::poll::Buffered; + + struct FakeChunks { data: VecDeque, } - impl crate::transport::poll::RecvStream for FakeRecvStream { + impl crate::transport::poll::Chunks for FakeChunks { type Error = FakeError; - fn poll_read(&mut self, _cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_read_chunk(&mut self, _cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { if self.data.is_empty() { return Poll::Ready(Ok(None)); } - let size = dst.len().min(self.data.len()); - for slot in dst.iter_mut().take(size) { - *slot = self.data.pop_front().unwrap(); - } - Poll::Ready(Ok(Some(size))) + let size = max.min(self.data.len()); + Poll::Ready(Ok(Some(self.data.drain(..size).collect()))) } fn stop(&mut self, _code: u32) {} - - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) - } } fn mock_server_setup(negotiated: Version) -> Vec { @@ -1151,17 +1146,18 @@ mod tests { impl crate::transport::poll::RecvStream for LocalRecv { type Error = FakeError; + type Lease<'a> = crate::transport::poll::BufferedLease<'a>; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { - self.inner.poll_read(cx, dst) + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Self::Error>> { + self.inner.poll_lease(cx, want) } - fn stop(&mut self, code: u32) { - self.inner.stop(code); + fn set_priority(&mut self, priority: i64) { + self.inner.set_priority(priority); } - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { - crate::transport::poll::RecvStream::poll_closed(&mut self.inner, cx) + fn stop(&mut self, code: u32) { + crate::transport::poll::RecvStream::stop(&mut self.inner, code); } } diff --git a/rs/moq-net/src/coding/decode.rs b/rs/moq-net/src/coding/decode.rs index dd756e7034a..d1ba55f70f8 100644 --- a/rs/moq-net/src/coding/decode.rs +++ b/rs/moq-net/src/coding/decode.rs @@ -137,13 +137,37 @@ impl From for DecodeError { pub struct Decoder<'a> { buf: &'a [u8], form: Form, + /// The length of the buffer this decoder started with. + len: usize, + /// See [`Self::need`]. + need: usize, } impl<'a> Decoder<'a> { /// Read `buf`, with varints in the given form. #[cfg_attr(not(feature = "async"), allow(dead_code))] pub fn new(buf: &'a [u8], form: Form) -> Self { - Self { buf, form } + Self { + buf, + form, + len: buf.len(), + need: 0, + } + } + + /// After a [`DecodeError::Short`], how many bytes from the start the failed read + /// needed: at least one more than the buffer when it can't tell, and 0 if no read of + /// this decoder ran short. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub fn need(&self) -> usize { + self.need + } + + /// Record that a read wanted `more` bytes past the current position. + fn short(&mut self, more: usize) -> DecodeError { + let need = (self.len - self.buf.len()).saturating_add(more); + self.need = self.need.max(need); + DecodeError::Short } /// The varint form this decoder reads. @@ -168,7 +192,7 @@ impl<'a> Decoder<'a> { #[cfg_attr(not(feature = "async"), allow(dead_code))] pub fn slice(&mut self, len: usize) -> Result<&'a [u8], DecodeError> { let Some((head, rest)) = self.buf.split_at_checked(len) else { - return Err(DecodeError::Short); + return Err(self.short(len)); }; self.buf = rest; Ok(head) @@ -214,9 +238,15 @@ impl<'a> Decoder<'a> { #[cfg_attr(not(target_arch = "wasm32"), inline(always))] #[cfg_attr(not(feature = "async"), allow(dead_code))] pub fn varint(&mut self) -> Result { - let (value, rest) = varint::read(self.buf, self.form)?; - self.buf = rest; - Ok(value) + match varint::read(self.buf, self.form) { + Ok((value, rest)) => { + self.buf = rest; + Ok(value) + } + // The width is in the first byte, but a byte at a time is short enough. + Err(DecodeError::Short) => Err(self.short(self.buf.len() + 1)), + Err(err) => Err(err), + } } /// Read an optional varint: 0 is `None`, and `n + 1` is `Some(n)`. @@ -256,4 +286,18 @@ mod tests { assert!(matches!(r.varint(), Err(DecodeError::Short))); assert_eq!(r.remaining(), 1); } + + /// A short read reports how much it needed from the start, so a reader can wait for + /// all of it rather than retry once per byte. + #[test] + fn short_reports_what_it_needed() { + let mut r = Decoder::new(&[0x01, 0x05, b'a', b'b'], Form::Quic); + assert_eq!(r.varint().unwrap(), 1); + assert!(matches!(r.bytes(), Err(DecodeError::Short))); + assert_eq!(r.need(), 7, "a 5 byte body after two prefix bytes"); + + let mut r = Decoder::new(&[0x40], Form::Quic); + assert!(matches!(r.varint(), Err(DecodeError::Short))); + assert_eq!(r.need(), 2); + } } diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index 4c8da6624c3..61a9ceb0d04 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -1,23 +1,31 @@ use std::{ - cmp, fmt::Debug, task::{Context, Poll, ready}, }; -use bytes::{Buf, Bytes, BytesMut}; +use bytes::{Buf, Bytes}; -use crate::{Error, StreamError, coding::*}; +use crate::{ + Error, StreamError, + coding::*, + transport::poll::{Lease, RecvStream}, +}; /// A reader for decoding messages from a stream. /// /// The `poll_*` methods are the implementation; the `async` methods are thin /// wrappers, so the reader can be driven from another poll function without -/// pinning a future. Partial bytes accumulate in the internal buffer, so a -/// `Pending` (or a cancelled wrapper) never desynchronizes the stream. -pub struct Reader { +/// pinning a future. Unconsumed bytes stay leased in the transport, which counts them +/// against its flow control until each unit is complete, so a `Pending` (or a cancelled +/// wrapper) never desynchronizes the stream and a parked reader holds no memory of its own. +pub struct Reader { stream: S, - buffer: BytesMut, version: V, + /// How many bytes the last short decode needed, so the retry waits for all of them + /// instead of waking once per chunk. + want: usize, + /// A contiguous copy of a message that spans transport chunks, empty between decodes. + scratch: Vec, } /// How much payload may pile up behind a parked consumer before @@ -31,17 +39,29 @@ pub struct Reader { /// enough that an ordinary burst is still one wake. const WAKE_BUDGET: usize = 64 * 1024; -impl Reader { +/// The largest [`Reader::scratch`] kept between decodes; one past this is freed. +const SCRATCH_KEEP: usize = 64 * 1024; + +impl Reader { pub fn new(stream: S, version: V) -> Self { Self { stream, - buffer: Default::default(), version, + want: 0, + scratch: Vec::new(), } } - /// Poll `decode` against the buffered bytes, reading more until it stops coming up - /// short. `consume` drops what it read; otherwise the bytes stay for the next read. + /// Poll the stream for a lease of at least `want` bytes. + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Error>> { + let version = &self.version; + self.stream + .poll_lease(cx, want) + .map_err(|err| version.transport_error(err)) + } + + /// Poll `decode` against the leased bytes, leasing more until it stops coming up + /// short. `consume` releases what it read; otherwise the bytes stay for the next read. fn poll_with( &mut self, cx: &mut Context<'_>, @@ -51,22 +71,67 @@ impl Reader { where V: Into
+ Copy, { + let form = self.version.into(); loop { - let mut r = Decoder::new(&self.buffer, self.version.into()); - match decode(&mut r) { - Ok(value) => { - if consume { - let used = self.buffer.len() - r.remaining(); - self.buffer.advance(used); + let want = self.want.max(1); + let version = &self.version; + let mut lease = ready!(self.stream.poll_lease(cx, want)).map_err(|err| version.transport_error(err))?; + + // Decode against the first chunk, or a copy of as much as the last attempt needed + // when that spans chunks. Only a message cut across chunks pays for the copy. + let mut size = lease.chunk().len().max(want.min(lease.remaining())); + loop { + let chunk = lease.chunk(); + let buf: &[u8] = match chunk.len() >= size { + true => chunk, + false => { + self.scratch.clear(); + for chunk in lease.chunks() { + let take = chunk.len().min(size - self.scratch.len()); + self.scratch.extend_from_slice(&chunk[..take]); + if self.scratch.len() == size { + break; + } + } + &self.scratch } - return Poll::Ready(Ok(value)); + }; + + let mut r = Decoder::new(buf, form); + let need = match decode(&mut r) { + Ok(value) => { + let used = buf.len() - r.remaining(); + self.want = 0; + if self.scratch.capacity() > SCRATCH_KEEP { + self.scratch = Vec::new(); + } + if consume { + lease.advance(used); + lease.release(); + } + return Poll::Ready(Ok(value)); + } + // A short read inside a complete body is malformed, but reports what it + // needed from the body, so always ask for at least one byte more. + Err(DecodeError::Short) => r.need().max(buf.len() + 1), + Err(e) => return Poll::Ready(Err(e.into())), + }; + + if need <= lease.remaining() { + size = need; + continue; } // Stream closed while we still need more data. - Err(DecodeError::Short) if !ready!(self.poll_read_more(cx))? => { + if lease.is_fin() { return Poll::Ready(Err(DecodeError::Short.into())); } - Err(DecodeError::Short) => {} - Err(e) => return Poll::Ready(Err(e.into())), + self.want = need; + // The lease keeps every byte for the retry, which copies them again, so a + // parked reader holds no more of its own than an idle one. + if self.scratch.capacity() > SCRATCH_KEEP { + self.scratch = Vec::new(); + } + break; } } } @@ -190,19 +255,21 @@ impl Reader { std::future::poll_fn(|cx| self.poll_varint_peek(cx)).await } - /// Poll for the next chunk, draining the reader's internal buffer first. + /// Poll for the next chunk of up to `max` bytes, or `None` once the stream finished. + /// Its credit returns as it is read. pub fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Error>> { - if !self.buffer.is_empty() { - let n = cmp::min(self.buffer.len(), max); - return Poll::Ready(Ok(Some(self.buffer.split_to(n).freeze()))); - } + self.want = 0; + let version = &self.version; self.stream .poll_read_chunk(cx, max) - .map_err(|err| self.version.transport_error(err)) + .map_err(|err| version.transport_error(err)) } /// Fill a frame's payload from the stream, returning `Pending` once it would block. /// + /// The payload reaches the frame as it arrives, but keeps its receive credit until the + /// frame is full, so incomplete frames stay bounded by the receive window. + /// /// `Ready(Ok(()))` means the frame is full and the caller should commit it; an error /// means the caller should abort it. Either way the frame is untouched otherwise, so /// the caller keeps ownership of the commit/abort decision. @@ -220,64 +287,82 @@ impl Reader { cx: &mut Context<'_>, frame: &mut crate::frame::Producer, ) -> Poll> { + self.want = 0; let mut owed = 0; let result = loop { - if frame.remaining() == 0 { - break Ok(()); - } - - match self.poll_read_chunk(cx, frame.remaining()) { + let mut lease = match self.poll_lease(cx, frame.remaining().min(1)) { Poll::Pending => { if owed > 0 { frame.notify(); } return Poll::Pending; } - Poll::Ready(Ok(Some(chunk))) if !chunk.is_empty() => { - owed += chunk.len(); - if let Err(err) = frame.write_deferred(chunk) { - break Err(err); - } - if owed >= WAKE_BUDGET { - frame.notify(); - owed = 0; - } - } - // A FIN mid-payload: the declared size never arrived. - Poll::Ready(Ok(_)) => break Err(Error::WrongSize), + Poll::Ready(Ok(lease)) => lease, Poll::Ready(Err(err)) => break Err(err), + }; + + // One chunk per lease: a lease may hold the transport's lock, so the frame is + // written, and its consumers woken, only once the lease is gone. + let chunk = lease.copy_to_bytes(lease.chunk().len().min(frame.remaining())); + let complete = chunk.len() == frame.remaining(); + let ended = lease.is_fin() && !lease.has_remaining(); + if complete { + // Every byte of the frame is in hand, so its credit returns. + lease.release(); + } else { + drop(lease); + } + + if !chunk.is_empty() { + owed += chunk.len(); + if let Err(err) = frame.write_deferred(chunk) { + return Poll::Ready(Err(err)); + } + if owed >= WAKE_BUDGET { + frame.notify(); + owed = 0; + } + } + if complete { + break Ok(()); + } + // A FIN mid-payload: the declared size never arrived. + if ended { + break Err(Error::WrongSize); } }; Poll::Ready(result) } - /// Poll for exactly `size` bytes, accumulating partial reads in the buffer. + /// Poll for exactly `size` bytes. pub fn poll_read_exact(&mut self, cx: &mut Context<'_>, size: usize) -> Poll> { - while self.buffer.len() < size { - if !ready!(self.poll_read_more(cx))? { - return Poll::Ready(Err(DecodeError::Short.into())); - } + self.want = 0; + let mut lease = ready!(self.poll_lease(cx, size))?; + if lease.remaining() < size { + return Poll::Ready(Err(DecodeError::Short.into())); } - Poll::Ready(Ok(self.buffer.split_to(size).freeze())) + let bytes = lease.copy_to_bytes(size); + lease.release(); + Poll::Ready(Ok(bytes)) } - /// Poll to discard `remaining` bytes, counting it down as they arrive. Nothing is - /// buffered past what one read brings, so a peer cannot make a skipped payload cost - /// memory. + /// Poll to discard `remaining` bytes, counting it down as they arrive. Each byte is + /// released as it is skipped, so a peer cannot make a skipped payload cost credit. pub(crate) fn poll_skip(&mut self, cx: &mut Context<'_>, remaining: &mut usize) -> Poll> { - loop { - let skipped = self.buffer.len().min(*remaining); - self.buffer.advance(skipped); - *remaining -= skipped; - if *remaining == 0 { - return Poll::Ready(Ok(())); - } - if !ready!(self.poll_read_more(cx))? { + self.want = 0; + while *remaining > 0 { + let mut lease = ready!(self.poll_lease(cx, 1))?; + if !lease.has_remaining() { return Poll::Ready(Err(DecodeError::Short.into())); } + let skipped = lease.remaining().min(*remaining); + lease.advance(skipped); + lease.release(); + *remaining -= skipped; } + Poll::Ready(Ok(())) } /// Read exactly the given number of bytes from the stream. @@ -294,23 +379,15 @@ impl Reader { Poll::Ready(Ok(())) } - /// Poll for whether data is available in the buffer or stream: `false` once it finishes. + /// Poll for whether data is available on the stream: `false` once it finishes. pub(crate) fn poll_has_more(&mut self, cx: &mut Context<'_>) -> Poll> { - if !self.buffer.is_empty() { - return Poll::Ready(Ok(true)); - } - - self.poll_read_more(cx) + let lease = ready!(self.poll_lease(cx, 1))?; + Poll::Ready(Ok(lease.has_remaining())) } - /// Poll for more data from the stream. `true` if data was read, `false` if the - /// stream finished. - fn poll_read_more(&mut self, cx: &mut Context<'_>) -> Poll> { - match ready!(self.stream.poll_read_buf(cx, &mut self.buffer)) { - Ok(Some(_)) => Poll::Ready(Ok(true)), - Ok(None) => Poll::Ready(Ok(false)), - Err(e) => Poll::Ready(Err(self.version.transport_error(e))), - } + /// Rank the stream for eviction under receive pressure. See [`RecvStream::set_priority`]. + pub fn set_priority(&mut self, priority: i64) { + self.stream.set_priority(priority); } /// Abort the stream with the given error. @@ -325,20 +402,22 @@ impl Reader { pub fn with_version(self, version: V2) -> Reader { Reader { stream: self.stream, - buffer: self.buffer, version, + want: self.want, + scratch: self.scratch, } } } #[cfg(test)] mod tests { - use std::sync::Arc; + use std::{collections::VecDeque, sync::Arc}; use futures::FutureExt; use super::*; use crate::StreamError; + use crate::transport::poll::{Buffered, Chunks}; /// Records the STOP_SENDING code so a test can assert which registry it came from. #[derive(Default)] @@ -346,24 +425,20 @@ mod tests { stops: Vec, } - impl crate::transport::poll::RecvStream for StopLog { + impl Chunks for StopLog { type Error = crate::lite::test_transport::SinkError; - fn poll_read( - &mut self, - _cx: &mut std::task::Context<'_>, - _dst: &mut [u8], - ) -> std::task::Poll, Self::Error>> { - std::task::Poll::Ready(Ok(None)) + fn poll_read_chunk(&mut self, _: &mut Context<'_>, _: usize) -> Poll, Self::Error>> { + Poll::Ready(Ok(None)) } fn stop(&mut self, code: u32) { self.stops.push(code); } + } - fn poll_closed(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll> { - std::task::Poll::Ready(Ok(())) - } + fn stop_log(version: V) -> Reader, V> { + Reader::new(Buffered::new(StopLog::default()), version) } /// STOP_SENDING refuses a stream, so it must carry a stream code. Reusing the session @@ -378,9 +453,13 @@ mod tests { (Error::Lagged, StreamError::TooFarBehind.to_code()), (Error::Unauthorized, StreamError::Unauthorized.to_code()), ] { - let mut reader = Reader::new(StopLog::default(), VERSION); + let mut reader = stop_log(VERSION); reader.abort(&err); - assert_eq!(reader.stream.stops, vec![expected], "{err:?} used the wrong registry"); + assert_eq!( + reader.stream.get_ref().stops, + vec![expected], + "{err:?} used the wrong registry" + ); } // The two registries disagree about 0 and 1, which is what makes the mix-up silent. @@ -392,15 +471,15 @@ mod tests { /// the same abort has to leave a different number on each wire. #[test] fn abort_stops_with_the_negotiated_protocols_code() { - let mut lite = Reader::new(StopLog::default(), crate::lite::Version::Lite05); + let mut lite = stop_log(crate::lite::Version::Lite05); lite.abort(&Error::Old); - let mut ietf = Reader::new(StopLog::default(), crate::ietf::Version::Draft20); + let mut ietf = stop_log(crate::ietf::Version::Draft20); ietf.abort(&Error::Old); - assert_eq!(lite.stream.stops, vec![StreamError::Old.to_code()]); - assert_eq!(ietf.stream.stops, vec![crate::ietf::error::INTERNAL_ERROR]); - assert_ne!(lite.stream.stops, ietf.stream.stops); + assert_eq!(lite.stream.get_ref().stops, vec![StreamError::Old.to_code()]); + assert_eq!(ietf.stream.get_ref().stops, vec![crate::ietf::error::INTERNAL_ERROR]); + assert_ne!(lite.stream.get_ref().stops, ietf.stream.get_ref().stops); } /// Counts wakes delivered to a parked consumer. @@ -445,32 +524,53 @@ mod tests { } /// Hands back one queued chunk per poll, then blocks: a stream whose socket has - /// delivered a burst and has nothing more for this turn. + /// delivered a burst and has nothing more for this turn. Finishes once `fin` is set and + /// the queue is empty. #[derive(Default)] - struct Chunks(std::collections::VecDeque<&'static [u8]>); + struct Queue { + chunks: VecDeque<&'static [u8]>, + fin: bool, + } - impl crate::transport::poll::RecvStream for Chunks { + impl Chunks for Queue { type Error = crate::lite::test_transport::SinkError; - fn poll_read(&mut self, _cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { - let Some(chunk) = self.0.pop_front() else { - return Poll::Pending; + fn poll_read_chunk(&mut self, _: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { + let Some(chunk) = self.chunks.pop_front() else { + return match self.fin { + true => Poll::Ready(Ok(None)), + false => Poll::Pending, + }; }; - // The caller sizes `dst`, so a chunk larger than it spans two reads. - let n = chunk.len().min(dst.len()); - dst[..n].copy_from_slice(&chunk[..n]); + // The caller sizes the read, so a chunk larger than it spans two reads. + let n = chunk.len().min(max); if n < chunk.len() { - self.0.push_front(&chunk[n..]); + self.chunks.push_front(&chunk[n..]); } - Poll::Ready(Ok(Some(n))) + Poll::Ready(Ok(Some(Bytes::from_static(&chunk[..n])))) } fn stop(&mut self, _code: u32) {} + } - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Pending + fn queue(chunks: &[&'static [u8]], version: V) -> Reader, V> { + Reader::new( + Buffered::new(Queue { + chunks: chunks.iter().copied().collect(), + fin: false, + }), + version, + ) + } + + /// How many bytes the reader still has leased. + fn leased(reader: &mut Reader, V>) -> usize { + let mut cx = Context::from_waker(std::task::Waker::noop()); + match reader.stream.poll_lease(&mut cx, 0) { + Poll::Ready(Ok(lease)) => lease.remaining(), + _ => panic!("a lease of nothing never waits"), } } @@ -478,28 +578,76 @@ mod tests { /// waiting on more has counted down what it already discarded. #[test] fn skip_discards_across_reads() { - let chunks = [b"abc".as_slice(), b"defg", b"hi"]; - let mut reader = Reader::new(Chunks(chunks.into()), crate::lite::Version::Lite05); + let mut reader = queue(&[b"abc", b"defg", b"hi"], crate::lite::Version::Lite05); let mut cx = Context::from_waker(std::task::Waker::noop()); let mut remaining = 6; assert!(matches!(reader.poll_skip(&mut cx, &mut remaining), Poll::Ready(Ok(())))); assert_eq!(&reader.read_exact(3).now_or_never().unwrap().unwrap()[..], b"ghi"); - let mut reader = Reader::new(Chunks([b"ab".as_slice(), b"cd"].into()), crate::lite::Version::Lite05); + let mut reader = queue(&[b"ab", b"cd"], crate::lite::Version::Lite05); let mut remaining = 5; assert!(reader.poll_skip(&mut cx, &mut remaining).is_pending()); assert_eq!(remaining, 1); - assert!(reader.buffer.is_empty(), "nothing skipped is kept"); + assert_eq!(leased(&mut reader), 0, "nothing skipped is kept"); + } + + /// A message cut across chunks decodes from a copy, waiting for exactly what its + /// prefix declared, and consumes only itself. + #[test] + fn a_message_across_chunks_waits_for_its_body() { + // A size-prefixed goaway: size 3, then a 2 byte URI. + let mut reader = queue(&[b"\x03", b"\x02a"], crate::lite::Version::Lite05); + let mut cx = Context::from_waker(std::task::Waker::noop()); + assert!(reader.poll_decode::(&mut cx).is_pending()); + assert_eq!(reader.want, 4, "the prefix declared the whole message"); + + reader.stream.get_mut().chunks.push_back(b"bnext"); + assert!(matches!( + reader.poll_decode::(&mut cx), + Poll::Ready(Ok(_)) + )); + assert_eq!(leased(&mut reader), 4, "what follows the message stays leased"); + assert!(reader.scratch.is_empty() || reader.scratch.capacity() <= SCRATCH_KEEP); + } + + /// A copy that spans chunks is freed when the reader parks for more: the lease still + /// holds those bytes, so keeping the copy would double what a peer can make it hold. + #[test] + fn a_parked_decode_frees_its_copy() { + const LEN: usize = 70_000; + let mut head = Vec::new(); + Encoder::new(&mut head, crate::lite::Version::Lite05.into()) + .varint(LEN as u64) + .unwrap(); + head.extend_from_slice(&[0; 10]); + let head: &'static [u8] = head.leak(); + let body: &'static [u8] = vec![0; LEN - 10].leak(); + + // Two length-prefixed fields: the first spans both chunks, the second never arrives. + let mut reader = queue(&[head, body], crate::lite::Version::Lite05); + let mut cx = Context::from_waker(std::task::Waker::noop()); + let res = reader.poll_with(&mut cx, true, |r| { + r.bytes()?; + r.bytes()?; + Ok(()) + }); + assert!(res.is_pending()); + const { assert!(SCRATCH_KEEP < LEN) }; + assert!(reader.scratch.capacity() <= SCRATCH_KEEP); + assert_eq!( + leased(&mut reader), + head.len() + body.len(), + "the lease keeps the bytes" + ); } fn refuses_frame + Debug, V: StreamCodes + Into + Copy>( - prefix: &[u8], + prefix: &'static [u8], version: V, oversized: bool, ) { - let mut reader = Reader::new(Chunks([b"following bytes".as_slice()].into()), version); - reader.buffer.extend_from_slice(prefix); + let mut reader = queue(&[prefix, b"following bytes"], version); let mut cx = Context::from_waker(std::task::Waker::noop()); let result = reader.poll_decode::(&mut cx); if oversized { @@ -516,8 +664,8 @@ mod tests { "{result:?}" ); } - assert_eq!(reader.buffer.len(), prefix.len()); - assert_eq!(reader.stream.0.len(), 1, "read past the complete frame"); + assert_eq!(reader.stream.get_ref().chunks.len(), 1, "read past the complete frame"); + assert_eq!(leased(&mut reader), prefix.len()); } #[test] @@ -531,17 +679,9 @@ mod tests { Version::Lite(lite::Version::Lite02), Version::Ietf(ietf::Version::Draft14), ] { - let prefix = if version.is_lite() { - &[0x20, 0][..] - } else { - &[0x20, 0, 0][..] - }; + let prefix: &'static [u8] = if version.is_lite() { &[0x20, 0] } else { &[0x20, 0, 0] }; refuses_frame::(prefix, version, false); - let prefix = if version.is_lite() { - &[0x21, 0][..] - } else { - &[0x21, 0, 0][..] - }; + let prefix: &'static [u8] = if version.is_lite() { &[0x21, 0] } else { &[0x21, 0, 0] }; refuses_frame::(prefix, version, false); } } @@ -553,15 +693,16 @@ mod tests { for size in [65536u64, (1 << 40) + 1] { let mut prefix = vec![0x20]; Encoder::new(&mut prefix, version.into()).varint(size).unwrap(); - refuses_frame::(&prefix, Version::Lite(version), true); + let client: &'static [u8] = prefix.clone().leak(); + refuses_frame::(client, Version::Lite(version), true); prefix[0] = 0x21; - refuses_frame::(&prefix, Version::Lite(version), true); + refuses_frame::(prefix.leak(), Version::Lite(version), true); } } for version in [lite::Version::Lite05, lite::Version::Lite06] { let mut prefix = Vec::new(); Encoder::new(&mut prefix, version.into()).varint(65536).unwrap(); - refuses_frame::(&prefix, version, true); + refuses_frame::(prefix.leak(), version, true); } } @@ -593,15 +734,28 @@ mod tests { #[test] fn fragmented_setup_waits_for_its_body() { - let mut reader = Reader::new(Chunks([b"\x01".as_slice()].into()), crate::lite::Version::Lite05); + let mut reader = queue(&[b"\x01"], crate::lite::Version::Lite05); let mut cx = Context::from_waker(std::task::Waker::noop()); assert!(reader.poll_decode::(&mut cx).is_pending()); - reader.stream.0.push_back(b"\x00"); + reader.stream.get_mut().chunks.push_back(b"\x00"); assert!(matches!( reader.poll_decode::(&mut cx), Poll::Ready(Ok(_)) )); - assert!(reader.buffer.is_empty()); + assert_eq!(leased(&mut reader), 0); + } + + /// A FIN before the declared payload arrived is a wrong size, not a complete frame. + #[test] + fn a_fin_mid_frame_is_a_wrong_size() { + let (mut frame, _payload, _waiter, _wakes) = parked_frame(9); + let mut reader = queue(&[b"foo"], crate::lite::Version::Lite05); + reader.stream.get_mut().fin = true; + let mut cx = Context::from_waker(std::task::Waker::noop()); + assert!(matches!( + reader.poll_read_frame(&mut cx, &mut frame), + Poll::Ready(Err(Error::WrongSize)) + )); } /// A consumer parked on the group cannot run until this task yields, so a burst of @@ -612,7 +766,7 @@ mod tests { let (mut frame, mut payload, waiter, wakes) = parked_frame(9); assert!(payload.poll_read_chunk(&waiter).is_pending()); - let mut reader = Reader::new(Chunks([b"foo".as_slice(), b"bar"].into()), crate::lite::Version::Lite05); + let mut reader = queue(&[b"foo", b"bar"], crate::lite::Version::Lite05); let mut cx = Context::from_waker(std::task::Waker::noop()); assert!(reader.poll_read_frame(&mut cx, &mut frame).is_pending()); @@ -630,7 +784,7 @@ mod tests { // The tail completes the frame, so the loop returns without a wake of its own. // The commit is what publishes it, and the commit is the caller's to make. - reader.stream.0.push_back(b"baz"); + reader.stream.get_mut().chunks.push_back(b"baz"); assert!(matches!( reader.poll_read_frame(&mut cx, &mut frame), Poll::Ready(Ok(())) @@ -645,26 +799,22 @@ mod tests { /// before handing over each chunk, so every wake the drain loop issues while the /// burst is in flight lands on a parked waiter and is counted. struct Burst { - chunks: Chunks, + chunks: Queue, payload: crate::frame::Consumer, waiter: kio::Waiter, } - impl crate::transport::poll::RecvStream for Burst { + impl Chunks for Burst { type Error = crate::lite::test_transport::SinkError; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { // Start each turn with none of the last turn's registrations left. self.waiter = kio::Waiter::new(self.waiter.waker().clone()); while self.payload.poll_read_chunk(&self.waiter).is_ready() {} - self.chunks.poll_read(cx, dst) + self.chunks.poll_read_chunk(cx, max) } fn stop(&mut self, _code: u32) {} - - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Pending - } } /// A sender at line rate can keep the drain loop `Ready` for as long as it likes, so @@ -680,11 +830,14 @@ mod tests { assert!(payload.poll_read_chunk(&waiter).is_pending()); let mut reader = Reader::new( - Burst { - chunks: Chunks(std::iter::repeat_n(CHUNK, chunks).collect()), + Buffered::new(Burst { + chunks: Queue { + chunks: std::iter::repeat_n(CHUNK, chunks).collect(), + fin: false, + }, payload, waiter, - }, + }), crate::lite::Version::Lite05, ); let mut cx = Context::from_waker(std::task::Waker::noop()); diff --git a/rs/moq-net/src/coding/stream.rs b/rs/moq-net/src/coding/stream.rs index 2eec91c3bb9..23c08969f26 100644 --- a/rs/moq-net/src/coding/stream.rs +++ b/rs/moq-net/src/coding/stream.rs @@ -72,10 +72,9 @@ impl Stream { { let mut writer = Writer::new(send, version.clone()); writer.set_priority(CONTROL_SEND_ORDER.into()); - Self { - writer, - reader: Reader::new(recv, version), - } + let mut reader = Reader::new(recv, version); + reader.set_priority(crate::priority::RECV_CONTROL); + Self { writer, reader } } /// Cast the stream to a different version, used during version negotiation. diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index b6dfb3b9790..0f4f1e052dc 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -489,9 +489,10 @@ pub enum Error { #[error("frame timestamp doesn't fit its track")] TimestampMismatch, - /// The group was evicted by its own track to pay eviction debt under memory - /// pressure (see [`cache::Pool`](crate::cache::Pool)). Unlike [`Self::Old`], - /// the group was still within the publisher's window; it can be re-fetched. + /// The group was evicted under memory pressure: by its own track to pay eviction debt + /// (see [`cache::Pool`](crate::cache::Pool)), or by the transport when incomplete + /// messages spent its receive window. Unlike [`Self::Old`], the group was still within + /// the publisher's window; it can be re-fetched. #[error("evicted")] Evicted, @@ -585,6 +586,7 @@ impl Error { match err.close_kind() { crate::transport::CloseKind::IdleTimeout => Self::IdleTimeout, + crate::transport::CloseKind::Evicted => Self::Evicted, // A connection ending must not become the model's clean end-of-content. crate::transport::CloseKind::Closed => SessionError::Cancel.into(), _ => Self::Transport(err.to_string()), diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 720a9fad46c..e917c812417 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -1193,6 +1193,8 @@ pub fn seeds() -> Vec { seeds } +type FrameReader = coding::Reader, lite::Version>; + /// Frames arriving on concurrent group streams, as the frame bench times them: one /// frame per stream, read a chunk per stream per poll turn so the streams interleave /// the way a session's do. @@ -1201,7 +1203,7 @@ pub fn seeds() -> Vec { /// every declared size when the first byte lands and `Some(0)` grows every buffer with /// the bytes received. pub struct FrameRecv { - streams: Vec, frame::Producer)>>, + streams: Vec>, // Keeps the groups' track alive. _track: track::Producer, _broadcast: broadcast::Producer, @@ -1237,7 +1239,10 @@ impl FrameRecv { chunk, ready: true, }; - Some((coding::Reader::new(stream, lite::Version::Lite05), frame)) + Some(( + coding::Reader::new(crate::transport::poll::Buffered::new(stream), lite::Version::Lite05), + frame, + )) }) .collect(); @@ -1274,23 +1279,9 @@ struct Chunks { ready: bool, } -impl crate::transport::poll::RecvStream for Chunks { +impl crate::transport::poll::Chunks for Chunks { type Error = NoError; - fn poll_read( - &mut self, - cx: &mut std::task::Context<'_>, - dst: &mut [u8], - ) -> std::task::Poll, NoError>> { - let max = dst.len(); - self.poll_read_chunk(cx, max).map_ok(|chunk| { - chunk.map(|chunk| { - dst[..chunk.len()].copy_from_slice(&chunk); - chunk.len() - }) - }) - } - // Zero-copy, like the QUIC stacks, so only the frame buffer's own copy is timed. fn poll_read_chunk( &mut self, @@ -1309,10 +1300,6 @@ impl crate::transport::poll::RecvStream for Chunks { } fn stop(&mut self, _code: u32) {} - - fn poll_closed(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll> { - std::task::Poll::Pending - } } /// [`Chunks`] never fails. diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index 20f4ec4403b..cbb57ec9a6c 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -4,7 +4,7 @@ use std::{ task::{Context, Poll}, }; -use bytes::{Buf, BufMut, Bytes, BytesMut}; +use bytes::{Buf, Bytes, BytesMut}; use crate::{ Error, PathOwned, @@ -160,44 +160,9 @@ impl VirtualRecvStream { } } -impl crate::transport::poll::RecvStream for VirtualRecvStream { +impl crate::transport::poll::Chunks for VirtualRecvStream { type Error = crate::Error; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { - if dst.is_empty() { - return Poll::Ready(Ok(Some(0))); - } - - std::task::ready!(self.poll_fill(cx)); - if self.buffer.is_empty() { - return Poll::Ready(Ok(None)); - } - - let n = dst.len().min(self.buffer.len()); - dst[..n].copy_from_slice(&self.buffer[..n]); - self.buffer.advance(n); - Poll::Ready(Ok(Some(n))) - } - - fn poll_read_buf( - &mut self, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll, Self::Error>> { - if !buf.has_remaining_mut() { - return Poll::Ready(Ok(Some(0))); - } - - std::task::ready!(self.poll_fill(cx)); - if self.buffer.is_empty() { - return Poll::Ready(Ok(None)); - } - - let n = buf.remaining_mut().min(self.buffer.len()); - buf.put(self.buffer.split_to(n)); - Poll::Ready(Ok(Some(n))) - } - fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { if max == 0 { return Poll::Ready(Ok(Some(Bytes::new()))); @@ -215,25 +180,6 @@ impl crate::transport::poll::RecvStream for VirtualRecvStream { fn stop(&mut self, _code: u32) { self.rx.close(); } - - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { - if self.closed { - return Poll::Ready(Ok(())); - } - - // Drain (and discard) remaining messages until the queue closes. - let waiter = self.park.hold(cx); - loop { - match self.rx.poll_pop(waiter) { - Poll::Ready(Some(_)) => {} - Poll::Ready(None) => { - self.closed = true; - return Poll::Ready(Ok(())); - } - Poll::Pending => return Poll::Pending, - } - } - } } impl Drop for VirtualRecvStream { @@ -557,50 +503,102 @@ impl crate::transport::poll::SendStream for } } +#[allow(clippy::large_enum_variant)] pub enum AdapterRecv { Real(S::RecvStream), - Virtual(VirtualRecvStream), + Virtual(crate::transport::poll::Buffered), } impl crate::transport::poll::RecvStream for AdapterRecv { type Error = AdapterError<::Error>; + type Lease<'a> + = AdapterLease<'a, S> + where + S: 'a; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Self::Error>> { match self { - Self::Real(s) => s.poll_read(cx, dst).map_err(AdapterError::Real), - Self::Virtual(s) => s.poll_read(cx, dst).map_err(AdapterError::Virtual), + Self::Real(s) => s + .poll_lease(cx, want) + .map(|r| r.map(AdapterLease::Real).map_err(AdapterError::Real)), + Self::Virtual(s) => s + .poll_lease(cx, want) + .map(|r| r.map(AdapterLease::Virtual).map_err(AdapterError::Virtual)), } } - fn poll_read_buf( - &mut self, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll, Self::Error>> { + fn set_priority(&mut self, priority: i64) { match self { - Self::Real(s) => s.poll_read_buf(cx, buf).map_err(AdapterError::Real), - Self::Virtual(s) => s.poll_read_buf(cx, buf).map_err(AdapterError::Virtual), + Self::Real(s) => s.set_priority(priority), + // Rides the control stream, which carries its credit. + Self::Virtual(_) => {} } } - fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { + fn stop(&mut self, code: u32) { match self { - Self::Real(s) => s.poll_read_chunk(cx, max).map_err(AdapterError::Real), - Self::Virtual(s) => s.poll_read_chunk(cx, max).map_err(AdapterError::Virtual), + Self::Real(s) => s.stop(code), + Self::Virtual(s) => s.stop(code), } } +} - fn stop(&mut self, code: u32) { +/// A [`Lease`](crate::transport::poll::Lease) of an [`AdapterRecv`]. +pub enum AdapterLease<'a, S: crate::transport::poll::Session + 'a> { + Real(::Lease<'a>), + Virtual(crate::transport::poll::BufferedLease<'a>), +} + +impl Buf for AdapterLease<'_, S> { + fn remaining(&self) -> usize { match self { - Self::Real(s) => s.stop(code), - Self::Virtual(s) => s.stop(code), + Self::Real(l) => l.remaining(), + Self::Virtual(l) => l.remaining(), } } - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + fn chunk(&self) -> &[u8] { match self { - Self::Real(s) => s.poll_closed(cx).map_err(AdapterError::Real), - Self::Virtual(s) => s.poll_closed(cx).map_err(AdapterError::Virtual), + Self::Real(l) => l.chunk(), + Self::Virtual(l) => l.chunk(), + } + } + + fn advance(&mut self, n: usize) { + match self { + Self::Real(l) => l.advance(n), + Self::Virtual(l) => l.advance(n), + } + } + + fn copy_to_bytes(&mut self, n: usize) -> Bytes { + match self { + Self::Real(l) => l.copy_to_bytes(n), + Self::Virtual(l) => l.copy_to_bytes(n), + } + } +} + +impl crate::transport::poll::Lease for AdapterLease<'_, S> { + fn is_fin(&self) -> bool { + match self { + Self::Real(l) => l.is_fin(), + Self::Virtual(l) => l.is_fin(), + } + } + + fn chunks(&self) -> impl Iterator { + let (real, virt) = match self { + Self::Real(l) => (Some(l.chunks()), None), + Self::Virtual(l) => (None, Some(l.chunks())), + }; + real.into_iter().flatten().chain(virt.into_iter().flatten()) + } + + fn release(self) { + match self { + Self::Real(l) => l.release(), + Self::Virtual(l) => l.release(), } } } @@ -1333,7 +1331,10 @@ impl crate::transport::poll::Session for Con let waiter = self.park.hold(cx); match std::task::ready!(self.shared.incoming.poll_pop(waiter)) { - Some((send, recv)) => Poll::Ready(Ok((AdapterSend::Virtual(send), AdapterRecv::Virtual(recv)))), + Some((send, recv)) => Poll::Ready(Ok(( + AdapterSend::Virtual(send), + AdapterRecv::Virtual(crate::transport::poll::Buffered::new(recv)), + ))), None => Poll::Ready(Err(crate::Error::Closed)), } } @@ -1345,7 +1346,10 @@ impl crate::transport::poll::Session for Con _: crate::transport::poll::SendOptions<'_, ::Group>, ) -> Poll> { let (send, recv) = self.shared.open_outgoing(self.version); - Poll::Ready(Ok((AdapterSend::Virtual(send), AdapterRecv::Virtual(recv)))) + Poll::Ready(Ok(( + AdapterSend::Virtual(send), + AdapterRecv::Virtual(crate::transport::poll::Buffered::new(recv)), + ))) } fn poll_open_uni( @@ -1500,8 +1504,13 @@ mod tests { } /// A receive stream with no request behind it, for exercising the reader alone. - fn detached_recv(initial: Bytes, rx: Queue) -> VirtualRecvStream { - VirtualRecvStream::new(initial, rx, Arc::new(Shared::default()), Arc::new(Mutex::new(None))) + fn detached_recv(initial: Bytes, rx: Queue) -> crate::transport::poll::Buffered { + crate::transport::poll::Buffered::new(VirtualRecvStream::new( + initial, + rx, + Arc::new(Shared::default()), + Arc::new(Mutex::new(None)), + )) } #[test] @@ -1589,14 +1598,14 @@ mod tests { .unwrap(); let (_, mut recv) = shared.incoming.pop().await.unwrap(); assert_eq!( - recv.read_chunk(usize::MAX).await.unwrap(), + read_virtual(&mut recv).await.unwrap(), Some(Bytes::from_static(b"request")) ); assert_eq!(routed, target_id, "{version}: update {update_id} routed to {routed}"); shared.push(routed, raw.clone()); assert_eq!( - recv.read_chunk(usize::MAX).await.unwrap(), + read_virtual(&mut recv).await.unwrap(), Some(raw), "{version}: update {update_id} did not reach {target_id}" ); @@ -1645,7 +1654,7 @@ mod tests { )); let ok = encode_msg(&ok, VERSION); assert!(shared.push(RequestId(6), ok.clone()), "the answer found no stream"); - assert_eq!(recv.read_chunk(usize::MAX).await.unwrap(), Some(ok.clone())); + assert_eq!(read_virtual(&mut recv).await.unwrap(), Some(ok.clone())); assert!(!shared.push(RequestId(6), ok), "a second answer was routed"); send.write_chunk(update(8)).await.unwrap(); @@ -1660,7 +1669,7 @@ mod tests { VERSION, ); shared.close(RequestId(8), error.clone()); - assert_eq!(recv.read_chunk(usize::MAX).await.unwrap(), Some(error)); + assert_eq!(read_virtual(&mut recv).await.unwrap(), Some(error)); assert!( shared.push(RequestId(4), Bytes::from_static(b"later")), "the refusal closed the request" @@ -1746,18 +1755,17 @@ mod tests { let mut stream = detached_recv(initial, follow); // Read initial data - let mut buf = [0u8; 32]; - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b"initial"); + let chunk = stream.read_chunk(32).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b"initial"); // Send follow-up tx.push(Bytes::from_static(b"followup")); - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b"followup"); + let chunk = stream.read_chunk(32).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b"followup"); // Drop the writer → FIN drop(tx); - let result = stream.read(&mut buf).await.unwrap(); + let result = stream.read_chunk(32).await.unwrap(); assert_eq!(result, None); } @@ -1767,16 +1775,14 @@ mod tests { let mut stream = detached_recv(initial, Queue::new()); // Read small chunks - let mut buf = [0u8; 5]; - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b"hello"); + let chunk = stream.read_chunk(5).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b"hello"); - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b" worl"); + let chunk = stream.read_chunk(5).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b" worl"); - let mut buf = [0u8; 1]; - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b"d"); + let chunk = stream.read_chunk(1).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b"d"); } #[moq_clock::test] @@ -1872,10 +1878,7 @@ mod tests { // Drain the advertisement the stream opened with, so both halves are quiescent // and a later read only sees what the terminal message routes. - assert_eq!( - recv.read_chunk(usize::MAX).await.unwrap(), - Some(encode_msg(&msg, version)) - ); + assert_eq!(read_virtual(&mut recv).await.unwrap(), Some(encode_msg(&msg, version))); recv } @@ -1906,10 +1909,15 @@ mod tests { (shared, ours, theirs) } + /// The next chunk of a virtual stream, read without a lease. + fn read_virtual(stream: &mut VirtualRecvStream) -> impl Future, crate::Error>> + '_ { + std::future::poll_fn(|cx| crate::transport::poll::Chunks::poll_read_chunk(stream, cx, usize::MAX)) + } + /// Read until the stream FINs, returning whether it did so within `limit` reads. async fn drained(stream: &mut VirtualRecvStream, limit: usize) -> bool { for _ in 0..limit { - if stream.read_chunk(usize::MAX).await.unwrap().is_none() { + if read_virtual(stream).await.unwrap().is_none() { return true; } } @@ -1956,7 +1964,7 @@ mod tests { shared.close(RequestId(7), encode_msg(&done, version)); assert!(drained(&mut theirs, 4).await); assert!(shared.streams.lock().unwrap().contains_key(&RequestId(4))); - assert!(ours.read_chunk(usize::MAX).now_or_never().is_none()); + assert!(read_virtual(&mut ours).now_or_never().is_none()); } #[moq_clock::test] @@ -1984,7 +1992,7 @@ mod tests { shared.close(RequestId(4), encode_msg(&cancel, version)); assert!(drained(&mut ours, 4).await); assert!(shared.streams.lock().unwrap().contains_key(&RequestId(7))); - assert!(theirs.read_chunk(usize::MAX).now_or_never().is_none()); + assert!(read_virtual(&mut theirs).now_or_never().is_none()); } /// Classify the peer withdrawing its own advertisement of `namespace`. diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 579cf6025b8..f2c5fde93c1 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -326,9 +326,11 @@ where Version::Draft16 => { let mut sub_ns_adapter = sub_ns_adapter; let (send, recv) = sub_ns_adapter.open_native_bi().await?; + let mut reader = crate::coding::Reader::new(recv, version); + reader.set_priority(crate::priority::RECV_CONTROL); Stream { writer: crate::coding::Writer::new(send, version), - reader: crate::coding::Reader::new(recv, version), + reader, } } _ => Stream::open(&mut sub_ns_adapter.clone(), version).await?, @@ -709,6 +711,7 @@ pub async fn accept_setup( continue; } + reader.set_priority(crate::priority::RECV_CONTROL); let setup: setup::Setup = reader.decode().await?; let (params, _) = ietf::Parameters::decode_slice(&setup.parameters, version)?; let path = match params.get_bytes(ietf::ParameterBytes::Path) { @@ -1089,6 +1092,7 @@ impl Uni { // The negotiation gates the announce and dispatch loops, so a SETUP we // cannot read must end the session rather than leave them parked on a // slot nothing will ever fill. + reader.set_priority(crate::priority::RECV_CONTROL); let msg = reader.decode::().await?; if let Some(peer_setup) = peer_setup { peer_setup.set(decode_peer_setup(msg.parameters, version, client)?); diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 7207dd3a500..7fb969e87df 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3986,6 +3986,8 @@ where pub async fn recv_group(&mut self, stream: &mut Reader) -> Result<(), Error> { let group: ietf::GroupHeader = stream.decode().await?; + // Ranked as the least urgent subscription until its alias resolves. + stream.set_priority(crate::priority::recv_group(i8::MIN, group.group_id, Default::default())); if group.sub_group_id != 0 { tracing::warn!(sub_group_id = %group.sub_group_id, "subgroup ID is not supported, dropping stream"); @@ -4028,6 +4030,10 @@ where Reading::open(&track.tail, Some(group.group_id), self.runtime.now()), ) }; + let (priority, order) = track + .subscription() + .map_or((0, Default::default()), |sub| (sub.priority, sub.order)); + stream.set_priority(crate::priority::recv_group(priority, group.group_id, order)); // Peek before creating a group: a standalone END_OF_TRACK has no payload // group, and a fill or resume may already own the prefix. diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index a9852496dd0..703b58f7924 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -104,6 +104,7 @@ impl SetupAccept { return Poll::Ready(Err(Error::ProtocolViolation)); } ready!(reader.poll_decode::(&mut cx))?; + reader.set_priority(crate::priority::RECV_CONTROL); *setup = Some(reader); } // A closed transport can still yield a complete buffered SETUP. Decode it diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 1dc19cf0777..817fb1d9f01 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -850,7 +850,11 @@ impl UniServe { }; self.state = match kind { lite::DataType::Group => UniState::Group(GroupRecv::new(self.subscriber.clone(), reader)), - lite::DataType::Setup => UniState::Setup { reader }, + lite::DataType::Setup => { + let mut reader = reader; + reader.set_priority(crate::priority::RECV_CONTROL); + UniState::Setup { reader } + } lite::DataType::Padding => UniState::Padding { reader, budget: kio::coop::Budget::new(32), @@ -1029,6 +1033,11 @@ impl GroupRecv { group.start_at(hdr.frame_start)?; (group, entry.producer.clone(), entry.framing, reading) }; + let (priority, order) = track + .subscription() + .map_or((0, Default::default()), |sub| (sub.priority, sub.order)); + self.reader + .set_priority(crate::priority::recv_group(priority, hdr.sequence, order)); // The timescale came from TRACK_INFO (read before this subscription was // even registered), so frames decode immediately. No SUBSCRIBE_OK to diff --git a/rs/moq-net/src/lite/test_transport.rs b/rs/moq-net/src/lite/test_transport.rs index a62306b3d3b..0a7644ad1c3 100644 --- a/rs/moq-net/src/lite/test_transport.rs +++ b/rs/moq-net/src/lite/test_transport.rs @@ -374,16 +374,16 @@ pub struct PendingRecv; impl poll::RecvStream for PendingRecv { type Error = SinkError; + // Never handed out. + type Lease<'a> = poll::BufferedLease<'a>; - fn poll_read(&mut self, _cx: &mut Context<'_>, _dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_lease(&mut self, _cx: &mut Context<'_>, _want: usize) -> Poll, Self::Error>> { Poll::Pending } - fn stop(&mut self, _code: u32) {} + fn set_priority(&mut self, _priority: i64) {} - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Pending - } + fn stop(&mut self, _code: u32) {} } /// A RESET_STREAM as the transport reports it. `Some(code)` decodes as @@ -421,16 +421,16 @@ pub struct DeadRecv(ResetError); impl poll::RecvStream for DeadRecv { type Error = ResetError; + // Never handed out. + type Lease<'a> = poll::BufferedLease<'a>; - fn poll_read(&mut self, _cx: &mut Context<'_>, _dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_lease(&mut self, _cx: &mut Context<'_>, _want: usize) -> Poll, Self::Error>> { Poll::Ready(Err(self.0)) } - fn stop(&mut self, _code: u32) {} + fn set_priority(&mut self, _priority: i64) {} - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Err(self.0)) - } + fn stop(&mut self, _code: u32) {} } /// Incoming streams that die before their first byte: `accept_uni` / `accept_bi` @@ -839,7 +839,10 @@ impl crate::transport::Stats for SinkStats { /// /// Parks once the script is exhausted rather than reporting EOF: a read loop that saw /// EOF would exit, and a test usually wants to assert against the loop still running. -pub struct ScriptedRecv { +pub type ScriptedRecv = poll::Buffered