diff --git a/quest/m1/unbounded-groups/README.md b/quest/m1/unbounded-groups/README.md index 20b989edbba..45fc83c7529 100644 --- a/quest/m1/unbounded-groups/README.md +++ b/quest/m1/unbounded-groups/README.md @@ -105,7 +105,6 @@ moq-transport, with memory held near the pool size. ## Required -- [Segmented frames](/quest/m1/unbounded-groups/segmented-frames.md) - large frames store as sealed segments, preallocate at most one, and release receive credit per segment - [Front eviction](/quest/m1/unbounded-groups/front-eviction.md) - groups shed whole frames from their front under pressure and refill them in place; the group caps go - [Byte holes](/quest/m1/unbounded-groups/byte-holes.md) - segments inside a frame evict and refill by `(frame, byte)` on the next lite version; the frame cap goes - [JS unbounded groups](/quest/m1/unbounded-groups/js.md) - `@moq/net` drops its caps, streams huge frames, and matches front eviction diff --git a/quest/m1/unbounded-groups/byte-holes.md b/quest/m1/unbounded-groups/byte-holes.md index 40516ecb018..4dce6b4d1bd 100644 --- a/quest/m1/unbounded-groups/byte-holes.md +++ b/quest/m1/unbounded-groups/byte-holes.md @@ -12,6 +12,14 @@ next unpublished lite version after lite-07. The frame cap See the [epic](/quest/m1/unbounded-groups/README.md) for the decisions. +- Segments are `frame::SEGMENT` (1 MiB) long, so segment `i` always holds + bytes from `i * SEGMENT`, and a completed chain is `frame::Payload::Chain`. + Only a frame past one segment written in pieces chains. One written whole + (`write_frame`, or a single write or transport chunk carrying all of it) + stays one `Bytes` at any size, so evicting part of it needs it split into + segment slices, whose memory frees only once every slice drops, or copied + into segments. + - `Position` gains a byte offset within the frame. A lite range end can carry it, and so can a group stream header, beside `frame_start`. A stream that starts mid-frame still states the frame's full size. Older diff --git a/quest/m1/unbounded-groups/front-eviction.md b/quest/m1/unbounded-groups/front-eviction.md index 0ef7418c03a..3d3432200da 100644 --- a/quest/m1/unbounded-groups/front-eviction.md +++ b/quest/m1/unbounded-groups/front-eviction.md @@ -100,6 +100,3 @@ Public API: removes `MAX_GROUP_FRAMES`, and `MAX_CACHE_BYTES` stops capping groups. A reader that used to see `GroupTooLarge` now refills or sees `Evicted`. Wire: none; it uses the frame-precise ranges. -## Required - -- [Segmented frames](/quest/m1/unbounded-groups/segmented-frames.md) - frame storage the cache can account for without whole-frame preallocation diff --git a/quest/m1/unbounded-groups/segmented-frames.md b/quest/m1/unbounded-groups/segmented-frames.md deleted file mode 100644 index 809508caf10..00000000000 --- a/quest/m1/unbounded-groups/segmented-frames.md +++ /dev/null @@ -1,31 +0,0 @@ -# [L] Segmented frames - -## Goal - -A large frame is stored in moq-net as a chain of sealed segments instead of -one contiguous buffer. Receiving it allocates at most one segment ahead, not -its full declared size. QUIC receive credit is released as each segment is -handed to the cache. Frames below the segment size keep today's single-`Bytes` -storage and cost nothing extra. Nothing user-visible changes, and the caps -stay for now. - -## Plan - -See the [epic](/quest/m1/unbounded-groups/README.md) for the decisions. - -- Today a completed frame is one `Bytes`, and a declared-size frame - preallocates up to the session `Budget` (`model/frame.rs`). Readers already - stream an in-flight frame by byte offset. Keep that cursor, and let it walk - across segment boundaries. -- Segments are the unit that [Byte holes](/quest/m1/unbounded-groups/byte-holes.md) - later evicts and refills, so give them a stable byte range within the frame. -- 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. -- 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. - -Public API: none. Wire: none. diff --git a/quest/m2/qmux-receive-lease.md b/quest/m2/qmux-receive-lease.md index ae1376b1e3d..addc11ad8b0 100644 --- a/quest/m2/qmux-receive-lease.md +++ b/quest/m2/qmux-receive-lease.md @@ -26,7 +26,8 @@ transport side should be wiring more than new machinery. The cost: a leasing transport must offer at least `moq_net::transport::MIN_RECEIVE_WINDOW`, the lite-01/02 `AnnounceInit` cap -(64 MiB) plus 16 bytes of prefixes, which also covers the 32 MiB frame cap. +(64 MiB) plus 16 bytes of prefixes. A frame holds credit only for the 1 MiB +storage segment being written, so its size no longer sets the floor. Native QUIC defaults to 128 MiB. qmux keeps reading its byte stream to see control frames, so the window, not the socket, is what bounds the bytes a slow reader's peer parks in userspace. Going from 1 MiB to 64 MiB multiplies @@ -36,9 +37,8 @@ Bring the options to the maintainer with a recommendation before building: - **The native floor.** Lease at `MIN_RECEIVE_WINDOW` or native's 128 MiB. Simplest, and one behavior across transports, at the largest allowance. -- **A smaller qmux floor.** Only lite-01/02 needs the 64 MiB term, and - [Segmented frames](/quest/m1/unbounded-groups/segmented-frames.md) removes - the frame term. MAX_DATA can only grow, so a session can start at the floor +- **A smaller qmux floor.** Only lite-01/02 needs the 64 MiB term; frames + need only a segment. MAX_DATA can only grow, so a session can start at the floor its version needs and extend once SETUP (or a WebSocket subprotocol) names a legacy version. A floor below some supported unit must refuse it loudly. - **Grow on demand.** Advertise a small window and extend MAX_DATA only when @@ -64,7 +64,6 @@ QMux MAX_DATA and STOP_SENDING only, no MoQ format change. ## Related -- [Segmented frames](/quest/m1/unbounded-groups/segmented-frames.md) - frame credit returns per segment, shrinking the floor's frame term - [WebSocket caps](/quest/m2/rate-websocket.md) - paced reads on the same sessions, inside the window - [Transport x version benches](/quest/m2/bench-transport-sweep.md) - where a per-session memory and throughput sweep belongs - [qmux drives its own socket](/quest/m2/perf/stream/driver.md) - replaces qmux's reader and writer tasks and its constructors' return type diff --git a/rs/moq-net/benches/frame.rs b/rs/moq-net/benches/frame.rs index 59a1a6ef919..3d8be4480f4 100644 --- a/rs/moq-net/benches/frame.rs +++ b/rs/moq-net/benches/frame.rs @@ -6,6 +6,9 @@ //! `budget` is the session default, which allocates up front until the frames in //! flight would pass it. Every stream reads a packet-sized chunk per turn. //! +//! `frame_recv_large` sweeps frames up to the group's byte cap over fewer streams, +//! where a frame spans many storage segments. +//! //! Run with `cargo bench -p moq-net --features fuzz --bench frame`. use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; @@ -17,12 +20,15 @@ const CHUNK: usize = 1200; const SIZES: [usize; 3] = [16 * 1024, 256 * 1024, 2 * 1024 * 1024]; const STREAMS: [usize; 3] = [1, 16, 64]; +const LARGE_SIZES: [usize; 3] = [4 * 1024 * 1024, 16 * 1024 * 1024, 32 * 1024 * 1024]; +const LARGE_STREAMS: [usize; 2] = [1, 4]; + const BUDGETS: [(&str, Option); 3] = [("upfront", Some(usize::MAX)), ("grow", Some(0)), ("budget", None)]; -fn bench(c: &mut Criterion) { - let mut group = c.benchmark_group("frame_recv"); - for size in SIZES { - for streams in STREAMS { +fn sweep(c: &mut Criterion, name: &str, sizes: &[usize], streams: &[usize]) { + let mut group = c.benchmark_group(name); + for &size in sizes { + for &streams in streams { group.throughput(Throughput::Bytes((size * streams) as u64)); for (name, budget) in BUDGETS { let id = format!("size={size}/streams={streams}"); @@ -39,5 +45,10 @@ fn bench(c: &mut Criterion) { group.finish(); } +fn bench(c: &mut Criterion) { + sweep(c, "frame_recv", &SIZES, &STREAMS); + sweep(c, "frame_recv_large", &LARGE_SIZES, &LARGE_STREAMS); +} + criterion_group!(benches, bench); criterion_main!(benches); diff --git a/rs/moq-net/benches/memory.rs b/rs/moq-net/benches/memory.rs index 06f159cadbd..ac45febd683 100644 --- a/rs/moq-net/benches/memory.rs +++ b/rs/moq-net/benches/memory.rs @@ -31,9 +31,21 @@ const VERSION: &str = "moq-lite-06"; /// Frames per group, so per-frame and per-group costs both appear. const FRAMES: usize = 4; -const SIZES: [usize; 3] = [64, 16 * 1024, 1024 * 1024]; +const SIZES: [usize; 4] = [64, 16 * 1024, 1024 * 1024, 4 * 1024 * 1024]; const STREAMS: [usize; 3] = [1, 16, 128]; +/// The most payload one round may write across every stream, so the largest frames skip +/// the widest sweep rather than hold gigabytes. +const ROUND_BYTES: usize = 512 * 1024 * 1024; + +/// Every `(streams, size)` pair to measure. +fn cases() -> impl Iterator { + STREAMS + .into_iter() + .flat_map(|streams| SIZES.into_iter().map(move |size| (streams, size))) + .filter(|(streams, size)| streams * FRAMES * size <= ROUND_BYTES) +} + struct Counter; static COUNTING: AtomicBool = AtomicBool::new(false); @@ -201,22 +213,18 @@ fn allocations() { const ROUNDS: usize = 64; let rt = runtime(); - for streams in STREAMS { - for size in SIZES { - let mut link = rt.block_on(Link::new(streams, size)); - ALLOCATIONS.store(0, Ordering::Relaxed); - ALLOCATED.store(0, Ordering::Relaxed); - COUNTING.store(true, Ordering::Relaxed); - rt.block_on(link.measure(ROUNDS as u64)); - COUNTING.store(false, Ordering::Relaxed); - - let frames = (ROUNDS * streams * FRAMES) as f64; - let allocations = ALLOCATIONS.load(Ordering::Relaxed) as f64 / frames; - let bytes = ALLOCATED.load(Ordering::Relaxed) as f64 / frames; - println!( - "streams={streams}/size={size}: {allocations:.1} allocations, {bytes:.0} bytes allocated per frame" - ); - } + for (streams, size) in cases() { + let mut link = rt.block_on(Link::new(streams, size)); + ALLOCATIONS.store(0, Ordering::Relaxed); + ALLOCATED.store(0, Ordering::Relaxed); + COUNTING.store(true, Ordering::Relaxed); + rt.block_on(link.measure(ROUNDS as u64)); + COUNTING.store(false, Ordering::Relaxed); + + let frames = (ROUNDS * streams * FRAMES) as f64; + let allocations = ALLOCATIONS.load(Ordering::Relaxed) as f64 / frames; + let bytes = ALLOCATED.load(Ordering::Relaxed) as f64 / frames; + println!("streams={streams}/size={size}: {allocations:.1} allocations, {bytes:.0} bytes allocated per frame"); } } @@ -228,18 +236,16 @@ fn bench(c: &mut Criterion) { let rt = runtime(); let mut group = c.benchmark_group("memory"); - for streams in STREAMS { - for size in SIZES { - group.throughput(Throughput::Bytes((streams * FRAMES * size) as u64)); - let mut link = None; - group.bench_function( - BenchmarkId::from_parameter(format!("streams={streams}/size={size}")), - |b| { - let link = link.get_or_insert_with(|| rt.block_on(Link::new(streams, size))); - b.iter_custom(|iters| rt.block_on(link.measure(iters))) - }, - ); - } + for (streams, size) in cases() { + group.throughput(Throughput::Bytes((streams * FRAMES * size) as u64)); + let mut link = None; + group.bench_function( + BenchmarkId::from_parameter(format!("streams={streams}/size={size}")), + |b| { + let link = link.get_or_insert_with(|| rt.block_on(Link::new(streams, size))); + b.iter_custom(|iters| rt.block_on(link.measure(iters))) + }, + ); } group.finish(); } diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index e62b9c1fbfa..fd35176776d 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -333,7 +333,8 @@ impl Reader { /// 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. + /// segment it lands in seals, so an incomplete frame holds at most one segment of the + /// receive window however large it is. /// /// `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 @@ -368,12 +369,19 @@ impl Reader { }; // 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(); + // written, and its consumers woken, only once the lease is gone. A chunk holding + // the rest of the frame is taken whole, so a frame delivered in one piece keeps + // its allocation; any other stops where the segment being written seals. + let remaining = frame.remaining(); + let take = match lease.chunk().len() >= remaining { + true => remaining, + false => lease.chunk().len().min(frame.unsealed()), + }; + let chunk = lease.copy_to_bytes(take); + let complete = chunk.len() == remaining; let ended = lease.is_fin() && !lease.has_remaining(); - if complete { - // Every byte of the frame is in hand, so its credit returns. + if complete || chunk.len() == frame.unsealed() { + // Every byte of the segment is in hand, so its credit returns. lease.release(); } else { drop(lease); @@ -946,4 +954,102 @@ mod tests { // Two budgets' worth, and the boundary is owed nothing by the time it arrives. assert_eq!(wakes.count(), 2, "the burst was withheld past its budget"); } + + /// A leasing stream that keeps the credit of every byte read until a release, the way + /// a QUIC connection window does, counting what is held and what came back. + #[derive(Default)] + struct Credited { + chunks: VecDeque, + held: usize, + released: usize, + } + + struct CreditedLease<'a>(&'a mut Credited); + + impl RecvStream for Credited { + type Error = crate::lite::test_transport::SinkError; + type Lease<'a> = CreditedLease<'a>; + + fn poll_lease(&mut self, _: &mut Context<'_>, want: usize) -> Poll, Self::Error>> { + match self.chunks.iter().map(Bytes::len).sum::() >= want { + true => Poll::Ready(Ok(CreditedLease(self))), + false => Poll::Pending, + } + } + + fn set_priority(&mut self, _: i64) {} + + fn stop(&mut self, _: u32) {} + } + + impl Buf for CreditedLease<'_> { + fn remaining(&self) -> usize { + self.0.chunks.iter().map(Bytes::len).sum() + } + + fn chunk(&self) -> &[u8] { + self.0.chunks.front().map_or(&[], |chunk| &chunk[..]) + } + + fn advance(&mut self, n: usize) { + let front = self.0.chunks.front_mut().expect("advanced past the lease"); + front.advance(n); + if front.is_empty() { + self.0.chunks.pop_front(); + } + self.0.held += n; + } + } + + impl Lease for CreditedLease<'_> { + fn chunks(&self) -> impl Iterator { + self.0.chunks.iter().map(|chunk| &chunk[..]) + } + + fn is_fin(&self) -> bool { + false + } + + fn release(self) { + self.0.released += std::mem::take(&mut self.0.held); + } + } + + /// A frame returns its receive credit as each segment seals, so however large it is, + /// it never holds more than one segment of the window, even when a chunk straddles + /// the boundary. + #[test] + fn a_frame_returns_credit_per_segment() { + use crate::frame::SEGMENT; + const SIZE: usize = 2 * SEGMENT + SEGMENT / 2; + const CHUNK: usize = 100_000; + + let (mut frame, mut payload, waiter, _wakes) = parked_frame(SIZE); + let data = Bytes::from((0..SIZE).map(|i| (i % 251) as u8).collect::>()); + let mut reader = Reader::new(Credited::default(), crate::lite::Version::Lite05); + let mut cx = Context::from_waker(std::task::Waker::noop()); + + let mut sent = 0; + let done = loop { + let next = (sent + CHUNK).min(SIZE); + reader.stream.chunks.push_back(data.slice(sent..next)); + sent = next; + let res = reader.poll_read_frame(&mut cx, &mut frame); + let stream = &reader.stream; + if res.is_ready() { + break res; + } + assert_eq!(stream.released, sent / SEGMENT * SEGMENT, "credit returned off a seal"); + assert!(stream.held < SEGMENT, "{} bytes held past a segment", stream.held); + }; + assert!(matches!(done, Poll::Ready(Ok(())))); + assert_eq!(reader.stream.released, SIZE); + + let _group = frame.finish().unwrap(); + let mut read = Vec::new(); + while let Poll::Ready(Ok(Some(chunk))) = payload.poll_read_chunk(&waiter) { + read.extend_from_slice(&chunk); + } + assert_eq!(read, data); + } } diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 23e73f0d42c..ef756f7101d 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -2790,7 +2790,7 @@ impl Publisher { /// Write a whole fetch object's length and payload, after its header. async fn write_fetch_payload( stream: &mut Writer, Version>, - payload: bytes::Bytes, + payload: crate::frame::Payload, version: Version, ) -> Result<(), Error> { stream.varint(payload.len() as u64).await?; @@ -5512,7 +5512,7 @@ enum GroupState { Serve { writer: Writer, frame: Option, - chunk: Option, + chunk: Option, batch: Box>, batch_pos: usize, }, @@ -5690,7 +5690,7 @@ impl GroupServe { .as_ref() .map_or(self.group.index() - 1, |fields| fields.id); if id >= self.slice.skip { - *chunk = Some(next); + *chunk = Some(next.into()); } } Poll::Ready(Ok(None)) => *frame = None, diff --git a/rs/moq-net/src/lite/ietf.rs b/rs/moq-net/src/lite/ietf.rs index f3f19a468fe..7da53fd149f 100644 --- a/rs/moq-net/src/lite/ietf.rs +++ b/rs/moq-net/src/lite/ietf.rs @@ -1201,7 +1201,7 @@ mod hop_tests { // The upstream group is still open. let frame = next(&mut received).await.unwrap(); - assert_eq!(&frame.payload[..], b"a"); + assert_eq!(&frame.payload.contiguous(0)[..], b"a"); assert_eq!(frame.timestamp, negotiated.then(|| ts(0))); match negotiated { true => { @@ -1236,7 +1236,7 @@ mod hop_tests { group.finish().unwrap(); let frame = next(&mut received).await.unwrap(); - assert_eq!(&frame.payload[..], b"b"); + assert_eq!(&frame.payload.contiguous(0)[..], b"b"); assert!(next(&mut received).await.is_none()); let info = moq_clock::timeout(TIMEOUT, copy.query()).await.unwrap().unwrap(); assert_eq!( @@ -1398,12 +1398,18 @@ mod hop_tests { group.write_frame(ts(5), b"key".to_vec()).unwrap(); let mut received = recv(&mut subscriber).await; let frame = next(&mut received).await.unwrap(); - assert_eq!((frame.timestamp, &frame.payload[..]), (Some(ts(5)), &b"key"[..])); + assert_eq!( + (frame.timestamp, &frame.payload.contiguous(0)[..]), + (Some(ts(5)), &b"key"[..]) + ); assert!(frame.fields.is_none() && received.wire().is_none()); group.write_frame(ts(6), b"delta".to_vec()).unwrap(); group.finish().unwrap(); let frame = next(&mut received).await.unwrap(); - assert_eq!((frame.timestamp, &frame.payload[..]), (Some(ts(6)), &b"delta"[..])); + assert_eq!( + (frame.timestamp, &frame.payload.contiguous(0)[..]), + (Some(ts(6)), &b"delta"[..]) + ); assert!(frame.fields.is_none()); assert!(next(&mut received).await.is_none()); assert!(received.end_object().is_none()); @@ -1480,7 +1486,7 @@ mod hop_tests { assert_eq!(received.sequence, 1); let frame = next(&mut received).await.unwrap(); assert_eq!(frame.timestamp, negotiated.then(|| ts(50)), "{offer:?}"); - assert_eq!(&frame.payload[..], b"frame"); + assert_eq!(&frame.payload.contiguous(0)[..], b"frame"); assert!(next(&mut received).await.is_none()); track.insert_datagram(2, ts(25), b"datagram".to_vec()).unwrap(); let datagram = moq_clock::timeout(TIMEOUT, kio::wait(|waiter| subscriber.poll_recv_wire_datagram(waiter))) @@ -1517,7 +1523,7 @@ mod hop_tests { for (millis, n) in [(5, 1), (6, 2)] { let frame = next(&mut received).await.unwrap(); assert_eq!( - (frame.timestamp, &frame.payload[..]), + (frame.timestamp, &frame.payload.contiguous(0)[..]), (Some(ts(millis)), &payload(n)[..]) ); assert!(frame.fields.is_none()); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 5208ba6bebd..6922d069e30 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2176,7 +2176,7 @@ enum FetchServe { stamp: Option, prev_ts: u64, frame: Option, - chunk: Option, + chunk: Option, batch: Box, batch_pos: usize, /// The `ietf` extension's object records, when negotiated. @@ -2289,7 +2289,7 @@ impl Request for FetchServe { } } else if let Some(pending) = frame { match ready!(pending.poll_read_chunk(waiter))? { - Some(next) => *chunk = Some(next), + Some(next) => *chunk = Some(next.into()), None => *frame = None, } } else if *batch_pos < batch.len() { @@ -4402,7 +4402,7 @@ enum GroupState { Serve { writer: Writer, frame: Option, - chunk: Option, + chunk: Option, batch: Box, batch_pos: usize, /// The `ietf` extension's object records, when negotiated. @@ -4619,7 +4619,7 @@ impl GroupServe { } } else if let Some(pending) = frame { match pending.poll_read_chunk(waiter) { - Poll::Ready(Ok(Some(next))) => *chunk = Some(next), + Poll::Ready(Ok(Some(next))) => *chunk = Some(next.into()), Poll::Ready(Ok(None)) => *frame = None, Poll::Ready(Err(err)) => break 'serve Err(err), Poll::Pending => return Poll::Pending, diff --git a/rs/moq-net/src/model/frame.rs b/rs/moq-net/src/model/frame.rs index 0f08d87d5e2..2f94bb847db 100644 --- a/rs/moq-net/src/model/frame.rs +++ b/rs/moq-net/src/model/frame.rs @@ -12,7 +12,7 @@ use std::sync::{Mutex, OnceLock, PoisonError}; use std::task::{Poll, ready}; use arrayvec::ArrayVec; -use bytes::Bytes; +use bytes::{Buf, Bytes, BytesMut}; use crate::group::{self, GroupState}; use crate::time::Instant; @@ -40,8 +40,10 @@ pub struct Info { /// A completed frame: an optional timestamp, its full, contiguous payload, and when it /// arrived. /// -/// This is the stored form of every finished frame in a group. The payload is a -/// single [`Bytes`], so a consumer gets it with one zero-copy slice. +/// Reading a frame whole is a zero-copy slice of what its group stores, except for a +/// frame larger than a storage segment that arrived in pieces: the group keeps that as +/// separate segments, so a whole read copies them into one buffer. Streaming it through +/// [`Consumer::poll_read_chunk`] hands out each segment without a copy. #[derive(Clone, Debug)] pub struct Frame { /// Presentation timestamp, at the parent track's timescale, or `None` when untimed. @@ -73,11 +75,184 @@ impl Frame { } } +/// Bytes in each sealed segment of a frame larger than one. +/// +/// A frame within one segment is stored as a single buffer. A larger one that arrives in +/// pieces is a chain of segments of exactly this size, the last shorter, so segment `i` +/// always holds the frame's bytes from `i * SEGMENT`. Receiving it allocates at most one +/// segment ahead of what arrived, and returns each segment's receive credit as it seals, +/// so a connection window only has to fit a segment rather than a frame. +/// +/// Sized from `benches/frame.rs`: smaller segments cost allocations per byte, and larger +/// ones a bigger allocation and credit held ahead of the bytes. Most media frames fit in +/// one, so a whole read of them never copies. +pub(crate) const SEGMENT: usize = 1024 * 1024; + +/// A completed frame's payload as its group stores it: one buffer, or the sealed +/// segments of a frame larger than [`SEGMENT`] written in pieces. +/// +/// A view, like [`Bytes`]: reading it as a [`Buf`] moves this copy past bytes, never the +/// stored one. +#[derive(Clone, Debug)] +pub(crate) enum Payload { + Whole(Bytes), + Chain(Chain), +} + +/// A frame's sealed segments, read from `start`. +#[derive(Clone, Debug)] +pub(crate) struct Chain { + // Each [`SEGMENT`] long but the last, and never empty. + segments: Arc<[Bytes]>, + // The frame offset this view starts at. + start: usize, +} + +impl Chain { + /// The bytes from `start` to the end. + pub fn len(&self) -> usize { + self.end() - self.start + } + + /// The whole chain copied into one buffer. + #[cold] + #[inline(never)] + pub fn concat(self) -> Bytes { + Payload::Chain(self).contiguous(0) + } + + /// The frame's length, past which nothing is stored. + fn end(&self) -> usize { + let last = self.segments.last().map_or(0, Bytes::len); + self.segments.len().saturating_sub(1) * SEGMENT + last + } + + /// The stored bytes from frame offset `at` to the end of its segment, zero-copy. + fn segment_at(&self, at: usize) -> Bytes { + let index = at / SEGMENT; + match self.segments.get(index) { + Some(segment) => segment.slice(at - index * SEGMENT..), + None => Bytes::new(), + } + } +} + +// A chain costs a cached frame nothing over the one buffer it holds otherwise. +const _: () = assert!(size_of::() == size_of::()); + +impl Default for Payload { + fn default() -> Self { + Self::Whole(Bytes::new()) + } +} + +impl From for Payload { + fn from(bytes: Bytes) -> Self { + Self::Whole(bytes) + } +} + +impl Payload { + pub fn len(&self) -> usize { + match self { + Self::Whole(bytes) => bytes.len(), + Self::Chain(chain) => chain.len(), + } + } + + #[cfg(feature = "async")] + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// The bytes from `offset` to the end of the segment holding it, zero-copy. + pub fn chunk_at(&self, offset: usize) -> Bytes { + match self { + Self::Whole(bytes) => bytes.slice(offset..), + Self::Chain(chain) => chain.segment_at(chain.start + offset), + } + } + + /// The bytes from `offset` on in one buffer, copied only when they span segments. + pub fn contiguous(&self, offset: usize) -> Bytes { + let len = self.len(); + let first = self.chunk_at(offset); + if offset + first.len() >= len { + return first; + } + let mut out = BytesMut::with_capacity(len - offset); + let mut at = offset; + while at < len { + let chunk = self.chunk_at(at); + at += chunk.len(); + out.extend_from_slice(&chunk); + } + out.freeze() + } +} + +impl Buf for Payload { + fn remaining(&self) -> usize { + self.len() + } + + fn chunk(&self) -> &[u8] { + match self { + Self::Whole(bytes) => bytes, + Self::Chain(chain) => { + let index = chain.start / SEGMENT; + chain + .segments + .get(index) + .map_or(&[], |segment| &segment[chain.start - index * SEGMENT..]) + } + } + } + + fn advance(&mut self, cnt: usize) { + assert!(cnt <= self.len(), "advance past the end of a payload"); + match self { + Self::Whole(bytes) => bytes.advance(cnt), + Self::Chain(chain) => chain.start += cnt, + } + } + + // Zero-copy within a segment, which is what a transport taking `chunk()` asks for. + fn copy_to_bytes(&mut self, len: usize) -> Bytes { + let chunk = self.chunk_at(0); + let out = match len <= chunk.len() { + true => chunk.slice(..len), + false => self.contiguous(0).slice(..len), + }; + self.advance(len); + out + } +} + +/// A completed frame as its group stores it. +#[derive(Clone, Debug)] +pub(crate) struct Cached { + pub timestamp: Option, + pub payload: Payload, + // When the frame's last byte landed in this node's cache; see [`Frame::arrived`]. + pub arrived: Instant, +} + +impl From for Cached { + fn from(frame: Frame) -> Self { + Self { + timestamp: frame.timestamp, + payload: Payload::Whole(frame.payload), + arrived: frame.arrived, + } + } +} + /// A frame with the relay metadata its group keeps beside it, read only by wire egress. #[cfg(feature = "async")] #[derive(Clone, Debug)] pub(crate) struct Object { - pub frame: Frame, + pub frame: Cached, pub fields: Option>, /// The inflated size of a compressed frame read in its stored form (see /// [`group::Consumer::compressed`]), whose `payload` is then the DEFLATE slice. @@ -86,7 +261,7 @@ pub(crate) struct Object { #[cfg(feature = "async")] impl std::ops::Deref for Object { - type Target = Frame; + type Target = Cached; fn deref(&self) -> &Self::Target { &self.frame } @@ -198,10 +373,11 @@ impl Buffer { /// /// A writer declares each frame's size before sending it, and allocating that size up /// front saves a reallocation and copy per doubling as the payload arrives. A peer's -/// declaration costs it nothing, though, so without a bound it could commit up to -/// [`group::MAX_CACHE_BYTES`] per stream without sending a byte. A frame whose declared -/// size fits what remains is allocated up front; past that, its buffer grows with the -/// bytes received. Each frame returns its share once it completes or aborts. +/// declaration costs it nothing, though, so without a bound it could commit a +/// [`SEGMENT`] per stream for a single byte. A frame reserves what one segment of it +/// takes, and while that fits what remains each segment is allocated whole; past that, +/// it grows with the bytes received. Each frame returns its share once it completes or +/// aborts. /// /// Every broadcast, track, and group minted from an [`crate::origin::Producer`] handle shares /// its budget. The built-in sessions each ingest through a handle with its own. @@ -263,10 +439,11 @@ impl Drop for Reservation { /// [`Producer`] and any streaming [`Consumer`]s. /// /// A whole-frame [`Bytes`] write is stored directly. Chunked writes are copied into a -/// heap [`Segment`], either sized to the declared frame up front or grown as bytes -/// arrive (see [`Budget`]). The producer writes through the raw pointer (sole writer, -/// guaranteed by the frame owning the group's only writer); `written` provides -/// happens-before for cross-thread reads. +/// heap [`Segment`], or a chain of them for a frame larger than [`SEGMENT`], each either +/// allocated at its full length by its first byte or grown as bytes arrive (see +/// [`Budget`]). The producer writes through the raw pointer (sole writer, guaranteed by +/// the frame owning the group's only writer); `written` provides happens-before for +/// cross-thread reads. #[derive(Clone)] pub(super) struct FrameBuf(Arc); @@ -283,6 +460,42 @@ enum FrameStorage { // Replaced by a larger copy whenever a write outgrows it. A reader clones the current // segment under the lock, so a slice it already handed out keeps the old one alive. Growing(Mutex>), + // A frame larger than one segment. + Chain(Mutex), +} + +/// A frame larger than one [`SEGMENT`], as it streams in. +#[derive(Default)] +struct Segments { + // Every full segment so far, in order. + sealed: Vec, + // The segment being written, allocated by its first byte and grown like + // [`FrameStorage::Growing`]. + open: Option>, +} + +impl Segments { + /// Copy `src` in at frame offset `offset`, sealing each segment it fills. + fn append(&mut self, mut offset: usize, mut src: &[u8], size: usize, grow: bool) { + while !src.is_empty() { + let base = offset - offset % SEGMENT; + let len = (size - base).min(SEGMENT); + let within = offset - base; + let take = src.len().min(len - within); + let open = self + .open + .get_or_insert_with(|| Arc::new(Segment::new(if grow { 0 } else { len }))); + Segment::reserve(open, within, within + take, len); + // Safety: sole writer, within the segment's length, past every byte handed out. + unsafe { open.write(within, &src[..take]) }; + if within + take == len { + let segment = self.open.take().expect("written above"); + self.sealed.push(Bytes::from_owner(Filled { segment, len })); + } + offset += take; + src = &src[take..]; + } + } } /// An owned, zero-initialized heap allocation that chunked writes are copied into. @@ -323,6 +536,25 @@ impl Segment { debug_assert!(offset + src.len() <= self.capacity); unsafe { std::ptr::copy_nonoverlapping(src.as_ptr(), self.data.add(offset), src.len()) }; } + + /// Replace `current` with a larger copy of its first `filled` bytes if it holds fewer + /// than `needed` of the `len` it grows to. + /// + /// Doubling keeps the copies linear in `len`, and the cap makes the last copy exactly + /// `len`, so it seals without another. + fn reserve(current: &mut Arc, filled: usize, needed: usize, len: usize) { + if current.capacity >= needed { + return; + } + let next = Segment::new(needed.max(current.capacity * 2).min(len)); + let filled = Filled { + segment: current.clone(), + len: filled, + }; + // Safety: `next` is new, so nothing reads it yet. + unsafe { next.write(0, filled.as_ref()) }; + *current = Arc::new(next); + } } /// The first `len` bytes of a [`Segment`], all written, so it can back a @@ -402,9 +634,11 @@ impl FrameBuf { return; } let prev = self.written(Ordering::Relaxed); + let size = self.size(); let storage = self.0.storage.get_or_init(|| match self.0.grow { + _ if size > SEGMENT => FrameStorage::Chain(Mutex::default()), true => FrameStorage::Growing(Mutex::new(Arc::new(Segment::new(0)))), - false => FrameStorage::Fixed(Arc::new(Segment::new(self.size()))), + false => FrameStorage::Fixed(Arc::new(Segment::new(size))), }); // Safety (every write below): sole writer; the caller bounds-checked `src` against // the remaining size, and consumers only read `[..written]`. @@ -415,47 +649,62 @@ impl FrameBuf { FrameStorage::Fixed(segment) => unsafe { segment.write(prev, src) }, FrameStorage::Growing(current) => { let mut current = current.lock().unwrap_or_else(PoisonError::into_inner); - let needed = prev + src.len(); - if current.capacity < needed { - // Doubling keeps the copies linear in the frame size, and the cap makes - // the last segment exactly the frame, so it freezes without a copy. - let next = Segment::new(needed.max(current.capacity * 2).min(self.size())); - let filled = Filled { - segment: current.clone(), - len: prev, - }; - unsafe { next.write(0, filled.as_ref()) }; - *current = Arc::new(next); - } + Segment::reserve(&mut current, prev, prev + src.len(), size); unsafe { current.write(prev, src) }; } + FrameStorage::Chain(segments) => { + let mut segments = segments.lock().unwrap_or_else(PoisonError::into_inner); + segments.append(prev, src, size, self.0.grow); + } } // Safety: sole writer, and the caller bounds-checked `src` against the size. unsafe { self.store_written(prev + src.len()) }; } - /// Freeze the buffer into the completed payload. - /// - /// Returns the shared [`Bytes`] directly for a whole-frame write (zero-copy), or - /// wraps the mutable allocation otherwise. - fn freeze(&self) -> Bytes { - self.slice(0, self.size()) + /// Freeze the written frame into the completed payload, without a copy. + fn freeze(&self) -> Payload { + match self.0.storage.get() { + Some(FrameStorage::Chain(segments)) => { + let segments = segments.lock().unwrap_or_else(PoisonError::into_inner); + debug_assert!(segments.open.is_none(), "froze a frame mid-segment"); + Payload::Chain(Chain { + segments: segments.sealed.as_slice().into(), + start: 0, + }) + } + _ => Payload::Whole(self.slice(0, self.size())), + } } - /// A zero-copy slice of the written region `[start..end]`. + /// A zero-copy slice of the written region from `start` to `end`, or to the end of + /// `start`'s segment if that comes first. /// /// `end` must not exceed a `written` value already loaded with `Acquire`, which is /// also what makes the segment read here at least as new as those bytes. fn slice(&self, start: usize, end: usize) -> Bytes { - let segment = match self.0.storage.get() { + if start >= end { + return Bytes::new(); + } + let (segment, base) = match self.0.storage.get() { Some(FrameStorage::Shared(bytes)) => return bytes.slice(start..end), - Some(FrameStorage::Fixed(segment)) => segment.clone(), - Some(FrameStorage::Growing(current)) => current.lock().unwrap_or_else(PoisonError::into_inner).clone(), + Some(FrameStorage::Fixed(segment)) => (segment.clone(), 0), + Some(FrameStorage::Growing(current)) => (current.lock().unwrap_or_else(PoisonError::into_inner).clone(), 0), + Some(FrameStorage::Chain(segments)) => { + let segments = segments.lock().unwrap_or_else(PoisonError::into_inner); + let index = start / SEGMENT; + let base = index * SEGMENT; + if let Some(sealed) = segments.sealed.get(index) { + return sealed.slice(start - base..end.min(base + SEGMENT) - base); + } + let open = segments.open.clone().expect("written bytes are sealed or open"); + (open, base) + } // Nothing written, so the range is empty. None => return Bytes::new(), }; - debug_assert!(end <= segment.capacity); - Bytes::from_owner(Filled { segment, len: end }).slice(start..) + let len = end.min(base + SEGMENT) - base; + debug_assert!(len <= segment.capacity); + Bytes::from_owner(Filled { segment, len }).slice(start - base..) } /// Heap bytes the buffer currently holds for the payload. @@ -465,6 +714,11 @@ impl FrameBuf { Some(FrameStorage::Shared(bytes)) => bytes.len(), Some(FrameStorage::Fixed(segment)) => segment.capacity, Some(FrameStorage::Growing(current)) => current.lock().unwrap_or_else(PoisonError::into_inner).capacity, + Some(FrameStorage::Chain(segments)) => { + let segments = segments.lock().unwrap_or_else(PoisonError::into_inner); + let sealed: usize = segments.sealed.iter().map(Bytes::len).sum(); + sealed + segments.open.as_ref().map_or(0, |open| open.capacity) + } None => 0, } } @@ -482,8 +736,10 @@ impl FrameBuf { /// the group, since an unfinished frame leaves the group's stream broken. /// /// A single whole-frame [`write`](Self::write) keeps the caller's allocation -/// (zero-copy); chunked writes copy into one buffer sized to the declared frame, or -/// grown as bytes arrive once the declared sizes in flight pass the origin's budget. +/// (zero-copy). Chunked writes copy into one buffer, or a chain of fixed-size segments +/// for a large frame, each allocated whole by its first byte, or grown as bytes arrive +/// once the frames in flight pass the origin's budget. Nothing is allocated past the +/// segment being written, whatever size was declared. pub struct Producer { group: Option, buf: FrameBuf, @@ -544,6 +800,15 @@ impl Producer { self.buf.size() - self.buf.written(Ordering::Acquire) } + /// Bytes still needed to seal the segment being written, which for a frame within one + /// segment is the rest of the frame. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn unsealed(&self) -> usize { + let written = self.buf.written(Ordering::Acquire); + let end = (written / SEGMENT + 1).saturating_mul(SEGMENT).min(self.buf.size()); + end - written + } + /// Write a chunk of data to the frame, waking its readers. /// /// Returns [`Error::WrongSize`] if the chunk would exceed the remaining bytes. @@ -602,7 +867,12 @@ impl Producer { } let payload = self.buf.freeze(); let mut group = self.group.take().expect("frame owns its group"); - let committed = group.frame_commit(Frame::new(self.info.timestamp, payload)); + let committed = group.frame_commit(Cached { + timestamp: self.info.timestamp, + payload, + // Stamped by the commit, once readable. + arrived: Instant::ZERO, + }); self.done = true; committed?; Ok(group) @@ -652,7 +922,7 @@ impl Drop for Producer { /// tail (streamed). #[derive(Clone)] pub(super) enum Source { - Complete(Bytes), + Complete(Payload), Partial(FrameBuf), } @@ -836,7 +1106,7 @@ impl Consumer { fn size(&self) -> usize { match &self.source { - Source::Complete(bytes) => bytes.len(), + Source::Complete(payload) => payload.len(), Source::Partial(_) => self.info.size as usize, } } @@ -884,12 +1154,12 @@ impl Consumer { return Poll::Ready(Err(Error::Old)); } let buf = match &self.source { - Source::Complete(bytes) => { - if self.read_idx >= bytes.len() { + Source::Complete(payload) => { + if self.read_idx >= payload.len() { return Poll::Ready(Ok(None)); } - let out = bytes.slice(self.read_idx..); - self.read_idx = bytes.len(); + let out = payload.chunk_at(self.read_idx); + self.read_idx += out.len(); self.stats.bytes(out.len() as u64); return Poll::Ready(Ok(Some(out))); } @@ -901,7 +1171,7 @@ impl Consumer { let written = buf.written(Ordering::Acquire); if written > self.read_idx { let out = buf.slice(self.read_idx, written); - self.read_idx = written; + self.read_idx += out.len(); self.stats.bytes(out.len() as u64); return Poll::Ready(Ok(Some(out))); } @@ -942,9 +1212,9 @@ impl Consumer { return Poll::Ready(Err(Error::Old)); } let buf = match &self.source { - Source::Complete(bytes) => { - let out = bytes.slice(self.read_idx..); - self.read_idx = bytes.len(); + Source::Complete(payload) => { + let out = payload.contiguous(self.read_idx); + self.read_idx = payload.len(); self.stats.bytes(out.len() as u64); return Poll::Ready(Ok(out)); } @@ -967,7 +1237,7 @@ impl Consumer { Poll::Pending } })?); - let out = buf.slice(read_idx, size); + let out = buf.freeze().contiguous(read_idx); self.read_idx = size; self.stats.bytes(out.len() as u64); Poll::Ready(Ok(out)) @@ -1018,7 +1288,7 @@ mod tests { assert_eq!(early, &b"abc"[..]); assert_eq!(buf.slice(3, 7), &b"defg"[..]); - assert_eq!(buf.freeze(), &b"abcdefghij"[..]); + assert_eq!(buf.freeze().contiguous(0), &b"abcdefghij"[..]); } /// A reader on another thread slicing while the writer reallocates only ever sees @@ -1048,7 +1318,71 @@ mod tests { buf.append(chunk); } reader.join().unwrap(); - assert_eq!(buf.freeze(), payload); + assert_eq!(buf.freeze().contiguous(0), payload); + } + + /// A frame larger than a segment allocates only the segment being written, whatever it + /// declared, and seals each at the same fixed boundary. Reads never cross one, and a + /// whole read copies the chain into one buffer. + #[test] + fn chain_allocates_a_segment_at_a_time() { + const SIZE: usize = 2 * SEGMENT + 100; + // Straddles every segment boundary. + const CHUNK: usize = 300_000; + let payload: Vec = (0..SIZE).map(|i| (i % 251) as u8).collect(); + + for grow in [false, true] { + let buf = FrameBuf::with_growth(SIZE, grow); + let mut read = 0; + for chunk in payload.chunks(CHUNK) { + buf.append(chunk); + let written = buf.written(Ordering::Acquire); + let ahead = written.next_multiple_of(SEGMENT).min(SIZE); + assert!( + buf.allocated() <= ahead, + "grow={grow}: {} past {written}", + buf.allocated() + ); + + while read < written { + let slice = buf.slice(read, written); + assert_eq!(slice, payload[read..read + slice.len()]); + assert!( + read / SEGMENT == (read + slice.len() - 1) / SEGMENT, + "a read crossed a segment" + ); + read += slice.len(); + } + } + + let Payload::Chain(chain) = buf.freeze() else { + panic!("grow={grow}: a frame past one segment is a chain"); + }; + let lens: Vec = chain.segments.iter().map(Bytes::len).collect(); + assert_eq!(lens, [SEGMENT, SEGMENT, 100]); + assert_eq!(buf.freeze().contiguous(0), payload); + assert_eq!(buf.freeze().contiguous(SEGMENT + 1), payload[SEGMENT + 1..]); + } + } + + /// A chain read as a [`Buf`] hands out each segment without a copy and advances only + /// its own view. + #[test] + fn chain_reads_as_a_buf() { + let first = Bytes::from(vec![1u8; SEGMENT]); + let stored = Payload::Chain(Chain { + segments: Arc::from([first.clone(), Bytes::from_static(b"tail")]), + start: 0, + }); + + let mut view = stored.clone(); + assert_eq!(view.chunk().len(), SEGMENT); + let head = view.copy_to_bytes(SEGMENT - 1); + assert_eq!(head.as_ptr(), first.as_ptr(), "a read within a segment copied"); + assert_eq!(view.copy_to_bytes(3), &[1, b't', b'a'][..], "a read across segments"); + assert_eq!(view.chunk(), b"il"); + assert_eq!(view.remaining(), 2); + assert_eq!(stored.len(), SEGMENT + 4, "the stored payload never moved"); } /// A reservation holds its bytes until it drops, and a request larger than what diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index a8229ddf461..7f76df61fd4 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -16,7 +16,7 @@ //! The stream is closed with [Error] when all writers or readers are dropped. use super::flate; use crate::cache; -use crate::frame::{self, Frame, FrameBuf}; +use crate::frame::{self, Cached, FrameBuf, Payload}; use crate::{Timescale, stats, track}; use bytes::Bytes; use std::collections::VecDeque; @@ -135,7 +135,7 @@ pub(super) struct Partial { /// A completed frame as a group stores it, with what egress needs beside it. #[derive(Clone)] pub(super) struct Entry { - frame: Frame, + frame: Cached, // The IETF object fields it arrived with, for a relay to forward. fields: Option>, // A compressed group's inflated size of the stored slice; zero on a plain group. @@ -144,7 +144,7 @@ pub(super) struct Entry { impl Entry { /// A frame without object fields. - fn plain(frame: Frame, inflated: u64) -> Self { + fn plain(frame: Cached, inflated: u64) -> Self { Self { frame, fields: None, @@ -444,7 +444,7 @@ struct Inflate { // Local index of the first slice: the copy's inflated length when the read took them. from: usize, // Each slice with its declared inflated size. - slices: Vec<(Bytes, u64)>, + slices: Vec<(Payload, u64)>, } impl Inflate { @@ -470,7 +470,7 @@ impl Inflate { bill.add(flate::DECODER_HEAP); flate::Decoder::new() }); - decoder.frame(slice, size) + decoder.frame(&slice.contiguous(0), size) }); match frame { Ok(frame) => { @@ -1266,9 +1266,9 @@ impl Producer { } None => 0, }; - let frame = Frame { + let frame = Cached { timestamp, - payload, + payload: payload.into(), arrived: super::clock::now(), }; let slots = state.push(Entry::plain(frame, inflated)); @@ -1377,7 +1377,7 @@ impl Producer { state.cache += size; state.stamp(frame.timestamp); latest = frame.timestamp; - let slots = state.push(Entry::plain(frame, inflated)); + let slots = state.push(Entry::plain(frame.into(), inflated)); now = state.charge.add(size + slots); } state.next_index = next_index; @@ -1429,7 +1429,8 @@ impl Producer { pub(crate) fn create_object(mut self, frame: frame::Header) -> Result { let meter = self.chunk_meter(frame.inflated); let size = frame.info.size as usize; - let reserved = self.handle.cache.frames().reserve(size); + // Only the segment being written is ever allocated ahead of its bytes. + let reserved = self.handle.cache.frames().reserve(size.min(frame::SEGMENT)); let buf = match reserved { Some(_) => FrameBuf::new(size), None => FrameBuf::growing(size), @@ -1570,7 +1571,7 @@ impl Producer { } /// Commit the in-flight frame as a completed frame (called by [`frame::Producer::finish`]). - pub(super) fn frame_commit(&mut self, frame: Frame) -> Result<()> { + pub(super) fn frame_commit(&mut self, frame: Cached) -> Result<()> { let mut state = modify(&self.handle.state)?; // Completing the frame charges whatever it wrote since the last notify, and is a // write access like any chunk, and the only one the payload is guaranteed to get: @@ -1585,7 +1586,7 @@ impl Producer { if partial.as_ref().is_some_and(|partial| partial.scratch) { let flate = state.flate.as_mut().expect("scratch only opens on a compressed group"); inflated = frame.payload.len() as u64; - frame.payload = flate.deflate(&frame.payload); + frame.payload = flate.deflate(&frame.payload.contiguous(0)).into(); flate.inflated += inflated; let size = frame.payload.len() as u64; // The frame already counts against the group's frames. @@ -2236,10 +2237,16 @@ impl Demand { /// consumer that never reads whole frames, or drains through a higher-level buffer, /// pays nothing. Inline storage would triple the size of every [`Consumer`], which /// moves by value through each layer that hands out a group. +/// +/// Slots hold the public [`frame::Frame`], with a segmented payload set aside in `chains`, +/// so a whole read of a small frame moves it out of its slot as it did before segments. #[derive(Default)] struct Prefetch { // Initialized, not-yet-taken frames are `frames[pos..len]`; the rest are uninitialized. - frames: Option; Self::CAP]>>, + // A slot whose frame is in `chains` holds an empty payload. + frames: Option; Self::CAP]>>, + // Segmented payloads for `frames`, allocated by the first batch that carries one. + chains: Option; Self::CAP]>>, // Relay metadata for `frames`, allocated by the first batch that carries any. fields: Option>; Self::CAP]>>, // Inflated sizes for `frames`, filled only by a batch of a compressed group's stored slices. @@ -2256,8 +2263,38 @@ impl Prefetch { self.inflated.as_ref()?.get(self.pos.checked_sub(1)?).copied() } - /// Take the next buffered frame, or `None` if the batch is drained. - fn pop(&mut self) -> Option { + /// Take the next buffered frame whole, copying a segmented payload into one buffer. + // Every small-frame read comes through here, so the chain copy stays out of line: + // with it inline, this measured ~30 more instructions a frame on `group_read_frames`. + #[inline(always)] + fn pop(&mut self) -> Option { + let mut frame = self.take()?; + if self.chains.is_some() { + self.join_chain(&mut frame); + } + Some(frame) + } + + #[cold] + #[inline(never)] + fn join_chain(&mut self, frame: &mut frame::Frame) { + if let Some(chain) = self.take_chain() { + frame.payload = chain.concat(); + } + } + + /// Take the next buffered frame as stored, keeping a segmented payload in segments. + fn pop_cached(&mut self) -> Option { + let frame = self.take()?; + Some(Cached { + timestamp: frame.timestamp, + payload: self.take_chain().map_or(Payload::Whole(frame.payload), Payload::Chain), + arrived: frame.arrived, + }) + } + + #[inline] + fn take(&mut self) -> Option { if self.pos == self.len { return None; } @@ -2268,6 +2305,12 @@ impl Prefetch { Some(frame) } + /// The segmented payload of the frame [`Self::take`] just returned. + #[inline] + fn take_chain(&mut self) -> Option { + self.chains.as_mut()?.get_mut(self.pos.checked_sub(1)?)?.take() + } + /// The relay metadata of the frame [`Self::pop`] just returned. fn take_fields(&mut self) -> Option> { self.fields.as_mut()?.get_mut(self.pos.checked_sub(1)?)?.take() @@ -2290,7 +2333,22 @@ impl Prefetch { let frames = self .frames .get_or_insert_with(|| Box::new([const { MaybeUninit::uninit() }; Self::CAP])); - frames[at].write(entry.frame.clone()); + let payload = match (&entry.frame.payload, &mut self.chains) { + (Payload::Whole(bytes), Some(slots)) => { + slots[at] = None; + bytes.clone() + } + (Payload::Whole(bytes), None) => bytes.clone(), + (Payload::Chain(chain), slots) => { + slots.get_or_insert_default()[at] = Some(chain.clone()); + Bytes::new() + } + }; + frames[at].write(frame::Frame { + timestamp: entry.frame.timestamp, + payload, + arrived: entry.frame.arrived, + }); // Plain frames never allocate the metadata table. match (&entry.fields, &mut self.fields) { (Some(fields), slots) => slots.get_or_insert_default()[at] = Some(fields.clone()), @@ -2349,8 +2407,13 @@ impl Prefetch { }; for i in self.pos..self.len { match job.get(local + i) { - // SAFETY: slots in `pos..len` are initialized (written by `fill`, not yet popped). - Ok(payload) => unsafe { frames[i].assume_init_mut() }.payload = payload, + Ok(payload) => { + // SAFETY: slots in `pos..len` are initialized (written by `fill`, not yet popped). + unsafe { frames[i].assume_init_mut() }.payload = payload; + if let Some(chains) = &mut self.chains { + chains[i] = None; + } + } Err(err) => { for slot in &mut frames[i..self.len] { // SAFETY: as above, and `len` drops below them right after. @@ -2375,6 +2438,13 @@ impl Prefetch { // SAFETY: slots in `pos..len` are initialized (written by `fill`, not yet popped). bytes += unsafe { slot.assume_init_ref() }.payload.len() as u64; } + if let Some(chains) = &self.chains { + bytes += chains[self.pos..self.len] + .iter() + .flatten() + .map(|chain| chain.len() as u64) + .sum::(); + } ((self.len - self.pos) as u64, bytes) } } @@ -2856,17 +2926,32 @@ impl Consumer { /// Read the next frame (timestamp and payload) all at once, without blocking. pub fn poll_read_frame(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.poll_read_as(waiter, Prefetch::pop) + } + + /// Read the next frame as stored, so a frame in segments stays in them. + #[cfg(feature = "async")] + fn poll_read_cached(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.poll_read_as(waiter, Prefetch::pop_cached) + } + + /// Read the next frame through the prefetch with `take`, which picks its form. + fn poll_read_as( + &mut self, + waiter: &kio::Waiter, + take: impl Fn(&mut Prefetch) -> Option + Copy, + ) -> Poll>> { if self.recover.is_none() { - return self.poll_read_frame_once(waiter); + return self.poll_read_frame_once(waiter, take); } - self.poll_resumed(waiter, Self::poll_read_frame_once) + self.poll_resumed(waiter, |this, waiter| this.poll_read_frame_once(waiter, take)) } /// Read the next frame with its relay metadata, for wire egress. #[cfg(feature = "async")] pub(crate) fn poll_read_object(&mut self, waiter: &kio::Waiter) -> Poll>> { // Every frame is served through the cursor's prefetch, which holds its fields. - let frame = ready!(self.poll_read_frame(waiter))?; + let frame = ready!(self.poll_read_cached(waiter))?; Poll::Ready(Ok(frame.map(|frame| frame::Object { frame, fields: self.cursor.prefetch.take_fields(), @@ -2874,11 +2959,15 @@ impl Consumer { }))) } - fn poll_read_frame_once(&mut self, waiter: &kio::Waiter) -> Poll>> { + fn poll_read_frame_once( + &mut self, + waiter: &kio::Waiter, + take: impl Fn(&mut Prefetch) -> Option + Copy, + ) -> Poll>> { if self.expired { return Poll::Ready(Err(Error::Old)); } - let res = self.cursor.poll_read_frame(waiter, &self.stats); + let res = self.cursor.poll_read_frame(waiter, &self.stats, take); self.poll_expired_if_blocked(waiter, res) } @@ -3067,7 +3156,7 @@ impl Cursor { // Hand out any frames a prior read_frame prefetched before touching the tail. // Their bytes were already counted at the batch fill, so the frame::Consumer // carries no meter. - if let Some(frame) = self.prefetch.pop() { + if let Some(frame) = self.prefetch.pop_cached() { self.refresh_if_stale(); self.index += 1; let tail = self.index.saturating_add(self.prefetch.buffered().0 as usize)..end; @@ -3127,7 +3216,7 @@ impl Cursor { fields, inflated: None, }; - (header, frame::Source::Complete(payload)) + (header, frame::Source::Complete(payload.into())) } }, }; @@ -3143,13 +3232,18 @@ impl Cursor { }))) } - fn poll_read_frame(&mut self, waiter: &kio::Waiter, stats: &stats::Meter) -> Poll>> { + fn poll_read_frame( + &mut self, + waiter: &kio::Waiter, + stats: &stats::Meter, + take: impl Fn(&mut Prefetch) -> Option + Copy, + ) -> Poll>> { if self.capped() { return Poll::Ready(Ok(None)); } // Fast path: serve from the prefetched batch without locking or allocating a waker. - if let Some(frame) = self.prefetch.pop() { + if let Some(frame) = take(&mut self.prefetch) { self.refresh_if_stale(); self.index += 1; return Poll::Ready(Ok(Some(frame))); @@ -3167,7 +3261,7 @@ impl Cursor { let (frames, bytes) = self.prefetch.buffered(); stats.frames(frames); stats.bytes(bytes); - return Poll::Ready(Ok(self.prefetch.pop().inspect(|_| { + return Poll::Ready(Ok(take(&mut self.prefetch).inspect(|_| { self.index += 1; }))); } @@ -3213,7 +3307,7 @@ impl Cursor { stats.frames(frames); stats.bytes(bytes); - Poll::Ready(Ok(self.prefetch.pop().inspect(|_| { + Poll::Ready(Ok(take(&mut self.prefetch).inspect(|_| { self.index += 1; }))) } @@ -3261,6 +3355,7 @@ impl Fetch { #[cfg(test)] mod test { use super::*; + use crate::frame::Frame; use crate::model::test_tracing::count_drop_warnings; use bytes::Bytes; #[cfg(feature = "async")] @@ -3787,7 +3882,7 @@ mod test { let head = Head { start: 0, first_object: true, - entries: vec![Entry::plain(Frame::new(None, Bytes::from_static(b"ab")), 0)], + entries: vec![Entry::plain(Frame::new(None, Bytes::from_static(b"ab")).into(), 0)], compressed: false, }; adopter.adopt_head(&head); @@ -3813,9 +3908,9 @@ mod test { start: 0, first_object: true, entries: vec![ - Entry::plain(Frame::new(None, Bytes::from_static(b"a")), 0), + Entry::plain(Frame::new(None, Bytes::from_static(b"a")).into(), 0), Entry { - frame: Frame::new(None, Bytes::from_static(b"b")), + frame: Frame::new(None, Bytes::from_static(b"b")).into(), fields: fields(1), inflated: 0, }, @@ -3863,7 +3958,10 @@ mod test { let mut whole = tail.consume(); let mut read = Vec::new(); while let Poll::Ready(Ok(Some(object))) = whole.poll_read_object(&waiter) { - read.push((object.frame.payload, object.fields.map(|fields| fields.id))); + read.push(( + object.frame.payload.contiguous(0), + object.fields.map(|fields| fields.id), + )); } assert_eq!( read, @@ -4707,6 +4805,89 @@ mod test { assert_eq!(third.allocated(), 1024, "the finished frame did not return its share"); } + /// A frame past one segment reserves and allocates a segment rather than its declared + /// size, so a peer declaring the largest frame commits no more than a small one. + #[test] + fn a_large_frame_allocates_one_segment_ahead() { + let budget = frame::Budget::new(2 * frame::SEGMENT); + let pool = cache::Pool::unbounded(); + let group = |sequence| { + let cache = cache::Track::new(pool.clone(), budget.clone(), kio::Weak::new()); + Producer::new(Info { sequence }, track::Info::timed(), cache) + }; + + let frames: Vec<_> = (0..2) + .map(|sequence| { + let mut frame = group(sequence).create_frame(sized(MAX_CACHE_BYTES)).unwrap(); + frame.write(Bytes::from_static(b"x")).unwrap(); + frame + }) + .collect(); + for frame in &frames { + assert_eq!(frame.allocated(), frame::SEGMENT, "reserved other than one segment"); + } + } + + /// A prefetched batch mixing small and segmented frames hands each one out intact, + /// whether it is read whole, streamed, or read as stored for the wire. + #[test] + #[cfg(feature = "async")] + fn a_batch_mixes_segmented_frames() { + let mut group = Info { sequence: 0 }.produce(); + let payloads: Vec = (0..Prefetch::CAP) + .map(|i| match i % 2 { + 0 => Bytes::from(vec![i as u8; 3]), + _ => Bytes::from((0..2 * frame::SEGMENT + 7).map(|b| (b + i) as u8).collect::>()), + }) + .collect(); + for payload in &payloads { + let mut frame = group + .create_frame(frame::Info { + size: payload.len() as u64, + timestamp: Some(Timestamp::ZERO), + }) + .unwrap(); + // Written in pieces, so the large ones chain. + for piece in payload.chunks(300_000) { + frame.write(payload.slice_ref(piece)).unwrap(); + } + group = frame.finish().unwrap(); + } + group.finish().unwrap(); + + let mut whole = group.consume(); + for (i, payload) in payloads.iter().enumerate() { + let frame = whole.read_frame().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(&frame.payload, payload, "whole frame {i}"); + } + + // The first read prefetches the batch, so the streamed rest come out of it. + let mut mixed = group.consume(); + mixed.read_frame().now_or_never().unwrap().unwrap().unwrap(); + for (i, payload) in payloads.iter().enumerate().skip(1) { + let mut frame = mixed.next_frame().now_or_never().unwrap().unwrap().unwrap(); + let first = frame.read_chunk().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!( + first.len(), + payload.len().min(frame::SEGMENT), + "streamed frame {i} lost its segments" + ); + let rest = frame.read_all().now_or_never().unwrap().unwrap(); + assert_eq!([first, rest].concat(), payload[..], "streamed frame {i}"); + } + + let mut objects = group.consume(); + let waiter = kio::Waiter::noop(); + for (i, payload) in payloads.iter().enumerate() { + let Poll::Ready(Ok(Some(object))) = objects.poll_read_object(&waiter) else { + panic!("frame {i} is ready"); + }; + let chained = matches!(object.frame.payload, Payload::Chain(_)); + assert_eq!(chained, payload.len() > frame::SEGMENT, "stored frame {i}"); + assert_eq!(object.frame.payload.contiguous(0), payload, "stored frame {i}"); + } + } + /// A repetitive frame DEFLATE shrinks, distinct per index. fn verbose(index: usize) -> Bytes { Bytes::from(format!("{index}:{}", "moq".repeat(300))) @@ -4795,7 +4976,9 @@ mod test { }; let size = object.inflated.expect("a stored slice carries its size"); assert_eq!(size, verbose(index).len() as u64); - let frame = decoder.frame(&object.frame.payload, size as usize).unwrap(); + let frame = decoder + .frame(&object.frame.payload.contiguous(0), size as usize) + .unwrap(); assert_eq!(frame, verbose(index), "slice {index}"); } diff --git a/rs/moq-net/src/transport.rs b/rs/moq-net/src/transport.rs index 7a44b10e87b..b956acd9bd3 100644 --- a/rs/moq-net/src/transport.rs +++ b/rs/moq-net/src/transport.rs @@ -180,13 +180,14 @@ impl Stats for StatsUnavailable {} /// /// A session keeps each unit it is still assembling leased, and a unit larger than the /// window can never complete, so the window has to fit the largest one with its framing: -/// a legacy announcement set, past any frame payload. +/// a legacy announcement set, past any frame segment. A frame returns its credit a +/// segment at a time, so its size never matters here. pub const MIN_RECEIVE_WINDOW: u64 = { // A type and a size prefix, each a varint of at most 8 bytes. const PREFIXES: u64 = 16; let message = as crate::lite::Message>::MAX_SIZE as u64 + PREFIXES; - let frame = crate::group::MAX_CACHE_BYTES; - if message > frame { message } else { frame } + let segment = crate::frame::SEGMENT as u64; + if message > segment { message } else { segment } }; /// The typed transport condition when no application code was received. diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index 89330aee100..8a8c9689137 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -50,6 +50,7 @@ mod request_caps; mod restart; mod route_change; mod routing_contracts; +mod segmented_frame; mod session_close; mod session_panic; mod settled_end_survives_abort; diff --git a/rs/moq-net/tests/integration/segmented_frame.rs b/rs/moq-net/tests/integration/segmented_frame.rs new file mode 100644 index 00000000000..0bfe4898972 --- /dev/null +++ b/rs/moq-net/tests/integration/segmented_frame.rs @@ -0,0 +1,81 @@ +#![cfg(feature = "async")] + +//! A frame larger than a storage segment, written in pieces so its group keeps it as a +//! chain of segments, crosses a session intact on each protocol, read whole and streamed. + +use std::time::Duration; + +use bytes::Bytes; +use moq_net::{Hop, Timestamp, Version, frame, origin, track}; + +use crate::support::harness::{MockConnectOptions, connect_mock, spawn, subscribed}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +const VERSIONS: &[&str] = &["moq-lite-06", "moq-lite-07-wip", "moq-transport-22"]; + +/// Past two segments, so the chain has a full middle and a short tail. +const SIZE: usize = 5 * 1024 * 1024 / 2; + +fn origin(hop: u64) -> origin::Producer { + let (origin, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + spawn(driver); + origin +} + +#[moq_clock::test] +async fn a_segmented_frame_crosses_whole() { + let payload = Bytes::from((0..SIZE).map(|i| (i % 251) as u8).collect::>()); + for version in VERSIONS { + let version: Version = version.parse().unwrap(); + let publisher = origin(1); + let viewer = origin(2); + let broadcast = publisher.create_broadcast("room").unwrap(); + let info = track::Info::default().with_timescale(moq_net::Timescale::MILLI); + let track = broadcast.create_track("file", info).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(viewer.clone()); + let _pair = connect_mock(options).await; + + let remote = moq_clock::timeout(TIMEOUT, viewer.consume().routed_broadcast("room")) + .await + .expect("broadcast timeout") + .unwrap(); + let mut sub = remote.track("file").unwrap().subscribe(None).await.unwrap(); + subscribed(&track).await; + + // Two frames: one read whole, one streamed. + let mut group = track.append_group().unwrap(); + for _ in 0..2 { + let mut frame = group + .create_frame(frame::Info { + size: SIZE as u64, + timestamp: Some(Timestamp::ZERO), + }) + .unwrap(); + for chunk in payload.chunks(300_000) { + frame.write(Bytes::copy_from_slice(chunk)).unwrap(); + } + group = frame.finish().unwrap(); + } + group.finish().unwrap(); + + let mut group = moq_clock::timeout(TIMEOUT, sub.recv_group()) + .await + .expect("group timeout") + .unwrap() + .expect("a group"); + let whole = group.read_frame().await.unwrap().expect("a frame"); + assert_eq!(whole.payload, payload, "{version}"); + + let mut streamed = group.next_frame().await.unwrap().expect("a frame"); + let mut read = Vec::new(); + while let Some(chunk) = streamed.read_chunk().await.unwrap() { + read.extend_from_slice(&chunk); + } + assert_eq!(read, payload, "{version}"); + } +} diff --git a/rs/moq-tokio/tests/integration/receive_credit.rs b/rs/moq-tokio/tests/integration/receive_credit.rs index e7adafb995b..7b150892b25 100644 --- a/rs/moq-tokio/tests/integration/receive_credit.rs +++ b/rs/moq-tokio/tests/integration/receive_credit.rs @@ -339,9 +339,27 @@ async fn bufferbloat_is_not_a_stuck_window() { assert!(stats.down.throttled > 0, "the bottleneck held the sender back: {stats}"); } -/// A frame larger than the whole window, which can never complete: the stuck window -/// evicts it like any other, so its reader learns the group was evicted and is not -/// fetched again. +/// A frame many windows long still completes, since its credit returns as each storage +/// segment seals: the window only has to fit one segment, so nothing is evicted. +#[tokio::test] +async fn lite_frame_of_many_windows_completes() { + // Room for one of moq-net's 1 MiB storage segments and its framing. + const SEGMENTED: usize = 4 << 20; + let setup = setup("moq-lite-06", transport(SEGMENTED as u64)).await; + let (track, mut sub) = setup.subscribe("file", 0).await; + + let payload = Bytes::from((0..4 * SEGMENTED).map(|i| (i % 251) as u8).collect::>()); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + group.finish().unwrap(); + + assert!(next_frame(&mut sub).await == payload, "the frame arrived corrupted"); + assert_eq!(setup.conn.stats().credit.evicted, 0); +} + +/// A frame within one storage segment but larger than the whole window, which can never +/// complete: the stuck window evicts it like any other, so its reader learns the group +/// was evicted and is not fetched again. #[tokio::test] async fn lite_frame_past_the_window_gives_way() { let setup = setup("moq-lite-06", transport(WINDOW as u64)).await;