From 778d6e21932dcaac9d4e58805074f86e735d7ede Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 16 Oct 2026 19:24:31 -0700 Subject: [PATCH 1/6] quest: claim m2/perf/uring-copies From c91ec99a55d4fb968b85ba7c9699151fbd203835 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:01:07 -0700 Subject: [PATCH 2/6] perf(uring): remove avoidable copies on egress, streams, and datagram receive - Egress: moq-quic encrypts each GSO train straight into the TX pool's buffer instead of a scratch Vec that was then copied. `udp::TxBuf` now derefs to the slot's `Vec` (empty on acquire) and `udp::Transmit` drops `len`: the buffer's length is what goes out. - Stream send: `poll_write_buf` hands moq-quic the chunk via `write_chunks`, so a `Bytes` source is kept by reference. A new `moq_quic::SendStream::write_limit` sizes the chunk first, since bytes taken from a `Buf` cannot be put back. - Stream receive: `poll_read_chunk` returns moq-quic's `Bytes` (or a split of the read-ahead backlog) instead of the trait default's 8 KiB copy. - Datagram receive: one `BytesMut` per completion, split per GRO segment, instead of one allocation per datagram. `udp::Packet::{payload_mut, segments}` lose their only caller and go. Co-Authored-By: Claude Opus 5.5 --- quest/m2/perf/README.md | 1 - quest/m2/perf/uring-copies.md | 25 ----- quest/m2/quic-io-boundary.md | 5 +- rs/moq-quic/src/connection/streams/mod.rs | 19 ++++ rs/moq-quic/src/tests/mod.rs | 7 ++ rs/moq-uring/benches/worker_turns.rs | 3 +- rs/moq-uring/src/quic/driver/connection.rs | 15 +-- rs/moq-uring/src/quic/driver/endpoint.rs | 33 +++--- rs/moq-uring/src/quic/driver/stream.rs | 75 ++++++++++---- rs/moq-uring/src/quic/web.rs | 7 ++ rs/moq-uring/src/udp.rs | 101 ++++++++---------- rs/moq-uring/src/worker.rs | 30 +++--- rs/moq-uring/tests/allocations.rs | 6 +- rs/moq-uring/tests/integration/credit.rs | 115 ++++++++++++++++++++- rs/moq-uring/tests/integration/ecn.rs | 3 +- rs/moq-uring/tests/integration/teardown.rs | 3 +- 16 files changed, 286 insertions(+), 162 deletions(-) delete mode 100644 quest/m2/perf/uring-copies.md diff --git a/quest/m2/perf/README.md b/quest/m2/perf/README.md index f288ab41a2b..e51748dc6ad 100644 --- a/quest/m2/perf/README.md +++ b/quest/m2/perf/README.md @@ -76,7 +76,6 @@ optimization. - [One enter per turn](/quest/m2/perf/uring-one-enter.md) - a parking turn pays one io_uring_enter, submits flush deferred completions, and SQEs per enter is a counter - [Run to quiescence](/quest/m2/perf/uring-quiescence.md) - a received packet's reply is staged in the same turn, under a pass and train budget that keeps the fairness rule - [Ingest batch](/quest/m2/perf/ingest-batch.md) - relay ingest pays one lock, wake, and clock read per burst of whole frames instead of per frame -- [Remove moq-uring copies](/quest/m2/perf/uring-copies.md) - egress, stream send and receive, and datagram receive stop copying where the QUIC core already allows it, with before/after measurements on `moq-quic` - [Shard by default](/quest/m2/perf/shard-default.md) - a saturating shard sweep decides whether an unset `runtime.workers` runs one worker per core on Linux ## Related diff --git a/quest/m2/perf/uring-copies.md b/quest/m2/perf/uring-copies.md deleted file mode 100644 index 77dd67eb337..00000000000 --- a/quest/m2/perf/uring-copies.md +++ /dev/null @@ -1,25 +0,0 @@ -# [S] Remove moq-uring's avoidable copies - -## Goal - -moq-uring stops copying and allocating on paths where the QUIC core already -supports zero-copy, with a measured before/after. - -## Plan - -Found in the 2026-09-30 fork planning; none needs a QUIC API change: - -- Egress: `quic/driver/connection.rs` copies each GSO train from `scratch` into - the TX buffer. Hand the TX pool's `Vec` to `poll_transmit` directly; it - encrypts in place and only reserves when capacity is short. -- Stream send: moq-net's `write_chunk(Bytes)` falls back to the trait's - default `poll_write_buf`, which copies through `poll_write(&[u8])`. Route - it to the core's `write_chunks` so large chunks are kept by reference. -- Stream receive: moq-uring uses the trait's default `poll_read_chunk`, which - allocates an 8 KiB `BytesMut` and copies each read. Return the core's - `Bytes` directly. -- Datagram receive: `endpoint.rs` allocates a `BytesMut` per GRO segment. - Copy once per completion and split it into segments. - -The clock-read fix is [moq-uring on moq-clock](/quest/m1/time/uring.md). -Measure with the echo benches and `just bench`. diff --git a/quest/m2/quic-io-boundary.md b/quest/m2/quic-io-boundary.md index 052c9394868..81f66265b40 100644 --- a/quest/m2/quic-io-boundary.md +++ b/quest/m2/quic-io-boundary.md @@ -12,8 +12,8 @@ those buffers with the ring is Decided in the 2026-09-30 plan: commit to the redesign, but start only after a `moq-quic` profile shows where ingress and egress spend their time. -No current profile is from `moq-quic`, and every other copy is removed without an -API change by [Remove moq-uring copies](/quest/m2/perf/uring-copies.md). +No current profile is from `moq-quic`, and moq-uring already avoids every +other copy the current API allows. Candidates, to be ranked by the profile: @@ -40,4 +40,3 @@ the same workloads. ## Required - [Relay profiling recipe](/quest/m2/perf/lock-profile.md) - the captures that rank the candidates -- [Remove moq-uring copies](/quest/m2/perf/uring-copies.md) - the cheap wins land first, so the profile shows what is left diff --git a/rs/moq-quic/src/connection/streams/mod.rs b/rs/moq-quic/src/connection/streams/mod.rs index 34f3df7dd9d..8756b98837b 100644 --- a/rs/moq-quic/src/connection/streams/mod.rs +++ b/rs/moq-quic/src/connection/streams/mod.rs @@ -233,6 +233,25 @@ impl<'a> SendStream<'a> { self.write_source(&mut BytesArray::from_chunks(data)) } + /// How many bytes a write would accept right now + /// + /// Zero when a write would block or fail. Lets a caller size a chunk before taking it from a + /// buffer it cannot put bytes back into. + pub fn write_limit(&self) -> usize { + if self.conn_state.is_closed() { + return 0; + } + let budget = match self.state.send.get(&self.id) { + None => return 0, + Some(None) => self.state.max_send_data(self.id).into_inner(), + Some(Some(stream)) if !stream.is_writable() || stream.stop_reason.is_some() => { + return 0; + } + Some(Some(stream)) => stream.max_data - stream.offset(), + }; + usize::try_from(budget.min(self.state.write_limit())).unwrap_or(usize::MAX) + } + fn write_source(&mut self, source: &mut B) -> Result { if self.conn_state.is_closed() { trace!(%self.id, "write blocked; connection draining"); diff --git a/rs/moq-quic/src/tests/mod.rs b/rs/moq-quic/src/tests/mod.rs index 28ebabca77e..de10533ea50 100644 --- a/rs/moq-quic/src/tests/mod.rs +++ b/rs/moq-quic/src/tests/mod.rs @@ -2339,7 +2339,9 @@ fn test_flow_control(config: TransportConfig, window_size: usize) { // Stream reset before read let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap(); info!("writing"); + assert_eq!(pair.client_send(client_ch, s).write_limit(), window_size); assert_eq!(pair.client_send(client_ch, s).write(&msg), Ok(window_size)); + assert_eq!(pair.client_send(client_ch, s).write_limit(), 0); assert_eq!( pair.client_send(client_ch, s).write(&msg[window_size..]), Err(WriteError::Blocked) @@ -2347,6 +2349,7 @@ fn test_flow_control(config: TransportConfig, window_size: usize) { pair.drive(); info!("resetting"); pair.client_send(client_ch, s).reset(VarInt(42)).unwrap(); + assert_eq!(pair.client_send(client_ch, s).write_limit(), 0); pair.drive(); let mut recv = pair.server_recv(server_ch, s); @@ -2360,7 +2363,9 @@ fn test_flow_control(config: TransportConfig, window_size: usize) { // Happy path info!("writing"); let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap(); + assert_eq!(pair.client_send(client_ch, s).write_limit(), window_size); assert_eq!(pair.client_send(client_ch, s).write(&msg), Ok(window_size)); + assert_eq!(pair.client_send(client_ch, s).write_limit(), 0); assert_eq!( pair.client_send(client_ch, s).write(&msg[window_size..]), Err(WriteError::Blocked) @@ -2392,7 +2397,9 @@ fn test_flow_control(config: TransportConfig, window_size: usize) { assert_eq!(cursor, window_size); pair.drive(); info!("writing"); + assert_eq!(pair.client_send(client_ch, s).write_limit(), window_size); assert_eq!(pair.client_send(client_ch, s).write(&msg), Ok(window_size)); + assert_eq!(pair.client_send(client_ch, s).write_limit(), 0); assert_eq!( pair.client_send(client_ch, s).write(&msg[window_size..]), Err(WriteError::Blocked) diff --git a/rs/moq-uring/benches/worker_turns.rs b/rs/moq-uring/benches/worker_turns.rs index 9d9e21cc14e..d6715742116 100644 --- a/rs/moq-uring/benches/worker_turns.rs +++ b/rs/moq-uring/benches/worker_turns.rs @@ -40,10 +40,9 @@ fn benchmark(c: &mut criterion::Criterion) { for receiver in &receivers { for _ in 0..burst { let mut tx = sender.acquire().await.expect("send buffer"); - tx[..4].copy_from_slice(b"turn"); + tx.extend_from_slice(b"turn"); tx.send(udp::Transmit { to: receiver.local_addr().expect("addr"), - len: 4, segment: 4, ecn: None, }) diff --git a/rs/moq-uring/src/quic/driver/connection.rs b/rs/moq-uring/src/quic/driver/connection.rs index 6f3721eef9d..e49b484bac4 100644 --- a/rs/moq-uring/src/quic/driver/connection.rs +++ b/rs/moq-uring/src/quic/driver/connection.rs @@ -400,7 +400,6 @@ pub(crate) fn launch( endpoint, key, deadline: owner.timer(), - scratch: Vec::with_capacity(TRAIN_SEGMENTS * SEGMENT), close_staged: false, }; let future = async move { kio::wait(|waiter| driver.poll(waiter)).await }; @@ -715,9 +714,6 @@ struct Driver { endpoint: Weak, key: ConnectionHandle, deadline: crate::Timer, - /// Egress staging: moq-quic writes into a `Vec`, so a train is built - /// here and copied into the socket's registered buffer. - scratch: Vec, /// A CONNECTION_CLOSE has been handed to the socket. Not implied by a /// flush that came back empty: moq-quic paces the close like any other packet, /// so it can sit behind a pacing timer after the application asked for it. @@ -866,31 +862,30 @@ impl Driver { Poll::Pending => return Poll::Pending, }; - let segments = (tx.len() / SEGMENT).min(TRAIN_SEGMENTS); + let segments = (tx.capacity() / SEGMENT).min(TRAIN_SEGMENTS); if segments == 0 { return Poll::Ready(Err(Error::Io(format!( "transmit buffer of {} bytes holds no {SEGMENT} byte segment", - tx.len() + tx.capacity() )))); } - self.scratch.clear(); let (transmit, closing) = { let mut conn = self.shared.conn.borrow_mut(); + // moq-quic encrypts the train straight into the socket's buffer. // Nothing to send, or paced; the buffer returns to the pool on drop. - let Some(transmit) = conn.poll_transmit(moq_clock::Instant::now(), segments, &mut self.scratch) else { + let Some(transmit) = conn.poll_transmit(moq_clock::Instant::now(), segments, &mut tx) else { return Poll::Pending; }; // Once closed, moq-quic transmits nothing but the CONNECTION_CLOSE. (transmit, conn.is_closed()) }; + debug_assert_eq!(transmit.size, tx.len()); - tx[..transmit.size].copy_from_slice(&self.scratch[..transmit.size]); // A lone datagram is its own segment size, and the socket's GSO // stride has to match what moq-quic actually packed. let transmit = udp::Transmit { to: transmit.destination, - len: transmit.size, segment: transmit.segment_size.unwrap_or(transmit.size), ecn: transmit.ecn.map(super::ecn_from_quic), }; diff --git a/rs/moq-uring/src/quic/driver/endpoint.rs b/rs/moq-uring/src/quic/driver/endpoint.rs index e91ad1950be..1b45c53b313 100644 --- a/rs/moq-uring/src/quic/driver/endpoint.rs +++ b/rs/moq-uring/src/quic/driver/endpoint.rs @@ -242,7 +242,7 @@ impl Inner { return Poll::Ready(()); } match self.socket.poll_recv(waiter) { - Poll::Ready(Ok(mut packet)) => self.demux(&mut packet), + Poll::Ready(Ok(packet)) => self.demux(&packet), Poll::Ready(Err(err)) => { self.fail(Error::Io(err.to_string())); return Poll::Ready(()); @@ -253,7 +253,7 @@ impl Inner { } /// Route every datagram in one receive, then kick the connections fed. - fn demux(self: &Rc, packet: &mut udp::Packet) { + fn demux(self: &Rc, packet: &udp::Packet) { let from = packet.from(); let ecn = packet.ecn().map(super::ecn_to_quic); // Where the endpoint writes its own answers (version negotiation, @@ -264,19 +264,25 @@ impl Inner { // coalesce is a single source, and ids rarely change mid-burst). let mut fed: Vec = Vec::new(); - for segment in packet.segments() { + // moq-quic takes each datagram as an owned `BytesMut`: copy the whole + // completion out of the receive pool once and split it per datagram. + // The last one is taken whole, since a split makes the allocation + // shared, which a lone datagram has no use for. + let mut payload = BytesMut::from(packet.payload()); + let stride = packet.stride(); + while !payload.is_empty() { + let segment = match payload.len() > stride { + true => payload.split_to(stride), + false => std::mem::take(&mut payload), + }; buf.clear(); // The socket does not report which local address a datagram // arrived on, so the four-tuple stays half known, exactly as it // does on a platform without `IP_PKTINFO`. - let event = self.endpoint.borrow_mut().handle( - moq_clock::Instant::now(), - from, - None, - ecn, - BytesMut::from(&segment[..]), - &mut buf, - ); + let event = + self.endpoint + .borrow_mut() + .handle(moq_clock::Instant::now(), from, None, ecn, segment, &mut buf); match event { Some(DatagramEvent::ConnectionEvent(key, event)) => { let conn = self.conns.borrow().get(&key).cloned(); @@ -408,14 +414,13 @@ impl Inner { let Poll::Ready(Ok(mut tx)) = self.socket.poll_acquire(&kio::Waiter::noop()) else { return; }; - if transmit.size > tx.len() { + if transmit.size > tx.capacity() { tracing::debug!(size = transmit.size, "dropping an oversized endpoint response"); return; } - tx[..transmit.size].copy_from_slice(&buf[..transmit.size]); + tx.extend_from_slice(&buf[..transmit.size]); let transmit = udp::Transmit { to: transmit.destination, - len: transmit.size, segment: transmit.segment_size.unwrap_or(transmit.size), ecn: transmit.ecn.map(super::ecn_from_quic), }; diff --git a/rs/moq-uring/src/quic/driver/stream.rs b/rs/moq-uring/src/quic/driver/stream.rs index 06baf6a7634..36157e1cee6 100644 --- a/rs/moq-uring/src/quic/driver/stream.rs +++ b/rs/moq-uring/src/quic/driver/stream.rs @@ -5,7 +5,7 @@ //! buffer; the handles just kick the driver so egress reaches the wire and //! park on the per-stream waiter lists the driver wakes. -use std::task::{Context, Poll}; +use std::task::{Context, Poll, ready}; use bytes::{Buf, Bytes, BytesMut}; use moq_quic::{Checkpoint, StreamId, VarInt}; @@ -101,7 +101,11 @@ impl SendStream { impl web_transport_trait::poll::SendStream for SendStream { type Error = Error; - fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + fn poll_write(&mut self, cx: &mut Context<'_>, mut buf: &[u8]) -> Poll> { + self.poll_write_buf(cx, &mut buf) + } + + fn poll_write_buf(&mut self, cx: &mut Context<'_>, buf: &mut B) -> Poll> { let waiter = self.park.hold(cx); if self.fin || self.reset { return Poll::Ready(Err(Error::Quic("stream already finished".to_string()))); @@ -109,7 +113,21 @@ impl web_transport_trait::poll::SendStream for SendStream { if let Some(err) = self.shared.closed() { return Poll::Ready(Err(err)); } - let result = self.shared.conn.borrow_mut().send_stream(self.id).write(buf); + let result = { + let mut conn = self.shared.conn.borrow_mut(); + let mut stream = conn.send_stream(self.id); + // Bytes taken from `buf` cannot go back, so take only what moq-quic + // accepts. It keeps the chunk by reference, which for a `Bytes` + // source means no copy at all. + match stream.write_limit().min(buf.chunk().len()) { + // Blocked, failing, or nothing to write; the empty write says which. + 0 => stream.write(&[]), + take => stream.write_chunks(&mut [buf.copy_to_bytes(take)]).map(|written| { + debug_assert_eq!(written.bytes, take, "moq-quic accepted less than its write limit"); + written.bytes + }), + } + }; match result { Ok(n) => { self.shared.kick(); @@ -276,22 +294,35 @@ impl RecvStream { dst: &mut [u8], held: Option<&mut moq_quic::Credit>, ) -> Poll, Error>> { + let Some(chunk) = ready!(self.poll_chunk(cx, dst.len(), held))? else { + return Poll::Ready(Ok(None)); + }; + dst[..chunk.len()].copy_from_slice(&chunk); + Poll::Ready(Ok(Some(chunk.len()))) + } + + /// [`poll_read_chunk`](web_transport_trait::poll::RecvStream::poll_read_chunk), + /// with the credit for the bytes read moved into `held` instead of + /// returned to the peer. + fn poll_chunk( + &mut self, + cx: &mut Context<'_>, + max: usize, + held: Option<&mut moq_quic::Credit>, + ) -> Poll, Error>> { let waiter = self.park.hold(cx); - if dst.is_empty() { - return Poll::Ready(Ok(Some(0))); + if max == 0 { + return Poll::Ready(Ok(Some(Bytes::new()))); } if !self.backlog.is_empty() { - return Poll::Ready(Ok(Some(self.drain(dst, held)))); + return Poll::Ready(Ok(Some(self.drain(max, held)))); } if self.finished { return Poll::Ready(Ok(None)); } - match Self::read(&self.shared, self.id, dst.len(), held) { - Read::Chunk(bytes) => { - // moq-quic hands out at most `dst.len()`. - dst[..bytes.len()].copy_from_slice(&bytes); - Poll::Ready(Ok(Some(bytes.len()))) - } + match Self::read(&self.shared, self.id, max, held) { + // moq-quic hands out at most `max`. + Read::Chunk(bytes) => Poll::Ready(Ok(Some(bytes))), Read::Finished => { self.finished = true; Poll::Ready(Ok(None)) @@ -365,24 +396,22 @@ impl RecvStream { read } - /// Move up to `dst.len()` read-ahead bytes out of the backlog, with their - /// credit moved into `held` or else returned to the peer. + /// Take up to `max` read-ahead bytes out of the backlog, with their credit + /// moved into `held` or else returned to the peer. /// /// Wakes a `poll_closed` parked at the read-ahead cap: the room it was /// waiting for is what this just made. - fn drain(&mut self, dst: &mut [u8], held: Option<&mut moq_quic::Credit>) -> usize { - let n = dst.len().min(self.backlog.len()); - dst[..n].copy_from_slice(&self.backlog[..n]); - self.backlog.advance(n); - let credit = self.backlog_credit.split(n as u64); + fn drain(&mut self, max: usize, held: Option<&mut moq_quic::Credit>) -> Bytes { + let chunk = self.backlog.split_to(max.min(self.backlog.len())).freeze(); + let credit = self.backlog_credit.split(chunk.len() as u64); match held { Some(held) => held.merge(credit), None => release(&self.shared, credit), } - if n > 0 { + if !chunk.is_empty() { self.shared.wake_readable(self.id); } - n + chunk } } @@ -393,6 +422,10 @@ impl web_transport_trait::poll::RecvStream for RecvStream { self.poll_read_into(cx, dst, None) } + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { + self.poll_chunk(cx, max, None) + } + fn stop(&mut self, code: u32) { // Giving up on the read side abandons whatever was read ahead, even // when the FIN is already in and only the backlog is left. diff --git a/rs/moq-uring/src/quic/web.rs b/rs/moq-uring/src/quic/web.rs index 02222e8c6e7..1e6a0afecd0 100644 --- a/rs/moq-uring/src/quic/web.rs +++ b/rs/moq-uring/src/quic/web.rs @@ -949,6 +949,13 @@ impl web_transport_trait::poll::RecvStream for RecvStream { } } + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { + match web_transport_trait::poll::RecvStream::poll_read_chunk(&mut self.inner, cx, max) { + Poll::Ready(Err(err)) => Poll::Ready(Err(self.map(err))), + other => other, + } + } + fn stop(&mut self, code: u32) { match self.web { true => self.inner.stop_code(proto::error_to_http3(code)), diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index 6f0fe55a74b..5008e17b5ae 100644 --- a/rs/moq-uring/src/udp.rs +++ b/rs/moq-uring/src/udp.rs @@ -177,9 +177,7 @@ impl Ecn { pub struct Transmit { /// The destination. pub to: SocketAddr, - /// How many bytes of the buffer to send. - pub len: usize, - /// The datagram size; the buffer is sent as `len / segment` datagrams, + /// The datagram size; the buffer is sent in datagrams of this many bytes, /// the last possibly short. pub segment: usize, /// The ECN codepoint every datagram carries, if any. @@ -263,8 +261,8 @@ fn grow_rx(rx: &mut Rx, config: &Config) -> bool { /// Allocate more send buffers, up to [`Config::tx_buffers_max`]. Returns /// whether the pool grew. /// -/// The `Box<[u8]>` allocations are stable across the `Vec` growth, which is -/// what lets a live [`TxBuf`] keep a raw pointer into one. +/// The payload allocations are stable across the `Vec` growth, which is what +/// lets an in-flight send keep a raw pointer into one. fn grow_tx(tx: &mut Tx, config: &Config) -> bool { let Some(target) = grown(tx.bufs.len(), config.tx_buffers_max) else { return false; @@ -363,7 +361,9 @@ struct Tx { /// Stable storage reused by every checkout of one transmit slot: the payload /// the kernel reads and the `sendmsg` headers pointing into it. struct TxSlot { - data: Box<[u8]>, + /// Empty while a [`TxBuf`] has it checked out; sending or dropping that + /// hands the payload back. + data: Vec, headers: Vec, /// Sends staged from this slot that the kernel has not completed. The slot /// returns to the free list when it hits zero. @@ -373,7 +373,7 @@ struct TxSlot { impl TxSlot { fn new(len: usize) -> Self { Self { - data: vec![0u8; len].into_boxed_slice(), + data: Vec::with_capacity(len), headers: Vec::new(), in_flight: 0, } @@ -450,9 +450,11 @@ impl SockShared { } } - fn release_tx(&self, id: u16) { + fn release_tx(&self, id: u16, data: Vec) { let mut tx = self.tx.borrow_mut(); - debug_assert_eq!(tx.bufs[id as usize].in_flight, 0); + let slot = &mut tx.bufs[id as usize]; + debug_assert_eq!(slot.in_flight, 0); + slot.data = data; tx.free.push(id); tx.stalled = false; tx.waiters.wake(); @@ -743,16 +745,12 @@ impl Socket { grow_tx(&mut tx, &self.shared.config); } if let Some(id) = tx.free.pop() { - let slot = &mut tx.bufs[id as usize]; - // SAFETY: `id` was exclusively checked out of the free list; the - // allocation is stable (see TxBuf). - let ptr = unsafe { NonNull::new_unchecked(slot.data.as_mut_ptr()) }; - let cap = slot.data.len(); + let mut data = std::mem::take(&mut tx.bufs[id as usize].data); + data.clear(); return Poll::Ready(Ok(TxBuf { sock: self.shared.clone(), id, - ptr, - cap, + data, armed: false, })); } @@ -828,18 +826,6 @@ impl Packet { // `sock` Rc keeps alive; the pool never touches it while outstanding. unsafe { std::slice::from_raw_parts(self.ptr.as_ptr(), self.len) } } - - /// The whole coalesced payload, mutably (QUIC decrypts in place). - pub fn payload_mut(&mut self) -> &mut [u8] { - // SAFETY: as `payload`, and `&mut self` forbids aliasing our slices. - unsafe { std::slice::from_raw_parts_mut(self.ptr.as_ptr(), self.len) } - } - - /// The individual datagrams. - pub fn segments(&mut self) -> impl Iterator { - let stride = self.stride; - self.payload_mut().chunks_mut(stride) - } } impl Drop for Packet { @@ -859,21 +845,21 @@ impl std::fmt::Debug for Packet { } } -/// A checked-out send-staging buffer: fill it, then [`send`](Self::send) it. +/// A checked-out send-staging buffer, empty with room for +/// [`Config::tx_buffer_len`] bytes: fill it, then [`send`](Self::send) it. /// /// The buffer belongs to the socket it was acquired from, and sending goes /// back through that socket. Dropping it unsent returns it to the pool. pub struct TxBuf { sock: Rc, id: u16, - ptr: NonNull, - cap: usize, + data: Vec, armed: bool, } impl TxBuf { - /// Send `self[..len]` on the owning socket as datagrams of `segment` - /// bytes (the last may be short), marked with `ecn`. Fire-and-forget: + /// Send the buffer on the owning socket as datagrams of `segment` bytes + /// (the last may be short), marked with `ecn`. Fire-and-forget: /// the buffer returns to the pool when the kernel completes, and a /// failed send surfaces on the next pool acquire. /// @@ -882,16 +868,14 @@ impl TxBuf { /// to submit staged datagrams and drain their completions, but does not /// guarantee kernel completion or delivery. pub fn send(mut self, transmit: Transmit) -> io::Result<()> { - let Transmit { to, len, segment, ecn } = transmit; + let Transmit { to, segment, ecn } = transmit; + let len = self.data.len(); // `UDP_SEGMENT` is a u16, so an oversized segment would silently // truncate into a tiny stride and explode the implied segment count. - if len == 0 || len > self.cap || segment == 0 || segment > usize::from(u16::MAX) { + if len == 0 || segment == 0 || segment > usize::from(u16::MAX) { return Err(io::Error::new( io::ErrorKind::InvalidInput, - format!( - "invalid send: {len} bytes in {segment} byte segments from a {} byte buffer", - self.cap - ), + format!("invalid send: {len} bytes in {segment} byte segments"), )); } let shared = match self.sock.owner.upgrade() { @@ -918,17 +902,21 @@ impl TxBuf { self.armed = true; let sock = self.sock.clone(); - let base = self.ptr.as_ptr(); - let headers = { + let (base, headers) = { let mut tx = sock.tx.borrow_mut(); - let headers = &mut tx.bufs[self.id as usize].headers; - if headers.len() < segments { - headers.resize_with(segments, SendHdr::zeroed); + let slot = &mut tx.bufs[self.id as usize]; + // The slot keeps the payload while the kernel reads it. Moving the + // `Vec` leaves its heap allocation where it is. + slot.data = std::mem::take(&mut self.data); + if slot.headers.len() < segments { + slot.headers.resize_with(segments, SendHdr::zeroed); } // SAFETY: the slot is checked out, so it cannot be sent from again - // (and its headers cannot grow again) until every send below - // completes and returns it to the free list. - unsafe { NonNull::new_unchecked(headers.as_mut_ptr()) } + // (and its payload and headers cannot change) until every send + // below completes and returns it to the free list. + (slot.data.as_mut_ptr(), unsafe { + NonNull::new_unchecked(slot.headers.as_mut_ptr()) + }) }; let staging = Staging { sock: sock.clone(), @@ -957,32 +945,30 @@ impl TxBuf { } impl std::ops::Deref for TxBuf { - type Target = [u8]; + type Target = Vec; - fn deref(&self) -> &[u8] { - // SAFETY: `id` is exclusively ours until release; stable allocation. - unsafe { std::slice::from_raw_parts(self.ptr.as_ptr(), self.cap) } + fn deref(&self) -> &Vec { + &self.data } } impl std::ops::DerefMut for TxBuf { - fn deref_mut(&mut self) -> &mut [u8] { - // SAFETY: as `deref`. - unsafe { std::slice::from_raw_parts_mut(self.ptr.as_ptr(), self.cap) } + fn deref_mut(&mut self) -> &mut Vec { + &mut self.data } } impl Drop for TxBuf { fn drop(&mut self) { if !self.armed { - self.sock.release_tx(self.id); + self.sock.release_tx(self.id, std::mem::take(&mut self.data)); } } } impl std::fmt::Debug for TxBuf { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("TxBuf").field("cap", &self.cap).finish() + f.debug_struct("TxBuf").field("len", &self.data.len()).finish() } } @@ -1694,10 +1680,9 @@ mod tests { let err = worker .block_on(async { let mut tx = sock.acquire().await.expect("send buffer"); - tx[..4].copy_from_slice(b"lost"); + tx.extend_from_slice(b"lost"); tx.send(Transmit { to: SocketAddr::new(Ipv4Addr::LOCALHOST.to_ipv6_mapped().into(), 1), - len: 4, segment: 4, ecn: None, }) diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index 6755e083e48..f772a88ee50 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -931,9 +931,10 @@ mod tests { let sock = handle.udp(bind(), udp::Config::default()).expect("socket"); let shared = sock.downgrade(); let to = sock.local_addr().expect("addr"); - let Poll::Ready(Ok(tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { + let Poll::Ready(Ok(mut tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("no tx buffer"); }; + tx.resize(1200, 0); drop(worker); // Every path a retained handle can reach fails instead of pending on @@ -944,7 +945,6 @@ mod tests { assert!( tx.send(udp::Transmit { to, - len: 1200, segment: 1200, ecn: None, }) @@ -1089,10 +1089,10 @@ mod tests { held.push(tx); } assert_eq!(held.len(), usize::from(ceiling)); - for tx in held.drain(..) { + for mut tx in held.drain(..) { + tx.resize(1200, 0); tx.send(udp::Transmit { to, - len: 1200, segment: 1200, ecn: None, }) @@ -1206,12 +1206,12 @@ mod tests { .udp(std::net::UdpSocket::bind("127.0.0.1:0").expect("bind"), config) .expect("socket"); let to = sock.local_addr().expect("addr"); - let Poll::Ready(Ok(tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { + let Poll::Ready(Ok(mut tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("no tx buffer"); }; + tx.resize(64 * 1024, 0); tx.send(udp::Transmit { to, - len: 64 * 1024, segment: 1000, ecn: None, }) @@ -1234,13 +1234,13 @@ mod tests { .udp(std::net::UdpSocket::bind("127.0.0.1:0").expect("bind"), config) .expect("socket"); let to = sock.local_addr().expect("addr"); - let Poll::Ready(Ok(tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { + let Poll::Ready(Ok(mut tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("no tx buffer"); }; + tx.resize(64 * 1024, 0); let err = tx .send(udp::Transmit { to, - len: 64 * 1024, segment: 1, ecn: None, }) @@ -1260,15 +1260,15 @@ mod tests { ) .expect("socket"); let to = sock.local_addr().expect("addr"); - let Poll::Ready(Ok(tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { + let Poll::Ready(Ok(mut tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("no tx buffer"); }; + tx.resize(60_000, 0); // `UDP_SEGMENT` is a u16: without validation this would truncate to a // one-byte stride instead of one segment. let err = tx .send(udp::Transmit { to, - len: 60_000, segment: usize::from(u16::MAX) + 2, ecn: None, }) @@ -1309,10 +1309,9 @@ mod tests { let Poll::Ready(Ok(mut tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("the socket failed after a rejected train"); }; - tx[..2 * 1200].fill(7); + tx.resize(2 * 1200, 7); tx.send(udp::Transmit { to, - len: 2 * 1200, segment: 1200, ecn: None, }) @@ -1359,10 +1358,9 @@ mod tests { let Poll::Ready(Ok(mut tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("no tx buffer"); }; - tx[..4 * 1200].fill(7); + tx.resize(4 * 1200, 7); tx.send(udp::Transmit { to, - len: 4 * 1200, segment: 1200, ecn: None, }) @@ -1436,16 +1434,16 @@ mod tests { .expect("socket"); let to = sock.local_addr().expect("addr"); - let Poll::Ready(Ok(tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { + let Poll::Ready(Ok(mut tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("no tx buffer"); }; + tx.resize(1200, 0); // The pool is one buffer deep and that one is checked out. assert!(sock.poll_acquire(&kio::Waiter::noop()).is_pending()); assert!(sock.poll_acquire(&kio::Waiter::noop()).is_pending()); assert_eq!(metrics.snapshot().tx_stalls, 1); tx.send(udp::Transmit { to, - len: 1200, segment: 1200, ecn: None, }) diff --git a/rs/moq-uring/tests/allocations.rs b/rs/moq-uring/tests/allocations.rs index 833d9d1b9fe..ddb86c5b624 100644 --- a/rs/moq-uring/tests/allocations.rs +++ b/rs/moq-uring/tests/allocations.rs @@ -80,10 +80,9 @@ fn steady_state_sends_do_not_allocate() { // Grow every reusable pool before observing the steady state. for _ in 0..100 { let mut tx = socket.acquire().await.expect("warmup acquire"); - tx[..4800].fill(0x5a); + tx.resize(4800, 0x5a); tx.send(udp::Transmit { to, - len: 4800, segment: 1200, ecn: None, }) @@ -101,10 +100,9 @@ fn steady_state_sends_do_not_allocate() { acquires += counted().0; count_from_zero(); - tx[..4800].fill(0x5a); + tx.resize(4800, 0x5a); let sent = tx.send(udp::Transmit { to, - len: 4800, segment: 1200, ecn: None, }); diff --git a/rs/moq-uring/tests/integration/credit.rs b/rs/moq-uring/tests/integration/credit.rs index b6d91de0e71..49cca1c4a90 100644 --- a/rs/moq-uring/tests/integration/credit.rs +++ b/rs/moq-uring/tests/integration/credit.rs @@ -1,6 +1,7 @@ -//! Held receive credit through the io_uring driver: held reads keep the -//! connection window spent until their credit drops, and bytes `poll_closed` -//! reads ahead keep theirs until the application reads them. +//! Credit through the io_uring driver: held reads keep the connection window +//! spent until their credit drops, bytes `poll_closed` reads ahead keep theirs +//! until the application reads them, and a chunked write takes no more from +//! its buffer than the peer's window admits. //! //! Unsupported local kernels skip loudly; MOQ_STRICT makes CI refuse them. @@ -9,9 +10,11 @@ use crate::support; use std::net::UdpSocket; -use std::task::Poll; +use std::task::{Context, Poll, Waker}; use std::time::Duration; +use bytes::Bytes; + use moq_uring::{Config, Error, Worker, quic, udp}; use web_transport_trait::poll::{RecvStream as _, SendStream as _, Session as _}; @@ -183,3 +186,107 @@ fn held_reads_spend_the_connection_window() { }) .expect("worker"); } + +#[test] +fn chunked_writes_stop_at_the_window() { + const STREAM_WINDOW: usize = 4096; + + let Some(mut worker) = worker() else { return }; + let handle = worker.handle(); + let certs = support::certs().expect("certificates"); + + let socket = |handle: &moq_uring::Handle| { + handle + .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) + .expect("socket") + }; + let mut server_config = quic::server::Config::new( + quic::Identity::open(&certs.cert, &certs.key) + .expect("identity") + .server(&[ALPN], quic::server::ClientAuth::None) + .expect("tls"), + ); + server_config.transport.stream_receive_window = Some(STREAM_WINDOW as u64); + let server = quic::Endpoint::new( + socket(&handle), + quic::endpoint::Config::default().with_server(server_config), + ) + .expect("server endpoint"); + let client = quic::Endpoint::new(socket(&handle), quic::endpoint::Config::default()).expect("dial-only endpoint"); + let mut dial = quic::client::Config::new(server.local_addr(), "localhost"); + dial.alpn = vec![ALPN.to_string()]; + dial.verify = false; + + let payload: Bytes = (0..3 * STREAM_WINDOW).map(|i| (i * 31 % 251) as u8).collect(); + + worker + .block_on(async move { + let mut dialed = client.connect(&dial).await.expect("dial"); + let mut accepted = server.accept().await.expect("accept"); + + let mut send = std::future::poll_fn(|cx| dialed.poll_open_uni(cx)) + .await + .expect("open_uni"); + + // The bytes past the window stay in the caller's buffer, untaken, + // and moq-quic keeps the ones it took by reference, not as a copy. + let mut cx = Context::from_waker(Waker::noop()); + let mut buf = Bytes::copy_from_slice(&payload); + assert!(buf.is_unique()); + assert!(matches!( + send.poll_write_buf(&mut cx, &mut buf), + Poll::Ready(Ok(STREAM_WINDOW)) + )); + assert_eq!(buf, payload.slice(STREAM_WINDOW..)); + assert!(!buf.is_unique(), "moq-quic copied the chunk"); + assert!(send.poll_write_buf(&mut cx, &mut buf).is_pending()); + assert_eq!(buf, payload.slice(STREAM_WINDOW..)); + + handle.spawn(async move { + while !buf.is_empty() { + std::future::poll_fn(|cx| send.poll_write_buf(cx, &mut buf)) + .await + .expect("write"); + } + send.finish().expect("finish"); + let _ = std::future::poll_fn(|cx| send.poll_closed(cx)).await; + }); + + let mut recv = within( + &handle, + "the stream", + std::future::poll_fn(|cx| accepted.poll_accept_uni(cx)), + ) + .await + .expect("accept_uni"); + + // Straight out of moq-quic, capped at `max`. + let first = within( + &handle, + "stream data", + std::future::poll_fn(|cx| recv.poll_read_chunk(cx, 1000)), + ) + .await + .expect("read") + .expect("not finished"); + assert!(!first.is_empty() && first.len() <= 1000); + + // Then out of the read-ahead backlog. + within(&handle, "the FIN", std::future::poll_fn(|cx| recv.poll_closed(cx))) + .await + .expect("finished"); + let mut received = first.to_vec(); + while let Some(chunk) = within( + &handle, + "stream data", + std::future::poll_fn(|cx| recv.poll_read_chunk(cx, usize::MAX)), + ) + .await + .expect("read") + { + received.extend_from_slice(&chunk); + } + assert_eq!(received, payload); + }) + .expect("worker"); +} diff --git a/rs/moq-uring/tests/integration/ecn.rs b/rs/moq-uring/tests/integration/ecn.rs index 1f89c4e7a77..579e239aa0d 100644 --- a/rs/moq-uring/tests/integration/ecn.rs +++ b/rs/moq-uring/tests/integration/ecn.rs @@ -40,10 +40,9 @@ fn round_trip(bind: &str, config: udp::Config, ecn: Option, datagrams: .block_on(async move { let mut buf = tx.acquire().await.expect("acquire"); let len = SEGMENT * datagrams; - buf[..len].fill(0xAB); + buf.resize(len, 0xAB); buf.send(udp::Transmit { to, - len, segment: SEGMENT, ecn, }) diff --git a/rs/moq-uring/tests/integration/teardown.rs b/rs/moq-uring/tests/integration/teardown.rs index 0cf74b4e035..aa437c8f533 100644 --- a/rs/moq-uring/tests/integration/teardown.rs +++ b/rs/moq-uring/tests/integration/teardown.rs @@ -57,10 +57,9 @@ fn a_send_staged_on_the_way_out_still_goes() { worker .block_on(async { let mut tx = sock.acquire().await.expect("acquire"); - tx[..3].copy_from_slice(b"bye"); + tx.extend_from_slice(b"bye"); tx.send(udp::Transmit { to: peer_addr, - len: 3, segment: 3, ecn: None, }) From 440072f39242e35a1d47417fed5708e08815d2c6 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 16 Oct 2026 22:00:21 -0700 Subject: [PATCH 3/6] fix(uring): refuse a send buffer grown past tx_buffer_len Co-Authored-By: Claude Opus 5.5 --- rs/moq-uring/src/udp.rs | 7 +++++-- rs/moq-uring/src/worker.rs | 27 +++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index 5008e17b5ae..1dfb5cbec0a 100644 --- a/rs/moq-uring/src/udp.rs +++ b/rs/moq-uring/src/udp.rs @@ -870,12 +870,15 @@ impl TxBuf { pub fn send(mut self, transmit: Transmit) -> io::Result<()> { let Transmit { to, segment, ecn } = transmit; let len = self.data.len(); + let cap = self.sock.config.tx_buffer_len; // `UDP_SEGMENT` is a u16, so an oversized segment would silently // truncate into a tiny stride and explode the implied segment count. - if len == 0 || segment == 0 || segment > usize::from(u16::MAX) { + // A buffer grown past the pool's size is refused rather than sent, so + // the configured ceiling holds for every train. + if len == 0 || len > cap || segment == 0 || segment > usize::from(u16::MAX) { return Err(io::Error::new( io::ErrorKind::InvalidInput, - format!("invalid send: {len} bytes in {segment} byte segments"), + format!("invalid send: {len} bytes in {segment} byte segments from a {cap} byte buffer"), )); } let shared = match self.sock.owner.upgrade() { diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index f772a88ee50..0b381a866cf 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -1249,6 +1249,33 @@ mod tests { drop(worker); } + #[test] + fn buffer_grown_past_the_pool_is_rejected() { + let Some(worker) = worker() else { return }; + let handle = worker.handle(); + let config = udp::Config { + tx_buffer_len: 2048, + ..Default::default() + }; + let sock = handle + .udp(std::net::UdpSocket::bind("127.0.0.1:0").expect("bind"), config) + .expect("socket"); + let to = sock.local_addr().expect("addr"); + let Poll::Ready(Ok(mut tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { + panic!("no tx buffer"); + }; + tx.resize(4096, 0); + let err = tx + .send(udp::Transmit { + to, + segment: 1024, + ecn: None, + }) + .expect_err("4096 bytes from a 2048 byte buffer"); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); + drop(worker); + } + #[test] fn oversized_gso_segment_is_rejected() { let Some(worker) = worker() else { return }; From e49790ebf5f9d99e3c181b871c34c3f20f57f27e Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 16 Oct 2026 22:57:55 -0700 Subject: [PATCH 4/6] fix(uring): hand out every TX checkout at the pool's size Co-Authored-By: Claude Opus 5.5 --- rs/moq-uring/src/udp.rs | 6 ++++++ rs/moq-uring/src/worker.rs | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index 1dfb5cbec0a..240b331a199 100644 --- a/rs/moq-uring/src/udp.rs +++ b/rs/moq-uring/src/udp.rs @@ -747,6 +747,12 @@ impl Socket { if let Some(id) = tx.free.pop() { let mut data = std::mem::take(&mut tx.bufs[id as usize].data); data.clear(); + // A checkout is a `Vec` its holder may have shrunk or regrown before + // dropping it; every checkout holds exactly one pool-sized train. + let cap = self.shared.config.tx_buffer_len; + if data.capacity() != cap { + data = Vec::with_capacity(cap); + } return Poll::Ready(Ok(TxBuf { sock: self.shared.clone(), id, diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index 0b381a866cf..7d4b44fed50 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -1249,6 +1249,39 @@ mod tests { drop(worker); } + #[test] + fn a_resized_checkout_comes_back_pool_sized() { + let Some(worker) = worker() else { return }; + let handle = worker.handle(); + let config = udp::Config { + tx_buffers_max: 1, + tx_buffer_len: 2048, + ..Default::default() + }; + let sock = handle + .udp(std::net::UdpSocket::bind("127.0.0.1:0").expect("bind"), config) + .expect("socket"); + let acquire = || { + let Poll::Ready(Ok(tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { + panic!("no tx buffer"); + }; + tx + }; + + let mut tx = acquire(); + assert_eq!(tx.capacity(), 2048); + tx.shrink_to_fit(); + drop(tx); + let mut tx = acquire(); + assert_eq!((tx.len(), tx.capacity()), (0, 2048)); + tx.resize(8192, 7); + drop(tx); + let tx = acquire(); + assert_eq!((tx.len(), tx.capacity()), (0, 2048)); + drop(tx); + drop(worker); + } + #[test] fn buffer_grown_past_the_pool_is_rejected() { let Some(worker) = worker() else { return }; From b275b703a142d294656d51dd02d68773adb89423 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:46:18 -0700 Subject: [PATCH 5/6] fix(uring): treat a zero UDP_GRO size as no stride Co-Authored-By: Claude Opus 5.5 --- rs/moq-uring/src/udp.rs | 34 ++++++++++++++++++++++++++++++---- 1 file changed, 30 insertions(+), 4 deletions(-) diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index 240b331a199..021b44ea055 100644 --- a/rs/moq-uring/src/udp.rs +++ b/rs/moq-uring/src/udp.rs @@ -1257,8 +1257,7 @@ pub(crate) fn on_recv( match received { Ok((_, Some(queued))) => { sock.metrics.rx_receives.add(1); - // A zero stride would be a kernel that reported a `UDP_GRO` size of - // zero; count the receive as one datagram rather than dividing by it. + // Only an empty datagram has a zero stride; it is still one datagram. sock.metrics .rx_datagrams .add(queued.len.div_ceil(queued.stride.max(1)) as u64); @@ -1449,7 +1448,8 @@ fn resend_unsegmented(op: &SendOp, base: NonNull, one: &SendOne) -> io::Resu /// What the kernel said about one receive, from its control buffer. #[derive(Default)] struct RecvMeta { - /// The `UDP_GRO` segment size, if the receive was coalesced. + /// The `UDP_GRO` segment size, if the receive was coalesced. Never zero: + /// splitting a payload into zero-byte datagrams would never finish. stride: Option, /// The ECN codepoint from `IP_TOS` or `IPV6_TCLASS`, if marked. ecn: Option, @@ -1473,7 +1473,9 @@ impl RecvMeta { let data = &control[offset + header_len..offset + message_len]; match (header.cmsg_level, header.cmsg_type) { (libc::SOL_UDP, libc::UDP_GRO) => { - meta.stride = read_int(data).and_then(|value| usize::try_from(value).ok()); + meta.stride = read_int(data) + .and_then(|value| usize::try_from(value).ok()) + .filter(|&stride| stride > 0); } // Linux reports the TOS byte itself, but the traffic class as an int. (libc::IPPROTO_IP, libc::IP_TOS) => { @@ -1588,6 +1590,30 @@ mod tests { use super::*; use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4}; + /// One `UDP_GRO` control message carrying `size`. + fn gro(size: libc::c_int) -> Vec { + let space = unsafe { libc::CMSG_SPACE(std::mem::size_of::() as _) } as usize; + let mut control = vec![0u8; space]; + let header = libc::cmsghdr { + cmsg_len: unsafe { libc::CMSG_LEN(std::mem::size_of::() as _) } as _, + cmsg_level: libc::SOL_UDP, + cmsg_type: libc::UDP_GRO, + }; + let header_len = unsafe { libc::CMSG_LEN(0) } as usize; + // SAFETY: `control` holds a header plus an int. + unsafe { + control.as_mut_ptr().cast::().write_unaligned(header); + control.as_mut_ptr().add(header_len).cast::().write_unaligned(size); + } + control + } + + #[test] + fn a_zero_gro_size_is_no_stride() { + assert_eq!(RecvMeta::parse(&gro(1200)).stride, Some(1200)); + assert_eq!(RecvMeta::parse(&gro(0)).stride, None); + } + /// Round-trip an address through the kernel wire encoding. fn roundtrip(addr: SocketAddr) -> Option { // SAFETY: all-zero is a valid sockaddr_storage. From e80e53df5d2b163487c21531fda33645ea8e15ae Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:59:25 -0700 Subject: [PATCH 6/6] style(uring): rustfmt the GRO parse test Co-Authored-By: Claude Opus 5.5 --- rs/moq-uring/src/udp.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index 021b44ea055..6cd04075d33 100644 --- a/rs/moq-uring/src/udp.rs +++ b/rs/moq-uring/src/udp.rs @@ -1603,7 +1603,11 @@ mod tests { // SAFETY: `control` holds a header plus an int. unsafe { control.as_mut_ptr().cast::().write_unaligned(header); - control.as_mut_ptr().add(header_len).cast::().write_unaligned(size); + control + .as_mut_ptr() + .add(header_len) + .cast::() + .write_unaligned(size); } control }