perf(uring): remove avoidable copies on egress, streams, and datagram receive

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

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

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<B: BytesSource>(&mut self, source: &mut B) -> Result<Written, WriteError> {
         if self.conn_state.is_closed() {
             trace!(%self.id, "write blocked; connection draining");

rs/moq-quic/src/tests/mod.rs

@@ -2339,14 +2339,17 @@ 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)
     );
     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)

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,
 									})

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<endpoint::Inner>,
 	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<u8>,
 	/// 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),
 		};

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<Self>, packet: &mut udp::Packet) {
+	fn demux(self: &Rc<Self>, 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<ConnectionHandle> = 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),
 		};

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,15 +101,33 @@ impl SendStream {
 impl web_transport_trait::poll::SendStream for SendStream {
 	type Error = Error;
 
-	fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize, Self::Error>> {
+	fn poll_write(&mut self, cx: &mut Context<'_>, mut buf: &[u8]) -> Poll<Result<usize, Self::Error>> {
+		self.poll_write_buf(cx, &mut buf)
+	}
+
+	fn poll_write_buf<B: Buf>(&mut self, cx: &mut Context<'_>, buf: &mut B) -> Poll<Result<usize, Self::Error>> {
 		let waiter = self.park.hold(cx);
 		if self.fin || self.reset {
 			return Poll::Ready(Err(Error::Quic("stream already finished".to_string())));
 		}
 		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<Result<Option<usize>, 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<Result<Option<Bytes>, 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<Result<Option<Bytes>, 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.

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<Result<Option<Bytes>, 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)),