From f41a4fda062dd519a537e2870fb8cc855ad820c7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 07:15:48 -0700 Subject: [PATCH 01/45] chore: claim quest/m1/quic/receive-memory (lease design) Co-Authored-By: Claude Opus 5.5 From 57aef679acc28660229c1bebb07e1a0c14ab08d5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 08:05:12 -0700 Subject: [PATCH 02/45] wip(net,quic): receive leases with transport-side eviction moq-net's RecvStream leases unconsumed bytes and ranks streams for eviction; transports without leases go through Buffered. moq-quic keeps leased bytes with their connection credit and evicts the lowest-priority holder once the window is stuck. Drivers are not ported yet. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/client.rs | 36 +- rs/moq-net/src/coding/decode.rs | 54 +- rs/moq-net/src/coding/reader.rs | 393 ++++++++----- rs/moq-net/src/coding/stream.rs | 7 +- rs/moq-net/src/error.rs | 8 +- rs/moq-net/src/fuzz.rs | 24 +- rs/moq-net/src/ietf/adapter.rs | 192 ++++--- rs/moq-net/src/ietf/session.rs | 6 +- rs/moq-net/src/ietf/subscriber.rs | 1 + rs/moq-net/src/lite/session.rs | 1 + rs/moq-net/src/lite/subscriber.rs | 7 +- rs/moq-net/src/lite/test_transport.rs | 63 +-- rs/moq-net/src/priority.rs | 25 + rs/moq-net/src/server.rs | 30 +- rs/moq-net/src/transport.rs | 520 +++++++++++++----- rs/moq-net/src/transport/memory.rs | 67 +-- rs/moq-net/tests/integration/goaway.rs | 3 +- .../tests/integration/subscribe_tracks.rs | 5 +- rs/moq-net/tests/integration/support/mock.rs | 47 +- rs/moq-quic/src/connection/mod.rs | 17 +- rs/moq-quic/src/connection/stats.rs | 5 +- rs/moq-quic/src/connection/streams/lease.rs | 381 +++++++++++++ rs/moq-quic/src/connection/streams/mod.rs | 7 +- rs/moq-quic/src/connection/streams/recv.rs | 80 +-- rs/moq-quic/src/connection/streams/state.rs | 440 +++++++++------ rs/moq-quic/src/fuzz/connection.rs | 68 ++- rs/moq-quic/src/lib.rs | 6 +- rs/moq-quic/src/tests/mod.rs | 105 ++-- 28 files changed, 1665 insertions(+), 933 deletions(-) create mode 100644 rs/moq-quic/src/connection/streams/lease.rs diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 00e3d42a6d8..556471a33e1 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -558,9 +558,9 @@ mod tests { fn new(protocol: Option<&'static str>, server_control_bytes: Vec) -> Self { let writes = Arc::new(Mutex::new(Vec::new())); let send = FakeSendStream { writes: writes.clone() }; - let recv = FakeRecvStream { + let recv = crate::transport::poll::Buffered::new(FakeChunks { data: VecDeque::from(server_control_bytes), - }; + }); let state = FakeSessionState { protocol, control_stream: Mutex::new(Some((send, recv))), @@ -711,30 +711,25 @@ mod tests { } } - struct FakeRecvStream { + type FakeRecvStream = crate::transport::poll::Buffered; + + struct FakeChunks { data: VecDeque, } - impl crate::transport::poll::RecvStream for FakeRecvStream { + impl crate::transport::poll::Chunks for FakeChunks { type Error = FakeError; - fn poll_read(&mut self, _cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_read_chunk(&mut self, _cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { if self.data.is_empty() { return Poll::Ready(Ok(None)); } - let size = dst.len().min(self.data.len()); - for slot in dst.iter_mut().take(size) { - *slot = self.data.pop_front().unwrap(); - } - Poll::Ready(Ok(Some(size))) + let size = max.min(self.data.len()); + Poll::Ready(Ok(Some(self.data.drain(..size).collect()))) } fn stop(&mut self, _code: u32) {} - - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) - } } fn mock_server_setup(negotiated: Version) -> Vec { @@ -1092,17 +1087,18 @@ mod tests { impl crate::transport::poll::RecvStream for LocalRecv { type Error = FakeError; + type Lease<'a> = crate::transport::poll::BufferedLease<'a>; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { - self.inner.poll_read(cx, dst) + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Self::Error>> { + self.inner.poll_lease(cx, want) } - fn stop(&mut self, code: u32) { - self.inner.stop(code); + fn set_priority(&mut self, priority: i64) { + self.inner.set_priority(priority); } - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { - crate::transport::poll::RecvStream::poll_closed(&mut self.inner, cx) + fn stop(&mut self, code: u32) { + crate::transport::poll::RecvStream::stop(&mut self.inner, code); } } diff --git a/rs/moq-net/src/coding/decode.rs b/rs/moq-net/src/coding/decode.rs index dd756e7034a..d1ba55f70f8 100644 --- a/rs/moq-net/src/coding/decode.rs +++ b/rs/moq-net/src/coding/decode.rs @@ -137,13 +137,37 @@ impl From for DecodeError { pub struct Decoder<'a> { buf: &'a [u8], form: Form, + /// The length of the buffer this decoder started with. + len: usize, + /// See [`Self::need`]. + need: usize, } impl<'a> Decoder<'a> { /// Read `buf`, with varints in the given form. #[cfg_attr(not(feature = "async"), allow(dead_code))] pub fn new(buf: &'a [u8], form: Form) -> Self { - Self { buf, form } + Self { + buf, + form, + len: buf.len(), + need: 0, + } + } + + /// After a [`DecodeError::Short`], how many bytes from the start the failed read + /// needed: at least one more than the buffer when it can't tell, and 0 if no read of + /// this decoder ran short. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub fn need(&self) -> usize { + self.need + } + + /// Record that a read wanted `more` bytes past the current position. + fn short(&mut self, more: usize) -> DecodeError { + let need = (self.len - self.buf.len()).saturating_add(more); + self.need = self.need.max(need); + DecodeError::Short } /// The varint form this decoder reads. @@ -168,7 +192,7 @@ impl<'a> Decoder<'a> { #[cfg_attr(not(feature = "async"), allow(dead_code))] pub fn slice(&mut self, len: usize) -> Result<&'a [u8], DecodeError> { let Some((head, rest)) = self.buf.split_at_checked(len) else { - return Err(DecodeError::Short); + return Err(self.short(len)); }; self.buf = rest; Ok(head) @@ -214,9 +238,15 @@ impl<'a> Decoder<'a> { #[cfg_attr(not(target_arch = "wasm32"), inline(always))] #[cfg_attr(not(feature = "async"), allow(dead_code))] pub fn varint(&mut self) -> Result { - let (value, rest) = varint::read(self.buf, self.form)?; - self.buf = rest; - Ok(value) + match varint::read(self.buf, self.form) { + Ok((value, rest)) => { + self.buf = rest; + Ok(value) + } + // The width is in the first byte, but a byte at a time is short enough. + Err(DecodeError::Short) => Err(self.short(self.buf.len() + 1)), + Err(err) => Err(err), + } } /// Read an optional varint: 0 is `None`, and `n + 1` is `Some(n)`. @@ -256,4 +286,18 @@ mod tests { assert!(matches!(r.varint(), Err(DecodeError::Short))); assert_eq!(r.remaining(), 1); } + + /// A short read reports how much it needed from the start, so a reader can wait for + /// all of it rather than retry once per byte. + #[test] + fn short_reports_what_it_needed() { + let mut r = Decoder::new(&[0x01, 0x05, b'a', b'b'], Form::Quic); + assert_eq!(r.varint().unwrap(), 1); + assert!(matches!(r.bytes(), Err(DecodeError::Short))); + assert_eq!(r.need(), 7, "a 5 byte body after two prefix bytes"); + + let mut r = Decoder::new(&[0x40], Form::Quic); + assert!(matches!(r.varint(), Err(DecodeError::Short))); + assert_eq!(r.need(), 2); + } } diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index 5c3bf08ec8a..df207dadc71 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -1,23 +1,31 @@ use std::{ - cmp, fmt::Debug, task::{Context, Poll, ready}, }; -use bytes::{Buf, Bytes, BytesMut}; +use bytes::{Buf, Bytes}; -use crate::{Error, StreamError, coding::*}; +use crate::{ + Error, StreamError, + coding::*, + transport::poll::{Lease, RecvStream}, +}; /// A reader for decoding messages from a stream. /// /// The `poll_*` methods are the implementation; the `async` methods are thin /// wrappers, so the reader can be driven from another poll function without -/// pinning a future. Partial bytes accumulate in the internal buffer, so a -/// `Pending` (or a cancelled wrapper) never desynchronizes the stream. -pub struct Reader { +/// pinning a future. Unconsumed bytes stay leased in the transport, which counts them +/// against its flow control until each unit is complete, so a `Pending` (or a cancelled +/// wrapper) never desynchronizes the stream and a parked reader holds no memory of its own. +pub struct Reader { stream: S, - buffer: BytesMut, version: V, + /// How many bytes the last short decode needed, so the retry waits for all of them + /// instead of waking once per chunk. + want: usize, + /// A contiguous copy of a message that spans transport chunks, empty between decodes. + scratch: Vec, } /// How much payload may pile up behind a parked consumer before @@ -31,17 +39,27 @@ pub struct Reader { /// enough that an ordinary burst is still one wake. const WAKE_BUDGET: usize = 64 * 1024; -impl Reader { +/// The largest [`Reader::scratch`] kept between decodes; one past this is freed. +const SCRATCH_KEEP: usize = 64 * 1024; + +impl Reader { pub fn new(stream: S, version: V) -> Self { Self { stream, - buffer: Default::default(), version, + want: 0, + scratch: Vec::new(), } } - /// Poll `decode` against the buffered bytes, reading more until it stops coming up - /// short. `consume` drops what it read; otherwise the bytes stay for the next read. + /// Poll the stream for a lease of at least `want` bytes. + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Error>> { + let version = &self.version; + self.stream.poll_lease(cx, want).map_err(|err| version.transport_error(err)) + } + + /// Poll `decode` against the leased bytes, leasing more until it stops coming up + /// short. `consume` releases what it read; otherwise the bytes stay for the next read. fn poll_with( &mut self, cx: &mut Context<'_>, @@ -51,22 +69,56 @@ impl Reader { where V: Into
+ Copy, { + let form = self.version.into(); loop { - let mut r = Decoder::new(&self.buffer, self.version.into()); - match decode(&mut r) { - Ok(value) => { - if consume { - let used = self.buffer.len() - r.remaining(); - self.buffer.advance(used); + let want = self.want.max(1); + let version = &self.version; + let lease = ready!(self.stream.poll_lease(cx, want)).map_err(|err| version.transport_error(err))?; + + // Decode against the first chunk, or a copy of as much as the last attempt needed + // when that spans chunks. Only a message cut across chunks pays for the copy. + let mut size = lease.chunk().len().max(want.min(lease.len())); + loop { + let chunk = lease.chunk(); + let buf: &[u8] = match chunk.len() >= size { + true => chunk, + false => { + self.scratch.clear(); + self.scratch.resize(size, 0); + lease.buf().copy_to_slice(&mut self.scratch); + &self.scratch + } + }; + + let mut r = Decoder::new(buf, form); + let need = match decode(&mut r) { + Ok(value) => { + let used = buf.len() - r.remaining(); + self.want = 0; + if self.scratch.capacity() > SCRATCH_KEEP { + self.scratch = Vec::new(); + } + if consume { + lease.consume(used); + } + return Poll::Ready(Ok(value)); } - return Poll::Ready(Ok(value)); + // A short read inside a complete body is malformed, but reports what it + // needed from the body, so always ask for at least one byte more. + Err(DecodeError::Short) => r.need().max(buf.len() + 1), + Err(e) => return Poll::Ready(Err(e.into())), + }; + + if need <= lease.len() { + size = need; + continue; } // Stream closed while we still need more data. - Err(DecodeError::Short) if !ready!(self.poll_read_more(cx))? => { + if lease.is_fin() { return Poll::Ready(Err(DecodeError::Short.into())); } - Err(DecodeError::Short) => {} - Err(e) => return Poll::Ready(Err(e.into())), + self.want = need; + break; } } } @@ -190,19 +242,21 @@ impl Reader { std::future::poll_fn(|cx| self.poll_varint_peek(cx)).await } - /// Poll for the next chunk, draining the reader's internal buffer first. + /// Poll for the next chunk of up to `max` bytes, or `None` once the stream finished. + /// Its credit returns as it is read. pub fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Error>> { - if !self.buffer.is_empty() { - let n = cmp::min(self.buffer.len(), max); - return Poll::Ready(Ok(Some(self.buffer.split_to(n).freeze()))); - } + self.want = 0; + let version = &self.version; self.stream .poll_read_chunk(cx, max) - .map_err(|err| self.version.transport_error(err)) + .map_err(|err| version.transport_error(err)) } /// Fill a frame's payload from the stream, returning `Pending` once it would block. /// + /// The payload reaches the frame as it arrives, but keeps its receive credit until the + /// frame is full, so incomplete frames stay bounded by the receive window. + /// /// `Ready(Ok(()))` means the frame is full and the caller should commit it; an error /// means the caller should abort it. Either way the frame is untouched otherwise, so /// the caller keeps ownership of the commit/abort decision. @@ -220,64 +274,80 @@ impl Reader { cx: &mut Context<'_>, frame: &mut crate::frame::ProducerOwned, ) -> Poll> { + self.want = 0; let mut owed = 0; let result = loop { - if frame.remaining() == 0 { - break Ok(()); - } - - match self.poll_read_chunk(cx, frame.remaining()) { + let mut lease = match self.poll_lease(cx, frame.remaining().min(1)) { Poll::Pending => { if owed > 0 { frame.notify(); } return Poll::Pending; } - Poll::Ready(Ok(Some(chunk))) if !chunk.is_empty() => { - owed += chunk.len(); - if let Err(err) = frame.write(chunk) { - break Err(err); - } - if owed >= WAKE_BUDGET { - frame.notify(); - owed = 0; - } - } - // A FIN mid-payload: the declared size never arrived. - Poll::Ready(Ok(_)) => break Err(Error::WrongSize), + Poll::Ready(Ok(lease)) => lease, Poll::Ready(Err(err)) => break Err(err), + }; + + while frame.remaining() > 0 && !lease.is_empty() { + let chunk = lease.take(frame.remaining()); + owed += chunk.len(); + if let Err(err) = frame.write(chunk) { + return Poll::Ready(Err(err)); + } + if owed >= WAKE_BUDGET { + frame.notify(); + owed = 0; + } + } + + if frame.remaining() == 0 { + // The frame is complete, so its credit returns with the lease. + lease.consume(0); + break Ok(()); + } + // A FIN mid-payload: the declared size never arrived. + if lease.is_fin() { + break Err(Error::WrongSize); } }; Poll::Ready(result) } - /// Poll for exactly `size` bytes, accumulating partial reads in the buffer. + /// Poll for exactly `size` bytes. pub fn poll_read_exact(&mut self, cx: &mut Context<'_>, size: usize) -> Poll> { - while self.buffer.len() < size { - if !ready!(self.poll_read_more(cx))? { - return Poll::Ready(Err(DecodeError::Short.into())); - } + self.want = 0; + let mut lease = ready!(self.poll_lease(cx, size))?; + if lease.len() < size { + return Poll::Ready(Err(DecodeError::Short.into())); } - Poll::Ready(Ok(self.buffer.split_to(size).freeze())) + let bytes = match lease.chunk().len() >= size { + true => lease.take(size), + false => { + let bytes = lease.buf().copy_to_bytes(size); + lease.consume(size); + return Poll::Ready(Ok(bytes)); + } + }; + lease.consume(0); + Poll::Ready(Ok(bytes)) } - /// Poll to discard `remaining` bytes, counting it down as they arrive. Nothing is - /// buffered past what one read brings, so a peer cannot make a skipped payload cost - /// memory. + /// Poll to discard `remaining` bytes, counting it down as they arrive. Each byte is + /// released as it is skipped, so a peer cannot make a skipped payload cost credit. pub(crate) fn poll_skip(&mut self, cx: &mut Context<'_>, remaining: &mut usize) -> Poll> { - loop { - let skipped = self.buffer.len().min(*remaining); - self.buffer.advance(skipped); - *remaining -= skipped; - if *remaining == 0 { - return Poll::Ready(Ok(())); - } - if !ready!(self.poll_read_more(cx))? { + self.want = 0; + while *remaining > 0 { + let lease = ready!(self.poll_lease(cx, 1))?; + if lease.is_empty() { return Poll::Ready(Err(DecodeError::Short.into())); } + let skipped = lease.len().min(*remaining); + lease.consume(skipped); + *remaining -= skipped; } + Poll::Ready(Ok(())) } /// Read exactly the given number of bytes from the stream. @@ -294,23 +364,15 @@ impl Reader { Poll::Ready(Ok(())) } - /// Poll for whether data is available in the buffer or stream: `false` once it finishes. + /// Poll for whether data is available on the stream: `false` once it finishes. pub(crate) fn poll_has_more(&mut self, cx: &mut Context<'_>) -> Poll> { - if !self.buffer.is_empty() { - return Poll::Ready(Ok(true)); - } - - self.poll_read_more(cx) + let lease = ready!(self.poll_lease(cx, 1))?; + Poll::Ready(Ok(!lease.is_empty())) } - /// Poll for more data from the stream. `true` if data was read, `false` if the - /// stream finished. - fn poll_read_more(&mut self, cx: &mut Context<'_>) -> Poll> { - match ready!(self.stream.poll_read_buf(cx, &mut self.buffer)) { - Ok(Some(_)) => Poll::Ready(Ok(true)), - Ok(None) => Poll::Ready(Ok(false)), - Err(e) => Poll::Ready(Err(self.version.transport_error(e))), - } + /// Rank the stream for eviction under receive pressure. See [`RecvStream::set_priority`]. + pub fn set_priority(&mut self, priority: i64) { + self.stream.set_priority(priority); } /// Abort the stream with the given error. @@ -325,20 +387,22 @@ impl Reader { pub fn with_version(self, version: V2) -> Reader { Reader { stream: self.stream, - buffer: self.buffer, version, + want: self.want, + scratch: self.scratch, } } } #[cfg(test)] mod tests { - use std::sync::Arc; + use std::{collections::VecDeque, sync::Arc}; use futures::FutureExt; use super::*; use crate::StreamError; + use crate::transport::poll::{Buffered, Chunks}; /// Records the STOP_SENDING code so a test can assert which registry it came from. #[derive(Default)] @@ -346,24 +410,20 @@ mod tests { stops: Vec, } - impl crate::transport::poll::RecvStream for StopLog { + impl Chunks for StopLog { type Error = crate::lite::test_transport::SinkError; - fn poll_read( - &mut self, - _cx: &mut std::task::Context<'_>, - _dst: &mut [u8], - ) -> std::task::Poll, Self::Error>> { - std::task::Poll::Ready(Ok(None)) + fn poll_read_chunk(&mut self, _: &mut Context<'_>, _: usize) -> Poll, Self::Error>> { + Poll::Ready(Ok(None)) } fn stop(&mut self, code: u32) { self.stops.push(code); } + } - fn poll_closed(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll> { - std::task::Poll::Ready(Ok(())) - } + fn stop_log(version: V) -> Reader, V> { + Reader::new(Buffered::new(StopLog::default()), version) } /// STOP_SENDING refuses a stream, so it must carry a stream code. Reusing the session @@ -378,9 +438,9 @@ mod tests { (Error::Lagged, StreamError::TooFarBehind.to_code()), (Error::Unauthorized, StreamError::Unauthorized.to_code()), ] { - let mut reader = Reader::new(StopLog::default(), VERSION); + let mut reader = stop_log(VERSION); reader.abort(&err); - assert_eq!(reader.stream.stops, vec![expected], "{err:?} used the wrong registry"); + assert_eq!(reader.stream.get_ref().stops, vec![expected], "{err:?} used the wrong registry"); } // The two registries disagree about 0 and 1, which is what makes the mix-up silent. @@ -392,15 +452,15 @@ mod tests { /// the same abort has to leave a different number on each wire. #[test] fn abort_stops_with_the_negotiated_protocols_code() { - let mut lite = Reader::new(StopLog::default(), crate::lite::Version::Lite05); + let mut lite = stop_log(crate::lite::Version::Lite05); lite.abort(&Error::Old); - let mut ietf = Reader::new(StopLog::default(), crate::ietf::Version::Draft20); + let mut ietf = stop_log(crate::ietf::Version::Draft20); ietf.abort(&Error::Old); - assert_eq!(lite.stream.stops, vec![StreamError::Old.to_code()]); - assert_eq!(ietf.stream.stops, vec![crate::ietf::error::INTERNAL_ERROR]); - assert_ne!(lite.stream.stops, ietf.stream.stops); + assert_eq!(lite.stream.get_ref().stops, vec![StreamError::Old.to_code()]); + assert_eq!(ietf.stream.get_ref().stops, vec![crate::ietf::error::INTERNAL_ERROR]); + assert_ne!(lite.stream.get_ref().stops, ietf.stream.get_ref().stops); } /// Counts wakes delivered to a parked consumer. @@ -448,32 +508,53 @@ mod tests { } /// Hands back one queued chunk per poll, then blocks: a stream whose socket has - /// delivered a burst and has nothing more for this turn. + /// delivered a burst and has nothing more for this turn. Finishes once `fin` is set and + /// the queue is empty. #[derive(Default)] - struct Chunks(std::collections::VecDeque<&'static [u8]>); + struct Queue { + chunks: VecDeque<&'static [u8]>, + fin: bool, + } - impl crate::transport::poll::RecvStream for Chunks { + impl Chunks for Queue { type Error = crate::lite::test_transport::SinkError; - fn poll_read(&mut self, _cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { - let Some(chunk) = self.0.pop_front() else { - return Poll::Pending; + fn poll_read_chunk(&mut self, _: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { + let Some(chunk) = self.chunks.pop_front() else { + return match self.fin { + true => Poll::Ready(Ok(None)), + false => Poll::Pending, + }; }; - // The caller sizes `dst`, so a chunk larger than it spans two reads. - let n = chunk.len().min(dst.len()); - dst[..n].copy_from_slice(&chunk[..n]); + // The caller sizes the read, so a chunk larger than it spans two reads. + let n = chunk.len().min(max); if n < chunk.len() { - self.0.push_front(&chunk[n..]); + self.chunks.push_front(&chunk[n..]); } - Poll::Ready(Ok(Some(n))) + Poll::Ready(Ok(Some(Bytes::from_static(&chunk[..n])))) } fn stop(&mut self, _code: u32) {} + } - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Pending + fn queue(chunks: &[&'static [u8]], version: V) -> Reader, V> { + Reader::new( + Buffered::new(Queue { + chunks: chunks.iter().copied().collect(), + fin: false, + }), + version, + ) + } + + /// How many bytes the reader still has leased. + fn leased(reader: &mut Reader, V>) -> usize { + let mut cx = Context::from_waker(std::task::Waker::noop()); + match reader.stream.poll_lease(&mut cx, 0) { + Poll::Ready(Ok(lease)) => lease.len(), + _ => panic!("a lease of nothing never waits"), } } @@ -481,28 +562,45 @@ mod tests { /// waiting on more has counted down what it already discarded. #[test] fn skip_discards_across_reads() { - let chunks = [b"abc".as_slice(), b"defg", b"hi"]; - let mut reader = Reader::new(Chunks(chunks.into()), crate::lite::Version::Lite05); + let mut reader = queue(&[b"abc", b"defg", b"hi"], crate::lite::Version::Lite05); let mut cx = Context::from_waker(std::task::Waker::noop()); let mut remaining = 6; assert!(matches!(reader.poll_skip(&mut cx, &mut remaining), Poll::Ready(Ok(())))); assert_eq!(&reader.read_exact(3).now_or_never().unwrap().unwrap()[..], b"ghi"); - let mut reader = Reader::new(Chunks([b"ab".as_slice(), b"cd"].into()), crate::lite::Version::Lite05); + let mut reader = queue(&[b"ab", b"cd"], crate::lite::Version::Lite05); let mut remaining = 5; assert!(reader.poll_skip(&mut cx, &mut remaining).is_pending()); assert_eq!(remaining, 1); - assert!(reader.buffer.is_empty(), "nothing skipped is kept"); + assert_eq!(leased(&mut reader), 0, "nothing skipped is kept"); + } + + /// A message cut across chunks decodes from a copy, waiting for exactly what its + /// prefix declared, and consumes only itself. + #[test] + fn a_message_across_chunks_waits_for_its_body() { + // A size-prefixed goaway: size 3, then a 2 byte URI. + let mut reader = queue(&[b"\x03", b"\x02a"], crate::lite::Version::Lite05); + let mut cx = Context::from_waker(std::task::Waker::noop()); + assert!(reader.poll_decode::(&mut cx).is_pending()); + assert_eq!(reader.want, 4, "the prefix declared the whole message"); + + reader.stream.get_mut().chunks.push_back(b"bnext"); + assert!(matches!( + reader.poll_decode::(&mut cx), + Poll::Ready(Ok(_)) + )); + assert_eq!(leased(&mut reader), 4, "what follows the message stays leased"); + assert!(reader.scratch.is_empty() || reader.scratch.capacity() <= SCRATCH_KEEP); } fn refuses_frame + Debug, V: StreamCodes + Into + Copy>( - prefix: &[u8], + prefix: &'static [u8], version: V, oversized: bool, ) { - let mut reader = Reader::new(Chunks([b"following bytes".as_slice()].into()), version); - reader.buffer.extend_from_slice(prefix); + let mut reader = queue(&[prefix, b"following bytes"], version); let mut cx = Context::from_waker(std::task::Waker::noop()); let result = reader.poll_decode::(&mut cx); if oversized { @@ -519,8 +617,8 @@ mod tests { "{result:?}" ); } - assert_eq!(reader.buffer.len(), prefix.len()); - assert_eq!(reader.stream.0.len(), 1, "read past the complete frame"); + assert_eq!(reader.stream.get_ref().chunks.len(), 1, "read past the complete frame"); + assert_eq!(leased(&mut reader), prefix.len()); } #[test] @@ -534,17 +632,9 @@ mod tests { Version::Lite(lite::Version::Lite02), Version::Ietf(ietf::Version::Draft14), ] { - let prefix = if version.is_lite() { - &[0x20, 0][..] - } else { - &[0x20, 0, 0][..] - }; + let prefix: &'static [u8] = if version.is_lite() { &[0x20, 0] } else { &[0x20, 0, 0] }; refuses_frame::(prefix, version, false); - let prefix = if version.is_lite() { - &[0x21, 0][..] - } else { - &[0x21, 0, 0][..] - }; + let prefix: &'static [u8] = if version.is_lite() { &[0x21, 0] } else { &[0x21, 0, 0] }; refuses_frame::(prefix, version, false); } } @@ -556,15 +646,16 @@ mod tests { for size in [65536u64, (1 << 40) + 1] { let mut prefix = vec![0x20]; Encoder::new(&mut prefix, version.into()).varint(size).unwrap(); - refuses_frame::(&prefix, Version::Lite(version), true); + let client: &'static [u8] = prefix.clone().leak(); + refuses_frame::(client, Version::Lite(version), true); prefix[0] = 0x21; - refuses_frame::(&prefix, Version::Lite(version), true); + refuses_frame::(prefix.leak(), Version::Lite(version), true); } } for version in [lite::Version::Lite05, lite::Version::Lite06] { let mut prefix = Vec::new(); Encoder::new(&mut prefix, version.into()).varint(65536).unwrap(); - refuses_frame::(&prefix, version, true); + refuses_frame::(prefix.leak(), version, true); } } @@ -596,15 +687,28 @@ mod tests { #[test] fn fragmented_setup_waits_for_its_body() { - let mut reader = Reader::new(Chunks([b"\x01".as_slice()].into()), crate::lite::Version::Lite05); + let mut reader = queue(&[b"\x01"], crate::lite::Version::Lite05); let mut cx = Context::from_waker(std::task::Waker::noop()); assert!(reader.poll_decode::(&mut cx).is_pending()); - reader.stream.0.push_back(b"\x00"); + reader.stream.get_mut().chunks.push_back(b"\x00"); assert!(matches!( reader.poll_decode::(&mut cx), Poll::Ready(Ok(_)) )); - assert!(reader.buffer.is_empty()); + assert_eq!(leased(&mut reader), 0); + } + + /// A FIN before the declared payload arrived is a wrong size, not a complete frame. + #[test] + fn a_fin_mid_frame_is_a_wrong_size() { + let (mut frame, _payload, _waiter, _wakes) = parked_frame(9); + let mut reader = queue(&[b"foo"], crate::lite::Version::Lite05); + reader.stream.get_mut().fin = true; + let mut cx = Context::from_waker(std::task::Waker::noop()); + assert!(matches!( + reader.poll_read_frame(&mut cx, &mut frame), + Poll::Ready(Err(Error::WrongSize)) + )); } /// A consumer parked on the group cannot run until this task yields, so a burst of @@ -615,7 +719,7 @@ mod tests { let (mut frame, mut payload, waiter, wakes) = parked_frame(9); assert!(payload.poll_read_chunk(&waiter).is_pending()); - let mut reader = Reader::new(Chunks([b"foo".as_slice(), b"bar"].into()), crate::lite::Version::Lite05); + let mut reader = queue(&[b"foo", b"bar"], crate::lite::Version::Lite05); let mut cx = Context::from_waker(std::task::Waker::noop()); assert!(reader.poll_read_frame(&mut cx, &mut frame).is_pending()); @@ -633,7 +737,7 @@ mod tests { // The tail completes the frame, so the loop returns without a wake of its own. // The commit is what publishes it, and the commit is the caller's to make. - reader.stream.0.push_back(b"baz"); + reader.stream.get_mut().chunks.push_back(b"baz"); assert!(matches!( reader.poll_read_frame(&mut cx, &mut frame), Poll::Ready(Ok(())) @@ -648,26 +752,22 @@ mod tests { /// before handing over each chunk, so every wake the drain loop issues while the /// burst is in flight lands on a parked waiter and is counted. struct Burst { - chunks: Chunks, + chunks: Queue, payload: crate::frame::Consumer, waiter: kio::Waiter, } - impl crate::transport::poll::RecvStream for Burst { + impl Chunks for Burst { type Error = crate::lite::test_transport::SinkError; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { // Start each turn with none of the last turn's registrations left. self.waiter = kio::Waiter::new(self.waiter.waker().clone()); while self.payload.poll_read_chunk(&self.waiter).is_ready() {} - self.chunks.poll_read(cx, dst) + self.chunks.poll_read_chunk(cx, max) } fn stop(&mut self, _code: u32) {} - - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Pending - } } /// A sender at line rate can keep the drain loop `Ready` for as long as it likes, so @@ -683,11 +783,14 @@ mod tests { assert!(payload.poll_read_chunk(&waiter).is_pending()); let mut reader = Reader::new( - Burst { - chunks: Chunks(std::iter::repeat_n(CHUNK, chunks).collect()), + Buffered::new(Burst { + chunks: Queue { + chunks: std::iter::repeat_n(CHUNK, chunks).collect(), + fin: false, + }, payload, waiter, - }, + }), crate::lite::Version::Lite05, ); let mut cx = Context::from_waker(std::task::Waker::noop()); diff --git a/rs/moq-net/src/coding/stream.rs b/rs/moq-net/src/coding/stream.rs index 36aa621ca31..9a7592895d1 100644 --- a/rs/moq-net/src/coding/stream.rs +++ b/rs/moq-net/src/coding/stream.rs @@ -68,10 +68,9 @@ impl Stream { { let mut writer = Writer::new(send, version.clone()); writer.set_priority(CONTROL_SEND_ORDER.into()); - Self { - writer, - reader: Reader::new(recv, version), - } + let mut reader = Reader::new(recv, version); + reader.set_priority(crate::priority::RECV_CONTROL); + Self { writer, reader } } /// Cast the stream to a different version, used during version negotiation. diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index eb86c552f40..5fb922f32fe 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -495,9 +495,10 @@ pub enum Error { #[error("frame timestamp doesn't match track timescale")] TimestampMismatch, - /// The group was evicted by its own track to pay eviction debt under memory - /// pressure (see [`cache::Pool`](crate::cache::Pool)). Unlike [`Self::Old`], - /// the group was still within the publisher's window; it can be re-fetched. + /// The group was evicted under memory pressure: by its own track to pay eviction debt + /// (see [`cache::Pool`](crate::cache::Pool)), or by the transport when incomplete + /// messages spent its receive window. Unlike [`Self::Old`], the group was still within + /// the publisher's window; it can be re-fetched. #[error("evicted")] Evicted, @@ -587,6 +588,7 @@ impl Error { match err.close_kind() { crate::transport::CloseKind::IdleTimeout => Self::IdleTimeout, + crate::transport::CloseKind::Evicted => Self::Evicted, // A connection ending must not become the model's clean end-of-content. crate::transport::CloseKind::Closed => SessionError::Cancel.into(), _ => Self::Transport(err.to_string()), diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 8b9ff9f52f9..95cd031b260 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -1088,7 +1088,7 @@ pub fn seeds() -> Vec { /// every declared size when the first byte lands and `Some(0)` grows every buffer with /// the bytes received. pub struct FrameRecv { - streams: Vec, frame::ProducerOwned)>>, + streams: Vec, lite::Version>, frame::ProducerOwned)>>, groups: Vec, // Keeps the groups' track alive. _track: track::Producer, @@ -1127,7 +1127,7 @@ impl FrameRecv { chunk, ready: true, }; - Some((coding::Reader::new(stream, lite::Version::Lite05), frame)) + Some((coding::Reader::new(crate::transport::poll::Buffered::new(stream), lite::Version::Lite05), frame)) }) .collect(); @@ -1167,23 +1167,9 @@ struct Chunks { ready: bool, } -impl crate::transport::poll::RecvStream for Chunks { +impl crate::transport::poll::Chunks for Chunks { type Error = NoError; - fn poll_read( - &mut self, - cx: &mut std::task::Context<'_>, - dst: &mut [u8], - ) -> std::task::Poll, NoError>> { - let max = dst.len(); - self.poll_read_chunk(cx, max).map_ok(|chunk| { - chunk.map(|chunk| { - dst[..chunk.len()].copy_from_slice(&chunk); - chunk.len() - }) - }) - } - // Zero-copy, like the QUIC stacks, so only the frame buffer's own copy is timed. fn poll_read_chunk( &mut self, @@ -1202,10 +1188,6 @@ impl crate::transport::poll::RecvStream for Chunks { } fn stop(&mut self, _code: u32) {} - - fn poll_closed(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll> { - std::task::Poll::Pending - } } /// [`Chunks`] never fails. diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index 7c7bb3959bb..b6ffd94e76c 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -4,7 +4,7 @@ use std::{ task::{Context, Poll}, }; -use bytes::{Buf, BufMut, Bytes, BytesMut}; +use bytes::{Buf, Bytes, BytesMut}; use crate::{ Error, PathOwned, @@ -159,44 +159,9 @@ impl VirtualRecvStream { } } -impl crate::transport::poll::RecvStream for VirtualRecvStream { +impl crate::transport::poll::Chunks for VirtualRecvStream { type Error = crate::Error; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { - if dst.is_empty() { - return Poll::Ready(Ok(Some(0))); - } - - std::task::ready!(self.poll_fill(cx)); - if self.buffer.is_empty() { - return Poll::Ready(Ok(None)); - } - - let n = dst.len().min(self.buffer.len()); - dst[..n].copy_from_slice(&self.buffer[..n]); - self.buffer.advance(n); - Poll::Ready(Ok(Some(n))) - } - - fn poll_read_buf( - &mut self, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll, Self::Error>> { - if !buf.has_remaining_mut() { - return Poll::Ready(Ok(Some(0))); - } - - std::task::ready!(self.poll_fill(cx)); - if self.buffer.is_empty() { - return Poll::Ready(Ok(None)); - } - - let n = buf.remaining_mut().min(self.buffer.len()); - buf.put(self.buffer.split_to(n)); - Poll::Ready(Ok(Some(n))) - } - fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { if max == 0 { return Poll::Ready(Ok(Some(Bytes::new()))); @@ -214,25 +179,6 @@ impl crate::transport::poll::RecvStream for VirtualRecvStream { fn stop(&mut self, _code: u32) { self.rx.close(); } - - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { - if self.closed { - return Poll::Ready(Ok(())); - } - - // Drain (and discard) remaining messages until the queue closes. - let waiter = self.park.hold(cx); - loop { - match self.rx.poll_pop(waiter) { - Poll::Ready(Some(_)) => {} - Poll::Ready(None) => { - self.closed = true; - return Poll::Ready(Ok(())); - } - Poll::Pending => return Poll::Pending, - } - } - } } impl Drop for VirtualRecvStream { @@ -466,48 +412,87 @@ impl crate::transport::poll::SendStream for pub enum AdapterRecv { Real(S::RecvStream), - Virtual(VirtualRecvStream), + Virtual(crate::transport::poll::Buffered), +} + +/// A real stream's failure: an eviction keeps its meaning, anything else closed the stream. +fn real_error(err: impl crate::transport::Error) -> crate::Error { + match err.close_kind() { + crate::transport::CloseKind::Evicted => crate::Error::Evicted, + _ => crate::Error::Closed, + } } impl crate::transport::poll::RecvStream for AdapterRecv { type Error = crate::Error; + type Lease<'a> + = AdapterLease<'a, S> + where + S: 'a; - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Self::Error>> { match self { - Self::Real(s) => s.poll_read(cx, dst).map(|r| r.map_err(|_| crate::Error::Closed)), - Self::Virtual(s) => s.poll_read(cx, dst), + Self::Real(s) => s.poll_lease(cx, want).map(|r| r.map(AdapterLease::Real).map_err(real_error)), + Self::Virtual(s) => s.poll_lease(cx, want).map_ok(AdapterLease::Virtual), } } - fn poll_read_buf( - &mut self, - cx: &mut Context<'_>, - buf: &mut B, - ) -> Poll, Self::Error>> { + fn set_priority(&mut self, priority: i64) { match self { - Self::Real(s) => s.poll_read_buf(cx, buf).map(|r| r.map_err(|_| crate::Error::Closed)), - Self::Virtual(s) => s.poll_read_buf(cx, buf), + Self::Real(s) => s.set_priority(priority), + // Rides the control stream, which carries its credit. + Self::Virtual(_) => {} } } - fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { + fn stop(&mut self, code: u32) { match self { - Self::Real(s) => s.poll_read_chunk(cx, max).map(|r| r.map_err(|_| crate::Error::Closed)), - Self::Virtual(s) => s.poll_read_chunk(cx, max), + Self::Real(s) => s.stop(code), + Self::Virtual(s) => s.stop(code), } } +} - fn stop(&mut self, code: u32) { +/// A [`Lease`](crate::transport::poll::Lease) of an [`AdapterRecv`]. +pub enum AdapterLease<'a, S: crate::transport::poll::Session + 'a> { + Real(::Lease<'a>), + Virtual(crate::transport::poll::BufferedLease<'a>), +} + +impl crate::transport::poll::Lease for AdapterLease<'_, S> { + fn len(&self) -> usize { match self { - Self::Real(s) => s.stop(code), - Self::Virtual(s) => s.stop(code), + Self::Real(l) => l.len(), + Self::Virtual(l) => l.len(), } } - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + fn is_fin(&self) -> bool { match self { - Self::Real(s) => s.poll_closed(cx).map(|r| r.map_err(|_| crate::Error::Closed)), - Self::Virtual(s) => s.poll_closed(cx), + Self::Real(l) => l.is_fin(), + Self::Virtual(l) => l.is_fin(), + } + } + + fn chunks(&self) -> impl Iterator { + let (real, virt) = match self { + Self::Real(l) => (Some(l.chunks()), None), + Self::Virtual(l) => (None, Some(l.chunks())), + }; + real.into_iter().flatten().chain(virt.into_iter().flatten()) + } + + fn take(&mut self, max: usize) -> Bytes { + match self { + Self::Real(l) => l.take(max), + Self::Virtual(l) => l.take(max), + } + } + + fn consume(self, n: usize) { + match self { + Self::Real(l) => l.consume(n), + Self::Virtual(l) => l.consume(n), } } } @@ -1148,7 +1133,7 @@ impl crate::transport::poll::Session for Con let waiter = self.park.hold(cx); match std::task::ready!(self.shared.incoming.poll_pop(waiter)) { - Some((send, recv)) => Poll::Ready(Ok((AdapterSend::Virtual(send), AdapterRecv::Virtual(recv)))), + Some((send, recv)) => Poll::Ready(Ok((AdapterSend::Virtual(send), AdapterRecv::Virtual(crate::transport::poll::Buffered::new(recv))))), None => Poll::Ready(Err(crate::Error::Closed)), } } @@ -1158,7 +1143,7 @@ impl crate::transport::poll::Session for Con _cx: &mut Context<'_>, ) -> Poll> { let (send, recv) = self.shared.open_outgoing(self.version); - Poll::Ready(Ok((AdapterSend::Virtual(send), AdapterRecv::Virtual(recv)))) + Poll::Ready(Ok((AdapterSend::Virtual(send), AdapterRecv::Virtual(crate::transport::poll::Buffered::new(recv))))) } fn poll_open_uni(&mut self, cx: &mut Context<'_>) -> Poll> { @@ -1304,8 +1289,13 @@ mod tests { } /// A receive stream with no request behind it, for exercising the reader alone. - fn detached_recv(initial: Bytes, rx: Queue) -> VirtualRecvStream { - VirtualRecvStream::new(initial, rx, Arc::new(Shared::default()), Arc::new(Mutex::new(None))) + fn detached_recv(initial: Bytes, rx: Queue) -> crate::transport::poll::Buffered { + crate::transport::poll::Buffered::new(VirtualRecvStream::new( + initial, + rx, + Arc::new(Shared::default()), + Arc::new(Mutex::new(None)), + )) } #[test] @@ -1392,14 +1382,14 @@ mod tests { .unwrap(); let (_, mut recv) = shared.incoming.pop().await.unwrap(); assert_eq!( - recv.read_chunk(usize::MAX).await.unwrap(), + read_virtual(&mut recv).await.unwrap(), Some(Bytes::from_static(b"request")) ); assert_eq!(routed, target_id, "{version}: update {update_id} routed to {routed}"); shared.push(routed, raw.clone()); assert_eq!( - recv.read_chunk(usize::MAX).await.unwrap(), + read_virtual(&mut recv).await.unwrap(), Some(raw), "{version}: update {update_id} did not reach {target_id}" ); @@ -1462,18 +1452,17 @@ mod tests { let mut stream = detached_recv(initial, follow); // Read initial data - let mut buf = [0u8; 32]; - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b"initial"); + let chunk = stream.read_chunk(32).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b"initial"); // Send follow-up tx.push(Bytes::from_static(b"followup")); - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b"followup"); + let chunk = stream.read_chunk(32).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b"followup"); // Drop the writer → FIN drop(tx); - let result = stream.read(&mut buf).await.unwrap(); + let result = stream.read_chunk(32).await.unwrap(); assert_eq!(result, None); } @@ -1483,16 +1472,14 @@ mod tests { let mut stream = detached_recv(initial, Queue::new()); // Read small chunks - let mut buf = [0u8; 5]; - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b"hello"); + let chunk = stream.read_chunk(5).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b"hello"); - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b" worl"); + let chunk = stream.read_chunk(5).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b" worl"); - let mut buf = [0u8; 1]; - let n = stream.read(&mut buf).await.unwrap().unwrap(); - assert_eq!(&buf[..n], b"d"); + let chunk = stream.read_chunk(1).await.unwrap().unwrap(); + assert_eq!(&chunk[..], b"d"); } #[moq_clock::test] @@ -1589,7 +1576,7 @@ mod tests { // Drain the advertisement the stream opened with, so both halves are quiescent // and a later read only sees what the terminal message routes. assert_eq!( - recv.read_chunk(usize::MAX).await.unwrap(), + read_virtual(&mut recv).await.unwrap(), Some(encode_msg(&msg, version)) ); recv @@ -1623,9 +1610,14 @@ mod tests { } /// Read until the stream FINs, returning whether it did so within `limit` reads. + /// The next chunk of a virtual stream, read without a lease. + fn read_virtual(stream: &mut VirtualRecvStream) -> impl Future, crate::Error>> + '_ { + std::future::poll_fn(|cx| crate::transport::poll::Chunks::poll_read_chunk(stream, cx, usize::MAX)) + } + async fn drained(stream: &mut VirtualRecvStream, limit: usize) -> bool { for _ in 0..limit { - if stream.read_chunk(usize::MAX).await.unwrap().is_none() { + if read_virtual(stream).await.unwrap().is_none() { return true; } } @@ -1672,7 +1664,7 @@ mod tests { shared.close(RequestId(7), encode_msg(&done, version)); assert!(drained(&mut theirs, 4).await); assert!(shared.streams.lock().unwrap().contains_key(&RequestId(4))); - assert!(ours.read_chunk(usize::MAX).now_or_never().is_none()); + assert!(read_virtual(&mut ours).now_or_never().is_none()); } #[moq_clock::test] @@ -1700,7 +1692,7 @@ mod tests { shared.close(RequestId(4), encode_msg(&cancel, version)); assert!(drained(&mut ours, 4).await); assert!(shared.streams.lock().unwrap().contains_key(&RequestId(7))); - assert!(theirs.read_chunk(usize::MAX).now_or_never().is_none()); + assert!(read_virtual(&mut theirs).now_or_never().is_none()); } /// Classify the peer withdrawing its own advertisement of `namespace`. diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index c04e850beef..a66bf59cbf6 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -319,9 +319,11 @@ where Version::Draft16 => { let mut sub_ns_adapter = sub_ns_adapter; let (send, recv) = sub_ns_adapter.open_native_bi().await?; + let mut reader = crate::coding::Reader::new(recv, version); + reader.set_priority(crate::priority::RECV_CONTROL); Stream { writer: crate::coding::Writer::new(send, version), - reader: crate::coding::Reader::new(recv, version), + reader, } } _ => Stream::open(&mut sub_ns_adapter.clone(), version).await?, @@ -709,6 +711,7 @@ pub async fn accept_setup( continue; } + reader.set_priority(crate::priority::RECV_CONTROL); let setup: setup::Setup = reader.decode().await?; let (params, _) = ietf::Parameters::decode_slice(&setup.parameters, version)?; let path = match params.get_bytes(ietf::ParameterBytes::Path) { @@ -1067,6 +1070,7 @@ impl Uni { // The negotiation gates the announce and dispatch loops, so a SETUP we // cannot read must end the session rather than leave them parked on a // slot nothing will ever fill. + reader.set_priority(crate::priority::RECV_CONTROL); let msg = reader.decode::().await?; if let Some(peer_setup) = peer_setup { peer_setup.set(decode_peer_setup(msg.parameters, version, client)?); diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index d237a236743..efb03f64244 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3591,6 +3591,7 @@ where pub async fn recv_group(&mut self, stream: &mut Reader) -> Result<(), Error> { let group: ietf::GroupHeader = stream.decode().await?; + stream.set_priority(crate::priority::recv_group(group.group_id)); if group.sub_group_id != 0 { tracing::warn!(sub_group_id = %group.sub_group_id, "subgroup ID is not supported, dropping stream"); diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 756871b1ac5..0e934ab7f24 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -104,6 +104,7 @@ impl SetupAccept { return Poll::Ready(Err(Error::ProtocolViolation)); } ready!(reader.poll_decode::(&mut cx))?; + reader.set_priority(crate::priority::RECV_CONTROL); *setup = Some(reader); } // A closed transport can still yield a complete buffered SETUP. Decode it diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 5f59f20715d..967bba1c4a9 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -786,7 +786,11 @@ impl UniServe { }; self.state = match kind { lite::DataType::Group => UniState::Group(GroupRecv::new(self.subscriber.clone(), reader)), - lite::DataType::Setup => UniState::Setup { reader }, + lite::DataType::Setup => { + let mut reader = reader; + reader.set_priority(crate::priority::RECV_CONTROL); + UniState::Setup { reader } + } }; } UniState::Setup { reader } => { @@ -866,6 +870,7 @@ impl GroupRecv { GroupRecvState::Header => { let mut cx = waiter.context(); let hdr = ready!(self.reader.poll_decode::(&mut cx))?; + self.reader.set_priority(crate::priority::recv_group(hdr.sequence)); let (group, track, timescale, reading) = { let mut subs = self.subscriber.subscribes.lock(); diff --git a/rs/moq-net/src/lite/test_transport.rs b/rs/moq-net/src/lite/test_transport.rs index 967f72e621a..23aa799c8d4 100644 --- a/rs/moq-net/src/lite/test_transport.rs +++ b/rs/moq-net/src/lite/test_transport.rs @@ -315,16 +315,16 @@ pub struct PendingRecv; impl poll::RecvStream for PendingRecv { type Error = SinkError; + // Never handed out. + type Lease<'a> = poll::BufferedLease<'a>; - fn poll_read(&mut self, _cx: &mut Context<'_>, _dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_lease(&mut self, _cx: &mut Context<'_>, _want: usize) -> Poll, Self::Error>> { Poll::Pending } - fn stop(&mut self, _code: u32) {} + fn set_priority(&mut self, _priority: i64) {} - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Pending - } + fn stop(&mut self, _code: u32) {} } /// A RESET_STREAM as the transport reports it. `Some(code)` decodes as @@ -362,16 +362,16 @@ pub struct DeadRecv(ResetError); impl poll::RecvStream for DeadRecv { type Error = ResetError; + // Never handed out. + type Lease<'a> = poll::BufferedLease<'a>; - fn poll_read(&mut self, _cx: &mut Context<'_>, _dst: &mut [u8]) -> Poll, Self::Error>> { + fn poll_lease(&mut self, _cx: &mut Context<'_>, _want: usize) -> Poll, Self::Error>> { Poll::Ready(Err(self.0)) } - fn stop(&mut self, _code: u32) {} + fn set_priority(&mut self, _priority: i64) {} - fn poll_closed(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Err(self.0)) - } + fn stop(&mut self, _code: u32) {} } /// Incoming streams that die before their first byte: `accept_uni` / `accept_bi` @@ -747,7 +747,10 @@ impl crate::transport::Stats for SinkStats { /// /// Parks once the script is exhausted rather than reporting EOF: a read loop that saw /// EOF would exit, and a test usually wants to assert against the loop still running. -pub struct ScriptedRecv { +pub type ScriptedRecv = poll::Buffered