diff --git a/quest/m2/quic/README.md b/quest/m2/quic/README.md index 081ef50ae2..cb3a40a790 100644 --- a/quest/m2/quic/README.md +++ b/quest/m2/quic/README.md @@ -45,8 +45,6 @@ This is a transport API change, not a MoQ wire change. written down - [Deliver the application close before io_uring teardown](/quest/m2/quic/uring-close.md) - the peer receives the final close when the client immediately stops its worker -- [ECN on the io_uring UDP path](/quest/m2/quic/ecn-uring.md) - the ring's - sends carry ECT(0) and its receives read the mark, matching `noq-udp` - [Measure ECN on the backbone](/quest/m2/quic/ecn-measure.md) - a written verdict on marking versus dropping, and whether Linode and OVH keep marks - [Per-stream ACK progress](/quest/m2/quic/ack-progress.md) - the fork reports diff --git a/quest/m2/quic/ecn-measure.md b/quest/m2/quic/ecn-measure.md index 83d8bc9b56..62ceede2e4 100644 --- a/quest/m2/quic/ecn-measure.md +++ b/quest/m2/quic/ecn-measure.md @@ -30,8 +30,3 @@ repeat it. and the tcpdump summaries beside the numbers in the L4S quest's Plan. If neither provider preserves the marks, say so there: L4S stays off and the marking response is only a lab result. - -## Required - -- [ECN on the io_uring UDP path](/quest/m2/quic/ecn-uring.md) - the relays - under test must mark diff --git a/quest/m2/quic/ecn-uring.md b/quest/m2/quic/ecn-uring.md deleted file mode 100644 index db39ec0a7a..0000000000 --- a/quest/m2/quic/ecn-uring.md +++ /dev/null @@ -1,53 +0,0 @@ -# [S] ECN on the io_uring UDP path - -## Goal - -A relay on the io_uring runtime keeps ECN negotiated the way it already does -on the tokio runtime: its packets leave marked ECT(0), and marks on arriving -packets reach the QUIC stack, so the peer's ACKs carry ECN counts and noq -never disables ECN on a uring-to-uring session. A viewer's session is -unaffected. - -## Plan - -noq-proto already does classic ECN end to end: every path starts with -`sending_ecn` on and marks ECT(0), `process_ecn` validates the peer's ACK -ECN counts, and an ACK without counts disables it. `noq-udp` carries the -mark through `IP_TOS` / `IPV6_TCLASS` on send and reads it back on receive -(`unix.rs:607-620`). - -The gap is ours. `rs/moq-uring/src/udp.rs` builds its own control messages -and carries only `UDP_SEGMENT` on send and `UDP_GRO` on receive: no TOS or -TCLASS out, no `IP_RECVTOS` / `IPV6_RECVTCLASS` in. Nothing in `moq-uring` -mentions ECN, and `udp::TxBuf::send(len, to, segment)` has no slot for the -codepoint, so on the io_uring path the peer never sees a mark, its ACKs -carry no counts, and noq turns ECN off within the first ACK. - -- Send: `TxBuf::send` takes the codepoint beside the segment size and writes - `IP_TOS` (v4) or `IPV6_TCLASS` (v6) into the same control buffer as - `UDP_SEGMENT`, so a GSO train carries the mark on every segment. The noq - and quinn adapter passes `Transmit::ecn` at both call sites, - `rs/moq-uring/src/quic/quinn/endpoint.rs:377` and - `rs/moq-uring/src/quic/quinn/connection.rs:797`. quiche has no ECN send - API, so its two callers (`rs/moq-uring/src/quic/quiche/endpoint.rs:476` - and `rs/moq-uring/src/quic/quiche/connection.rs:843`) pass no codepoint - and change only to match the signature. -- Receive: enable `IP_RECVTOS` and `IPV6_RECVTCLASS` on the socket and parse - the TOS or TCLASS control message next to `UDP_GRO`. `udp::Packet` gains - an `ecn` accessor (one mark per completion; a GRO batch shares it), and - the noq and quinn adapter threads it into both `Endpoint::handle` calls at - `quinn/endpoint.rs:240-252`, which pass `None` today. -- `moq-uring` is 0.0.1 and unpublished, so the `udp` module's signature - changes on `main`. No config, wire, or doc changes. -- Regression, in `rs/moq-uring/tests`: a socket pair through the worker - sends with ECT(0) and the receiving `Packet::ecn` reads it back, over v4 - and v6, for a single datagram and for a GSO train; and a datagram sent - with CE arrives as CE. Fails today. noq-proto 1.2 exposes no - ECN state, so the session-level check waits for the fork; the `rs uring` - nightly lane already runs this crate and gates on the kernel floor. - -## Related - -- [Measure ECN on the backbone](/quest/m2/quic/ecn-measure.md) - the study - this fix unblocks -- [L4S on the backbone](/quest/m2/quic/ecn.md) - the fork-side half diff --git a/quest/m2/quic/ecn.md b/quest/m2/quic/ecn.md index 126b60703a..c12717f723 100644 --- a/quest/m2/quic/ecn.md +++ b/quest/m2/quic/ecn.md @@ -9,8 +9,7 @@ marks fall back to no ECN, and a viewer's session is unaffected. ## Plan -Classic ECN is already end to end once the -[io_uring path marks](/quest/m2/quic/ecn-uring.md); this quest is the +Classic ECN is already end to end on both runtimes; this quest is the fork-side half. noq-proto has no ECN knob: `sending_ecn` is hardcoded on per path and only an ACK without counts turns it off, so both `off` and `ect1` need the fork. @@ -36,7 +35,5 @@ per path and only an ACK without counts turns it off, so both `off` and - [Fork noq](/quest/m2/quic/fork.md) - the `Ect1` option and the `off` knob live there -- [ECN on the io_uring UDP path](/quest/m2/quic/ecn-uring.md) - the relays - must mark before a response can be measured - [Measure ECN on the backbone](/quest/m2/quic/ecn-measure.md) - the provider verdict this quest acts on diff --git a/rs/moq-uring/src/quic/quiche/connection.rs b/rs/moq-uring/src/quic/quiche/connection.rs index 585fbde3a4..609e31fd45 100644 --- a/rs/moq-uring/src/quic/quiche/connection.rs +++ b/rs/moq-uring/src/quic/quiche/connection.rs @@ -840,7 +840,12 @@ impl Driver { // Nothing to send; the buffer returns to the pool on drop. return Poll::Pending; }; - if let Err(err) = tx.send(filled, to, SEGMENT) { + if let Err(err) = tx.send(udp::Transmit { + to, + len: filled, + segment: SEGMENT, + ecn: None, + }) { return Poll::Ready(Err(Error::Io(err.to_string()))); } Poll::Ready(Ok(())) diff --git a/rs/moq-uring/src/quic/quiche/endpoint.rs b/rs/moq-uring/src/quic/quiche/endpoint.rs index fa562592fd..f121e4e8bc 100644 --- a/rs/moq-uring/src/quic/quiche/endpoint.rs +++ b/rs/moq-uring/src/quic/quiche/endpoint.rs @@ -473,7 +473,12 @@ impl Inner { }; match quiche::negotiate_version(&hdr.scid, &hdr.dcid, &mut tx) { Ok(len) => { - if let Err(err) = tx.send(len, to, len) { + if let Err(err) = tx.send(udp::Transmit { + to, + len, + segment: len, + ecn: None, + }) { tracing::debug!(%err, "failed to send a version negotiation packet"); } } diff --git a/rs/moq-uring/src/quic/quinn/connection.rs b/rs/moq-uring/src/quic/quinn/connection.rs index 55fd53d61c..95ded1782f 100644 --- a/rs/moq-uring/src/quic/quinn/connection.rs +++ b/rs/moq-uring/src/quic/quinn/connection.rs @@ -793,8 +793,13 @@ impl Driver { 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 quinn actually packed. - let segment = transmit.segment_size.unwrap_or(transmit.size); - if let Err(err) = tx.send(transmit.size, transmit.destination, segment) { + 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_quinn), + }; + if let Err(err) = tx.send(transmit) { return Poll::Ready(Err(Error::Io(err.to_string()))); } // A flush frees datagram-send queue space. diff --git a/rs/moq-uring/src/quic/quinn/endpoint.rs b/rs/moq-uring/src/quic/quinn/endpoint.rs index 0911403078..7f55313226 100644 --- a/rs/moq-uring/src/quic/quinn/endpoint.rs +++ b/rs/moq-uring/src/quic/quinn/endpoint.rs @@ -223,6 +223,7 @@ impl Inner { /// Route every datagram in one receive, then kick the connections fed. fn demux(self: &Rc, packet: &mut udp::Packet) { let from = packet.from(); + let ecn = packet.ecn().map(super::ecn_to_quinn); // Where the endpoint writes its own answers (version negotiation, // retry, a refusal), reused across the whole receive. let mut buf = Vec::new(); @@ -240,7 +241,7 @@ impl Inner { let event = self.endpoint.borrow_mut().handle( Instant::now(), from.into(), - None, + ecn, BytesMut::from(&segment[..]), &mut buf, ); @@ -249,7 +250,7 @@ impl Inner { Instant::now(), from, None, - None, + ecn, BytesMut::from(&segment[..]), &mut buf, ); @@ -373,8 +374,13 @@ impl Inner { return; } tx[..transmit.size].copy_from_slice(&buf[..transmit.size]); - let segment = transmit.segment_size.unwrap_or(transmit.size); - if let Err(err) = tx.send(transmit.size, transmit.destination, segment) { + 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_quinn), + }; + if let Err(err) = tx.send(transmit) { tracing::debug!(%err, "failed to send an endpoint response"); } } diff --git a/rs/moq-uring/src/quic/quinn/mod.rs b/rs/moq-uring/src/quic/quinn/mod.rs index f1a6b29201..93bd4d870d 100644 --- a/rs/moq-uring/src/quic/quinn/mod.rs +++ b/rs/moq-uring/src/quic/quinn/mod.rs @@ -31,6 +31,7 @@ use rustls::pki_types::{CertificateDer, PrivateKeyDer}; #[cfg(feature = "qlog")] use super::qlog; use super::{Congestion, Error, Identity, SEGMENT, Transport, client, endpoint::CID_LEN, server}; +use crate::udp; /// Per-stream flow control credit, matching the quiche backend. const STREAM_WINDOW: u32 = 4 * 1024 * 1024; @@ -40,6 +41,22 @@ const CONNECTION_WINDOW: u32 = 16 * 1024 * 1024; /// backend's 64. const DATAGRAM_WINDOW: usize = 64 * SEGMENT; +fn ecn_to_quinn(ecn: udp::Ecn) -> quinn_proto::EcnCodepoint { + match ecn { + udp::Ecn::Ect0 => quinn_proto::EcnCodepoint::Ect0, + udp::Ecn::Ect1 => quinn_proto::EcnCodepoint::Ect1, + udp::Ecn::Ce => quinn_proto::EcnCodepoint::Ce, + } +} + +fn ecn_from_quinn(ecn: quinn_proto::EcnCodepoint) -> udp::Ecn { + match ecn { + quinn_proto::EcnCodepoint::Ect0 => udp::Ecn::Ect0, + quinn_proto::EcnCodepoint::Ect1 => udp::Ecn::Ect1, + quinn_proto::EcnCodepoint::Ce => udp::Ecn::Ce, + } +} + impl From for Error { fn from(err: quinn_proto::ConnectionError) -> Self { use quinn_proto::ConnectionError; diff --git a/rs/moq-uring/src/udp.rs b/rs/moq-uring/src/udp.rs index 6068c86a9f..a52ddf70e6 100644 --- a/rs/moq-uring/src/udp.rs +++ b/rs/moq-uring/src/udp.rs @@ -14,7 +14,10 @@ //! Send stages datagrams in a pool of buffers owned by id and released //! explicitly on completion, the shape `SENDMSG_ZC`'s deferred-reclaim NOTIF //! model needs later. Every GSO `sendmsg` carries its `UDP_SEGMENT` control -//! message explicitly; the socket default is never relied on. +//! message explicitly; the socket default is never relied on. The ECN mark +//! rides beside it as `IP_TOS` or `IPV6_TCLASS`, and receives read the mark +//! back through `IP_RECVTOS` and `IPV6_RECVTCLASS`, so a QUIC stack's ECN +//! validation sees what the network did to its packets. //! //! Both pools are queues of concurrent operations, not byte budgets: one send //! buffer holds one GSO train and one receive buffer holds one completion, @@ -31,7 +34,7 @@ use std::alloc::{Layout, alloc_zeroed, dealloc, handle_alloc_error}; use std::cell::{Cell, RefCell}; use std::collections::VecDeque; use std::io; -use std::net::{SocketAddr, SocketAddrV6, UdpSocket}; +use std::net::{IpAddr, SocketAddr, SocketAddrV6, UdpSocket}; use std::os::fd::AsRawFd; use std::ptr::NonNull; use std::rc::{Rc, Weak}; @@ -44,7 +47,8 @@ use crate::Error; use crate::metrics::Counters; use crate::shared::{Cqe, Op, Shared}; -/// Space reserved for received control messages (`UDP_GRO` needs one int). +/// Space reserved for received control messages: `UDP_GRO` plus the packet's +/// `IP_TOS` or `IPV6_TCLASS` (the kernel emits one or the other), two ints. const CONTROL_LEN: usize = 64; /// Space reserved for the source address of each received datagram. const NAME_LEN: usize = std::mem::size_of::(); @@ -140,6 +144,45 @@ struct Queued { len: usize, from: SocketAddr, stride: usize, + ecn: Option, +} + +/// The ECN codepoint carried in the IP header's TOS or traffic class byte. +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Ecn { + /// ECN-capable transport, the codepoint classic ECN marks with. + Ect0 = 0b10, + /// ECN-capable transport, the codepoint L4S marks with. + Ect1 = 0b01, + /// Congestion experienced: a queue on the path marked the packet. + Ce = 0b11, +} + +impl Ecn { + /// The codepoint in the low two bits of a TOS byte, if any. + fn from_bits(bits: u8) -> Option { + match bits & 0b11 { + 0b10 => Some(Self::Ect0), + 0b01 => Some(Self::Ect1), + 0b11 => Some(Self::Ce), + _ => None, + } + } +} + +/// What one [`TxBuf::send`] puts on the wire. +#[derive(Debug, Clone, Copy)] +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 last possibly short. + pub segment: usize, + /// The ECN codepoint every datagram carries, if any. + pub ecn: Option, } /// Free `bid` back to its pool if nothing borrows it any more. Returns whether @@ -460,21 +503,14 @@ impl Socket { } if config.gro { - let on: libc::c_int = 1; - // SAFETY: valid fd, valid option buffer. - let ret = unsafe { - libc::setsockopt( - io.as_raw_fd(), - libc::SOL_UDP, - libc::UDP_GRO, - (&raw const on).cast(), - std::mem::size_of::() as libc::socklen_t, - ) - }; - if ret != 0 { - return Err(io::Error::last_os_error().into()); - } + set_option(&io, libc::SOL_UDP, libc::UDP_GRO)?; } + // Receive the ECN codepoint with each datagram. A v6 socket takes + // both: Linux reports a v4-mapped datagram's mark as `IP_TOS`. + if io.local_addr()?.is_ipv6() { + set_option(&io, libc::IPPROTO_IPV6, libc::IPV6_RECVTCLASS)?; + } + set_option(&io, libc::IPPROTO_IP, libc::IP_RECVTOS)?; // The ring's entry count is fixed at registration, so it is sized for // the ceiling; the buffers behind it are allocated as the pool grows. @@ -581,6 +617,7 @@ impl Socket { len: queued.len, stride: queued.stride, from: queued.from, + ecn: queued.ecn, })); } if let Some(code) = rx.error { @@ -673,6 +710,7 @@ pub struct Packet { len: usize, stride: usize, from: SocketAddr, + ecn: Option, } impl Packet { @@ -681,6 +719,12 @@ impl Packet { self.from } + /// The ECN codepoint the datagrams arrived with; GRO only coalesces + /// datagrams that share one. + pub fn ecn(&self) -> Option { + self.ecn + } + /// The datagram size GRO coalesced with; the final datagram may be short. pub fn stride(&self) -> usize { self.stride @@ -718,6 +762,7 @@ impl std::fmt::Debug for Packet { .field("from", &self.from) .field("len", &self.len) .field("stride", &self.stride) + .field("ecn", &self.ecn) .finish() } } @@ -735,16 +780,17 @@ pub struct TxBuf { } impl TxBuf { - /// Send `self[..len]` on the owning socket, to `to`, as datagrams of - /// `segment` bytes (the last may be short). Fire-and-forget: the buffer - /// returns to the pool when the kernel completes, and a failed send - /// surfaces on the next pool acquire. + /// Send `self[..len]` 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. /// /// This only stages an SQE, so the datagram reaches the kernel when the /// worker next enters the ring. Dropping the worker makes a bounded attempt /// to submit staged datagrams and drain their completions, but does not /// guarantee kernel completion or delivery. - pub fn send(mut self, len: usize, to: SocketAddr, segment: usize) -> io::Result<()> { + pub fn send(mut self, transmit: Transmit) -> io::Result<()> { + let Transmit { to, len, segment, ecn } = transmit; // `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) { @@ -798,14 +844,19 @@ impl TxBuf { headers, }; + let one = SendOne { + to, + ecn, + segment: sock.config.gso.then_some(segment as u16), + }; if sock.config.gso { - send_one(&shared, &staging, 0, base, len, to, Some(segment as u16))?; + send_one(&shared, &staging, 0, base, len, &one)?; } else { for index in 0..segments { let offset = index * segment; let chunk = segment.min(len - offset); // SAFETY: offset stays within the leased buffer. - send_one(&shared, &staging, index, unsafe { base.add(offset) }, chunk, to, None)?; + send_one(&shared, &staging, index, unsafe { base.add(offset) }, chunk, &one)?; } } sock.metrics.tx_datagrams.add(segments as u64); @@ -884,15 +935,23 @@ impl Drop for SendOp { } } +/// What every datagram of one [`TxBuf::send`] shares. +struct SendOne { + to: SocketAddr, + ecn: Option, + /// The `UDP_SEGMENT` size, when the call is one GSO train. + segment: Option, +} + fn send_one( shared: &Rc, staging: &Staging, index: usize, base: *mut u8, len: usize, - to: SocketAddr, - segment: Option, + one: &SendOne, ) -> io::Result<()> { + let SendOne { to, ecn, segment } = *one; // SAFETY: `index` is within the headers `TxBuf::send` reserved, and every // operation gets its own. let hdr = unsafe { &mut *staging.headers.as_ptr().add(index) }; @@ -907,17 +966,44 @@ fn send_one( hdr.hdr.msg_iov = &raw mut hdr.iov; hdr.hdr.msg_iovlen = 1; - if let Some(segment) = segment { - hdr.hdr.msg_control = hdr.control.0.as_mut_ptr().cast(); - // SAFETY: the control buffer is zeroed, aligned, and large enough for - // one u16 control message. - unsafe { - hdr.hdr.msg_controllen = libc::CMSG_SPACE(std::mem::size_of::() as _) as usize; - let cmsg = libc::CMSG_FIRSTHDR(&hdr.hdr); + // SAFETY: the control buffer is zeroed, aligned, and holds both messages + // (`CMSG_SPACE` of a u16 and of an int fit twice over in `CONTROL_LEN`); + // `msg_controllen` is set to the total first so `CMSG_NXTHDR` walks it. + unsafe { + let mut space = 0; + if segment.is_some() { + space += libc::CMSG_SPACE(std::mem::size_of::() as _) as usize; + } + if ecn.is_some() { + space += libc::CMSG_SPACE(std::mem::size_of::() as _) as usize; + } + if space > 0 { + hdr.hdr.msg_control = hdr.control.0.as_mut_ptr().cast(); + hdr.hdr.msg_controllen = space; + } + let mut cmsg = libc::CMSG_FIRSTHDR(&hdr.hdr); + if let Some(segment) = segment { (*cmsg).cmsg_level = libc::SOL_UDP; (*cmsg).cmsg_type = libc::UDP_SEGMENT; (*cmsg).cmsg_len = libc::CMSG_LEN(std::mem::size_of::() as _) as usize; std::ptr::write_unaligned(libc::CMSG_DATA(cmsg).cast::(), segment); + cmsg = libc::CMSG_NXTHDR(&hdr.hdr, cmsg); + } + if let Some(ecn) = ecn { + // A v4-mapped destination on a v6 socket leaves as IPv4, so the + // mark rides `IP_TOS`; a native v6 destination takes `IPV6_TCLASS`. + let is_ipv4 = match to.ip() { + IpAddr::V4(_) => true, + IpAddr::V6(v6) => v6.to_ipv4_mapped().is_some(), + }; + let (level, kind) = match is_ipv4 { + true => (libc::IPPROTO_IP, libc::IP_TOS), + false => (libc::IPPROTO_IPV6, libc::IPV6_TCLASS), + }; + (*cmsg).cmsg_level = level; + (*cmsg).cmsg_type = kind; + (*cmsg).cmsg_len = libc::CMSG_LEN(std::mem::size_of::() as _) as usize; + std::ptr::write_unaligned(libc::CMSG_DATA(cmsg).cast::(), ecn as libc::c_int); } } let hdr_ptr = &raw const hdr.hdr; @@ -1135,7 +1221,7 @@ fn on_recv_multi(sock: &Rc, cqe: Cqe) -> Result<(u16, Option if payload.is_empty() { return Ok((bid, None)); } - let stride = gro_stride(out.control_data()).unwrap_or(payload.len()); + let meta = RecvMeta::parse(out.control_data()); let payload_start = payload.as_ptr() as usize - buf.data.as_ptr() as usize; Ok(( bid, @@ -1144,7 +1230,8 @@ fn on_recv_multi(sock: &Rc, cqe: Cqe) -> Result<(u16, Option start: payload_start, len: payload.len(), from, - stride, + stride: meta.stride.unwrap_or(payload.len()), + ecn: meta.ecn, }), )) } @@ -1173,7 +1260,7 @@ fn on_recv_oneshot(one: OneshotRecv, cqe: Cqe) -> Result<(u16, Option), return Ok((bid, None)); } let control = &one.control.0[..one.hdr.msg_controllen.min(CONTROL_LEN)]; - let stride = gro_stride(control).unwrap_or(len); + let meta = RecvMeta::parse(control); // `claimed` stays set: the packet owns the buffer until released. Ok(( bid, @@ -1182,7 +1269,8 @@ fn on_recv_oneshot(one: OneshotRecv, cqe: Cqe) -> Result<(u16, Option), start: 0, len, from, - stride, + stride: meta.stride.unwrap_or(len), + ecn: meta.ecn, }), )) } @@ -1206,38 +1294,76 @@ pub(crate) fn on_send(op: SendOp, cqe: Cqe) { } } -/// The `UDP_GRO` segment size in a received control buffer, if present. -fn gro_stride(control: &[u8]) -> Option { - let header_len = unsafe { libc::CMSG_LEN(0) as usize }; - let mut offset = 0; - - while offset + header_len <= control.len() { - // SAFETY: bounds-checked read of a cmsghdr-sized prefix. - let header = unsafe { control.as_ptr().add(offset).cast::().read_unaligned() }; - let message_len = header.cmsg_len; - if message_len < header_len || offset + message_len > control.len() { - return None; - } - if header.cmsg_level == libc::SOL_UDP && header.cmsg_type == libc::UDP_GRO { - if message_len < header_len + std::mem::size_of::() { - return None; +/// 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. + stride: Option, + /// The ECN codepoint from `IP_TOS` or `IPV6_TCLASS`, if marked. + ecn: Option, +} + +impl RecvMeta { + /// Walk the control messages; a malformed buffer ends the walk with what + /// was read so far. + fn parse(control: &[u8]) -> Self { + let mut meta = Self::default(); + let header_len = unsafe { libc::CMSG_LEN(0) as usize }; + let mut offset = 0; + + while offset + header_len <= control.len() { + // SAFETY: bounds-checked read of a cmsghdr-sized prefix. + let header = unsafe { control.as_ptr().add(offset).cast::().read_unaligned() }; + let message_len = header.cmsg_len; + if message_len < header_len || offset + message_len > control.len() { + return meta; } - // SAFETY: length-checked just above. - let value = unsafe { - control - .as_ptr() - .add(offset + header_len) - .cast::() - .read_unaligned() - }; - return usize::try_from(value).ok(); + 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()); + } + // Linux reports the TOS byte itself, but the traffic class as an int. + (libc::IPPROTO_IP, libc::IP_TOS) => { + meta.ecn = data.first().and_then(|bits| Ecn::from_bits(*bits)); + } + (libc::IPPROTO_IPV6, libc::IPV6_TCLASS) => { + meta.ecn = read_int(data).and_then(|value| Ecn::from_bits(value as u8)); + } + _ => {} + } + // SAFETY: CMSG_SPACE is a pure size computation. + let aligned = unsafe { libc::CMSG_SPACE((message_len - header_len) as _) as usize }; + offset = offset.saturating_add(aligned.max(header_len)); } - // SAFETY: CMSG_SPACE is a pure size computation. - let aligned = unsafe { libc::CMSG_SPACE((message_len - header_len) as _) as usize }; - offset = offset.saturating_add(aligned.max(header_len)); + + meta } +} + +/// A control message's payload as the int the kernel wrote, if it is one. +fn read_int(data: &[u8]) -> Option { + let bytes = data.get(..std::mem::size_of::())?; + Some(libc::c_int::from_ne_bytes(bytes.try_into().ok()?)) +} - None +/// Turn a boolean socket option on. +fn set_option(io: &UdpSocket, level: libc::c_int, name: libc::c_int) -> io::Result<()> { + let on: libc::c_int = 1; + // SAFETY: valid fd, valid option buffer. + let ret = unsafe { + libc::setsockopt( + io.as_raw_fd(), + level, + name, + (&raw const on).cast(), + std::mem::size_of::() as libc::socklen_t, + ) + }; + match ret { + 0 => Ok(()), + _ => Err(io::Error::last_os_error()), + } } /// Write `addr` into `out`, returning the length the kernel wants. diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index 43c2302da4..9a3362e4da 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -732,7 +732,15 @@ mod tests { assert!(handle.udp(bind(), udp::Config::default()).is_err()); assert!(matches!(sock.poll_recv(&kio::Waiter::noop()), Poll::Ready(Err(_)))); assert!(matches!(sock.poll_acquire(&kio::Waiter::noop()), Poll::Ready(Err(_)))); - assert!(tx.send(1200, to, 1200).is_err()); + assert!( + tx.send(udp::Transmit { + to, + len: 1200, + segment: 1200, + ecn: None, + }) + .is_err() + ); // And a late spawn is dropped rather than parked forever. handle.spawn(async {}); drop(sock); @@ -870,7 +878,13 @@ mod tests { } assert_eq!(held.len(), usize::from(ceiling)); for tx in held.drain(..) { - tx.send(1200, to, 1200).expect("send"); + tx.send(udp::Transmit { + to, + len: 1200, + segment: 1200, + ecn: None, + }) + .expect("send"); } // The point of the test is the overflow, so prove it happened: the @@ -983,7 +997,13 @@ mod tests { let Poll::Ready(Ok(tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("no tx buffer"); }; - tx.send(64 * 1024, to, 1000).expect("send 66 datagrams"); + tx.send(udp::Transmit { + to, + len: 64 * 1024, + segment: 1000, + ecn: None, + }) + .expect("send 66 datagrams"); drop(worker); } @@ -1005,7 +1025,14 @@ mod tests { let Poll::Ready(Ok(tx)) = sock.poll_acquire(&kio::Waiter::noop()) else { panic!("no tx buffer"); }; - let err = tx.send(64 * 1024, to, 1).expect_err("65536 datagrams from one buffer"); + let err = tx + .send(udp::Transmit { + to, + len: 64 * 1024, + segment: 1, + ecn: None, + }) + .expect_err("65536 datagrams from one buffer"); assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); drop(worker); } @@ -1027,7 +1054,12 @@ mod tests { // `UDP_SEGMENT` is a u16: without validation this would truncate to a // one-byte stride instead of one segment. let err = tx - .send(60_000, to, usize::from(u16::MAX) + 2) + .send(udp::Transmit { + to, + len: 60_000, + segment: usize::from(u16::MAX) + 2, + ecn: None, + }) .expect_err("oversized segment"); assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); drop(worker); @@ -1059,7 +1091,13 @@ mod tests { panic!("no tx buffer"); }; tx[..4 * 1200].fill(7); - tx.send(4 * 1200, to, 1200).expect("send"); + tx.send(udp::Transmit { + to, + len: 4 * 1200, + segment: 1200, + ecn: None, + }) + .expect("send"); // Drive the worker until the loopback delivers, parking on a timer each // turn so the park and timer counters see traffic too. @@ -1136,7 +1174,13 @@ mod tests { 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(1200, to, 1200).expect("send"); + tx.send(udp::Transmit { + to, + len: 1200, + segment: 1200, + ecn: None, + }) + .expect("send"); // Hold the received packet: its buffer is the pool, so the re-arm has // nowhere to receive into. diff --git a/rs/moq-uring/tests/allocations.rs b/rs/moq-uring/tests/allocations.rs index b7fe1e70d8..20c7ebdb35 100644 --- a/rs/moq-uring/tests/allocations.rs +++ b/rs/moq-uring/tests/allocations.rs @@ -82,7 +82,13 @@ fn steady_state_sends_do_not_allocate() { for _ in 0..100 { let mut tx = socket.acquire().await.expect("warmup acquire"); tx[..4800].fill(0x5a); - tx.send(4800, to, 1200).expect("warmup send"); + tx.send(udp::Transmit { + to, + len: 4800, + segment: 1200, + ecn: None, + }) + .expect("warmup send"); } let mut acquires = 0; @@ -97,7 +103,12 @@ fn steady_state_sends_do_not_allocate() { count_from_zero(); tx[..4800].fill(0x5a); - let sent = tx.send(4800, to, 1200); + let sent = tx.send(udp::Transmit { + to, + len: 4800, + segment: 1200, + ecn: None, + }); let (allocs, deallocs) = counted(); sent.expect("send"); stage_allocs += allocs; diff --git a/rs/moq-uring/tests/ecn.rs b/rs/moq-uring/tests/ecn.rs new file mode 100644 index 0000000000..50e053bad2 --- /dev/null +++ b/rs/moq-uring/tests/ecn.rs @@ -0,0 +1,114 @@ +//! ECN through the worker's UDP path: the codepoint a send asks for is what +//! the peer's `Packet::ecn` reads back, over IPv4 and IPv6, as a lone +//! datagram and as a GSO train, with CE surviving too. Without the TOS and +//! traffic-class control messages the peer sees no mark, its ACKs carry no +//! ECN counts, and noq disables ECN on the first one. +//! +//! Kernel-gated: skips loudly below the Linux 6.12 floor (GitHub-hosted CI), +//! and runs everywhere else. + +#![cfg(target_os = "linux")] + +use std::net::UdpSocket; + +use moq_uring::{Config, Error, Worker, udp}; + +fn worker() -> Option { + match Worker::new(Config::default()) { + Ok(worker) => Some(worker), + Err(Error::Unsupported(reason)) => { + eprintln!("skipping io_uring ecn test: {reason}"); + None + } + Err(err) => panic!("worker setup failed: {err}"), + } +} + +/// Send `datagrams` datagrams of `segment` bytes marked `ecn` from one +/// worker socket to another and return what the receiver read back. +fn round_trip(bind: &str, config: udp::Config, ecn: Option, datagrams: usize) -> Vec { + const SEGMENT: usize = 1200; + let Some(mut worker) = worker() else { return Vec::new() }; + let handle = worker.handle(); + + let rx = UdpSocket::bind(bind).expect("bind receiver"); + let tx = UdpSocket::bind(bind).expect("bind sender"); + let rx = handle.udp(rx, config.clone()).expect("receiver socket"); + let to = rx.local_addr().expect("receiver addr"); + let tx = handle.udp(tx, config).expect("sender socket"); + + worker + .block_on(async move { + let mut buf = tx.acquire().await.expect("acquire"); + let len = SEGMENT * datagrams; + buf[..len].fill(0xAB); + buf.send(udp::Transmit { + to, + len, + segment: SEGMENT, + ecn, + }) + .expect("send"); + + let mut packets = Vec::new(); + let mut received = 0; + while received < datagrams { + let packet = rx.recv().await.expect("recv"); + received += packet.payload().len().div_ceil(packet.stride()); + packets.push(packet); + } + packets + }) + .expect("worker") +} + +fn assert_marked(bind: &str, config: udp::Config, ecn: Option, datagrams: usize) { + let packets = round_trip(bind, config, ecn, datagrams); + for packet in &packets { + assert_eq!(packet.ecn(), ecn, "codepoint read back from {packet:?}"); + } +} + +fn ablated() -> udp::Config { + let mut config = udp::Config::default(); + config.gro = false; + config.gso = false; + config.multishot = false; + config +} + +#[test] +fn ect0_v4() { + assert_marked("127.0.0.1:0", udp::Config::default(), Some(udp::Ecn::Ect0), 1); +} + +#[test] +fn ect0_v6() { + assert_marked("[::1]:0", udp::Config::default(), Some(udp::Ecn::Ect0), 1); +} + +#[test] +fn ect0_gso_train() { + // One `sendmsg` carries the mark on every segment, and GRO keeps it. + assert_marked("127.0.0.1:0", udp::Config::default(), Some(udp::Ecn::Ect0), 8); + assert_marked("[::1]:0", udp::Config::default(), Some(udp::Ecn::Ect0), 8); +} + +#[test] +fn ect0_ablated() { + // One `sendmsg` per datagram and oneshot receives read the same mark. + assert_marked("127.0.0.1:0", ablated(), Some(udp::Ecn::Ect0), 4); + assert_marked("[::1]:0", ablated(), Some(udp::Ecn::Ect0), 4); +} + +#[test] +fn ce_survives() { + assert_marked("127.0.0.1:0", udp::Config::default(), Some(udp::Ecn::Ce), 1); + assert_marked("[::1]:0", udp::Config::default(), Some(udp::Ecn::Ce), 1); +} + +#[test] +fn unmarked() { + assert_marked("127.0.0.1:0", udp::Config::default(), None, 1); + assert_marked("[::1]:0", udp::Config::default(), None, 1); +} diff --git a/rs/moq-uring/tests/support/quiche.rs b/rs/moq-uring/tests/support/quiche.rs index f6e48eb0a9..015a7b5909 100644 --- a/rs/moq-uring/tests/support/quiche.rs +++ b/rs/moq-uring/tests/support/quiche.rs @@ -253,7 +253,12 @@ impl Peer { let Some(to) = dest else { return Ok(()); }; - tx.send(filled, to, SEGMENT)?; + tx.send(udp::Transmit { + to, + len: filled, + segment: SEGMENT, + ecn: None, + })?; } } diff --git a/rs/moq-uring/tests/teardown.rs b/rs/moq-uring/tests/teardown.rs index b66a647453..e7f0dee077 100644 --- a/rs/moq-uring/tests/teardown.rs +++ b/rs/moq-uring/tests/teardown.rs @@ -60,7 +60,13 @@ fn a_send_staged_on_the_way_out_still_goes() { .block_on(async { let mut tx = sock.acquire().await.expect("acquire"); tx[..3].copy_from_slice(b"bye"); - tx.send(3, peer_addr, 3).expect("send"); + tx.send(udp::Transmit { + to: peer_addr, + len: 3, + segment: 3, + ecn: None, + }) + .expect("send"); }) .expect("worker"); drop(sock);