diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 0810b45e59..535934e319 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -197,7 +197,7 @@ Each subscription carries the knobs that decide behavior under congestion: | Knob | Effect | | --- | --- | -| **Priority** (0..255) | Higher-priority tracks get bandwidth first. Audio above video, base layer above enhancement. From a native Rust publisher over QUIC or WebTransport (iroh excepted), subscriptions at the same priority split stream bandwidth evenly by bytes, however many groups each has queued. Datagrams are not part of that split yet. | +| **Priority** (0..255) | Higher-priority tracks get bandwidth first. Audio above video, base layer above enhancement. From a native Rust publisher over QUIC or WebTransport (iroh excepted), subscriptions at the same priority split bandwidth evenly by bytes, streams and datagrams alike, however many groups each has queued. A browser publisher and other transports without send groups are not covered. | | **Order** | Which group to send first when several are pending. Newest first for live (the default), oldest first for catch-up. Needs moq-lite 07 or moq-transport; older moq-lite always sends newest first. | | **Max delay** | How long a non-latest group stays worth delivering before it is skipped. Zero means "live edge only", and raising it is also what asks for history. | @@ -255,6 +255,12 @@ datagram: unreliable, unordered, under about 1200 bytes, and never retransmitted. It suits real-time audio and sensor data. There is no stream fallback, so a datagram that doesn't fit isn't delivered that way. +A datagram shares its subscription's priority and group order with streams, +so a low-priority datagram waits behind higher-priority groups instead of +jumping the queue. From a native Rust publisher over QUIC or WebTransport +(iroh excepted), a full send buffer drops the lowest-priority subscription's +oldest datagram. + Nothing caches a datagram. A subscription gets the ones inside its group range, and a new one may get the few still in the publisher's short send buffer, but `FETCH` never returns one. A publisher that knows the sequence was diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 910e20648f..b9de0849d2 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -596,6 +596,7 @@ The subscriber's `Group Order` determines which Group within the Track is transm A publisher SHOULD attempt to transmit streams based on these rules. This depends on the QUIC implementation and it may not be possible to get fine-grained control. +A publisher SHOULD order a [datagram](#datagrams) among the subscription's streams as if it were its Group's stream, ahead of that stream when both exist. ### Priority The `Subscriber Priority` is scoped to the connection and MAY change over the life of the subscription via SUBSCRIBE_UPDATE. @@ -725,7 +726,7 @@ A publisher sends a datagram to a subscription only when its Group Sequence is i A datagram is never cached or retransmitted, and FETCH never returns one. A publisher MAY send a new subscription the datagrams still in a short send buffer, so a subscriber cannot assume a datagram was sent after it subscribed. A publisher that knows a FETCH names a group delivered only as a datagram resets the stream with NOT_FETCHABLE; otherwise it answers as for a group that does not exist, with NOT_FOUND. -A publisher SHOULD only send a datagram if the congestion controller can transmit it immediately. +A publisher SHOULD drop a datagram rather than block when the transport has no room for it. There is no separate subscription for datagram delivery: datagrams are routed to existing subscriptions via the Subscribe ID, and a subscriber receiving the same group via both a stream and a datagram MUST deduplicate by group sequence. Each datagram body has the following encoding (note: there is no message length prefix; the QUIC datagram boundary delimits the payload): @@ -1586,6 +1587,7 @@ The `Message Length` describes the payload size on the wire. - SUBSCRIBE_OK `Group` names where delivery starts when the subscription resolves, which need not be the first group sent. It is not a new floor. A subscriber does not wait for a group between the requested floor and it, but a publisher still delivers one that arrives later within Subscriber Max Age. - Corrected the SUBSCRIBE note that offset `Group Start` by 1. Only `Group End` and `Frame End` are offset so 0 can mean absent. `Group Start` is an absolute sequence, and 0 is group 0. - A subscription's range bounds datagrams like groups, and FETCH never returns a datagram. +- A publisher SHOULD order a datagram among the subscription's streams as if it were its Group's stream, ahead of that stream when both exist. It SHOULD drop a datagram rather than block when the transport has no room for it, where it previously sent one only if the congestion controller could transmit it immediately. - Assigned 0x3A NOT_FETCHABLE in the stream error table: a FETCH for a group delivered only as a datagram. - The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset. - A subscriber MAY cancel by FIN on its Subscribe Stream, and the publisher then FINs without SUBSCRIBE_END. diff --git a/rs/moq-ffi/src/transport/adapter.rs b/rs/moq-ffi/src/transport/adapter.rs index bb2e27bfdf..46a2c8ff25 100644 --- a/rs/moq-ffi/src/transport/adapter.rs +++ b/rs/moq-ffi/src/transport/adapter.rs @@ -80,8 +80,14 @@ impl + Clone + 'st fn send_group(&mut self, _: i64) -> Option { None } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { - backend::Session::poll_send_datagram(&mut self.0, cx, payload).map_err(Error) + /// The backend cannot rank a datagram against a stream, so the group and order are ignored, + /// and a datagram it has no room for is dropped. + fn send_datagram(&mut self, payload: Bytes, _: Option<&poll::Ungrouped>, _: i64) -> Result<(), Self::Error> { + let mut cx = Context::from_waker(std::task::Waker::noop()); + match backend::Session::poll_send_datagram(&mut self.0, &mut cx, &payload) { + Poll::Ready(res) => res.map_err(Error), + Poll::Pending => Ok(()), + } } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { backend::Session::poll_recv_datagram(&mut self.0, cx).map_err(Error) diff --git a/rs/moq-net/benches/schedule.rs b/rs/moq-net/benches/schedule.rs index 0a531ce542..e02c647e6c 100644 --- a/rs/moq-net/benches/schedule.rs +++ b/rs/moq-net/benches/schedule.rs @@ -14,6 +14,8 @@ //! //! `fanout` sweeps subscriptions against groups in flight per subscription, timing one frame, //! so a scheduling cost that grows with the table rather than the touched path is a slope. +//! `datagram` sweeps send groups, domains, and queued datagrams per group against one +//! datagram sent and one evicted, for the same reason. use std::collections::HashMap; use std::sync::{Arc, Mutex}; @@ -38,7 +40,7 @@ mod track { mod schedule; use priority::{Priority, PriorityHandle, PriorityQueue}; -use schedule::{Schedule, Slot}; +use schedule::{Next, Schedule, Slot}; /// The most a frame carries, one packet. const PACKET: usize = 1200; @@ -188,7 +190,7 @@ struct Sim { kind: Kind, tracks: Vec, capacity: usize, - schedule: Schedule, + schedule: Schedule, groups: Vec, queue: PriorityQueue, moved: Arc>>, @@ -324,7 +326,7 @@ impl Sim { let mut budget = self.capacity; while budget > 0 { - let Some(id) = self.schedule.pop() else { + let Some(Next::Stream(id)) = self.schedule.pop() else { break; }; let stream = self.streams.get_mut(&id).unwrap(); @@ -420,7 +422,7 @@ fn bench_workloads(c: &mut Criterion) { } /// A saturated schedule: `subscriptions` groups at one priority, each with `streams` queued. -fn saturated(subscriptions: usize, streams: usize) -> (Schedule, Vec) { +fn saturated(subscriptions: usize, streams: usize) -> (Schedule, Vec) { let mut schedule = Schedule::new(); let mut slots = Vec::with_capacity(subscriptions * streams); for _ in 0..subscriptions { @@ -449,7 +451,9 @@ fn bench_fanout(c: &mut Criterion) { &(), |b, ()| { b.iter(|| { - let id = schedule.pop().unwrap(); + let Some(Next::Stream(id)) = schedule.pop() else { + unreachable!("only streams are queued"); + }; let slot = &mut slots[id as usize]; schedule.sent(id, slot, PACKET, Some((id % streams as u64) as i64), true); }); @@ -485,5 +489,75 @@ fn bench_fanout(c: &mut Criterion) { group.finish(); } -criterion_group!(benches, bench_workloads, bench_fanout); +/// A full datagram budget: `groups` send groups at one priority, each in its own domain when +/// `domains`, with `datagrams` queued in each, one packet apiece. +fn datagrams(groups: usize, datagrams: usize, domains: bool) -> (Schedule, Vec) { + let mut schedule = Schedule::new(); + let ids: Vec<_> = (0..groups) + .map(|_| { + let domain = domains.then(|| schedule.open_domain()); + schedule.open_group(0, domain) + }) + .collect(); + let budget = groups * datagrams * PACKET; + for seq in 0..datagrams { + for &group in &ids { + assert!(schedule.push_datagram(Some(group), seq as i64, (), PACKET, budget)); + } + } + (schedule, ids) +} + +fn bench_datagram(c: &mut Criterion) { + for (name, domains) in [("groups", false), ("domains", true)] { + // A datagram into a full budget, which evicts the oldest of the busiest domain's + // lowest-priority group. + let mut group = c.benchmark_group(format!("schedule_datagram_evict_{name}")); + for groups in [1, 16, 256, 4_096] { + for queued in [1, 16, 256] { + if groups * queued > 100_000 { + continue; + } + let (mut schedule, ids) = datagrams(groups, queued, domains); + let budget = schedule.datagram_bytes(); + let mut seq = queued as i64; + let mut next = 0; + group.bench_with_input(BenchmarkId::new(format!("{name}_{groups}"), queued), &(), |b, ()| { + b.iter(|| { + // Round robin, so every group keeps its share of the budget. + schedule.push_datagram(Some(ids[next]), seq, (), PACKET, budget); + next = (next + 1) % ids.len(); + seq += 1; + }); + }); + } + } + group.finish(); + } + + // A datagram sent and another queued in its place, through every round robin. + let mut group = c.benchmark_group("schedule_datagram_send"); + for groups in [1, 16, 256, 4_096] { + for queued in [1, 16, 256] { + if groups * queued > 100_000 { + continue; + } + let (mut schedule, _) = datagrams(groups, queued, false); + let mut seq = queued as i64; + group.bench_with_input(BenchmarkId::new(format!("groups_{groups}"), queued), &(), |b, ()| { + b.iter(|| { + let Some(Next::Datagram(at)) = schedule.pop() else { + unreachable!("only datagrams are queued"); + }; + schedule.sent_datagram(at); + schedule.push_datagram(at, seq, (), PACKET, usize::MAX); + seq += 1; + }); + }); + } + } + group.finish(); +} + +criterion_group!(benches, bench_workloads, bench_fanout, bench_datagram); criterion_main!(benches); diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 1a9d3a8b9e..edf5d0d8c9 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -624,8 +624,13 @@ mod tests { None } - fn poll_send_datagram(&mut self, _cx: &mut Context<'_>, _payload: &[u8]) -> Poll> { - Poll::Ready(Ok(())) + fn send_datagram( + &mut self, + _: Bytes, + _: Option<&crate::transport::poll::Ungrouped>, + _: i64, + ) -> Result<(), Self::Error> { + Ok(()) } fn poll_recv_datagram(&mut self, _cx: &mut Context<'_>) -> Poll> { @@ -1019,8 +1024,13 @@ mod tests { None } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { - self.inner.poll_send_datagram(cx, payload) + fn send_datagram( + &mut self, + payload: Bytes, + group: Option<&crate::transport::poll::Ungrouped>, + order: i64, + ) -> Result<(), Self::Error> { + self.inner.send_datagram(payload, group, order) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index dc7a2d80aa..286866a86e 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -1176,10 +1176,15 @@ impl crate::transport::poll::Session for Con self.inner.send_group(priority) } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { + fn send_datagram( + &mut self, + payload: Bytes, + group: Option<&::Group>, + order: i64, + ) -> Result<(), Self::Error> { self.inner - .poll_send_datagram(cx, payload) - .map(|r| r.map_err(|_| crate::Error::Closed)) + .send_datagram(payload, group, order) + .map_err(|_| crate::Error::Closed) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index d7891df15b..1a9dc30ef6 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -3937,7 +3937,10 @@ impl TrackServe { ); continue; } - let _ = self.session.send_datagram(&body); + // Ranked like a stream carrying the group from its start, so it merges into the + // subscription's group order and wins a tie with that group's stream. + let order = crate::priority::in_group(sequence, 0, self.ranking.order); + let _ = self.session.send_datagram(body, self.ranking.group.as_ref(), order); } Ok(()) } @@ -5123,6 +5126,49 @@ mod group_priority_test { assert_eq!(log.orders().len(), 2, "the streams keep their place in the group"); } + /// An OBJECT_DATAGRAM goes in its subscription's send group, ranked like the stream + /// carrying its group from the start, so the transport merges it into the group order. + #[test] + fn datagrams_rank_within_their_subscription() { + let log = crate::lite::test_transport::Log::default(); + let session = SinkSession::new(log.clone()) + .with_groups() + .with_datagrams(1200) + .with_unacked_fin(); + let mut track = track::Producer::new( + std::sync::Arc::new(crate::broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let subscriber = track.subscribe( + track::Subscription::default() + .with_priority(3) + .with_max_delay(std::time::Duration::from_secs(3600)), + ); + let mut serve = TrackServe::new( + session, + subscriber, + RequestId(0), + Version::Draft14, + ServeRange::default(), + None, + ); + let mut group = track.create_group(group::Info { sequence: 0 }).unwrap(); + group + .write_frame(crate::Timestamp::from_millis(0).unwrap(), b"x".as_slice()) + .unwrap(); + group.finish().unwrap(); + let sequence = track + .append_datagram(crate::Timestamp::from_millis(1).unwrap(), b"d".as_slice()) + .unwrap(); + + let waiter = kio::Waiter::noop(); + assert!(serve.poll(&waiter).is_pending()); + let order = crate::priority::in_group(sequence, 0, Order::Descending); + assert_eq!(log.datagram_ranks(), [(Some(0), order)]); + assert_eq!(log.orders(), [(0, 1)], "the stream for group 0 ranks below"); + } + /// A group with more objects ready than one poll's budget drains over several polls. #[test] fn a_long_group_drains_within_the_budget() { diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 1ab54f6a7d..c9665cb4c4 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3189,6 +3189,8 @@ struct Opens { impl Subscription { /// Send one datagram best-effort over a QUIC datagram (lite-05 §6.4). /// + /// It goes in the subscription's send group, ranked like a stream carrying its group, so + /// it merges into the subscription's group order and wins a tie with that group's stream. /// The datagram is dropped (there is no group fallback) if the encoded body doesn't fit the /// transport's datagram limit or the send fails (congestion / no capacity right now). /// Returns whether it was handed to the transport. @@ -3223,7 +3225,9 @@ impl Subscription { return false; } - let _ = self.session.send_datagram(&body); + // The run loop may not have served a group since the last SUBSCRIBE_UPDATE. + let order = crate::priority::in_group(datagram.sequence, 0, self.steer_current().order); + let _ = self.session.send_datagram(body, self.group.as_ref(), order); true } @@ -4592,6 +4596,84 @@ mod serve_group_test { assert_eq!(flipped, [(0, 1), (1, -1)], "older groups first"); } + /// A datagram goes in its subscription's send group, ranked like the stream carrying + /// its group from the start, so the transport merges it into the group order. + #[moq_net_sim::test] + async fn datagrams_rank_within_their_subscription() { + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let subscriber = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + let mut session = SinkSession::new(Log::default()) + .with_groups() + .with_datagrams(1200) + .with_unacked_fin(); + let log = session.log.clone(); + let (mut run, mut writer, _) = lite_run(Version::Lite06, session.clone(), subscriber); + run.ctx.group = crate::transport::poll::Session::send_group(&mut session, 3); + + write_group(&mut track, 0, 0); + let sequence = track + .append_datagram(Timestamp::from_millis(1).unwrap(), b"d".as_slice()) + .unwrap(); + { + let mut poll = std::pin::pin!(kio::wait(|waiter| run.poll(&mut writer, waiter))); + assert!(futures::poll!(poll.as_mut()).is_pending()); + } + let order = crate::priority::in_group(sequence, 0, Order::Descending); + assert_eq!(log.datagram_ranks(), [(Some(0), order)]); + assert_eq!(log.orders(), [(0, 1)], "the stream for group 0 ranks below"); + assert!(order > 1); + } + + /// A SUBSCRIBE_UPDATE that flips the group order reranks the next datagram, even with no + /// group served in between. + #[moq_net_sim::test] + async fn datagrams_follow_an_updated_group_order() { + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let subscriber = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + let mut session = SinkSession::new(Log::default()).with_groups().with_datagrams(1200); + let log = session.log.clone(); + let (mut run, mut writer, _) = lite_run(Version::Lite06, session.clone(), subscriber); + run.ctx.group = crate::transport::poll::Session::send_group(&mut session, 3); + + let first = track + .append_datagram(Timestamp::from_millis(1).unwrap(), b"a".as_slice()) + .unwrap(); + { + let mut poll = std::pin::pin!(kio::wait(|waiter| run.poll(&mut writer, waiter))); + assert!(futures::poll!(poll.as_mut()).is_pending()); + } + + run.update(lite::SubscribeUpdate { + order: Order::Ascending, + priority: 3, + max_delay: Duration::from_secs(3600), + start_group: None, + end_group: None, + start_frame: 0, + end_frame: None, + }); + let second = track + .append_datagram(Timestamp::from_millis(2).unwrap(), b"b".as_slice()) + .unwrap(); + { + let mut poll = std::pin::pin!(kio::wait(|waiter| run.poll(&mut writer, waiter))); + assert!(futures::poll!(poll.as_mut()).is_pending()); + } + + let descending = crate::priority::in_group(first, 0, Order::Descending); + let ascending = crate::priority::in_group(second, 0, Order::Ascending); + assert_ne!(ascending, crate::priority::in_group(second, 0, Order::Descending)); + assert_eq!(log.datagram_ranks(), [(Some(0), descending), (Some(0), ascending)]); + } + /// A group with more frames ready than one poll's budget drains over several polls, /// so a sibling (the transport's driver, say) runs in between. #[test] diff --git a/rs/moq-net/src/lite/test_transport.rs b/rs/moq-net/src/lite/test_transport.rs index 5c32cef64c..4772814717 100644 --- a/rs/moq-net/src/lite/test_transport.rs +++ b/rs/moq-net/src/lite/test_transport.rs @@ -35,11 +35,16 @@ impl crate::transport::Error for SinkError { } } +/// A datagram's send group id and in-group order. +pub type Rank = (Option, i64); + /// What the session's send streams recorded. Resets are a list, not a last-value, so a /// stray second reset (a drop fallback firing behind an abort) is visible. #[derive(Clone, Default)] pub struct Log { pub datagrams: Arc>>, + /// The send group and order each datagram was sent with, beside [`Self::datagrams`]. + datagram_ranks: Arc>>, pub writes: Arc>>, pub resets: Arc>>, /// Every write, FIN and reset on every send stream, in call order. See [`Self::trail`]. @@ -116,6 +121,19 @@ impl Log { self.orders.lock().unwrap().clone() } + /// The send group and order of every datagram, as (group, order) in call order. + pub fn datagram_ranks(&self) -> Vec { + self.datagram_ranks.lock().unwrap().clone() + } + + fn record_datagram(&self, payload: bytes::Bytes, group: Option<&SinkGroup>, order: i64) { + self.datagrams.lock().unwrap().push(payload); + self.datagram_ranks + .lock() + .unwrap() + .push((group.map(|group| group.id), order)); + } + /// The session-level closes, as (code, reason). A list rather than a last-value so /// a second close racing the first is visible. pub fn closes(&self) -> Vec<(u32, String)> { @@ -412,8 +430,14 @@ impl poll::Session for DeadStreamSession { None } - fn poll_send_datagram(&mut self, _cx: &mut Context<'_>, _payload: &[u8]) -> Poll> { - Poll::Ready(Ok(())) + fn send_datagram( + &mut self, + payload: bytes::Bytes, + group: Option<&SinkGroup>, + order: i64, + ) -> Result<(), Self::Error> { + self.log.record_datagram(payload, group, order); + Ok(()) } fn poll_recv_datagram(&mut self, _cx: &mut Context<'_>) -> Poll> { @@ -464,6 +488,8 @@ pub struct SinkSession { unacked_fin: bool, /// Set by [`Self::with_groups`]. groups: bool, + /// Set by [`Self::with_datagrams`]; zero disables datagrams. + datagram_size: usize, /// The ALPN to report, for a test that needs a specific negotiated version rather /// than the SETUP-negotiated fallback an absent one selects. protocol: Option<&'static str>, @@ -494,6 +520,12 @@ impl SinkSession { self } + /// Accept datagrams up to `max_size` bytes, recording them in the log. + pub fn with_datagrams(mut self, max_size: usize) -> Self { + self.datagram_size = max_size; + self + } + /// Report these connection statistics, as a real transport would. pub fn with_stats(self, stats: SinkStats) -> Self { self.set_stats(stats); @@ -612,8 +644,14 @@ impl poll::Session for SinkSession { self.groups.then(|| SinkGroup::open(self.log.clone(), priority)) } - fn poll_send_datagram(&mut self, _cx: &mut Context<'_>, _payload: &[u8]) -> Poll> { - Poll::Ready(Ok(())) + fn send_datagram( + &mut self, + payload: bytes::Bytes, + group: Option<&SinkGroup>, + order: i64, + ) -> Result<(), Self::Error> { + self.log.record_datagram(payload, group, order); + Ok(()) } fn poll_recv_datagram(&mut self, _cx: &mut Context<'_>) -> Poll> { @@ -621,7 +659,7 @@ impl poll::Session for SinkSession { } fn max_datagram_size(&self) -> usize { - 0 + self.datagram_size } fn binding(&self) -> crate::transport::Binding { @@ -937,13 +975,14 @@ impl poll::Session for ScriptedSession { None } - fn poll_send_datagram(&mut self, _cx: &mut Context<'_>, payload: &[u8]) -> Poll> { - self.log - .datagrams - .lock() - .unwrap() - .push(bytes::Bytes::copy_from_slice(payload)); - Poll::Ready(Ok(())) + fn send_datagram( + &mut self, + payload: bytes::Bytes, + group: Option<&SinkGroup>, + order: i64, + ) -> Result<(), Self::Error> { + self.log.record_datagram(payload, group, order); + Ok(()) } fn poll_recv_datagram(&mut self, _cx: &mut Context<'_>) -> Poll> { diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 4e3b2cdbfb..3ce727a171 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -1069,12 +1069,13 @@ mod tests { None } - fn poll_send_datagram( + fn send_datagram( &mut self, - _cx: &mut std::task::Context<'_>, - _payload: &[u8], - ) -> std::task::Poll> { - std::task::Poll::Ready(Ok(())) + _: Bytes, + _: Option<&crate::transport::poll::Ungrouped>, + _: i64, + ) -> Result<(), Self::Error> { + Ok(()) } fn poll_recv_datagram( &mut self, diff --git a/rs/moq-net/src/transport.rs b/rs/moq-net/src/transport.rs index cdd084eee1..2a88a3b9e6 100644 --- a/rs/moq-net/src/transport.rs +++ b/rs/moq-net/src/transport.rs @@ -164,24 +164,27 @@ pub mod poll { /// handle is dropped and its streams finish. fn send_group(&mut self, priority: i64) -> Option<::Group>; - /// Poll to send a datagram over the network. + /// Send a datagram in `group` at `order`, best-effort. /// - /// Returns [`Poll::Pending`] while the transport has no room for it, so a caller - /// can wait for capacity rather than having the payload dropped underneath it. + /// `order` ranks the datagram among the group's streams like a stream's own + /// [priority](SendStream::set_priority), and a datagram goes ahead of a stream of equal + /// order. Without a group it is ranked among the ungrouped streams. A transport that + /// cannot rank datagrams against streams ignores both. /// - /// `payload` is taken by reference, not by value or as a [`Buf`]: a - /// [`Poll::Pending`] return means the caller retries with the same datagram, and - /// both of those would have consumed it. (A datagram also needs *contiguous* - /// bytes, and the only way to get those from a generic [`Buf`] is - /// [`Buf::copy_to_bytes`], which consumes.) + /// Never blocks: a transport with no room drops a queued datagram, the lowest priority + /// first where it can tell, exactly as the network is allowed to. /// /// Accepting a datagram is not delivery. QUIC datagrams may still be dropped: /// - Network congestion. /// - Random packet loss. /// - Payload is larger than `max_datagram_size()` /// - Peer is not receiving datagrams. - /// - ??? - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll>; + fn send_datagram( + &mut self, + payload: Bytes, + group: Option<&::Group>, + order: i64, + ) -> Result<(), Self::Error>; /// Poll for a datagram from the network. fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll>; @@ -242,16 +245,6 @@ pub mod poll { RecvDatagram(self) } - /// Send a datagram, best-effort: if the transport has no room for it right - /// now, the datagram is dropped, exactly as the network is allowed to do. - fn send_datagram(&mut self, payload: &[u8]) -> Result<(), Self::Error> { - let mut cx = Context::from_waker(std::task::Waker::noop()); - match self.poll_send_datagram(&mut cx, payload) { - Poll::Ready(res) => res, - Poll::Pending => Ok(()), - } - } - /// Wait until the session is closed by either side, returning the reason. fn closed(&mut self) -> SessionClosed<'_, Self> { SessionClosed(self) diff --git a/rs/moq-net/src/transport/memory.rs b/rs/moq-net/src/transport/memory.rs index c9229b31d8..e1713331c4 100644 --- a/rs/moq-net/src/transport/memory.rs +++ b/rs/moq-net/src/transport/memory.rs @@ -13,8 +13,8 @@ //! with a code and drops what it queued. Closing the session keeps queued messages //! readable, and leaves streams as they were, so the peer reads them, then the close. //! -//! Datagrams are copied, since [`poll::Session::poll_send_datagram`] borrows them, and -//! dropped when oversized or when the peer's queue is full, as a network would. +//! Datagrams ignore their send group and order, and are dropped when oversized or when the +//! peer's queue is full, as a network would. use std::{ collections::VecDeque, @@ -298,15 +298,15 @@ impl poll::Session for Session { None } - fn poll_send_datagram(&mut self, _cx: &mut Context<'_>, payload: &[u8]) -> Poll> { + fn send_datagram(&mut self, payload: Bytes, _: Option<&poll::Ungrouped>, _: i64) -> Result<(), Error> { if let Some(err) = closed(&self.conn().close) { - return Poll::Ready(Err(err)); + return Err(err); } // An oversized datagram, or one the peer has no room for, is lost as on a network. if payload.len() <= MAX_DATAGRAM_SIZE { - let _ = self.side.outgoing().datagrams.try_push(Bytes::copy_from_slice(payload)); + let _ = self.side.outgoing().datagrams.try_push(payload); } - Poll::Ready(Ok(())) + Ok(()) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { @@ -877,12 +877,12 @@ mod tests { #[test] fn datagrams() { let (mut a, mut b) = pair(Config::default()); - ready(a.poll_send_datagram(&mut noop(), &[0; MAX_DATAGRAM_SIZE + 1])).unwrap(); + a.send_datagram(vec![0; MAX_DATAGRAM_SIZE + 1].into(), None, 0).unwrap(); for _ in 0..DATAGRAM_QUEUE { - ready(a.poll_send_datagram(&mut noop(), b"one")).unwrap(); + a.send_datagram("one".into(), None, 0).unwrap(); } // The peer's queue is full, so this one is lost, like the oversized one. - ready(a.poll_send_datagram(&mut noop(), b"two")).unwrap(); + a.send_datagram("two".into(), None, 0).unwrap(); for _ in 0..DATAGRAM_QUEUE { assert_eq!(ready(b.poll_recv_datagram(&mut noop())).unwrap(), "one"); } diff --git a/rs/moq-net/tests/integration/support/mock.rs b/rs/moq-net/tests/integration/support/mock.rs index d98423ff31..d5ca709ac4 100644 --- a/rs/moq-net/tests/integration/support/mock.rs +++ b/rs/moq-net/tests/integration/support/mock.rs @@ -600,11 +600,16 @@ impl poll::Session for MockSession { None } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { + fn send_datagram( + &mut self, + payload: Bytes, + group: Option<&poll::Ungrouped>, + order: i64, + ) -> Result<(), Self::Error> { if self.side.lossy.load(Ordering::Relaxed) { - return Poll::Ready(Ok(())); + return Ok(()); } - self.inner.poll_send_datagram(cx, payload) + self.inner.send_datagram(payload, group, order) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { diff --git a/rs/moq-quic/README.md b/rs/moq-quic/README.md index 0b17ed7206..8d12d1e703 100644 --- a/rs/moq-quic/README.md +++ b/rs/moq-quic/README.md @@ -67,6 +67,7 @@ Changes on top of the upstream commit, besides the renames: quinn's `handshake_timeout` test in its async layer expects the handshake to end at the idle timeout, so the driver's copy sets both timeouts. - **Instants on [moq-clock](../moq-clock)**: the crate's `Instant` is `moq_clock::Instant`, so one clock steps every timer, tokio's paused clock and the simulator included. qlog stamps events relative to `QlogConfig::start_time` on that clock rather than the streamer's own `std::time::Instant`. - **Send groups** (`streams/schedule.rs`, `Connection::send_groups`): streams outside every group go first, in quinn's order; groups in one scheduling domain are served strictly by priority, with equal priorities sharing bytes by deficit round robin; domains share bytes the same way. A stream's priority is an `i64` ordering it only within its group. This replaces quinn's single pending-stream heap, so a cherry-pick touching `PendingStreamsQueue` has to be redone against the schedule. moq-tokio's `quic::driver` and moq-uring's driver wrap a group in a shared `SendGroup` handle that closes on drop. +- **Scheduled datagrams** (`Datagrams::send`): a datagram is queued in a send group (or ungrouped) at an `i64` order and scheduled with the group's streams, winning a tie with a stream of equal order and spending the group's byte credit, instead of quinn's datagram-first FIFO in `populate_packet`. Sending never blocks: `datagram_send_buffer_size` is shared, and an overflow evicts the oldest datagram of the lowest-priority group in the domain holding the most datagram bytes. `SendDatagramError::Blocked` and `Event::DatagramsUnblocked` are gone, and a cherry-pick touching the `// DATAGRAM` loop has to be redone against `StreamsState::write_frames`. - **`RESET_STREAM_AT`** ([draft-ietf-quic-reliable-stream-reset](https://datatracker.ietf.org/doc/draft-ietf-quic-reliable-stream-reset/)): the `reset_stream_at` transport parameter is always advertised, `Checkpoint::reset` (from `SendStream::checkpoint`) keeps the data written before the checkpoint reliable, holding a plain `RESET_STREAM` until that data is acknowledged when the peer lacks the extension, and a received reset surfaces only once the reliable prefix has been read. Stopping a stream after it was reset no longer credits its unread data twice. - **Fuzzing** runs through the hidden `fuzz` module behind a `fuzz` feature, with one-line shims in [`fuzz/`](fuzz/README.md): upstream's `packet`, `params`, `streamid`, and `streams` targets, plus `frame` and a connection op sequence. Upstream's `cfg(fuzzing)` `fuzzing` module and `arbitrary` feature are gone, so its fuzz patches port by hand. - **Draining endpoint event**: `EndpointEvent::is_draining` fires once per connection when it stops being active, before the drained event, so the driver's `Endpoint::wait_all_draining` can end a shutdown without sitting out every closing period. diff --git a/rs/moq-quic/benches/wire.rs b/rs/moq-quic/benches/wire.rs index cf167d371b..a651516fe9 100644 --- a/rs/moq-quic/benches/wire.rs +++ b/rs/moq-quic/benches/wire.rs @@ -50,7 +50,7 @@ fn queue(pair: &mut Pair, packets: usize, frames: usize) -> usize { let mut datagrams = conn.datagrams(); let empty = datagrams.send_buffer_space(); for _ in 0..packets * frames { - datagrams.send(data.clone(), false).unwrap(); + datagrams.send(data.clone(), None, 0).unwrap(); } empty } diff --git a/rs/moq-quic/src/config/transport.rs b/rs/moq-quic/src/config/transport.rs index 2876ca7c48..554080ef16 100644 --- a/rs/moq-quic/src/config/transport.rs +++ b/rs/moq-quic/src/config/transport.rs @@ -326,9 +326,12 @@ impl TransportConfig { /// While datagrams are sent ASAP, it is possible for an application to generate data faster /// than the link, or even the underlying hardware, can transmit them. This limits the amount of /// memory that may be consumed in that case. When the send buffer is full and a new datagram is - /// sent, older datagrams are dropped until sufficient space is available. + /// sent, the domain holding the most datagram bytes drops the oldest datagram of its + /// lowest-priority send group until the new one fits; see + /// [`Datagrams::send`](crate::Datagrams::send). /// - /// The amount of payload data buffered may be smaller than `value` due to overhead. + /// Counts each datagram's frame bytes, so the payload buffered is a little smaller than + /// `value`. pub fn datagram_send_buffer_size(&mut self, value: usize) -> &mut Self { self.datagram_send_buffer_size = value; self diff --git a/rs/moq-quic/src/connection/datagrams.rs b/rs/moq-quic/src/connection/datagrams.rs index d116d8c81a..6cd3222f72 100644 --- a/rs/moq-quic/src/connection/datagrams.rs +++ b/rs/moq-quic/src/connection/datagrams.rs @@ -4,7 +4,7 @@ use bytes::Bytes; use thiserror::Error; use tracing::{debug, trace}; -use super::{Connection, Event}; +use super::{Connection, SendGroup}; use crate::{ TransportError, frame::{Datagram, FrameStruct}, @@ -16,55 +16,57 @@ pub struct Datagrams<'a> { } impl Datagrams<'_> { - /// Queue an unreliable, unordered datagram for immediate transmission + /// Queue an unreliable, unordered datagram in `group` at `order` /// - /// If `drop` is true, previously queued datagrams which are still unsent may be discarded to - /// make space for this datagram, in order of oldest to newest. If `drop` is false, and there - /// isn't enough space due to previously queued datagrams, this function will return - /// `SendDatagramError::Blocked`. `Event::DatagramsUnblocked` will be emitted once datagrams - /// have been sent. + /// The datagram is scheduled like one of the group's streams at priority `order`: it waits + /// behind higher orders, goes ahead of a stream of equal order, and spends the group's + /// share of bytes. Without a group, it is ordered among the ungrouped streams, which go + /// before every group. + /// + /// Sending never blocks. While queued datagrams overflow + /// [`TransportConfig::datagram_send_buffer_size`](crate::TransportConfig::datagram_send_buffer_size), + /// the domain holding the most datagram bytes drops the oldest datagram of its + /// lowest-priority group, and this datagram is dropped instead once it is the one to go. /// /// Returns `Err` iff a `len`-byte datagram cannot currently be sent. - pub fn send(&mut self, data: Bytes, drop: bool) -> Result<(), SendDatagramError> { + /// + /// # Panics + /// - when `group` was opened on another connection + pub fn send( + &mut self, + data: Bytes, + group: Option<&SendGroup>, + order: i64, + ) -> Result<(), SendDatagramError> { if self.conn.config.datagram_receive_buffer_size.is_none() { return Err(SendDatagramError::Disabled); } let max = self .max_size() .ok_or(SendDatagramError::UnsupportedByPeer)?; - let send_buffer_size = self.conn.config.datagram_send_buffer_size; - if data.len() > Ord::min(max, send_buffer_size) { + let budget = self.conn.config.datagram_send_buffer_size; + let datagram = Datagram { data }; + if datagram.data.len() > max || datagram.size(true) > budget { return Err(SendDatagramError::TooLarge); } - if drop { - self.conn - .datagrams - .make_space_for(data.len(), send_buffer_size); - } else if !self + if !self .conn - .datagrams - .has_send_buffer_space(data.len(), send_buffer_size) + .streams + .queue_datagram(datagram, group, order, budget) { - self.conn.datagrams.send_blocked = true; - return Err(SendDatagramError::Blocked(data)); + trace!("dropping outgoing datagram"); } - self.conn.datagrams.outgoing.push_back(Datagram { data }); Ok(()) } - /// Discard queued datagrams that no longer fit the active path, waking blocked senders. + /// Discard queued datagrams that no longer fit the active path. pub(super) fn drop_oversized(&mut self) { let Some(max_datagram_size) = self.max_size() else { return; }; - if !self.conn.datagrams.drop_oversized(max_datagram_size) - || !self.conn.datagrams.send_blocked - { - return; - } - - self.conn.datagrams.send_blocked = false; - self.conn.events.push_back(Event::DatagramsUnblocked); + self.conn + .streams + .drop_oversized_datagrams(max_datagram_size); } /// Compute the maximum size of datagrams that may passed to `send_datagram` @@ -100,21 +102,20 @@ impl Datagrams<'_> { /// Bytes available in the outgoing datagram buffer /// /// When greater than zero, [`send`](Self::send)ing a datagram of at most this size is - /// guaranteed not to cause older datagrams to be dropped. + /// guaranteed not to cause queued datagrams to be dropped. pub fn send_buffer_space(&self) -> usize { self.conn .config .datagram_send_buffer_size - .saturating_sub(self.conn.datagrams.outgoing.payload_bytes) - .saturating_sub(size_of::()) + .saturating_sub(self.conn.streams.datagram_bytes()) + .saturating_sub(Datagram::SIZE_BOUND) } } +/// Incoming datagrams; outgoing ones are scheduled with the streams. #[derive(Default)] pub(super) struct DatagramState { pub(super) incoming: DatagramBuffer, - pub(super) outgoing: DatagramBuffer, - pub(super) send_blocked: bool, } impl DatagramState { @@ -148,68 +149,6 @@ impl DatagramState { Ok(was_empty) } - fn make_space_for(&mut self, datagram_len: usize, send_buffer_size: usize) { - while !self.has_send_buffer_space(datagram_len, send_buffer_size) { - let Some(prev) = self.outgoing.pop_front() else { - break; - }; - trace!(len = prev.data.len(), "dropping outgoing datagram"); - } - } - - fn has_send_buffer_space(&self, datagram_len: usize, send_buffer_size: usize) -> bool { - let Some(total) = self.outgoing.memory_used().checked_add(datagram_len) else { - return false; - }; - - total <= send_buffer_size - } - - /// Discard outgoing datagrams with a payload larger than `max_payload` bytes - /// - /// Returns whether any datagrams were dropped. - /// - /// Used to ensure that reductions in MTU don't get us stuck in a state where we have a datagram - /// queued but can't send it. - pub(super) fn drop_oversized(&mut self, max_payload: usize) -> bool { - let mut dropped_any = false; - self.outgoing.queue.retain(|datagram| { - let result = datagram.data.len() <= max_payload; - if !result { - trace!( - "dropping {} byte datagram violating {} byte limit", - datagram.data.len(), - max_payload - ); - self.outgoing.payload_bytes -= datagram.data.len(); - dropped_any = true; - } - result - }); - dropped_any - } - - /// Attempt to write a datagram frame into `buf`, consuming it from `self.outgoing` - /// - /// Returns whether a frame was written. At most `max_size` bytes will be written, including - /// framing. - pub(super) fn write(&mut self, buf: &mut Vec, max_size: usize) -> bool { - let Some(datagram) = self.outgoing.pop_front() else { - return false; - }; - - if buf.len() + datagram.size(true) > max_size { - // Future work: we could be more clever about cramming small datagrams into - // mostly-full packets when a larger one is queued first - self.outgoing.push_front(datagram); - return false; - } - - trace!(len = datagram.data.len(), "DATAGRAM"); - datagram.encode(true, buf); - true - } - pub(super) fn recv(&mut self) -> Option { let x = self.incoming.pop_front()?.data; Some(x) @@ -234,20 +173,11 @@ impl DatagramBuffer { Some(datagram) } - fn push_front(&mut self, datagram: Datagram) { - self.payload_bytes += datagram.data.len(); - self.queue.push_front(datagram); - } - fn memory_used(&self) -> usize { self.payload_bytes .saturating_add(self.queue.len() * size_of::()) } - pub(super) fn can_send_1rtt(&self, max_size: usize) -> bool { - self.queue.front().is_some_and(|x| x.size(true) <= max_size) - } - pub(super) fn is_empty(&self) -> bool { self.queue.is_empty() } @@ -257,37 +187,6 @@ impl DatagramBuffer { mod tests { use super::*; - #[test] - fn make_space_for_accounts_for_new_datagram() { - let mut state = DatagramState::default(); - state.outgoing.push_back(Datagram { - data: Bytes::from_static(&[0; 7]), - }); - state.outgoing.push_back(Datagram { - data: Bytes::from_static(&[0; 2]), - }); - - state.make_space_for(4, 10 + 2 * size_of::()); - - assert_eq!(state.outgoing.queue.len(), 1); - assert_eq!(state.outgoing.queue[0].data.len(), 2); - assert_eq!(state.outgoing.payload_bytes, 2); - } - - #[test] - fn make_space_for_handles_overflowing_capacity_check() { - let mut state = DatagramState::default(); - state.outgoing.queue.push_back(Datagram { - data: Bytes::from_static(&[0]), - }); - state.outgoing.payload_bytes = usize::MAX - 1; - - state.make_space_for(2, usize::MAX); - - assert!(state.outgoing.is_empty()); - assert_eq!(state.outgoing.payload_bytes, usize::MAX - 2); - } - #[test] fn empty_frame_flood_limit() { let mut state = DatagramState::default(); @@ -303,29 +202,12 @@ mod tests { } } } - - #[test] - fn drop_oversized_keeps_datagrams_at_limit() { - let mut state = DatagramState::default(); - state.outgoing.push_back(Datagram { - data: Bytes::from_static(&[0; 10]), - }); - state.outgoing.push_back(Datagram { - data: Bytes::from_static(&[0; 11]), - }); - - assert!(state.drop_oversized(10)); - - assert_eq!(state.outgoing.queue.len(), 1); - assert_eq!(state.outgoing.queue[0].data.len(), 10); - assert_eq!(state.outgoing.payload_bytes, 10); - } } /// Errors that can arise when sending a datagram #[derive(Debug, Error, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)] pub enum SendDatagramError { - /// The peer does not support receiving datagram frames + /// The peer does not support receiving datagrams #[error("datagrams not supported by peer")] UnsupportedByPeer, /// Datagram support is disabled locally @@ -337,7 +219,4 @@ pub enum SendDatagramError { /// exceeded. #[error("datagram too large")] TooLarge, - /// Send would block - #[error("datagram send blocked")] - Blocked(Bytes), } diff --git a/rs/moq-quic/src/connection/mod.rs b/rs/moq-quic/src/connection/mod.rs index 3b45a80d40..c1869ecb9d 100644 --- a/rs/moq-quic/src/connection/mod.rs +++ b/rs/moq-quic/src/connection/mod.rs @@ -23,7 +23,7 @@ use crate::{ config::{ServerConfig, TransportConfig}, connection::spaces::LostPacket, crypto::{self, KeyPair, Keys, PacketKey}, - frame::{self, Close, Datagram, FrameStruct, NewConnectionId, NewToken}, + frame::{self, Close, FrameStruct, NewConnectionId, NewToken}, packet::{ FixedLengthConnectionIdParser, Header, InitialHeader, InitialPacket, LongType, Packet, PacketNumber, PartialDecode, SpaceId, @@ -964,7 +964,7 @@ impl Connection { && !can_send.acks && can_send.other && (buf_capacity - builder.datagram_start) == self.path.current_mtu() as usize - && self.datagrams.outgoing.is_empty()), + && self.streams.datagram_bytes() == 0), "SendableFrames was {can_send:?}, but only ACKs have been written" ); pad_datagram |= sent.requires_padding; @@ -3626,23 +3626,6 @@ impl Connection { self.stats.frame_tx.retire_connection_id += 1; } - // DATAGRAM - let mut sent_datagrams = false; - while buf.len() + Datagram::SIZE_BOUND < max_size && space_id == SpaceId::Data { - match self.datagrams.write(buf, max_size) { - true => { - sent_datagrams = true; - sent.non_retransmits = true; - self.stats.frame_tx.datagram += 1; - } - false => break, - } - } - if self.datagrams.send_blocked && sent_datagrams { - self.events.push_back(Event::DatagramsUnblocked); - self.datagrams.send_blocked = false; - } - // NEW_TOKEN while let Some(remote_addr) = space.pending.new_tokens.pop() { debug_assert_eq!(space_id, SpaceId::Data); @@ -3682,12 +3665,17 @@ impl Connection { self.stats.frame_tx.new_token += 1; } - // STREAM + // STREAM and DATAGRAM, interleaved in schedule order if space_id == SpaceId::Data { - sent.stream_frames = - self.streams - .write_stream_frames(buf, max_size, self.config.send_fairness); + let written = self.streams.write_frames( + buf, + pre_payload_len..max_size, + self.config.send_fairness, + ); + sent.stream_frames = written.streams; + sent.non_retransmits |= written.datagrams > 0; self.stats.frame_tx.stream += sent.stream_frames.len() as u64; + self.stats.frame_tx.datagram += written.datagrams; } // Bundle ACK with other frames when there is room for them. @@ -3999,14 +3987,13 @@ impl Connection { /// /// See also `self.space(SpaceId::Data).can_send()` fn can_send_1rtt(&self, max_size: usize) -> bool { - self.streams.can_send_stream_data() + self.streams.can_send_frames(max_size) || self.path.challenge_pending || self .prev_path .as_ref() .is_some_and(|(_, x)| x.challenge_pending) || !self.path_responses.is_empty() - || self.datagrams.outgoing.can_send_1rtt(max_size) } /// Update counters to account for a packet becoming acknowledged, lost, or abandoned @@ -4356,8 +4343,6 @@ pub enum Event { Stream(StreamEvent), /// One or more application datagrams have been received DatagramReceived, - /// One or more application datagrams have been sent after blocking - DatagramsUnblocked, /// The currently active path was updated PathUpdated, } diff --git a/rs/moq-quic/src/connection/streams/mod.rs b/rs/moq-quic/src/connection/streams/mod.rs index 34f3df7dd9..efaab5f2f8 100644 --- a/rs/moq-quic/src/connection/streams/mod.rs +++ b/rs/moq-quic/src/connection/streams/mod.rs @@ -18,7 +18,7 @@ use recv::Recv; mod saturated; mod schedule; pub use recv::{Chunks, Credit, ReadError, ReadableError}; -use schedule::Schedule; +use schedule::{Next, Schedule}; mod send; pub(crate) use send::{ByteSlice, BytesArray}; @@ -502,7 +502,8 @@ impl Checkpoint { /// A send group shares bytes with the other groups of its domain: a higher priority group /// preempts every lower one, and groups of equal priority get an equal share of bytes no /// matter how many streams each has queued. Domains share bytes equally whatever their -/// groups' priorities. Streams outside any group go before all of it. +/// groups' priorities. Streams outside any group go before all of it. A group's datagrams, +/// from [`Datagrams::send`](crate::Datagrams::send), share its bytes in the same order. pub struct SendGroups<'a> { pub(super) state: &'a mut StreamsState, } @@ -567,6 +568,47 @@ impl SendGroups<'_> { } } +impl StreamsState { + /// Queue a datagram in `group` at `order`, within `budget` bytes of queued datagrams. + /// + /// Returns whether it stayed queued rather than being evicted itself. + /// + /// # Panics + /// - when `group` was opened on another connection + pub(crate) fn queue_datagram( + &mut self, + datagram: frame::Datagram, + group: Option<&SendGroup>, + order: i64, + budget: usize, + ) -> bool { + let group = group.map(|g| g.index(self.tag)); + let frame = datagram.size(true); + self.pending + .push_datagram(group, order, datagram, frame, budget) + } + + /// Drop queued datagrams with a payload over `max_payload` bytes, returning whether any was. + pub(crate) fn drop_oversized_datagrams(&mut self, max_payload: usize) -> bool { + self.pending.retain_datagrams(|datagram| { + let keep = datagram.data.len() <= max_payload; + if !keep { + trace!( + "dropping {} byte datagram violating {} byte limit", + datagram.data.len(), + max_payload + ); + } + keep + }) + } + + /// Frame bytes queued datagrams hold against the send buffer. + pub(crate) fn datagram_bytes(&self) -> usize { + self.pending.datagram_bytes() + } +} + /// A send group, from [`SendGroups::open`] /// /// Release it with [`SendGroups::close`] on the connection that opened it. diff --git a/rs/moq-quic/src/connection/streams/saturated.rs b/rs/moq-quic/src/connection/streams/saturated.rs index fb561e5fb6..0ae5297667 100644 --- a/rs/moq-quic/src/connection/streams/saturated.rs +++ b/rs/moq-quic/src/connection/streams/saturated.rs @@ -1,24 +1,37 @@ -//! Scheduling proofs through the real send path: streams written into send groups and -//! drained into packets by `write_stream_frames`, with loss and resets applied the way +//! Scheduling proofs through the real send path: streams and datagrams queued into send +//! groups and drained into packets by `write_frames`, with loss and resets applied the way //! the connection applies them. use std::collections::HashMap; +use bytes::{BufMut, Bytes, BytesMut}; + use super::state::StreamsState; use super::{SendDomain, SendGroup, SendGroups, SendStream, Streams}; use crate::{ Dir, Side, StreamId, VarInt, connection::{State as ConnState, spaces::Retransmits}, - frame::StreamMeta, + frame::{self, Frame, StreamMeta}, transport_parameters::TransportParameters, }; +/// One frame of a packet, in the order it was written. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Sent { + Stream(StreamId), + /// A datagram, by the tag it was queued with. + Datagram(u64), +} + const PACKET: usize = 1200; /// How far two equally entitled groups may drift apart: each round robin may /// overdraw by a frame, at both the group and the domain tier. const SLACK: u64 = 2 * PACKET as u64; +/// A PING frame's one byte, standing in for the control frames ahead of the streams. +const PING: u8 = 0x01; + /// A sender with more queued than the path takes, drained a packet at a time. struct Sender { state: StreamsState, @@ -29,6 +42,12 @@ struct Sender { /// Bytes each stream's frames took on the wire, headers included, which is what /// the round robins share. wire: HashMap, + /// Bytes each datagram tag took on the wire, headers included. + datagram_wire: HashMap, + /// Every frame sent, in order. + order: Vec, + /// The next datagram tag. + tag: u64, } impl Sender { @@ -58,6 +77,9 @@ impl Sender { conn: ConnState::Established, sent: HashMap::new(), wire: HashMap::new(), + datagram_wire: HashMap::new(), + order: Vec::new(), + tag: 0, } } @@ -99,10 +121,62 @@ impl Sender { assert_eq!(self.send(id).write(&vec![0; bytes]), Ok(bytes)); } - /// Send one packet, returning its frames. + /// Queue a `bytes` payload datagram in `group` at `order`, returning its tag. + fn datagram(&mut self, group: Option<&SendGroup>, order: i64, bytes: usize) -> u64 { + let tag = self.tag; + self.tag += 1; + let mut data = BytesMut::with_capacity(bytes); + data.put_u64(tag); + data.resize(bytes, 0); + let datagram = frame::Datagram { + data: data.freeze(), + }; + assert!( + self.state + .queue_datagram(datagram, group, order, usize::MAX) + ); + tag + } + + /// Send one packet, returning its stream frames. fn packet(&mut self) -> Vec { - let mut buf = Vec::with_capacity(PACKET); - let frames = self.state.write_stream_frames(&mut buf, PACKET, true); + self.packet_of(PACKET) + } + + /// Send one packet of `size` bytes, returning its stream frames. + fn packet_of(&mut self, size: usize) -> Vec { + self.packet_after(&[], size) + } + + /// Send one packet of `size` bytes that already holds the control frames `prefix`, + /// returning its stream frames. + fn packet_after(&mut self, prefix: &[u8], size: usize) -> Vec { + let mut buf = Vec::with_capacity(size); + buf.extend_from_slice(prefix); + let written = self.state.write_frames(&mut buf, 0..size, true); + let frames = written.streams; + + // The packet's frames in order, and the wire bytes its datagrams took. + let mut datagram_bytes = 0; + let frames_in = match buf.is_empty() { + true => None, + false => Some(frame::Iter::new(Bytes::from(buf.clone())).unwrap()), + }; + for frame in frames_in.into_iter().flatten() { + match frame.unwrap() { + Frame::Stream(stream) => self.order.push(Sent::Stream(stream.id)), + Frame::Datagram(datagram) => { + let wire = datagram.size(true) as u64; + let tag = u64::from_be_bytes(datagram.data[..8].try_into().unwrap()); + *self.datagram_wire.entry(tag).or_default() += wire; + datagram_bytes += wire; + self.order.push(Sent::Datagram(tag)); + } + Frame::Ping => {} + frame => panic!("unexpected {frame:?}"), + } + } + let stream_bytes = (buf.len() - prefix.len()) as u64 - datagram_bytes; // Every frame carries its length but perhaps the last, which can run to the // end of the packet instead. @@ -114,14 +188,11 @@ impl Sender { (header.len() as u64, frame.offsets.end - frame.offsets.start) }) .collect(); - if sizes.iter().map(|(h, p)| h + p).sum::() != buf.len() as u64 { + if sizes.iter().map(|(h, p)| h + p).sum::() != stream_bytes { let (header, payload) = sizes.last_mut().unwrap(); *header -= VarInt::from_u64(*payload).unwrap().size() as u64; } - assert_eq!( - sizes.iter().map(|(h, p)| h + p).sum::(), - buf.len() as u64 - ); + assert_eq!(sizes.iter().map(|(h, p)| h + p).sum::(), stream_bytes); for (frame, (header, payload)) in frames.iter().zip(sizes) { *self.sent.entry(frame.id).or_default() += payload; @@ -152,9 +223,25 @@ impl Sender { .sum() } + /// Send `packets` packets, returning every frame in order. + fn drain_frames(&mut self, packets: usize) -> Vec { + let start = self.order.len(); + for _ in 0..packets { + self.packet(); + } + self.order[start..].to_vec() + } + + fn datagram_wire(&self, tags: &[u64]) -> u64 { + tags.iter() + .map(|tag| self.datagram_wire.get(tag).copied().unwrap_or(0)) + .sum() + } + fn clear(&mut self) { self.sent.clear(); self.wire.clear(); + self.datagram_wire.clear(); } } @@ -493,3 +580,186 @@ fn streams_can_leave_their_group() { assert!(sender.state.pending.is_clean()); assert_eq!(sender.drain(1), [id]); } + +/// A low-priority datagram track waits behind a higher-priority stream rather than +/// going out ahead of every stream. +#[test] +fn low_priority_datagrams_wait_behind_streams() { + let mut sender = Sender::new(); + let high = sender.group(1); + let low = sender.group(0); + let tags: Vec<_> = (0..20) + .map(|seq| sender.datagram(Some(&low), seq, 1_000)) + .collect(); + let stream = sender.open(Some(&high), 0, 20_000); + + let order = sender.drain_frames(100); + let first = order + .iter() + .position(|sent| matches!(sent, Sent::Datagram(_))) + .unwrap(); + assert!( + order[..first] + .iter() + .all(|&sent| sent == Sent::Stream(stream)) + ); + assert!( + order[first..] + .iter() + .all(|sent| matches!(sent, Sent::Datagram(_))) + ); + assert_eq!(sender.bytes(&[stream]), 20_000); + assert_eq!(order[first..].len(), tags.len()); +} + +/// A high-priority datagram track still goes ahead of lower-priority streams. +#[test] +fn high_priority_datagrams_preempt_streams() { + let mut sender = Sender::new(); + let high = sender.group(1); + let low = sender.group(0); + let stream = sender.open(Some(&low), 0, 20_000); + let tags: Vec<_> = (0..20) + .map(|seq| sender.datagram(Some(&high), seq, 1_000)) + .collect(); + + let order = sender.drain_frames(100); + // Newest first, as the group orders them. + let expected: Vec<_> = tags.iter().rev().map(|&tag| Sent::Datagram(tag)).collect(); + assert_eq!(order[..tags.len()], expected[..]); + assert!( + order[tags.len()..] + .iter() + .all(|&sent| sent == Sent::Stream(stream)) + ); +} + +/// Inside one group a datagram merges into the group order: it waits behind a stream +/// for a newer group and goes ahead of a stream of its own group. +#[test] +fn datagrams_merge_into_their_group_order() { + let mut sender = Sender::new(); + let group = sender.group(0); + let older = sender.datagram(Some(&group), 10, 100); + let newer = sender.open(Some(&group), 11, 3_000); + let tied = sender.open(Some(&group), 12, 3_000); + let ahead = sender.datagram(Some(&group), 12, 100); + + let order = sender.drain_frames(20); + let position = |sent| order.iter().position(|&x| x == sent).unwrap(); + let last = |sent| order.iter().rposition(|&x| x == sent).unwrap(); + assert_eq!(position(Sent::Datagram(ahead)), 0); + assert!(last(Sent::Stream(tied)) < position(Sent::Stream(newer))); + assert!(last(Sent::Stream(newer)) < position(Sent::Datagram(older))); +} + +/// A group's datagrams spend its fair share, so sending datagrams buys no extra bytes over +/// an equal group of streams. +#[test] +fn datagrams_and_streams_share_a_groups_credit() { + let mut sender = Sender::new(); + let datagrams = sender.group(0); + let streams = sender.group(0); + let tags: Vec<_> = (0..1_000) + .map(|seq| sender.datagram(Some(&datagrams), seq, 500)) + .collect(); + let stream = sender.open(Some(&streams), 0, 1_000_000); + + sender.drain(200); + assert_fair(sender.datagram_wire(&tags), sender.wire(&[stream])); +} + +/// With one domain per broadcast, two saturated datagram broadcasts both keep sending, +/// one of many small datagrams and one of few large ones, each getting half the bytes. +#[test] +fn saturated_datagram_domains_share_bytes() { + let mut sender = Sender::new(); + let small = sender.groups().open_domain(); + let large = sender.groups().open_domain(); + let small = sender.groups().open(0, Some(&small)); + let large = sender.groups().open(9, Some(&large)); + let small: Vec<_> = (0..2_000) + .map(|seq| sender.datagram(Some(&small), seq, 50)) + .collect(); + let large: Vec<_> = (0..200) + .map(|seq| sender.datagram(Some(&large), seq, 1_000)) + .collect(); + + sender.drain(100); + assert_fair(sender.datagram_wire(&small), sender.datagram_wire(&large)); +} + +/// A datagram outside every group goes ahead of every group, like a control stream. +#[test] +fn ungrouped_datagrams_preempt_groups() { + let mut sender = Sender::new(); + let group = sender.group(i64::MAX); + let stream = sender.open(Some(&group), i64::MAX, 3_000); + let tag = sender.datagram(None, 0, 100); + + let order = sender.drain_frames(5); + assert_eq!(order[0], Sent::Datagram(tag)); + assert!(order[1..].iter().all(|&sent| sent == Sent::Stream(stream))); +} + +/// A datagram too large for the room left in a packet ends it and leads the next one, +/// rather than letting what queues behind it go first. +#[test] +fn a_datagram_that_does_not_fit_leads_the_next_packet() { + let mut sender = Sender::new(); + let group = sender.group(0); + let stream = sender.open(Some(&group), 0, 100); + let small = sender.datagram(Some(&group), 2, 300); + let large = sender.datagram(Some(&group), 1, 1_000); + + assert_eq!(sender.drain_frames(1), [Sent::Datagram(small)]); + assert_eq!( + sender.drain_frames(1), + [Sent::Datagram(large), Sent::Stream(stream)] + ); + assert!(!sender.state.can_send_stream_data()); +} + +/// A packet clamped below a queued datagram, such as a loss probe, still carries what fits +/// behind it, and the datagram keeps its place. +#[test] +fn a_clamped_packet_sends_what_fits() { + let mut sender = Sender::new(); + let group = sender.group(0); + let large = sender.datagram(Some(&group), 1, 1_000); + assert!(!sender.state.can_send_frames(500)); + let stream = sender.open(Some(&group), 0, 2_000); + assert!(sender.state.can_send_frames(500)); + + sender.packet_of(500); + assert_eq!(sender.order, [Sent::Stream(stream)]); + assert_eq!(sender.drain_frames(1)[0], Sent::Datagram(large)); +} + +/// A packet already holding a control frame is not empty, so a datagram that no longer +/// fits ends it rather than letting lower-ranked data ahead. +#[test] +fn a_datagram_behind_control_frames_ends_the_packet() { + let mut sender = Sender::new(); + let group = sender.group(0); + let large = sender.datagram(Some(&group), 1, 1_000); + sender.open(Some(&group), 0, 2_000); + + // The DATAGRAM frame (type, two-byte length, payload) would fill the packet alone, + // but a PING is already in it. + sender.packet_after(&[PING], 1 + 2 + 1_000); + assert_eq!(sender.order, []); + assert_eq!(sender.drain_frames(1)[0], Sent::Datagram(large)); +} + +/// A released group keeps its queued datagrams and frees once they are sent. +#[test] +fn a_released_group_sends_its_datagrams() { + let mut sender = Sender::new(); + let group = sender.group(0); + let tag = sender.datagram(Some(&group), 0, 100); + sender.groups().close(group); + assert!(!sender.state.pending.is_clean()); + assert_eq!(sender.drain_frames(1), [Sent::Datagram(tag)]); + assert!(sender.state.pending.is_clean()); +} diff --git a/rs/moq-quic/src/connection/streams/schedule.rs b/rs/moq-quic/src/connection/streams/schedule.rs index 7d31645298..a4c2499e67 100644 --- a/rs/moq-quic/src/connection/streams/schedule.rs +++ b/rs/moq-quic/src/connection/streams/schedule.rs @@ -1,20 +1,30 @@ -//! The order streams with queued data are sent in. +//! The order streams and datagrams with queued data are sent in. //! -//! Streams outside any send group go first, in quinn's order: highest priority first, round -//! robin between equal priorities. Everything else is a hierarchy: +//! Streams and datagrams outside any send group go first, in quinn's order: highest priority +//! first, round robin between equal priorities. Everything else is a hierarchy: //! //! 1. Domains share the remaining bytes by deficit round robin. //! 2. Inside a domain, a higher-priority send group preempts every lower one. //! 3. Send groups of equal priority share their domain's bytes by deficit round robin. -//! 4. Inside a group, streams go in their own priority order, round robin between ties. +//! 4. Inside a group, streams and datagrams go in their own order, round robin between tied +//! streams. A datagram goes ahead of a stream of equal order, and tied datagrams oldest +//! first. //! //! Both round robins charge the bytes actually written, so a group of many small streams and a -//! group of one large stream get the same rate. A turn may overdraw its credit by one frame; -//! the debt carries into the next turn, so the error stays bounded by a frame. +//! group of one large stream get the same rate, and a group's datagrams spend the same credit +//! as its streams. A turn may overdraw its credit by one frame; the debt carries into the next +//! turn, so the error stays bounded by a frame. +//! +//! Queued datagrams share one byte budget. When it overflows, the domain holding the most +//! datagram bytes gives up the oldest datagram of its lowest-priority group. Priorities are +//! only compared inside a domain, so one busy domain cannot evict another's datagrams. //! //! Self-contained (std and slab only) so the scheduler benchmark can compile it from source. -use std::collections::{BTreeMap, VecDeque}; +use std::{ + cmp::Reverse, + collections::{BTreeMap, BTreeSet, VecDeque}, +}; use slab::Slab; @@ -42,65 +52,118 @@ enum Queued { At(Key), } -/// A queued stream's priority, then a decreasing counter so ties go round robin. -type Key = (i64, u64); +/// A queued item's order, then whether it is a datagram, which wins a tie, then a decreasing +/// counter so the rest of a tie goes oldest first: round robin for requeued streams. +type Key = (i64, bool, u64); + +/// What to send next, from [`Schedule::pop`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum Next { + /// A stream, out of the schedule until [`Schedule::sent`] reports back. + Stream(K), + /// The datagram heading a group, or the ungrouped queue with `None`. It stays queued + /// until [`Schedule::sent_datagram`] takes it. + Datagram(Option), +} + +#[derive(Debug)] +enum Item { + Stream(K), + /// A datagram and the bytes of its frame, which it holds against the budget. + Datagram(D, usize), +} #[derive(Debug)] -struct Queue { - streams: BTreeMap, +struct Queue { + items: BTreeMap>, next: Option, + /// The order of every queued datagram by its counter, so the oldest is the last. + datagrams: BTreeMap, } -impl Queue { +impl Queue { fn new() -> Self { Self { - streams: BTreeMap::new(), + items: BTreeMap::new(), next: None, + datagrams: BTreeMap::new(), } } - fn pop(&mut self) -> Option { - self.next - .take() - .or_else(|| self.streams.pop_last().map(|(_, id)| id)) + fn pop(&mut self, group: Option) -> Option> { + if let Some(id) = self.next.take() { + return Some(Next::Stream(id)); + } + let head = self.items.last_entry()?; + if let Item::Datagram(..) = head.get() { + return Some(Next::Datagram(group)); + } + let Item::Stream(id) = head.remove() else { + unreachable!("checked for a stream"); + }; + Some(Next::Stream(id)) } fn remove(&mut self, queued: Queued) { match queued { Queued::Next => self.next = None, Queued::At(key) => { - self.streams.remove(&key); + self.items.remove(&key); } } } + /// The datagram heading the queue, its frame bytes, and its counter. + fn head_datagram(&self) -> (&D, usize, u64) { + match self.items.last_key_value() { + Some((&(_, _, recency), Item::Datagram(datagram, frame))) => { + (datagram, *frame, recency) + } + _ => panic!("no datagram heads the queue"), + } + } + + /// The counter of the oldest queued datagram. + fn oldest_datagram(&self) -> Option { + self.datagrams.last_key_value().map(|(&recency, _)| recency) + } + + fn take_datagram(&mut self, recency: u64) -> (D, usize) { + let order = self.datagrams.remove(&recency).expect("datagram is queued"); + match self.items.remove(&(order, true, recency)) { + Some(Item::Datagram(datagram, size)) => (datagram, size), + _ => unreachable!("an indexed datagram is queued"), + } + } + fn is_empty(&self) -> bool { - self.next.is_none() && self.streams.is_empty() + self.next.is_none() && self.items.is_empty() } #[cfg(test)] fn len(&self) -> usize { - self.streams.len() + self.next.is_some() as usize + self.items.len() + self.next.is_some() as usize } } #[derive(Debug)] -struct Group { +struct Group { domain: usize, priority: i64, - queue: Queue, + queue: Queue, /// Bytes this group may still send in its current turn; negative after an overdraw. deficit: i64, - /// Whether the group has queued streams, or is being served one. It stays set between + /// Whether the group has queued items, or is being served a stream. It stays set between /// [`Schedule::pop`] and [`Schedule::sent`], so requeueing the stream is not a new turn. active: bool, /// Whether an entry for this group sits in its level's round robin. Usually equal to - /// having queued streams, but a group that runs dry away from the front of the round + /// having queued items, but a group that runs dry away from the front of the round /// robin keeps its entry until the round robin reaches it. ringed: bool, /// Streams that name this group, queued or not. members: usize, - /// Whether the owner released the group, which is freed once no stream names it. + /// Whether the owner released the group, which is freed once no stream names it and its + /// datagrams are gone. closed: bool, } @@ -112,13 +175,24 @@ struct Domain { deficit: i64, /// Whether an entry for this domain sits in the domain round robin, as for groups. ringed: bool, - /// Groups in this domain with queued streams. + /// Groups in this domain with queued items. backlogged: usize, /// Groups in this domain that are not yet freed. groups: usize, closed: bool, + /// Datagram bytes queued in the domain. + datagram_bytes: usize, + /// The counter of every datagram queued in the domain, so the last is its oldest. + datagrams: BTreeSet, + /// The order the domain's queues give up datagrams in: groups before the ungrouped queue + /// (`None`, in the default domain), lowest priority first, then the oldest datagram. + victims: BTreeSet, } +/// A queue holding datagrams, keyed for eviction: whether it is the ungrouped queue, its +/// group's priority, its oldest datagram, and the group. +type Victim = (bool, i64, Reverse, Option); + impl Domain { fn new() -> Self { Self { @@ -128,15 +202,18 @@ impl Domain { backlogged: 0, groups: 0, closed: false, + datagram_bytes: 0, + datagrams: BTreeSet::new(), + victims: BTreeSet::new(), } } } -/// Every stream with data to send, in the order to send it. +/// Every stream and datagram with data to send, in the order to send it. #[derive(Debug)] -pub(super) struct Schedule { - ungrouped: Queue, - groups: Slab>, +pub(super) struct Schedule { + ungrouped: Queue, + groups: Slab>, domains: Slab, /// The domain round robin. ring: VecDeque, @@ -144,9 +221,20 @@ pub(super) struct Schedule { backlogged: usize, /// Decreases with every push, so a requeued stream goes behind its equal-priority peers. recency: u64, + /// Domains holding datagrams, by their datagram bytes then their oldest datagram, so the + /// last gives one up first. + hoards: BTreeSet<(usize, u64, usize)>, + /// Datagram bytes queued across every domain, held against the budget. + datagram_bytes: usize, + /// How many queued datagrams have each frame size, so the smallest is the first. + frames: BTreeMap, + /// Streams queued, not counting one between [`Self::pop`] and [`Self::sent`]. + streams: usize, + /// Datagrams [`Self::defer_datagram`] took out of line until [`Self::restore`]. + deferred: Vec<(Option, Key, Item)>, } -impl Schedule { +impl Schedule { pub(super) fn new() -> Self { let mut domains = Slab::new(); assert_eq!(domains.insert(Domain::new()), DEFAULT_DOMAIN); @@ -157,14 +245,35 @@ impl Schedule { ring: VecDeque::new(), backlogged: 0, recency: u64::MAX, + hoards: BTreeSet::new(), + datagram_bytes: 0, + frames: BTreeMap::new(), + streams: 0, + deferred: Vec::new(), } } - /// Whether no stream is queued. + /// Whether nothing is queued. + #[cfg(test)] pub(super) fn is_empty(&self) -> bool { self.ungrouped.is_empty() && self.backlogged == 0 } + /// Whether a packet with room for a `space`-byte frame would carry something: any stream, + /// which can be split to fit, or a datagram no larger than that. + pub(super) fn can_send(&self, space: usize) -> bool { + self.streams > 0 + || self + .frames + .first_key_value() + .is_some_and(|(&frame, _)| frame <= space) + } + + /// Frame bytes held by queued datagrams. + pub(super) fn datagram_bytes(&self) -> usize { + self.datagram_bytes + } + #[cfg(test)] pub(super) fn len(&self) -> usize { self.ungrouped.len() @@ -211,7 +320,7 @@ impl Schedule { }) } - /// Release a group; it is freed once no stream names it. + /// Release a group; it is freed once no stream names it and its datagrams are gone. pub(super) fn close_group(&mut self, group: usize) { self.groups[group].closed = true; self.free_group(group); @@ -223,11 +332,12 @@ impl Schedule { /// Move a group to another priority level, keeping its credit. pub(super) fn set_group_priority(&mut self, group: usize, priority: i64) { - let state = &mut self.groups[group]; - if state.priority == priority { + if self.groups[group].priority == priority { return; } + self.unindex(Some(group)); + let state = &mut self.groups[group]; let old = std::mem::replace(&mut state.priority, priority); let domain = state.domain; if state.ringed { @@ -242,6 +352,7 @@ impl Schedule { .push_back(group); } } + self.index(Some(group)); } /// Move a stream to `group`, or out of every group, keeping its place in line if queued. @@ -279,9 +390,10 @@ impl Schedule { return; } self.recency -= 1; - let key = (priority, self.recency); + let key = (priority, false, self.recency); slot.queued = Some(Queued::At(key)); - self.queue(slot.group).streams.insert(key, id); + self.queue(slot.group).items.insert(key, Item::Stream(id)); + self.streams += 1; self.activate(slot.group); } @@ -292,6 +404,7 @@ impl Schedule { assert!(queue.next.is_none()); queue.next = Some(id); slot.queued = Some(Queued::Next); + self.streams += 1; self.activate(slot.group); } @@ -301,16 +414,147 @@ impl Schedule { return; }; self.queue(slot.group).remove(queued); + self.streams -= 1; self.deactivate(slot.group); } - /// Take the next stream to send from. + /// Queue a datagram in `group`, or ungrouped, at `order` among that queue's streams and + /// datagrams. Its `frame` bytes are what it sends and holds of the `budget` every queued + /// datagram shares. + /// + /// Returns whether it stayed queued: while the budget overflows, the domain holding the + /// most datagram bytes evicts the oldest datagram of its lowest-priority group, and this + /// datagram is dropped once it is the one to go. + pub(super) fn push_datagram( + &mut self, + group: Option, + order: i64, + datagram: D, + frame: usize, + budget: usize, + ) -> bool { + if let Some(group) = group { + assert!( + !self.groups[group].closed, + "datagram sent on a closed group" + ); + } + + self.recency -= 1; + let recency = self.recency; + self.unindex(group); + let queue = self.queue(group); + queue + .items + .insert((order, true, recency), Item::Datagram(datagram, frame)); + queue.datagrams.insert(recency, order); + let domain = self.domain_of(group); + let domain = &mut self.domains[domain]; + domain.datagram_bytes += frame; + domain.datagrams.insert(recency); + self.datagram_bytes += frame; + *self.frames.entry(frame).or_default() += 1; + self.index(group); + self.activate(group); + + while self.datagram_bytes > budget { + let &(_, _, domain) = self.hoards.last().expect("queued bytes are indexed"); + let &(_, _, Reverse(oldest), victim) = self.domains[domain] + .victims + .first() + .expect("a domain holding datagram bytes has a victim"); + self.remove_datagram(victim, oldest); + if oldest == recency { + return false; + } + } + true + } + + /// The datagram [`Self::pop`] returned as [`Next::Datagram`], and its frame bytes. + pub(super) fn datagram(&self, group: Option) -> (&D, usize) { + let (datagram, frame, _) = self.head(group); + (datagram, frame) + } + + /// Take the datagram [`Self::pop`] returned as [`Next::Datagram`], charging its frame to + /// its group. + pub(super) fn sent_datagram(&mut self, group: Option) -> D { + let (_, frame, recency) = self.head(group); + if let Some(group) = group { + let state = &mut self.groups[group]; + state.deficit -= frame as i64; + self.domains[state.domain].deficit -= frame as i64; + } + self.remove_datagram(group, recency) + } + + /// Take the datagram [`Self::pop`] returned out of line, so what queues behind it can fill + /// a packet it does not fit. [`Self::restore`] puts it back in its place. + /// + /// A group this empties restarts its turn when restored, gaining up to a quantum of credit, + /// so this is for the rare packet clamped below the path MTU, not every full one. + pub(super) fn defer_datagram(&mut self, group: Option) { + let (key, item) = self + .queue(group) + .items + .pop_last() + .expect("a datagram heads"); + self.deferred.push((group, key, item)); + // Its group stays allocated: the datagram is still indexed for eviction. + self.deactivate(group); + } + + /// Requeue every datagram [`Self::defer_datagram`] took out of line. + pub(super) fn restore(&mut self) { + for (group, key, item) in std::mem::take(&mut self.deferred) { + self.queue(group).items.insert(key, item); + self.activate(group); + } + } + + fn head(&self, group: Option) -> (&D, usize, u64) { + match group { + Some(group) => self.groups[group].queue.head_datagram(), + None => self.ungrouped.head_datagram(), + } + } + + /// Drop every queued datagram `keep` refuses, returning whether any was. + pub(super) fn retain_datagrams(&mut self, mut keep: impl FnMut(&D) -> bool) -> bool { + let groups = self.groups.iter().map(|(index, g)| (Some(index), &g.queue)); + let mut doomed = Vec::new(); + for (group, queue) in std::iter::once((None, &self.ungrouped)).chain(groups) { + for (&recency, &order) in &queue.datagrams { + if let Some(Item::Datagram(datagram, _)) = queue.items.get(&(order, true, recency)) + && !keep(datagram) + { + doomed.push((group, recency)); + } + } + } + for &(group, recency) in &doomed { + self.remove_datagram(group, recency); + } + !doomed.is_empty() + } + + /// Take the next stream or datagram to send. /// - /// The caller writes from it and then reports back with [`Self::sent`], which requeues - /// it and charges the bytes to its group. - pub(super) fn pop(&mut self) -> Option { - if let Some(id) = self.ungrouped.pop() { - return Some(id); + /// The caller writes from a stream and then reports back with [`Self::sent`], which + /// requeues it and charges the bytes to its group. A datagram stays queued until + /// [`Self::sent_datagram`] takes it, so one that does not fit waits for the next packet. + pub(super) fn pop(&mut self) -> Option> { + let next = self.pop_next(); + if let Some(Next::Stream(_)) = next { + self.streams -= 1; + } + next + } + + fn pop_next(&mut self) -> Option> { + if let Some(next) = self.ungrouped.pop(None) { + return Some(next); } loop { @@ -347,7 +591,12 @@ impl Schedule { continue; } - return Some(group_state.queue.pop().expect("checked non-empty")); + return Some( + group_state + .queue + .pop(Some(group)) + .expect("checked non-empty"), + ); } } @@ -379,14 +628,78 @@ impl Schedule { } } - fn queue(&mut self, group: Option) -> &mut Queue { + fn queue(&mut self, group: Option) -> &mut Queue { match group { Some(group) => &mut self.groups[group].queue, None => &mut self.ungrouped, } } - /// Give a group that just got its first queued stream a turn. + fn domain_of(&self, group: Option) -> usize { + group.map_or(DEFAULT_DOMAIN, |group| self.groups[group].domain) + } + + /// Dequeue a datagram, leaving the round robin with the group's last item. + fn remove_datagram(&mut self, group: Option, recency: u64) -> D { + self.unindex(group); + let (datagram, frame) = self.queue(group).take_datagram(recency); + let domain = self.domain_of(group); + let domain = &mut self.domains[domain]; + domain.datagram_bytes -= frame; + domain.datagrams.remove(&recency); + self.datagram_bytes -= frame; + if let std::collections::btree_map::Entry::Occupied(mut count) = self.frames.entry(frame) { + *count.get_mut() -= 1; + if *count.get() == 0 { + count.remove(); + } + } + self.index(group); + + self.deactivate(group); + if let Some(group) = group { + self.free_group(group); + } + datagram + } + + /// Withdraw a queue and its domain from the eviction order, ahead of changing its + /// datagrams or priority; [`Self::index`] puts them back. + fn unindex(&mut self, group: Option) { + self.reindex(group, false); + } + + fn index(&mut self, group: Option) { + self.reindex(group, true); + } + + fn reindex(&mut self, group: Option, insert: bool) { + let (queue, priority) = match group { + Some(index) => { + let state = &self.groups[index]; + (&state.queue, state.priority) + } + None => (&self.ungrouped, i64::MAX), + }; + let index = self.domain_of(group); + let domain = &mut self.domains[index]; + if let Some(oldest) = queue.oldest_datagram() { + let victim = (group.is_none(), priority, Reverse(oldest), group); + match insert { + true => domain.victims.insert(victim), + false => domain.victims.remove(&victim), + }; + } + if let Some(&oldest) = domain.datagrams.last() { + let hoard = (domain.datagram_bytes, oldest, index); + match insert { + true => self.hoards.insert(hoard), + false => self.hoards.remove(&hoard), + }; + } + } + + /// Give a group that just got its first queued item a turn. fn activate(&mut self, group: Option) { let Some(group) = group else { return; @@ -470,11 +783,11 @@ impl Schedule { fn free_group(&mut self, group: usize) { let state = &self.groups[group]; - if !state.closed || state.members > 0 { + if !state.closed || state.members > 0 || !state.queue.datagrams.is_empty() { return; } - // No members means nothing queued, but a stale round-robin entry may remain. + // Nothing names or fills the group, but a stale round-robin entry may remain. let state = self.groups.remove(group); let domain = &mut self.domains[state.domain]; if state.ringed { @@ -516,9 +829,10 @@ mod tests { use super::*; - /// Streams with a fixed number of bytes each, drained a frame at a time. + /// Streams with a fixed number of bytes each, drained a frame at a time, and datagrams + /// sent whole. Datagrams carry their id and size. struct Sim { - schedule: Schedule, + schedule: Schedule, streams: HashMap, next: u64, fair: bool, @@ -554,9 +868,30 @@ mod tests { id } - /// Send one frame of up to `frame` bytes, returning the stream it came from. + /// Queue a datagram of `bytes` within `budget`, returning its id and whether it stayed. + fn datagram( + &mut self, + group: Option, + order: i64, + bytes: usize, + budget: usize, + ) -> (u64, bool) { + let id = self.next; + self.next += 1; + let queued = self + .schedule + .push_datagram(group, order, (id, bytes), bytes, budget); + (id, queued) + } + + /// Send one frame of up to `frame` bytes or one datagram, returning its id. fn step(&mut self, frame: usize) -> Option { - let id = self.schedule.pop()?; + let id = match self.schedule.pop()? { + Next::Stream(id) => id, + Next::Datagram(group) => { + return Some(self.schedule.sent_datagram(group).0); + } + }; let stream = self.streams.get_mut(&id).unwrap(); let bytes = frame.min(stream.remaining); stream.remaining -= bytes; @@ -569,6 +904,11 @@ mod tests { } Some(id) } + + /// Send everything left, returning the ids in order. + fn drain(&mut self) -> Vec { + std::iter::from_fn(|| self.step(1000)).collect() + } } #[test] @@ -657,4 +997,138 @@ mod tests { let ids: Vec<_> = (0..4).map(|_| sim.step(1000).unwrap()).collect(); assert!(ids.contains(&first) && ids.contains(&second), "{ids:?}"); } + + #[test] + fn datagrams_follow_their_order_and_win_a_tie() { + let mut sim = Sim::new(); + let group = sim.schedule.open_group(0, None); + let older = sim.datagram(Some(group), 10, 100, usize::MAX).0; + let stream = sim.stream(Some(group), 11, 100); + let tied = sim.stream(Some(group), 10, 100); + let first = sim.datagram(Some(group), 10, 100, usize::MAX).0; + let second = sim.datagram(Some(group), 10, 100, usize::MAX).0; + + // The newer group first, then its tie: datagrams oldest first, ahead of the stream. + assert_eq!(sim.drain(), [stream, older, first, second, tied]); + assert!(sim.schedule.is_empty()); + assert_eq!(sim.schedule.datagram_bytes(), 0); + } + + #[test] + fn a_closed_group_sends_its_datagrams_then_frees() { + let mut sim = Sim::new(); + let group = sim.schedule.open_group(0, None); + let id = sim.datagram(Some(group), 0, 100, usize::MAX).0; + sim.schedule.close_group(group); + assert!(!sim.schedule.is_clean()); + + assert_eq!(sim.drain(), [id]); + assert!(sim.schedule.is_clean()); + assert!(sim.schedule.groups.is_empty()); + } + + #[test] + fn eviction_takes_the_lowest_priority_oldest_datagram() { + let mut sim = Sim::new(); + let high = sim.schedule.open_group(1, None); + let low = sim.schedule.open_group(0, None); + let low_old = sim.datagram(Some(low), 0, 100, 400).0; + let high_old = sim.datagram(Some(high), 0, 100, 400).0; + // Newer but higher order: eviction goes by age, not order. + let low_new = sim.datagram(Some(low), 5, 100, 400).0; + let ungrouped = sim.datagram(None, 0, 100, 400).0; + + // Full: the next evicts the low group's oldest, then its newest. + let (high_new, queued) = sim.datagram(Some(high), 0, 100, 400); + assert!(queued); + assert!(sim.datagram(Some(high), 0, 100, 400).1); + assert_eq!(sim.drain(), [ungrouped, high_old, high_new, high_new + 1]); + let _ = (low_old, low_new); + } + + #[test] + fn the_new_datagram_drops_when_it_is_the_victim() { + let mut sim = Sim::new(); + let high = sim.schedule.open_group(1, None); + let low = sim.schedule.open_group(0, None); + let kept = sim.datagram(Some(high), 0, 100, 100).0; + assert!(!sim.datagram(Some(low), 0, 100, 100).1); + assert_eq!(sim.schedule.datagram_bytes(), 100); + assert!(sim.schedule.is_clean()); + assert_eq!(sim.drain(), [kept]); + } + + #[test] + fn eviction_stays_inside_the_busiest_domain() { + let mut sim = Sim::new(); + let quiet = sim.schedule.open_domain(); + let busy = sim.schedule.open_domain(); + let quiet_low = sim.schedule.open_group(0, Some(quiet)); + let busy_high = sim.schedule.open_group(9, Some(busy)); + + // The quiet domain's low-priority datagram survives a busy high-priority domain. + let survivor = sim.datagram(Some(quiet_low), 0, 100, 500).0; + let mut busy_ids = Vec::new(); + for _ in 0..10 { + busy_ids.push(sim.datagram(Some(busy_high), 0, 100, 500).0); + } + let sent = sim.drain(); + assert!(sent.contains(&survivor), "{sent:?}"); + // The busy domain kept its newest four. + assert_eq!(sent.len(), 5); + assert!(busy_ids[6..].iter().all(|id| sent.contains(id)), "{sent:?}"); + } + + #[test] + fn domains_tied_on_bytes_evict_from_the_older() { + let mut sim = Sim::new(); + let groups: Vec<_> = (0..3) + .map(|_| { + let domain = sim.schedule.open_domain(); + sim.schedule.open_group(0, Some(domain)) + }) + .collect(); + let ids: Vec<_> = groups + .iter() + .map(|&group| sim.datagram(Some(group), 0, 100, 250)) + .collect(); + // Each domain holds 100 bytes, so the one with the oldest datagram gives it up. + assert!(ids.iter().all(|&(_, queued)| queued)); + let sent = sim.drain(); + assert!(!sent.contains(&ids[0].0), "{sent:?}"); + assert_eq!(sent.len(), 2); + } + + #[test] + fn a_deferred_datagram_keeps_its_place() { + let mut sim = Sim::new(); + let group = sim.schedule.open_group(0, None); + let large = sim.datagram(Some(group), 2, 1000, usize::MAX).0; + let small = sim.datagram(Some(group), 1, 10, usize::MAX).0; + let stream = sim.stream(Some(group), 0, 100); + + // Only 100 bytes of room: the large datagram steps aside for what fits behind it. + assert_eq!(sim.schedule.pop(), Some(Next::Datagram(Some(group)))); + sim.schedule.defer_datagram(Some(group)); + assert_eq!(sim.step(1000), Some(small)); + assert_eq!(sim.step(1000), Some(stream)); + assert_eq!(sim.step(1000), None); + sim.schedule.restore(); + assert!(!sim.schedule.can_send(999)); + assert!(sim.schedule.can_send(1000)); + assert_eq!(sim.drain(), [large]); + } + + #[test] + fn retained_datagrams_drop_what_no_longer_fits() { + let mut sim = Sim::new(); + let group = sim.schedule.open_group(0, None); + let small = sim.datagram(Some(group), 0, 10, usize::MAX).0; + sim.datagram(Some(group), 0, 1000, usize::MAX); + sim.datagram(None, 0, 1000, usize::MAX); + assert!(sim.schedule.retain_datagrams(|&(_, bytes)| bytes <= 100)); + assert!(!sim.schedule.retain_datagrams(|&(_, bytes)| bytes <= 100)); + assert_eq!(sim.schedule.datagram_bytes(), 10); + assert_eq!(sim.drain(), [small]); + } } diff --git a/rs/moq-quic/src/connection/streams/state.rs b/rs/moq-quic/src/connection/streams/state.rs index fb6d2895ca..6c1f57c8d9 100644 --- a/rs/moq-quic/src/connection/streams/state.rs +++ b/rs/moq-quic/src/connection/streams/state.rs @@ -2,6 +2,7 @@ use std::{ collections::{VecDeque, hash_map}, convert::TryFrom, mem, + ops::Range, }; use bytes::BufMut; @@ -9,8 +10,8 @@ use rustc_hash::FxHashMap; use tracing::{debug, trace}; use super::{ - Credit, Recv, Retransmits, Schedule, Send, SendState, ShouldTransmit, StreamEvent, StreamHalf, - ThinRetransmits, + Credit, Next, Recv, Retransmits, Schedule, Send, SendState, ShouldTransmit, StreamEvent, + StreamHalf, ThinRetransmits, }; use crate::{ Dir, MAX_STREAM_COUNT, Side, StreamId, TransportError, VarInt, @@ -21,6 +22,15 @@ use crate::{ transport_parameters::TransportParameters, }; +/// The frames [`StreamsState::write_frames`] wrote into a packet. +#[derive(Debug, Default)] +pub(crate) struct WrittenFrames { + /// The STREAM frames, which are retransmitted when lost. + pub(crate) streams: StreamMetaVec, + /// How many DATAGRAM frames, which are not. + pub(crate) datagrams: u64, +} + /// Wrapper around `Recv` that facilitates reusing `Recv` instances #[derive(Debug)] pub(super) enum StreamRecv { @@ -100,8 +110,8 @@ pub(crate) struct StreamsState { /// This differs from `self.send.len()` in that it does not include streams that the peer is /// permitted to open but which have not yet been opened. pub(super) send_streams: usize, - /// Streams with outgoing data queued, in the order to send them - pub(super) pending: Schedule, + /// Streams with outgoing data queued and outgoing datagrams, in the order to send them + pub(super) pending: Schedule, events: VecDeque, /// Streams blocked on connection-level flow control or stream window space @@ -439,11 +449,18 @@ impl StreamsState { } } - /// Whether any stream data is queued, regardless of control frames + /// Whether any stream data or datagram is queued, regardless of control frames + #[cfg(test)] pub(crate) fn can_send_stream_data(&self) -> bool { !self.pending.is_empty() } + /// Whether [`Self::write_frames`] would write a frame into `space` bytes: any queued + /// stream, or a queued datagram whose frame fits + pub(crate) fn can_send_frames(&self, space: usize) -> bool { + self.pending.can_send(space) + } + /// Whether MAX_STREAM_DATA frames could be sent for stream `id` pub(crate) fn can_send_flow_control(&self, id: StreamId) -> bool { self.recv @@ -652,25 +669,42 @@ impl StreamsState { } } - pub(crate) fn write_stream_frames( + /// Write STREAM and DATAGRAM frames in schedule order until the packet is full. + /// + /// `payload` is the packet's frame space in `buf`, which may already hold control frames. + /// A datagram cannot be split, so one too large for the room left ends the packet and + /// leads the next one. Only when it would leave the packet empty, which a packet clamped + /// below the path MTU can, does it step aside for what fits behind it, since + /// [`Self::can_send_frames`] promised a frame. + pub(crate) fn write_frames( &mut self, buf: &mut Vec, - max_buf_size: usize, + payload: Range, fair: bool, - ) -> StreamMetaVec { - let mut stream_frames = StreamMetaVec::new(); + ) -> WrittenFrames { + let max_buf_size = payload.end; + let mut written = WrittenFrames::default(); while buf.len() + frame::Stream::SIZE_BOUND < max_buf_size { - if max_buf_size - .checked_sub(buf.len() + frame::Stream::SIZE_BOUND) - .is_none() - { - break; - } - - // Pop the next stream to send from. A stream leaves the schedule once it has - // nothing left to send, so a reset one stays only for data below its Reliable Size. - let Some(id) = self.pending.pop() else { - break; + // Pop the next frame to send. A stream leaves the schedule once it has nothing + // left to send, so a reset one stays only for data below its Reliable Size. + let id = match self.pending.pop() { + None => break, + Some(Next::Stream(id)) => id, + Some(Next::Datagram(group)) => { + let (_, frame) = self.pending.datagram(group); + if buf.len() + frame > max_buf_size { + if buf.len() > payload.start { + break; + } + self.pending.defer_datagram(group); + continue; + } + let datagram = self.pending.sent_datagram(group); + trace!(len = datagram.data.len(), "DATAGRAM"); + datagram.encode(true, buf); + written.datagrams += 1; + continue; + } }; let stream = self .send @@ -709,10 +743,11 @@ impl StreamsState { let requeue = stream.is_pending().then_some(stream.priority); self.pending .sent(id, &mut stream.slot, buf.len() - start, requeue, fair); - stream_frames.push(meta); + written.streams.push(meta); } - stream_frames + self.pending.restore(); + written } /// Process the acknowledgement of stream data @@ -1675,7 +1710,7 @@ mod tests { high.write(b"high").unwrap(); let mut buf = Vec::with_capacity(40); - let meta = server.write_stream_frames(&mut buf, 40, true); + let meta = server.write_frames(&mut buf, 0..40, true).streams; assert_eq!(meta[0].id, id_high); assert_eq!(meta[1].id, id_mid); assert_eq!(meta[2].id, id_low); @@ -1713,7 +1748,7 @@ mod tests { assert_eq!(low.write(b"low").unwrap(), 3); let mut buf = Vec::with_capacity(40); - let mut sent = server.write_stream_frames(&mut buf, 40, fair); + let mut sent = server.write_frames(&mut buf, 0..40, fair).streams; assert_eq!(sent.len(), 1); let lost = sent.pop().unwrap(); assert_eq!(lost.id, id_low); @@ -1735,7 +1770,7 @@ mod tests { assert_eq!(high.write(b"high").unwrap(), 4); buf.clear(); - let sent = server.write_stream_frames(&mut buf, 40, fair); + let sent = server.write_frames(&mut buf, 0..40, fair).streams; assert_eq!(sent.len(), 2); assert_eq!(sent[0].id, id_high); assert_eq!(sent[0].offsets, 0..4); @@ -1796,7 +1831,7 @@ mod tests { high.set_priority(-1).unwrap(); let mut buf = Vec::with_capacity(1000); - let meta = server.write_stream_frames(&mut buf, 40, true); + let meta = server.write_frames(&mut buf, 0..40, true).streams; assert_eq!(meta.len(), 1); assert_eq!(meta[0].id, id_high); @@ -1804,7 +1839,7 @@ mod tests { assert_eq!(server.pending.len(), 2); // Send the remaining data. The initial mid priority one should go first now - let meta = server.write_stream_frames(&mut buf, 1000, true); + let meta = server.write_frames(&mut buf, 0..1000, true).streams; assert_eq!(meta.len(), 2); assert_eq!(meta[0].id, id_mid); assert_eq!(meta[1].id, id_high); @@ -1865,7 +1900,9 @@ mod tests { // loop until all the streams are written loop { let buf_len = buf.len(); - let meta = server.write_stream_frames(&mut buf, buf_len + 40, fair); + let meta = server + .write_frames(&mut buf, buf_len..buf_len + 40, fair) + .streams; if meta.is_empty() { break; } @@ -1936,7 +1973,9 @@ mod tests { // Write the first chunk of stream_a let buf_len = buf.len(); - let meta = server.write_stream_frames(&mut buf, buf_len + 40, false); + let meta = server + .write_frames(&mut buf, buf_len..buf_len + 40, false) + .streams; assert!(!meta.is_empty()); metas.extend(meta); @@ -1953,7 +1992,9 @@ mod tests { // loop until all the streams are written loop { let buf_len = buf.len(); - let meta = server.write_stream_frames(&mut buf, buf_len + 40, false); + let meta = server + .write_frames(&mut buf, buf_len..buf_len + 40, false) + .streams; if meta.is_empty() { break; } @@ -2621,7 +2662,7 @@ mod tests { } } - fn receive(state: &mut StreamsState, id: StreamId, data: std::ops::Range) { + fn receive(state: &mut StreamsState, id: StreamId, data: Range) { let len = (data.end - data.start) as usize; let _ = state .received( @@ -2876,7 +2917,7 @@ mod tests { }; stream.write(b"hello world").unwrap(); let mut buf = Vec::new(); - let sent = stream.state.write_stream_frames(&mut buf, 100, true); + let sent = stream.state.write_frames(&mut buf, 0..100, true).streams; stream.reset_to(7u32.into(), 4).unwrap(); assert!(pending.reset_stream.is_empty()); assert_eq!(server.buffered_data, 4); @@ -2884,7 +2925,7 @@ mod tests { // The prefix is lost and resent, without what followed it server.retransmit(sent[0].clone()); buf.clear(); - let resent = server.write_stream_frames(&mut buf, 100, true); + let resent = server.write_frames(&mut buf, 0..100, true).streams; assert_eq!(resent[0].offsets, 0..4); assert_eq!(server.received_ack_of(resent[0].clone()), Some(4)); assert_eq!(server.buffered_data, 0); @@ -2949,7 +2990,7 @@ mod tests { }; stream.write(b"hello world").unwrap(); let mut buf = Vec::new(); - let sent = stream.state.write_stream_frames(&mut buf, 100, true); + let sent = stream.state.write_frames(&mut buf, 0..100, true).streams; stream.reset_to(7u32.into(), 4).unwrap(); server.reset_acked(id, 4); @@ -3025,7 +3066,7 @@ mod tests { }; stream.write(b"hello world").unwrap(); let mut buf = Vec::new(); - let sent = stream.state.write_stream_frames(&mut buf, 100, true); + let sent = stream.state.write_frames(&mut buf, 0..100, true).streams; assert_eq!(sent[0].offsets, 0..11); stream.reset_to(7u32.into(), 4).unwrap(); assert_eq!(server.buffered_data, 4); @@ -3033,7 +3074,7 @@ mod tests { // Only the lost prefix is retransmitted, without the FIN of a stream that was never finished server.retransmit(sent[0].clone()); buf.clear(); - let resent = server.write_stream_frames(&mut buf, 100, true); + let resent = server.write_frames(&mut buf, 0..100, true).streams; assert_eq!(resent.len(), 1); assert_eq!(resent[0].offsets, 0..4); assert!(!resent[0].fin); diff --git a/rs/moq-quic/src/fuzz/connection.rs b/rs/moq-quic/src/fuzz/connection.rs index 63cdd15a40..6c419ad0bc 100644 --- a/rs/moq-quic/src/fuzz/connection.rs +++ b/rs/moq-quic/src/fuzz/connection.rs @@ -138,7 +138,7 @@ enum Op { Datagram { who: Who, len: u16, - drop: bool, + order: i8, }, Ping { who: Who, @@ -554,7 +554,7 @@ impl Pair { conn.set_send_window(window.into()); } } - Op::Datagram { who, len, drop } => { + Op::Datagram { who, len, order } => { let peer = self.peer(who); let seq = peer.datagrams_sent.len() as u64; let Some(conn) = peer.conn() else { @@ -566,7 +566,7 @@ impl Pair { .chain((0..len).map(|i| datagram_byte(seq, i.into()))) .collect(); let size = data.len(); - if conn.datagrams().send(data, drop).is_ok() { + if conn.datagrams().send(data, None, order.into()).is_ok() { peer.datagrams_sent.push(size); } } @@ -820,7 +820,7 @@ pub(super) fn datagrams() -> Vec> { Op::Datagram { who: Who::Server, len: 10, - drop: false, + order: 0, }, Op::KeyUpdate { who: Who::Client }, Op::Ping { who: Who::Client }, diff --git a/rs/moq-quic/src/tests/mod.rs b/rs/moq-quic/src/tests/mod.rs index 28ebabca77..ac9fcdb255 100644 --- a/rs/moq-quic/src/tests/mod.rs +++ b/rs/moq-quic/src/tests/mod.rs @@ -2865,7 +2865,7 @@ fn tail_loss_small_segment_size() { info!("Sending datagram batch"); for _ in 0..DGRAM_NUM { pair.client_datagrams(client_ch) - .send(vec![0; DGRAM_LEN].into(), false) + .send(vec![0; DGRAM_LEN].into(), None, 0) .unwrap(); } @@ -2918,7 +2918,7 @@ fn tail_loss_probe_keeps_ping_when_datagram_does_not_fit() { let datagram_len = pair.client_datagrams(client_ch).max_size().unwrap(); assert!(datagram_len > INITIAL_MTU as usize); pair.client_datagrams(client_ch) - .send(vec![0; datagram_len].into(), false) + .send(vec![0; datagram_len].into(), None, 0) .unwrap(); pair.drive(); @@ -2965,7 +2965,7 @@ fn tail_loss_respect_max_datagrams() { info!("Sending datagram batch"); for _ in 0..DGRAM_NUM { pair.client_datagrams(client_ch) - .send(vec![0; DGRAM_LEN].into(), false) + .send(vec![0; DGRAM_LEN].into(), None, 0) .unwrap(); } @@ -2986,7 +2986,7 @@ fn datagram_send_recv() { const DATA: &[u8] = b"whee"; pair.client_datagrams(client_ch) - .send(DATA.into(), true) + .send(DATA.into(), None, 0) .unwrap(); pair.drive(); assert_matches!( @@ -2997,6 +2997,54 @@ fn datagram_send_recv() { assert_matches!(pair.server_datagrams(server_ch).recv(), None); } +/// Writes more than a congestion window to a stream in a group at `stream`, queues datagrams in +/// a group at `datagram`, and returns how many datagrams the first flight carried. +fn first_flight_datagrams(stream: i64, datagram: i64) -> usize { + const DATAGRAMS: usize = 8; + let _guard = subscribe(); + let mut pair = Pair::default(); + let (client_ch, server_ch) = pair.connect(); + let conn = pair.server_conn_mut(server_ch); + let streams = conn.send_groups().open(stream, None); + let datagrams = conn.send_groups().open(datagram, None); + let s = conn.streams().open(Dir::Uni).unwrap(); + conn.send_stream(s).set_group(Some(&streams)).unwrap(); + conn.send_stream(s).write(&[0; 256 * 1024]).unwrap(); + for _ in 0..DATAGRAMS { + conn.datagrams() + .send(vec![1; 100].into(), Some(&datagrams), 0) + .unwrap(); + } + + // One congestion-limited flight. + pair.drive_server(); + pair.drive_client(); + assert!(pair.client_streams(client_ch).accept(Dir::Uni).is_some()); + let mut first = 0; + while pair.client_datagrams(client_ch).recv().is_some() { + first += 1; + } + + // Nothing is dropped, only ordered. + pair.drive(); + let mut rest = 0; + while pair.client_datagrams(client_ch).recv().is_some() { + rest += 1; + } + assert_eq!(first + rest, DATAGRAMS); + first +} + +#[test] +fn low_priority_datagrams_wait_behind_streams() { + assert_eq!(first_flight_datagrams(1, 0), 0); +} + +#[test] +fn high_priority_datagrams_preempt_streams() { + assert_eq!(first_flight_datagrams(0, 1), 8); +} + #[test] fn datagram_recv_buffer_overflow() { let _guard = subscribe(); @@ -3023,13 +3071,13 @@ fn datagram_recv_buffer_overflow() { const DATA2: &[u8] = &[0xBC; (PAYLOAD_WINDOW / 3) + 1]; const DATA3: &[u8] = &[0xCD; (PAYLOAD_WINDOW / 3) + 1]; pair.client_datagrams(client_ch) - .send(DATA1.into(), true) + .send(DATA1.into(), None, 0) .unwrap(); pair.client_datagrams(client_ch) - .send(DATA2.into(), true) + .send(DATA2.into(), None, 0) .unwrap(); pair.client_datagrams(client_ch) - .send(DATA3.into(), true) + .send(DATA3.into(), None, 0) .unwrap(); pair.drive(); assert_matches!( @@ -3041,7 +3089,7 @@ fn datagram_recv_buffer_overflow() { assert_matches!(pair.server_datagrams(server_ch).recv(), None); pair.client_datagrams(client_ch) - .send(DATA1.into(), true) + .send(DATA1.into(), None, 0) .unwrap(); pair.drive(); assert_eq!(pair.server_datagrams(server_ch).recv().unwrap(), DATA1); @@ -3060,12 +3108,12 @@ fn datagram_larger_than_send_buffer_is_too_large() { assert_matches!( pair.client_datagrams(client_ch) - .send(Bytes::from_static(&[0; 2]), true), + .send(Bytes::from_static(&[0; 2]), None, 0), Err(SendDatagramError::TooLarge) ); assert_matches!( pair.client_datagrams(client_ch) - .send(Bytes::from_static(&[0; 2]), false), + .send(Bytes::from_static(&[0; 2]), None, 0), Err(SendDatagramError::TooLarge) ); } @@ -3085,7 +3133,7 @@ fn datagram_unsupported() { assert_matches!(pair.server_conn_mut(server_ch).poll(), None); assert_matches!(pair.client_datagrams(client_ch).max_size(), None); - match pair.client_datagrams(client_ch).send(Bytes::new(), true) { + match pair.client_datagrams(client_ch).send(Bytes::new(), None, 0) { Err(SendDatagramError::UnsupportedByPeer) => {} Err(e) => panic!("unexpected error: {e}"), Ok(_) => panic!("unexpected success"), @@ -4378,7 +4426,7 @@ fn pure_sender_voluntarily_acks() { for _ in 0..100 { const MSG: &[u8] = b"hello"; pair.client_datagrams(client_ch) - .send(Bytes::from_static(MSG), true) + .send(Bytes::from_static(MSG), None, 0) .unwrap(); pair.drive(); assert_eq!(pair.server_datagrams(server_ch).recv().unwrap(), MSG); @@ -4509,7 +4557,7 @@ fn datagram_gso() { // Sending ack-eliciting packet from server let client send ACK, which prevents // sending bundled ACK for a while. pair.server_datagrams(server_ch) - .send(Bytes::new(), false) + .send(Bytes::new(), None, 0) .unwrap(); pair.drive(); @@ -4523,7 +4571,7 @@ fn datagram_gso() { const DATAGRAMS: usize = 10; for _ in 0..DATAGRAMS { pair.client_datagrams(client_ch) - .send(Bytes::from_static(&[0; DATAGRAM_LEN]), false) + .send(Bytes::from_static(&[0; DATAGRAM_LEN]), None, 0) .unwrap(); } pair.drive(); @@ -4552,7 +4600,7 @@ fn gso_truncation() { const SIZES: [usize; 3] = [1024, 768, 768]; for len in SIZES { pair.client_datagrams(client_ch) - .send(vec![0; len].into(), false) + .send(vec![0; len].into(), None, 0) .unwrap(); } pair.drive(); @@ -4617,7 +4665,7 @@ fn min_mtu_does_not_pad_application_data() { let message = Bytes::from_static(b"hello"); pair.client_datagrams(client_ch) - .send(message.clone(), false) + .send(message.clone(), None, 0) .unwrap(); pair.client.drive(pair.time, pair.server.addr); assert_eq!(pair.client.outbound.len(), 1); @@ -4802,10 +4850,10 @@ fn pad_to_mtu() { const LEN_1: usize = 800; const LEN_2: usize = 600; pair.client_datagrams(client_ch) - .send(vec![0; LEN_1].into(), false) + .send(vec![0; LEN_1].into(), None, 0) .unwrap(); pair.client_datagrams(client_ch) - .send(vec![0; LEN_2].into(), false) + .send(vec![0; LEN_2].into(), None, 0) .unwrap(); pair.client.drive(pair.time, pair.server.addr); @@ -4861,7 +4909,7 @@ fn large_datagram_with_acks() { let max_size = pair.client_datagrams(client_ch).max_size().unwrap(); let msg = Bytes::from(vec![0; max_size]); pair.client_datagrams(client_ch) - .send(msg.clone(), true) + .send(msg.clone(), None, 0) .unwrap(); let initial_datagrams = pair.client_conn_mut(client_ch).stats().udp_tx.datagrams; pair.drive(); @@ -4886,7 +4934,7 @@ fn voluntary_ack_with_large_datagrams() { for _ in 0..COUNT { let max_size = pair.client_datagrams(client_ch).max_size().unwrap(); pair.client_datagrams(client_ch) - .send(vec![0; max_size].into(), true) + .send(vec![0; max_size].into(), None, 0) .unwrap(); pair.drive(); } @@ -4908,7 +4956,7 @@ fn ack_bundled_with_datagrams() { // Send packet from client and then send from server. the packet from server should include ACKs pair.client_datagrams(client_ch) - .send(vec![0; 1].into(), false) + .send(vec![0; 1].into(), None, 0) .unwrap(); pair.drive_client(); pair.drive_server(); @@ -4918,7 +4966,7 @@ fn ack_bundled_with_datagrams() { pair.server_conn_mut(server_ch).stats().udp_tx.datagrams; pair.server_datagrams(server_ch) - .send(vec![0; 1].into(), false) + .send(vec![0; 1].into(), None, 0) .unwrap(); pair.drive_server(); @@ -4956,12 +5004,10 @@ fn path_changes_unblock_oversized_datagrams() { assert!(old_max > 1200); let empty_space = pair.server_datagrams(server_ch).send_buffer_space(); let data = Bytes::from(vec![42; old_max]); - loop { - match pair.server_datagrams(server_ch).send(data.clone(), false) { - Ok(()) => {} - Err(SendDatagramError::Blocked(_)) => break, - Err(error) => panic!("unexpected send error: {error}"), - } + while pair.server_datagrams(server_ch).send_buffer_space() >= old_max { + pair.server_datagrams(server_ch) + .send(data.clone(), None, 0) + .unwrap(); } while pair.server_conn_mut(server_ch).poll().is_some() {} @@ -4998,14 +5044,10 @@ fn path_changes_unblock_oversized_datagrams() { pair.server_datagrams(server_ch).send_buffer_space(), empty_space ); - assert!( - iter::from_fn(|| pair.server_conn_mut(server_ch).poll()) - .any(|event| matches!(event, Event::DatagramsUnblocked)) - ); let small = Bytes::from_static(b"small"); pair.server_datagrams(server_ch) - .send(small.clone(), false) + .send(small.clone(), None, 0) .unwrap(); pair.drive(); assert_eq!(pair.client_datagrams(client_ch).recv(), Some(small)); @@ -5013,9 +5055,9 @@ fn path_changes_unblock_oversized_datagrams() { } } -/// Verify that dropping oversized datagrams will trigger a DatagramsUnblocked event. +/// Verify that datagrams that no longer fit the path MTU are dropped from the send buffer. #[test] -fn oversized_datagrams_trigger_unblock() { +fn oversized_datagrams_are_dropped() { let _guard = subscribe(); let mut pair = Pair::default(); // Start the connection with a large MTU. @@ -5033,44 +5075,27 @@ fn oversized_datagrams_trigger_unblock() { // Send datagrams until the send buffer is full. let max_size = pair.client_datagrams(client_ch).max_size().unwrap(); let data = vec![0; max_size]; - loop { - match pair - .client_datagrams(client_ch) - .send(data.clone().into(), false) - { - Ok(_) => {} - Err(SendDatagramError::Blocked(_)) => { - break; - } - Err(e) => panic!("unexpected error: {e}"), - } + while pair.client_datagrams(client_ch).send_buffer_space() >= max_size { + pair.client_datagrams(client_ch) + .send(data.clone().into(), None, 0) + .unwrap(); } // Set the MTU to a smaller value so the queued datagrams cannot be sent. pair.mtu = 1200; // Drive the pair until black hole detection kicks in and the path MTU is adjusted. while pair.step() { - let err = loop { - if let Err(e) = pair + // Keep the buffer full until the MTU is adjusted and the datagram no longer fits. + let mut sent = Ok(()); + while sent.is_ok() && pair.client_datagrams(client_ch).send_buffer_space() >= max_size { + sent = pair .client_datagrams(client_ch) - .send(data.clone().into(), false) - { - break e; - } - }; - match err { - SendDatagramError::Blocked(_) => { - // continue with the next step but drain the DatagramsUnblocked events - // emitted datagrams were sent out. - while let Some(event) = pair.client_conn_mut(client_ch).poll() { - tracing::info!("ignoring connection event: {event:?}"); - } - } - SendDatagramError::TooLarge => { - // mtu adjusted, break the loop - break; - } - _ => panic!("unexpected error: {err}"), + .send(data.clone().into(), None, 0); + } + match sent { + Ok(()) => {} + Err(SendDatagramError::TooLarge) => break, + Err(err) => panic!("unexpected error: {err}"), } } @@ -5085,13 +5110,9 @@ fn oversized_datagrams_trigger_unblock() { assert_eq!( pair.client_datagrams(client_ch).send_buffer_space(), - send_buffer_size - size_of::(), + send_buffer_size - Datagram::SIZE_BOUND, "expected the send buffer to be empty after too large datagrams were dropped", ); - match pair.client_conn_mut(client_ch).poll() { - Some(Event::DatagramsUnblocked) => {} - _ => panic!("expected DatagramsUnblocked event"), - } } #[test] @@ -5991,7 +6012,7 @@ fn stream_starvation_reported_before_resumed_send() { fn datagram_starvation_reported_before_resumed_send() { check_starvation_reported_before_resumed_send(|pair, ch| { pair.client_datagrams(ch) - .send(Bytes::from_static(&[42; 1000]), true) + .send(Bytes::from_static(&[42; 1000]), None, 0) .unwrap(); }); } diff --git a/rs/moq-tokio/src/quic/driver/connection.rs b/rs/moq-tokio/src/quic/driver/connection.rs index d5e80f4bfd..8de19c7613 100644 --- a/rs/moq-tokio/src/quic/driver/connection.rs +++ b/rs/moq-tokio/src/quic/driver/connection.rs @@ -521,27 +521,36 @@ impl Connection { } } - /// Transmit `data` as an unreliable, unordered application datagram + /// Transmit `data` as an unreliable, unordered application datagram in `group` at `order` /// /// Application datagrams are a low-level primitive. They may be lost or delivered out of order, /// and `data` must both fit inside a single QUIC packet and be smaller than the maximum /// dictated by the peer. /// - /// Previously queued datagrams which are still unsent may be discarded to make space for this - /// datagram, in order of oldest to newest. - pub fn send_datagram(&self, data: Bytes) -> Result<(), SendDatagramError> { + /// The datagram is scheduled like one of the group's streams at priority `order`, or among + /// the ungrouped streams without a group. Sending never blocks: a full buffer drops the + /// oldest datagram of the lowest-priority group, see [`moq_quic::Datagrams::send`]. + pub fn send_datagram( + &self, + data: Bytes, + group: Option<&SendGroup>, + order: i64, + ) -> Result<(), SendDatagramError> { let conn = &mut *self.0.state.lock("send_datagram"); if let Some(x) = &conn.error { return Err(SendDatagramError::ConnectionLost(x.clone())); } use moq_quic::SendDatagramError::*; - match conn.inner.datagrams().send(data, true) { + match conn + .inner + .datagrams() + .send(data, group.map(SendGroup::id), order) + { Ok(()) => { conn.wake(); Ok(()) } Err(e) => Err(match e { - Blocked(..) => unreachable!(), UnsupportedByPeer => SendDatagramError::UnsupportedByPeer, Disabled => SendDatagramError::Disabled, TooLarge => SendDatagramError::TooLarge, @@ -549,22 +558,6 @@ impl Connection { } } - /// Transmit `data` as an unreliable, unordered application datagram - /// - /// Unlike [`send_datagram()`], this method will wait for buffer space during congestion - /// conditions, which effectively prioritizes old datagrams over new datagrams. - /// - /// See [`send_datagram()`] for details. - /// - /// [`send_datagram()`]: Connection::send_datagram - pub fn send_datagram_wait(&self, data: Bytes) -> SendDatagram<'_> { - SendDatagram { - conn: &self.0, - data: Some(data), - notify: self.0.shared.datagrams_unblocked.notified(), - } - } - /// Compute the maximum size of datagrams that may be passed to [`send_datagram()`]. /// /// Returns `None` if datagrams are unsupported by the peer or disabled locally. @@ -971,55 +964,6 @@ impl Future for ReadDatagram<'_> { } } -pin_project! { - /// Future produced by [`Connection::send_datagram_wait`] - pub struct SendDatagram<'a> { - conn: &'a ConnectionRef, - data: Option, - #[pin] - notify: Notified<'a>, - } -} - -impl Future for SendDatagram<'_> { - type Output = Result<(), SendDatagramError>; - fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll { - let mut this = self.project(); - let mut state = this.conn.state.lock("SendDatagram::poll"); - if let Some(e) = &state.error { - return Poll::Ready(Err(SendDatagramError::ConnectionLost(e.clone()))); - } - use moq_quic::SendDatagramError::*; - match state - .inner - .datagrams() - .send(this.data.take().unwrap(), false) - { - Ok(()) => { - state.wake(); - Poll::Ready(Ok(())) - } - Err(e) => Poll::Ready(Err(match e { - Blocked(data) => { - this.data.replace(data); - loop { - match this.notify.as_mut().poll(ctx) { - Poll::Pending => return Poll::Pending, - // Spurious wakeup, get a new future - Poll::Ready(()) => this - .notify - .set(this.conn.shared.datagrams_unblocked.notified()), - } - } - } - UnsupportedByPeer => SendDatagramError::UnsupportedByPeer, - Disabled => SendDatagramError::Disabled, - TooLarge => SendDatagramError::TooLarge, - })), - } - } -} - /// Watcher produced by [`Connection::watch_remote_address`] pub struct RemoteAddressWatcher { receiver: watch::Receiver, @@ -1095,7 +1039,6 @@ pub(crate) struct Shared { /// Notified when the peer has initiated a new stream stream_incoming: [Notify; 2], datagram_received: Notify, - datagrams_unblocked: Notify, closed: Notify, connected: Arc, /// Number of live handles that can used to initiate or handle I/O; excludes the driver @@ -1299,9 +1242,6 @@ impl State { DatagramReceived => { shared.datagram_received.notify_waiters(); } - DatagramsUnblocked => { - shared.datagrams_unblocked.notify_waiters(); - } Stream(StreamEvent::Readable { id }) => wake_stream(id, &mut self.blocked_readers), Stream(StreamEvent::Available { dir }) => { // Might mean any number of streams are ready, so we wake up everyone @@ -1383,7 +1323,6 @@ impl State { shared.stream_incoming[Dir::Uni as usize].notify_waiters(); shared.stream_incoming[Dir::Bi as usize].notify_waiters(); shared.datagram_received.notify_waiters(); - shared.datagrams_unblocked.notify_waiters(); shared.handshake_confirmed.notify_waiters(); wake_all_notify(&mut self.stopped); shared.closed.notify_waiters(); diff --git a/rs/moq-tokio/src/quic/driver/mod.rs b/rs/moq-tokio/src/quic/driver/mod.rs index b248a9fe09..a20c51650f 100644 --- a/rs/moq-tokio/src/quic/driver/mod.rs +++ b/rs/moq-tokio/src/quic/driver/mod.rs @@ -69,7 +69,7 @@ pub use moq_quic::{QlogConfig, QlogStream}; pub use crate::quic::driver::connection::{ AcceptBi, AcceptUni, Connecting, Connection, OpenBi, OpenUni, ReadDatagram, - RemoteAddressWatcher, SendDatagram, SendDatagramError, + RemoteAddressWatcher, SendDatagramError, }; pub use crate::quic::driver::endpoint::{Accept, Endpoint, EndpointStats}; pub use crate::quic::driver::incoming::{Incoming, IncomingFuture, RetryError}; diff --git a/rs/moq-tokio/src/quic/driver/send_group.rs b/rs/moq-tokio/src/quic/driver/send_group.rs index 41cd2e0ab4..9e8adc793f 100644 --- a/rs/moq-tokio/src/quic/driver/send_group.rs +++ b/rs/moq-tokio/src/quic/driver/send_group.rs @@ -7,11 +7,12 @@ use crate::quic::driver::connection::ConnectionInner; /// Streams that [join](crate::quic::driver::SendStream::set_group) a group share bytes with /// the connection's other groups: a higher priority group preempts every lower one, and groups /// of equal priority get an equal share of bytes however many streams each has queued. A -/// stream's own priority orders it only against the group's other streams. Streams outside any -/// group go before every group. See [`moq_quic::SendGroups`]. +/// stream's own priority orders it only against the group's other streams, and a datagram sent +/// in the group is ordered among them by its `order`. Streams and datagrams outside any group go +/// before every group. See [`moq_quic::SendGroups`]. /// -/// Clones share one group, released once the last clone is dropped and its streams finish. A -/// group does not keep its connection open. +/// Clones share one group, released once the last clone is dropped and its streams finish +/// and its datagrams are sent. A group does not keep its connection open. #[derive(Debug, Clone)] pub struct SendGroup(Arc); diff --git a/rs/moq-tokio/src/quic/driver/tests.rs b/rs/moq-tokio/src/quic/driver/tests.rs index de22289b70..d7846d30e9 100755 --- a/rs/moq-tokio/src/quic/driver/tests.rs +++ b/rs/moq-tokio/src/quic/driver/tests.rs @@ -1570,9 +1570,9 @@ async fn two_datagram_readers() { x }, async { - server.send_datagram(b"one"[..].into()).unwrap(); + server.send_datagram(b"one"[..].into(), None, 0).unwrap(); done.notified().await; - server.send_datagram_wait(b"two"[..].into()).await.unwrap(); + server.send_datagram(b"two"[..].into(), None, 0).unwrap(); } ); assert!(*a == *b"one" || *b == *b"one"); diff --git a/rs/moq-tokio/src/quic/web/session.rs b/rs/moq-tokio/src/quic/web/session.rs index 0fdcb2fcc9..3e6f84a071 100644 --- a/rs/moq-tokio/src/quic/web/session.rs +++ b/rs/moq-tokio/src/quic/web/session.rs @@ -326,34 +326,18 @@ impl Session { Ok(datagram) } - /// Sends an application datagram to the remote peer. + /// Sends an application datagram to the remote peer in `group` at `order`. /// /// Datagrams are unreliable and may be dropped or delivered out of order. /// The data must be smaller than [`max_datagram_size`](Self::max_datagram_size). - pub fn send_datagram(&self, data: Bytes) -> Result<(), SessionError> { - let result = if !self.header_datagram.is_empty() { - // Unfortunately, we need to allocate/copy each datagram because of the driver API. - // Pls go +1 if you care: https://github.com/quinn-rs/quinn/issues/1724 - let mut buf = BytesMut::with_capacity(self.header_datagram.len() + data.len()); - - // Prepend the datagram with the header indicating the session ID. - buf.extend_from_slice(&self.header_datagram); - buf.extend_from_slice(&data); - - self.conn.send_datagram(buf.into()) - } else { - self.conn.send_datagram(data) - }; - - result.map_err(|e| self.map_error(e))?; - Ok(()) - } - - /// Sends an application datagram to the remote peer, waiting for buffer space if needed. - /// - /// Datagrams are unreliable and may be dropped or delivered out of order. - /// The data must be smaller than [`max_datagram_size`](Self::max_datagram_size). - pub async fn send_datagram_wait(&self, data: Bytes) -> Result<(), SessionError> { + /// See [`Connection::send_datagram`](crate::quic::driver::Connection::send_datagram) for how + /// `group` and `order` schedule it. + pub fn send_datagram( + &self, + data: Bytes, + group: Option<&crate::quic::driver::SendGroup>, + order: i64, + ) -> Result<(), SessionError> { let result = if !self.header_datagram.is_empty() { // Unfortunately, we need to allocate/copy each datagram because of the driver API. // Pls go +1 if you care: https://github.com/quinn-rs/quinn/issues/1724 @@ -363,9 +347,9 @@ impl Session { buf.extend_from_slice(&self.header_datagram); buf.extend_from_slice(&data); - self.conn.send_datagram_wait(buf.into()).await + self.conn.send_datagram(buf.into(), group, order) } else { - self.conn.send_datagram_wait(data).await + self.conn.send_datagram(data, group, order) }; result.map_err(|e| self.map_error(e))?; @@ -1023,7 +1007,7 @@ impl web_transport_trait::Session for Session { } fn send_datagram(&self, data: Bytes) -> Result<(), Self::Error> { - Self::send_datagram(self, data) + Self::send_datagram(self, data, None, 0) } async fn recv_datagram(&self) -> Result { diff --git a/rs/moq-tokio/src/transport.rs b/rs/moq-tokio/src/transport.rs index 06e6b0652e..e908eb41d1 100644 --- a/rs/moq-tokio/src/transport.rs +++ b/rs/moq-tokio/src/transport.rs @@ -206,7 +206,10 @@ where fn poll_send_datagram(&mut self, _cx: &mut Context<'_>, payload: &[u8]) -> Poll> { // The async interface's send is non-blocking (the transport queues or // drops), so there is never a reason to return Pending. - Poll::Ready(self.session.send_datagram(Bytes::copy_from_slice(payload))) + Poll::Ready(web_transport_trait::Session::send_datagram( + &self.session, + Bytes::copy_from_slice(payload), + )) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { @@ -859,6 +862,15 @@ impl Grouped for crate::quic::web::Session { fn send_group(&self, priority: i64) -> Option { Some(Self::send_group(self, priority)) } + + fn send_datagram( + &self, + payload: Bytes, + group: Option<&crate::quic::driver::SendGroup>, + order: i64, + ) -> Result<(), Self::Error> { + Self::send_datagram(self, payload, group, order) + } } #[cfg(feature = "quic")] diff --git a/rs/moq-tokio/src/transport/grouped.rs b/rs/moq-tokio/src/transport/grouped.rs index acdf01cddf..42751e5fcb 100644 --- a/rs/moq-tokio/src/transport/grouped.rs +++ b/rs/moq-tokio/src/transport/grouped.rs @@ -15,6 +15,17 @@ pub trait Grouped: web_transport_trait::Session { let _ = priority; None } + + /// Send a datagram in `group` at `order`; a backend without groups ignores both. + fn send_datagram( + &self, + payload: bytes::Bytes, + group: Option<&::Group>, + order: i64, + ) -> Result<(), Self::Error> { + let _ = (group, order); + web_transport_trait::Session::send_datagram(self, payload) + } } /// A backend send stream that may join its [`Grouped`] session's send groups. diff --git a/rs/moq-tokio/src/transport/owned.rs b/rs/moq-tokio/src/transport/owned.rs index 0c32203738..a61ea36af5 100644 --- a/rs/moq-tokio/src/transport/owned.rs +++ b/rs/moq-tokio/src/transport/owned.rs @@ -63,8 +63,13 @@ impl> poll::Session fn send_group(&mut self, priority: i64) -> Option<::Group> { self.session.send_group(priority) } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { - backend::Session::poll_send_datagram(self, cx, payload).map_err(Error) + fn send_datagram( + &mut self, + payload: Bytes, + group: Option<&::Group>, + order: i64, + ) -> Result<(), Self::Error> { + super::Grouped::send_datagram(&self.session, payload, group, order).map_err(Error) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { backend::Session::poll_recv_datagram(self, cx).map_err(Error) diff --git a/rs/moq-uring/src/quic/driver/connection.rs b/rs/moq-uring/src/quic/driver/connection.rs index 6f3721eef9..272cf985a6 100644 --- a/rs/moq-uring/src/quic/driver/connection.rs +++ b/rs/moq-uring/src/quic/driver/connection.rs @@ -72,7 +72,6 @@ pub(crate) struct State { sends: FxHashMap>, datagram_recv_waiters: kio::WaiterList, - datagram_send_waiters: kio::WaiterList, /// The terminal error, set exactly once; everything fails with it after. closed: Option, @@ -103,7 +102,6 @@ impl State { finishing: FxHashMap::default(), sends: FxHashMap::default(), datagram_recv_waiters: kio::WaiterList::new(), - datagram_send_waiters: kio::WaiterList::new(), closed: None, closed_waiters: kio::WaiterList::new(), dead: false, @@ -139,7 +137,6 @@ impl State { self.accept_uni_waiters.wake(); self.open_waiters.wake(); self.datagram_recv_waiters.wake(); - self.datagram_send_waiters.wake(); for waiters in self.readable.values_mut() { waiters.wake(); } @@ -321,6 +318,25 @@ impl Connection { super::SendGroup::new(self.shared.clone(), priority) } + /// Send a datagram in `group` at `order`, scheduled like one of the group's streams at + /// priority `order`, or among the ungrouped streams without a group. + /// + /// Never blocks: a full buffer drops the oldest datagram of the lowest-priority group, + /// see [`moq_quic::Datagrams::send`]. + pub fn send_datagram(&self, payload: Bytes, group: Option<&super::SendGroup>, order: i64) -> Result<(), Error> { + if let Some(err) = self.shared.closed() { + return Err(err); + } + self.shared + .conn + .borrow_mut() + .datagrams() + .send(payload, group.map(super::SendGroup::id), order) + .map_err(|err| Error::Quic(err.to_string()))?; + self.shared.kick(); + Ok(()) + } + /// The peer's certificate chain in DER, leaf first, or `None` if it /// presented none. /// @@ -570,25 +586,8 @@ impl web_transport_trait::poll::Session for Connection { } } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { - let waiter = self.park.hold(cx); - if let Some(err) = self.shared.closed() { - return Poll::Ready(Err(err)); - } - let payload = Bytes::copy_from_slice(payload); - match self.shared.conn.borrow_mut().datagrams().send(payload, false) { - Ok(()) => { - self.shared.kick(); - Poll::Ready(Ok(())) - } - // The send queue is full; a flush frees space. - Err(moq_quic::SendDatagramError::Blocked(_)) => { - let mut state = self.shared.state.borrow_mut(); - waiter.register(&mut state.datagram_send_waiters); - Poll::Pending - } - Err(err) => Poll::Ready(Err(Error::Quic(err.to_string()))), - } + fn poll_send_datagram(&mut self, _: &mut Context<'_>, payload: &[u8]) -> Poll> { + Poll::Ready(Self::send_datagram(self, Bytes::copy_from_slice(payload), None, 0)) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { @@ -813,7 +812,6 @@ impl Driver { } moq_quic::Event::ConnectionLost { reason } => state.fail(reason.into()), moq_quic::Event::DatagramReceived => state.datagram_recv_waiters.wake(), - moq_quic::Event::DatagramsUnblocked => state.datagram_send_waiters.wake(), moq_quic::Event::HandshakeDataReady => {} moq_quic::Event::Stream(event) => sweep_stream(&mut state, event), // Non-exhaustive upstream; nothing else needs a wakeup. @@ -898,8 +896,6 @@ impl Driver { return Poll::Ready(Err(Error::Io(err.to_string()))); } self.close_staged |= closing; - // A flush frees datagram-send queue space. - self.shared.state.borrow_mut().datagram_send_waiters.wake(); Poll::Ready(Ok(())) } } diff --git a/rs/moq-uring/src/quic/driver/group.rs b/rs/moq-uring/src/quic/driver/group.rs index a90b1e2e97..bbe81976a8 100644 --- a/rs/moq-uring/src/quic/driver/group.rs +++ b/rs/moq-uring/src/quic/driver/group.rs @@ -9,10 +9,12 @@ use super::Shared; /// Streams that [join](super::SendStream::set_group) a group share bytes with the /// connection's other groups: a higher priority group preempts every lower one, and groups of /// equal priority get an equal share of bytes however many streams each has queued. A stream's -/// own priority orders it only against the group's other streams. Streams outside any group go +/// own priority orders it only against the group's other streams, and a datagram sent in the +/// group is ordered among them by its `order`. Streams and datagrams outside any group go /// before every group. See [`moq_quic::SendGroups`]. /// -/// Clones share one group, released once the last clone is dropped and its streams finish. +/// Clones share one group, released once the last clone is dropped and its streams finish +/// and its datagrams are sent. #[derive(Clone)] pub struct SendGroup(Rc); diff --git a/rs/moq-uring/src/quic/web.rs b/rs/moq-uring/src/quic/web.rs index 02222e8c6e..b33dfefd53 100644 --- a/rs/moq-uring/src/quic/web.rs +++ b/rs/moq-uring/src/quic/web.rs @@ -450,6 +450,21 @@ impl Session { self.conn.send_group(priority) } + /// Send a datagram in `group` at `order`, scheduled like one of the group's streams. + /// + /// See [`Connection::send_datagram`]. + pub fn send_datagram(&self, payload: Bytes, group: Option<&super::SendGroup>, order: i64) -> Result<(), Error> { + let Some(web) = &self.web else { + return self.conn.send_datagram(payload, group, order); + }; + let mut framed = BytesMut::with_capacity(web.header_datagram.len() + payload.len()); + framed.extend_from_slice(&web.header_datagram); + framed.extend_from_slice(&payload); + self.conn + .send_datagram(framed.freeze(), group, order) + .map_err(unmap_err) + } + /// Wrap a raw-QUIC connection in a session, with no /// WebTransport layering: streams and datagrams pass through untouched. pub fn raw(conn: Connection) -> Self { @@ -657,14 +672,8 @@ impl web_transport_trait::poll::Session for Session { Poll::Ready(result.map(|()| (send, recv))) } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { - let Some(web) = &self.web else { - return web_transport_trait::poll::Session::poll_send_datagram(&mut self.conn, cx, payload); - }; - let mut framed = Vec::with_capacity(web.header_datagram.len() + payload.len()); - framed.extend_from_slice(&web.header_datagram); - framed.extend_from_slice(payload); - web_transport_trait::poll::Session::poll_send_datagram(&mut self.conn, cx, &framed).map_err(unmap_err) + fn poll_send_datagram(&mut self, _: &mut Context<'_>, payload: &[u8]) -> Poll> { + Poll::Ready(Self::send_datagram(self, Bytes::copy_from_slice(payload), None, 0)) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { diff --git a/rs/moq-uring/src/transport.rs b/rs/moq-uring/src/transport.rs index f4492304b3..e9142e658d 100644 --- a/rs/moq-uring/src/transport.rs +++ b/rs/moq-uring/src/transport.rs @@ -9,6 +9,15 @@ impl Grouped for quic::web::Session { fn send_group(&mut self, priority: i64) -> Option { Some(quic::web::Session::send_group(self, priority)) } + + fn send_datagram( + &mut self, + payload: bytes::Bytes, + group: Option<&quic::SendGroup>, + order: i64, + ) -> Result<(), Self::Error> { + quic::web::Session::send_datagram(self, payload, group, order) + } } impl GroupedSend for quic::web::SendStream { diff --git a/rs/moq-uring/src/transport/adapter.rs b/rs/moq-uring/src/transport/adapter.rs index a09b5e417a..ac0f1bea92 100644 --- a/rs/moq-uring/src/transport/adapter.rs +++ b/rs/moq-uring/src/transport/adapter.rs @@ -59,6 +59,22 @@ pub trait Grouped: backend::Session { let _ = priority; None } + + /// Send a datagram in `group` at `order`; a backend without groups ignores both and + /// drops the datagram when it has no room. + fn send_datagram( + &mut self, + payload: Bytes, + group: Option<&::Group>, + order: i64, + ) -> Result<(), Self::Error> { + let _ = (group, order); + let mut cx = Context::from_waker(std::task::Waker::noop()); + match backend::Session::poll_send_datagram(self, &mut cx, &payload) { + Poll::Ready(res) => res, + Poll::Pending => Ok(()), + } + } } /// A backend send stream that may join its [`Grouped`] session's send groups. @@ -113,8 +129,13 @@ impl + Clone + 'static> pol fn send_group(&mut self, priority: i64) -> Option<::Group> { Grouped::send_group(&mut self.0, priority) } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { - backend::Session::poll_send_datagram(&mut self.0, cx, payload).map_err(Error) + fn send_datagram( + &mut self, + payload: Bytes, + group: Option<&::Group>, + order: i64, + ) -> Result<(), Self::Error> { + Grouped::send_datagram(&mut self.0, payload, group, order).map_err(Error) } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { backend::Session::poll_recv_datagram(&mut self.0, cx).map_err(Error) diff --git a/rs/moq-uring/tests/integration/web.rs b/rs/moq-uring/tests/integration/web.rs index 33783c3c84..daa8fd525b 100644 --- a/rs/moq-uring/tests/integration/web.rs +++ b/rs/moq-uring/tests/integration/web.rs @@ -141,7 +141,9 @@ fn webtransport_echo_end_to_end() { assert_eq!(read_all(&mut recv).await, PAYLOAD, "uni echo"); // Datagram echo. - session.send_datagram(PAYLOAD.to_vec().into()).expect("send datagram"); + session + .send_datagram(PAYLOAD.to_vec().into(), None, 0) + .expect("send datagram"); let echoed = session.read_datagram().await.expect("read datagram"); assert_eq!(&echoed[..], PAYLOAD, "datagram echo"); diff --git a/rs/moq-wasm/src/transport/adapter.rs b/rs/moq-wasm/src/transport/adapter.rs index bb2e27bfdf..46a2c8ff25 100644 --- a/rs/moq-wasm/src/transport/adapter.rs +++ b/rs/moq-wasm/src/transport/adapter.rs @@ -80,8 +80,14 @@ impl + Clone + 'st fn send_group(&mut self, _: i64) -> Option { None } - fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { - backend::Session::poll_send_datagram(&mut self.0, cx, payload).map_err(Error) + /// The backend cannot rank a datagram against a stream, so the group and order are ignored, + /// and a datagram it has no room for is dropped. + fn send_datagram(&mut self, payload: Bytes, _: Option<&poll::Ungrouped>, _: i64) -> Result<(), Self::Error> { + let mut cx = Context::from_waker(std::task::Waker::noop()); + match backend::Session::poll_send_datagram(&mut self.0, &mut cx, &payload) { + Poll::Ready(res) => res.map_err(Error), + Poll::Pending => Ok(()), + } } fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { backend::Session::poll_recv_datagram(&mut self.0, cx).map_err(Error)