perf(moq-net): store large frames as sealed segments

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

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

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

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

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<usize>); 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);

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<Item = (usize, usize)> {
+	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();
 }

rs/moq-net/src/coding/reader.rs

@@ -333,7 +333,8 @@ impl<S: RecvStream, V: StreamCodes> Reader<S, V> {
 	/// 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<S: RecvStream, V: StreamCodes> Reader<S, V> {
 			};
 
 			// 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<Bytes>,
+		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<Result<CreditedLease<'_>, Self::Error>> {
+			match self.chunks.iter().map(Bytes::len).sum::<usize>() >= 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<Item = &[u8]> {
+			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::<Vec<_>>());
+		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);
+	}
 }

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<Ranked<crate::transport::boxed::SendStream>, 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<crate::transport::boxed::SendStream, Version>,
 		frame: Option<frame::Consumer>,
-		chunk: Option<bytes::Bytes>,
+		chunk: Option<crate::frame::Payload>,
 		batch: Box<arrayvec::ArrayVec<frame::Object, 8>>,
 		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,