From ceeb2ac32126368df6bee1e43402590b726def32 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 04:45:59 -0800 Subject: [PATCH 1/3] chore(quest): claim quest/m1/quic-readable-dedupe.md Quest-Claim: e0d9ed3f7c04ac03 From 5a1d0209f8cc9f5a2b892bed747634aa7d94d985 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 04:59:03 -0800 Subject: [PATCH 2/3] fix(quic): release stream table and control queue capacity at rest Co-Authored-By: GPT-6.1 Sol --- quest/m1/README.md | 1 - quest/m1/quic-readable-dedupe.md | 19 --- rs/moq-quic/README.md | 2 +- rs/moq-quic/benches/receive_bursts.rs | 9 +- rs/moq-quic/src/connection/mod.rs | 2 +- rs/moq-quic/src/connection/spaces.rs | 10 +- rs/moq-quic/src/connection/streams/send.rs | 3 + rs/moq-quic/src/connection/streams/state.rs | 170 ++++++++++++++++++++ rs/moq-quic/src/connection/waits.rs | 36 ++++- 9 files changed, 224 insertions(+), 28 deletions(-) delete mode 100644 quest/m1/quic-readable-dedupe.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 360826fa74b..a07e764970f 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -203,7 +203,6 @@ per-frame CPU growth since #5926. - [A returning route serves past the cap its old copy was given](/quest/m1/returning-route-cap.md) - when a resumed reader's route returns, the reused serving copy asks with the reader's current resolution, not the capped resolution it was left with when it was replaced - [The resume model passes its long run again](/quest/m1/resume-model-7542.md) - fix #7542's regression that drops groups a long-budget reader can still get - [A resume more than one group behind never hangs](/quest/m1/ietf-resume-behind.md) - a resumed read gets every group or a drop on every wire version -- [moq-quic's idle connections keep no burst-sized state](/quest/m1/quic-readable-dedupe.md) - readable events are deduplicated and a quiet connection keeps no per-peak-stream bytes - [Stats name the route instance's epoch with a type](/quest/m1/stats-epoch-type.md) - `GroupFrame::epoch` gets a dedicated type and tests stop mislabelling the instance ## Related diff --git a/quest/m1/quic-readable-dedupe.md b/quest/m1/quic-readable-dedupe.md deleted file mode 100644 index 3b584bcd667..00000000000 --- a/quest/m1/quic-readable-dedupe.md +++ /dev/null @@ -1,19 +0,0 @@ -# [S] moq-quic's idle connections keep no burst-sized state - -## Goal - -A moq-quic connection's per-stream "readable" event queue stays bounded by the -number of streams with unread data, not by the size of a burst, and a quiet -connection keeps no per-stream bytes for streams it no longer has. - -## Plan - -Planned 2026-11-02 (maintainer, follow-up planning after the 2026-11-02 /quest-complete run), from [moq-dev/moq#7528](https://github.com/moq-dev/moq/pull/7528), which -stopped recycling freed receive state. - -- Deduplicate the per-stream readable events the connection queues; today the - queue can grow to the size of a burst. -- Trace the roughly 70 to 90 bytes per peak concurrent stream that a quiet - connection still keeps after #7528. The stream maps holding their peak - capacity are the likely part; release or shrink them at rest. -- Measure retained bytes after a burst with #7528's `receive_bursts` bench. diff --git a/rs/moq-quic/README.md b/rs/moq-quic/README.md index d68b6687180..cb6a2925aeb 100644 --- a/rs/moq-quic/README.md +++ b/rs/moq-quic/README.md @@ -97,7 +97,7 @@ Changes on top of the upstream commit, besides the renames: - **Namespaced streams** (`send`, `recv`, `stream`): quinn's `SendStream`, `RecvStream`, and their error zoo become `send::Stream` and `recv::Stream` with one `send::Error` (streams and datagrams) and one `recv::Error` (reads and leases), adding `Blocked`, and `Closed` for an ID the stream no longer has state for. `send::Stream::write` takes `&mut impl Buf`, splitting a `Bytes` source off without a copy, in place of `write` and `write_chunks`, and `Datagrams::send` takes `impl Buf` the same way, so a chained header costs no copy. An outgoing stream's `send::End` (finished and acknowledged, stopped, or reset) stays queryable until `send::Stream::release`, so a driver needs no table of its own for how a stream ended. A `send::Offset` comes only from `offset()` and is tagged to its stream, so `reset_at` and `acked` cannot be handed one past the data written. A port touching a public stream method has to be redone against these names. - **Connection limits** (`Limits`, `Connection::set_limits`, and the drivers' `set_limits`): one validated call changes a live connection's stream limit and windows, refusing values a peer would reject and changing nothing then. moq-tokio and moq-uring re-export the type, so both runtimes check limits the same way. Not offered upstream: it only bundles quinn's existing setters. - **Indexed send buffer** (`SendBuffer::get`): each retained write keeps its stream offset, and a STREAM frame's data comes from one binary search, instead of quinn walking every write from the oldest unacknowledged one for each piece, which costs a triangular number of visits when many small writes are in flight. Not offered upstream yet. -- **No receive-state recycling** (`StreamsState::stream_recv_freed`): a completed stream's `Recv` is dropped, where quinn keeps it in `free_recv` with its chunk list at peak capacity for the next stream. Recycled state stayed allocated per peak concurrent stream for the connection's lifetime: about 550 B per stream, and about 113 KiB for one that once buffered a 2,048-packet backlog. Reusing it saved only 0.1 to 0.3% of a burst's instructions. `benches/receive_bursts.rs` measures both. A cherry-pick touching `StreamRecv`, `free_recv`, or `reinit` has nothing to apply to. Not offered upstream yet. +- **No receive-state recycling** (`StreamsState::stream_recv_freed`): a completed stream's `Recv` is dropped, where quinn keeps it in `free_recv` with its chunk list at peak capacity for the next stream. Recycled state stayed allocated per peak concurrent stream for the connection's lifetime: about 550 B per stream, and about 113 KiB for one that once buffered a 2,048-packet backlog. Reusing it saved only 0.1 to 0.3% of a burst's instructions. `benches/receive_bursts.rs` measures both. Stream and waiter tables contract geometrically as streams go away and release every slot when empty; drained control queues and obsolete credit updates are dropped when nothing remains to send. A surviving stream cannot pin a burst's table capacity. A cherry-pick touching `StreamRecv`, `free_recv`, or `reinit` has nothing to apply to. Not offered upstream yet. - **Bandwidth probing** (`Connection::set_probe_target`, `connection/bandwidth_probe.rs`): an application-limited sender below its target fills the headroom BBR3 opens, in Startup and in one ProbeBW\_UP per probe wait, with duplicates of STREAM data still in flight, paced to the target, so the controller's samples measure the path rather than the application. That inflight bounds an attempt like a window once it sends its first duplicate; media alone past it, as beside another flow's standing queue, leaves the sender application-limited. `Controller::probes` and `Metrics::probe_inflight` tell the sender when and how far it may fill; CUBIC never does. A duplicate is never retransmitted, a reset or stopped stream's data is never resent, and a loss retransmits only the ranges no copy delivered (`SendBuffer::retransmit`). An attempt that raises nothing for two round trips, or ends short of the target, sits out 3 s and then 4 s before retrying the same target; reaching the target or a new one retries at once. `PathStats::sent_bandwidth_probe_bytes` counts the redundant bytes. A cherry-pick touching `poll_transmit`'s space loop or `populate_packet` has to keep the probe after the ordinary frames. Not offered upstream. - **Exact datagram overhead** (`Datagram::max_payload`): `Datagrams::max_size` subtracts the frame type and the length's actual varint size from the path budget and the peer's `max_datagram_frame_size`, instead of `Datagram::SIZE_BOUND`'s worst-case 8-byte length. RFC 9221 section 3 counts the frame type and length in the limit, so a peer advertising 1200 bytes (quicly does) accepts a 1197-byte payload, where reserving 8 bytes allows only 1191. quinn has the same behaviour; not offered upstream yet. A limit too small for even an empty frame, including RFC 9221's 0 for no support, makes `max_size` return `None` instead of accepting an empty payload. - **Waits** (`connection/waits.rs`): every wait is a `poll_*` taking the caller's `&kio::Waiter`, so a driver keeps no waiter tables of its own: `recv::Chunks::poll_next`, `recv::Stream::poll_lease` and `poll_received_reset`, `send::Stream::poll_write`, `poll_end`, and `poll_acked`, `Streams::poll_accept` and `poll_open`, `Datagrams::poll_recv`, and `Connection::poll_handshake_data`, `poll_connected`, `poll_handshake_confirmed`, and `poll_closed`, over QUIC and QMux alike, with `Connection::error` for why it was lost. quinn's `StreamEvent` and the `Event` variants a handle waited on are gone (`Event` keeps `PathUpdated`, and QMux has no events): the stream engine wakes each list where quinn queued the matching event, and wakes connection-blocked writers as soon as the window has room (`StreamsState::unblock`), so a port touching one of quinn's event sites has to wake the matching list instead. A lost connection wakes every waiter, and each wait drains what arrived (accepted streams, readable bytes, datagrams) before it fails, a stream wait with `ConnectionLost`. Not offered upstream: quinn keeps its wakers in the async layer. diff --git a/rs/moq-quic/benches/receive_bursts.rs b/rs/moq-quic/benches/receive_bursts.rs index 01df7c6d5eb..48858e15e25 100644 --- a/rs/moq-quic/benches/receive_bursts.rs +++ b/rs/moq-quic/benches/receive_bursts.rs @@ -13,7 +13,8 @@ //! `server_` columns are what dropping the server's connections frees: when new, when quiet after //! one burst, and when quiet after the resume and a repeat burst. `kept_b` is what quiet holds over //! new, per stream of the burst. The CPU columns time the server; the allocation counts include the -//! clients', which a receive change leaves alone. `teardown_b` is what the case leaks. +//! clients'. `quiet_delta_b` and `repeat_delta_b` keep the exact difference from cold, so table +//! rounding cannot hide a small retained allocation. `teardown_b` is what the case leaks. //! //! Criterion then times the server's side of a burst on a connection that already freed the //! previous one's streams, so the cost of allocating fresh receive state for each stream shows. @@ -267,7 +268,7 @@ fn server_heap(mut pair: Pair) -> isize { fn table() { println!( - "mode conns streams packets peak_kib server_cold_kib server_quiet_kib kept_b resume_allocs resume_cpu_us repeat_allocs repeat_cpu_us server_repeat_kib teardown_b rss_kib" + "mode conns streams packets peak_kib server_cold_kib server_quiet_kib kept_b resume_allocs resume_cpu_us repeat_allocs repeat_cpu_us server_repeat_kib quiet_delta_b repeat_delta_b teardown_b rss_kib" ); let kib = |bytes: isize| bytes / 1024; for mode in Mode::ALL { @@ -299,7 +300,7 @@ fn table() { let kept = (quiet - cold) / (conns * streams) as isize; println!( - "{} {conns} {streams} {packets} {} {} {} {kept} {resume_allocs} {} {repeat_allocs} {} {} {teardown} {}", + "{} {conns} {streams} {packets} {} {} {} {kept} {resume_allocs} {} {repeat_allocs} {} {} {} {} {teardown} {}", mode.name(), kib(peak), kib(cold), @@ -307,6 +308,8 @@ fn table() { resume_cpu.as_micros(), repeat_cpu.as_micros(), kib(repeat), + quiet - cold, + repeat - cold, rss_kib(), ); } diff --git a/rs/moq-quic/src/connection/mod.rs b/rs/moq-quic/src/connection/mod.rs index f04957267cf..2f2799ac0d1 100644 --- a/rs/moq-quic/src/connection/mod.rs +++ b/rs/moq-quic/src/connection/mod.rs @@ -1269,7 +1269,7 @@ impl Connection { } /// Indicate what types of frames are ready to send for the given space - fn space_can_send(&self, space_id: SpaceId, frame_space_1rtt: usize) -> SendableFrames { + fn space_can_send(&mut self, space_id: SpaceId, frame_space_1rtt: usize) -> SendableFrames { if self.spaces[space_id].crypto.is_none() { // No keys available for this space return SendableFrames::empty(); diff --git a/rs/moq-quic/src/connection/spaces.rs b/rs/moq-quic/src/connection/spaces.rs index f4741066d12..61ed210cfad 100644 --- a/rs/moq-quic/src/connection/spaces.rs +++ b/rs/moq-quic/src/connection/spaces.rs @@ -168,9 +168,15 @@ impl PacketSpace { x } - pub(super) fn can_send(&self, streams: &StreamsState) -> SendableFrames { + pub(super) fn can_send(&mut self, streams: &StreamsState) -> SendableFrames { let acks = self.pending_acks.can_send(); - let other = !self.pending.is_empty(streams) || self.ping_pending || self.immediate_ack_pending; + let pending = !self.pending.is_empty(streams); + if !pending { + // Drop both drained queue capacity and obsolete per-stream credit updates. Merely + // testing them as unsendable would retain their slots until more traffic arrives. + self.pending = Retransmits::default(); + } + let other = pending || self.ping_pending || self.immediate_ack_pending; SendableFrames { acks, other } } diff --git a/rs/moq-quic/src/connection/streams/send.rs b/rs/moq-quic/src/connection/streams/send.rs index eb815152b69..8b6aab1a823 100644 --- a/rs/moq-quic/src/connection/streams/send.rs +++ b/rs/moq-quic/src/connection/streams/send.rs @@ -460,6 +460,9 @@ impl<'a> Stream<'a> { /// kept for the connection's life. pub fn release(self) { if self.state.ends.remove(&self.id).is_some() { + if self.state.ends.capacity() > self.state.ends.len().saturating_mul(4) { + self.state.ends.shrink_to_fit(); + } return; } let max_data = self.state.max_send_data(self.id); diff --git a/rs/moq-quic/src/connection/streams/state.rs b/rs/moq-quic/src/connection/streams/state.rs index 1f147533e1f..7b8c1461cf1 100644 --- a/rs/moq-quic/src/connection/streams/state.rs +++ b/rs/moq-quic/src/connection/streams/state.rs @@ -672,6 +672,9 @@ impl StreamsState { buf.write_var(limit); stats.stream_data_blocked += 1; } + if pending.is_empty(self) { + *pending = Retransmits::default(); + } } /// Write STREAM and DATAGRAM frames in schedule order until the packet is full. @@ -1284,6 +1287,21 @@ impl StreamsState { if half == StreamHalf::Send { self.connection_blocked.swap_remove(&id); self.send_streams -= 1; + if self.send.capacity() > self.send.len().saturating_mul(4) { + self.send.shrink_to_fit(); + } + if self.connection_blocked.capacity() > self.connection_blocked.len().saturating_mul(4) { + self.connection_blocked.shrink_to_fit(); + } + } else { + // Contract geometrically, so a surviving stream cannot pin a burst's empty slots and + // removing a burst costs linear work overall. Empty tables release every slot. + if self.recv.capacity() > self.recv.len().saturating_mul(4) { + self.recv.shrink_to_fit(); + } + if self.unlooked.capacity() > self.unlooked.len().saturating_mul(4) { + self.unlooked.shrink_to_fit(); + } } self.waits.freed(id); } @@ -1350,6 +1368,158 @@ mod tests { ) } + /// Notifications coalesce while a reader is parked, and freed streams leave no table storage. + #[test] + fn receive_bursts_release_stream_storage() { + for dir in [Dir::Uni, Dir::Bi] { + for streams in [1, 16, 128] { + for packets in [1, 16, 256] { + let mut server = make(Side::Server); + let mut pending = Retransmits::default(); + for round in 0..3 { + let readers: Vec<_> = (0..streams).map(|_| Counter::new()).collect(); + let ids: Vec<_> = (0..streams) + .map(|i| StreamId::new(Side::Client, dir, round * streams + i)) + .collect(); + for (&id, reader) in ids.iter().zip(&readers) { + data(&mut server, id, 0, 0, false); + server.waits.register(id, Wait::Read, reader); + assert_eq!(Streams::new(&mut server, &ConnState::Established).accept(dir), Some(id)); + } + for packet in 0..packets { + for &id in &ids { + data(&mut server, id, packet, 1, false); + } + } + assert!(readers.iter().all(|reader| reader.wakes() == 1)); + assert_eq!(server.waits.stream_entries(), 0); + assert_eq!(server.unlooked.len(), streams as usize); + for id in ids { + data(&mut server, id, packets, 0, true); + // Read, cancel, and observe a reset on successive bursts of one connection. + if round == 0 { + let mut stream = recv(&mut server, &mut pending, id); + let mut chunks = stream.read().unwrap(); + let mut bytes = 0; + while let Some(chunk) = chunks.next(usize::MAX).unwrap() { + bytes += chunk.bytes.len(); + } + assert_eq!(bytes as u64, packets); + let _ = chunks.finalize(); + } else if round == 1 { + recv(&mut server, &mut pending, id).stop(0u32.into()).unwrap(); + } else { + let _ = server.received_reset(reset_at(id, 7, packets as u32, None)).unwrap(); + assert_eq!( + recv(&mut server, &mut pending, id).received_reset(), + Ok(Some(7u32.into())) + ); + } + assert!(server.recv.capacity() <= server.recv.len() * 4); + assert!(server.unlooked.capacity() <= server.unlooked.len() * 4); + if dir == Dir::Bi { + let mut stream = send::Stream { + id, + state: &mut server, + pending: &mut pending, + conn_state: &ConnState::Established, + }; + stream.finish().unwrap(); + stream.release(); + server.received_ack_of(frame::StreamMeta { + id, + offsets: 0..0, + fin: true, + }); + assert!(server.send.capacity() <= server.send.len() * 4); + } + } + assert_eq!(server.recv.capacity(), 0, "receive slots outlived their streams"); + assert_eq!(server.send.capacity(), 0, "send slots outlived their streams"); + assert_eq!(server.unlooked.capacity(), 0, "unread index outlived its streams"); + } + } + } + } + } + + #[test] + fn control_bursts_release_stream_storage() { + for streams in [1, 16, 128] { + let mut server = make(Side::Server); + server.set_params(&TransportParameters { + initial_max_streams_uni: (streams as u32).into(), + initial_max_data: 1024u32.into(), + ..TransportParameters::default() + }); + server.initial_stream_receive_window = 1; + server.stream_receive_window = 1; + let mut pending = Retransmits::default(); + let mut sent = Vec::new(); + for index in 0..streams { + let id = Streams::new(&mut server, &ConnState::Established) + .open(Dir::Uni, Default::default()) + .unwrap(); + let mut stream = send::Stream { + id, + state: &mut server, + pending: &mut pending, + conn_state: &ConnState::Established, + }; + assert_eq!(stream.write(&mut &b"x"[..]), Err(send::Error::Blocked)); + stream.reset(0u32.into()).unwrap(); + sent.push(id); + + let id = StreamId::new(Side::Client, Dir::Uni, index); + data(&mut server, id, 0, 1, false); + let mut stream = recv(&mut server, &mut pending, id); + let mut chunks = stream.read().unwrap(); + assert_eq!(chunks.next(1).unwrap().unwrap().bytes.len(), 1); + assert!(matches!(chunks.next(1), Err(recv::Error::Blocked))); + let _ = chunks.finalize(); + assert!(pending.max_stream_data.contains(&id)); + recv(&mut server, &mut pending, id).stop(0u32.into()).unwrap(); + data(&mut server, id, 1, 0, true); + } + // Control frames retain their obligations until sent, but no capacity after draining. + let mut stats = FrameStats::default(); + while !pending.is_empty(&server) { + let mut written = ThinRetransmits::default(); + server.write_control_frames(&mut Vec::new(), &mut pending, &mut written, &mut stats, 128); + if let Some(written) = written.get() { + for &(id, size) in &written.reset_stream { + server.reset_acked(id, size); + } + } + } + // QMux also serializes an idle batch, which discards obsolete credit updates. + server.write_control_frames( + &mut Vec::new(), + &mut pending, + &mut ThinRetransmits::default(), + &mut stats, + 128, + ); + for id in sent { + send::Stream { + id, + state: &mut server, + pending: &mut pending, + conn_state: &ConnState::Established, + } + .release(); + assert!(server.ends.capacity() <= server.ends.len() * 4); + } + assert_eq!(stats.reset_stream, streams); + assert_eq!(stats.stop_sending, streams); + assert_eq!(server.ends.capacity(), 0); + assert_eq!(pending.reset_stream.capacity(), 0); + assert_eq!(pending.stop_sending.capacity(), 0); + assert_eq!(pending.max_stream_data.capacity(), 0); + assert_eq!(pending.stream_data_blocked.capacity(), 0); + } + } + #[test] fn connection_blocked_reset_churn_is_bounded() { let mut server = make(Side::Server); diff --git a/rs/moq-quic/src/connection/waits.rs b/rs/moq-quic/src/connection/waits.rs index 3b27094e835..98fdddde846 100644 --- a/rs/moq-quic/src/connection/waits.rs +++ b/rs/moq-quic/src/connection/waits.rs @@ -112,6 +112,11 @@ impl Waits { waits.list(wait).wake(); } } + // Wakes may already have removed entries. Contract only as streams go away, rather than + // on every wake, so a reader that repeatedly parks does not churn the allocation. + if self.streams.capacity() > self.streams.len().saturating_mul(4) { + self.streams.shrink_to_fit(); + } } /// Register `waiter` for a stream the peer opens in `dir` @@ -222,7 +227,7 @@ impl Waits { return; } self.lost = Some(reason); - for (_, mut waits) in self.streams.drain() { + for (_, mut waits) in std::mem::take(&mut self.streams) { for wait in [Wait::Read, Wait::Write, Wait::End, Wait::Acked] { waits.list(wait).wake(); } @@ -296,3 +301,32 @@ impl std::ops::Deref for Counter { &self.waiter } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::Side; + + #[test] + fn freed_and_lost_streams_release_waiter_storage() { + for lost in [false, true] { + let mut waits = Waits::default(); + let readers: Vec<_> = (0..128).map(|_| Counter::new()).collect(); + for (index, reader) in readers.iter().enumerate() { + waits.register(StreamId::new(Side::Client, Dir::Uni, index as u64), Wait::Read, reader); + } + if lost { + waits.lose(Error::TimedOut); + } else { + for index in 0..128 { + let id = StreamId::new(Side::Client, Dir::Uni, index); + waits.wake(id, Wait::Read); + waits.freed(id); + assert!(waits.streams.capacity() <= waits.streams.len() * 4); + } + } + assert!(readers.iter().all(|reader| reader.wakes() == 1)); + assert_eq!(waits.streams.capacity(), 0, "waiter slots outlived their streams"); + } + } +} From fe6a25343119400171b83b4f1f12223b72f24a17 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 05:08:28 -0800 Subject: [PATCH 3/3] bench(quic): include single-stream allocation churn Co-Authored-By: GPT-6.1 Sol --- rs/moq-quic/README.md | 2 +- rs/moq-quic/benches/receive_bursts.rs | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/rs/moq-quic/README.md b/rs/moq-quic/README.md index cb6a2925aeb..31981d63440 100644 --- a/rs/moq-quic/README.md +++ b/rs/moq-quic/README.md @@ -97,7 +97,7 @@ Changes on top of the upstream commit, besides the renames: - **Namespaced streams** (`send`, `recv`, `stream`): quinn's `SendStream`, `RecvStream`, and their error zoo become `send::Stream` and `recv::Stream` with one `send::Error` (streams and datagrams) and one `recv::Error` (reads and leases), adding `Blocked`, and `Closed` for an ID the stream no longer has state for. `send::Stream::write` takes `&mut impl Buf`, splitting a `Bytes` source off without a copy, in place of `write` and `write_chunks`, and `Datagrams::send` takes `impl Buf` the same way, so a chained header costs no copy. An outgoing stream's `send::End` (finished and acknowledged, stopped, or reset) stays queryable until `send::Stream::release`, so a driver needs no table of its own for how a stream ended. A `send::Offset` comes only from `offset()` and is tagged to its stream, so `reset_at` and `acked` cannot be handed one past the data written. A port touching a public stream method has to be redone against these names. - **Connection limits** (`Limits`, `Connection::set_limits`, and the drivers' `set_limits`): one validated call changes a live connection's stream limit and windows, refusing values a peer would reject and changing nothing then. moq-tokio and moq-uring re-export the type, so both runtimes check limits the same way. Not offered upstream: it only bundles quinn's existing setters. - **Indexed send buffer** (`SendBuffer::get`): each retained write keeps its stream offset, and a STREAM frame's data comes from one binary search, instead of quinn walking every write from the oldest unacknowledged one for each piece, which costs a triangular number of visits when many small writes are in flight. Not offered upstream yet. -- **No receive-state recycling** (`StreamsState::stream_recv_freed`): a completed stream's `Recv` is dropped, where quinn keeps it in `free_recv` with its chunk list at peak capacity for the next stream. Recycled state stayed allocated per peak concurrent stream for the connection's lifetime: about 550 B per stream, and about 113 KiB for one that once buffered a 2,048-packet backlog. Reusing it saved only 0.1 to 0.3% of a burst's instructions. `benches/receive_bursts.rs` measures both. Stream and waiter tables contract geometrically as streams go away and release every slot when empty; drained control queues and obsolete credit updates are dropped when nothing remains to send. A surviving stream cannot pin a burst's table capacity. A cherry-pick touching `StreamRecv`, `free_recv`, or `reinit` has nothing to apply to. Not offered upstream yet. +- **No receive-state recycling** (`StreamsState::stream_recv_freed`): a completed stream's `Recv` is dropped, where quinn keeps it in `free_recv` with its chunk list at peak capacity for the next stream. Recycled state stayed allocated per peak concurrent stream for the connection's lifetime: about 550 B per stream, and about 113 KiB for one that once buffered a 2,048-packet backlog. Reusing it saved only 0.1 to 0.3% of a burst's instructions. `benches/receive_bursts.rs` measures both. Stream ID hash tables and waiter tables contract geometrically as streams go away and release every slot when empty; drained control queues and obsolete credit updates are dropped when nothing remains to send. A surviving stream cannot pin a burst's hash-table capacity. A cherry-pick touching `StreamRecv`, `free_recv`, or `reinit` has nothing to apply to. Not offered upstream yet. - **Bandwidth probing** (`Connection::set_probe_target`, `connection/bandwidth_probe.rs`): an application-limited sender below its target fills the headroom BBR3 opens, in Startup and in one ProbeBW\_UP per probe wait, with duplicates of STREAM data still in flight, paced to the target, so the controller's samples measure the path rather than the application. That inflight bounds an attempt like a window once it sends its first duplicate; media alone past it, as beside another flow's standing queue, leaves the sender application-limited. `Controller::probes` and `Metrics::probe_inflight` tell the sender when and how far it may fill; CUBIC never does. A duplicate is never retransmitted, a reset or stopped stream's data is never resent, and a loss retransmits only the ranges no copy delivered (`SendBuffer::retransmit`). An attempt that raises nothing for two round trips, or ends short of the target, sits out 3 s and then 4 s before retrying the same target; reaching the target or a new one retries at once. `PathStats::sent_bandwidth_probe_bytes` counts the redundant bytes. A cherry-pick touching `poll_transmit`'s space loop or `populate_packet` has to keep the probe after the ordinary frames. Not offered upstream. - **Exact datagram overhead** (`Datagram::max_payload`): `Datagrams::max_size` subtracts the frame type and the length's actual varint size from the path budget and the peer's `max_datagram_frame_size`, instead of `Datagram::SIZE_BOUND`'s worst-case 8-byte length. RFC 9221 section 3 counts the frame type and length in the limit, so a peer advertising 1200 bytes (quicly does) accepts a 1197-byte payload, where reserving 8 bytes allows only 1191. quinn has the same behaviour; not offered upstream yet. A limit too small for even an empty frame, including RFC 9221's 0 for no support, makes `max_size` return `None` instead of accepting an empty payload. - **Waits** (`connection/waits.rs`): every wait is a `poll_*` taking the caller's `&kio::Waiter`, so a driver keeps no waiter tables of its own: `recv::Chunks::poll_next`, `recv::Stream::poll_lease` and `poll_received_reset`, `send::Stream::poll_write`, `poll_end`, and `poll_acked`, `Streams::poll_accept` and `poll_open`, `Datagrams::poll_recv`, and `Connection::poll_handshake_data`, `poll_connected`, `poll_handshake_confirmed`, and `poll_closed`, over QUIC and QMux alike, with `Connection::error` for why it was lost. quinn's `StreamEvent` and the `Event` variants a handle waited on are gone (`Event` keeps `PathUpdated`, and QMux has no events): the stream engine wakes each list where quinn queued the matching event, and wakes connection-blocked writers as soon as the window has room (`StreamsState::unblock`), so a port touching one of quinn's event sites has to wake the matching list instead. A lost connection wakes every waiter, and each wait drains what arrived (accepted streams, readable bytes, datagrams) before it fails, a stream wait with `ConnectionLost`. Not offered upstream: quinn keeps its wakers in the async layer. diff --git a/rs/moq-quic/benches/receive_bursts.rs b/rs/moq-quic/benches/receive_bursts.rs index 48858e15e25..480dc115249 100644 --- a/rs/moq-quic/benches/receive_bursts.rs +++ b/rs/moq-quic/benches/receive_bursts.rs @@ -273,7 +273,7 @@ fn table() { let kib = |bytes: isize| bytes / 1024; for mode in Mode::ALL { for conns in [1, 4] { - for streams in [16, 256, 1024] { + for streams in [1, 16, 256, 1024] { for packets in [1, 16, 256, 2048] { if conns * streams * packets > BURST_PACKETS { continue; @@ -322,8 +322,8 @@ fn table() { fn bench_bursts(c: &mut Criterion) { let mut group = c.benchmark_group("receive_bursts"); for mode in Mode::ALL { - for streams in [16, 256] { - for packets in [16, 256] { + for streams in [1, 16, 256] { + for packets in [1, 16, 256] { let mut pair = None; group.throughput(Throughput::Bytes((streams * packets * PACKET) as u64)); group.bench_function(