diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f5bf02ef79..1aec54ce1c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -64,6 +64,48 @@ jobs: - name: Build and test noq-udp (posix_minimal) run: cargo test --locked -p noq-udp + test_musl: + name: Test ${{ matrix.target }} + if: "github.event_name != 'pull_request' || ! contains(github.event.pull_request.labels.*.name, 'flaky-test')" + timeout-minutes: 30 + strategy: + fail-fast: false + matrix: + include: + # musl binaries are statically linked, and each target runs on a runner + # of its own architecture, so the tests run natively. + - target: aarch64-unknown-linux-musl + runner: ubuntu-24.04-arm + - target: x86_64-unknown-linux-musl + runner: ubuntu-latest + runs-on: ${{ matrix.runner }} + env: + RUSTC_WRAPPER: "sccache" + SCCACHE_GHA_ENABLED: "on" + # When cross-compiling to musl, cc-rs looks for `-linux-musl-gcc`, + # but `musl-tools` only ships the host-native `musl-gcc`. + CC_aarch64_unknown_linux_musl: musl-gcc + CC_x86_64_unknown_linux_musl: musl-gcc + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@stable + with: + target: ${{ matrix.target }} + - uses: mozilla-actions/sccache-action@v0.0.9 + - name: Install musl toolchain + run: | + sudo apt-get update + sudo apt-get install -y musl-tools + - name: Install cargo-nextest + uses: taiki-e/install-action@v2 + with: + tool: nextest@0.9.80 + - name: Run tests + run: | + cargo nextest run --locked --workspace --exclude fuzz --lib --bins --tests --target ${{ matrix.target }} --profile ci + env: + RUST_LOG: ${{ runner.debug && 'TRACE' || 'DEBUG'}} + esp32_check: name: ESP32-C3 build check (noq-udp, noq-proto, noq) if: "github.event_name != 'pull_request' || ! contains(github.event.pull_request.labels.*.name, 'flaky-test')" diff --git a/noq-udp/src/apple_fast.rs b/noq-udp/src/apple_fast.rs index ec9e23d08a..cb276178a7 100644 --- a/noq-udp/src/apple_fast.rs +++ b/noq-udp/src/apple_fast.rs @@ -33,7 +33,7 @@ fn send_via_sendmsg_x( ) -> io::Result<()> { let mut hdrs = unsafe { mem::zeroed::<[msghdr_x; BATCH_SIZE]>() }; let mut iovs = unsafe { mem::zeroed::<[libc::iovec; BATCH_SIZE]>() }; - let mut ctrls = [cmsg::Aligned([0u8; cmsg::LEN]); BATCH_SIZE]; + let mut ctrls = [cmsg::SendBuf::zeroed(); BATCH_SIZE]; let addr = socket2::SockAddr::from(transmit.destination); let segment_size = transmit.segment_size.unwrap_or(transmit.contents.len()); let mut cnt = 0; @@ -75,7 +75,7 @@ fn prepare_msg_x( dst_addr: &socket2::SockAddr, hdr: &mut msghdr_x, iov: &mut libc::iovec, - ctrl: &mut cmsg::Aligned<[u8; cmsg::LEN]>, + ctrl: &mut cmsg::SendBuf, #[allow(unused_variables)] encode_src_ip: bool, sendmsg_einval: bool, ) { @@ -89,18 +89,18 @@ fn prepare_msg_x( hdr.msg_iov = iov; hdr.msg_iovlen = 1; - hdr.msg_control = ctrl.0.as_mut_ptr() as _; - hdr.msg_controllen = cmsg::LEN as _; + hdr.msg_control = ctrl.as_mut_ptr() as _; + hdr.msg_controllen = ctrl.len() as _; let mut encoder = unsafe { cmsg::Encoder::new(hdr) }; let ecn = transmit.ecn.map_or(0, |x| x as libc::c_int); let is_ipv4 = transmit.destination.is_ipv4() || matches!(transmit.destination.ip(), IpAddr::V6(addr) if addr.to_ipv4_mapped().is_some()); if is_ipv4 { if !sendmsg_einval { - encoder.push(libc::IPPROTO_IP, libc::IP_TOS, ecn as IpTosTy); + encoder.push_ecn_v4(ecn as IpTosTy); } } else { - encoder.push(libc::IPPROTO_IPV6, libc::IPV6_TCLASS, ecn); + encoder.push_ecn_v6(ecn); } if let Some(ip) = &transmit.src_ip { @@ -110,7 +110,7 @@ fn prepare_msg_x( let addr = libc::in_addr { s_addr: u32::from_ne_bytes(v4.octets()), }; - encoder.push(libc::IPPROTO_IP, libc::IP_RECVDSTADDR, addr); + encoder.push_src_addr_v4(addr); } } IpAddr::V6(v6) => { @@ -120,7 +120,7 @@ fn prepare_msg_x( s6_addr: v6.octets(), }, }; - encoder.push(libc::IPPROTO_IPV6, libc::IPV6_PKTINFO, pktinfo); + encoder.push_pktinfo_v6(pktinfo); } } } @@ -156,7 +156,7 @@ pub(crate) fn recv_via_recvmsg_x( // uninitialized memory, do not use `MaybeUninit` for `ctrls`, instead // initialize `ctrls` with `0`s. A control message of all `0`s is // automatically skipped by `libc::CMSG_NXTHDR`. - let mut ctrls = [cmsg::Aligned([0u8; cmsg::LEN]); BATCH_SIZE]; + let mut ctrls = [cmsg::RecvBuf::zeroed(); BATCH_SIZE]; let mut hdrs = unsafe { mem::zeroed::<[msghdr_x; BATCH_SIZE]>() }; let max_msg_count = bufs.len().min(BATCH_SIZE); for i in 0..max_msg_count { @@ -178,15 +178,15 @@ pub(crate) fn recv_via_recvmsg_x( fn prepare_recv_x( buf: &mut IoSliceMut<'_>, name: &mut MaybeUninit, - ctrl: &mut cmsg::Aligned<[u8; cmsg::LEN]>, + ctrl: &mut cmsg::RecvBuf, hdr: &mut msghdr_x, ) { hdr.msg_name = name.as_mut_ptr() as _; hdr.msg_namelen = size_of::() as _; hdr.msg_iov = buf as *mut IoSliceMut<'_> as *mut libc::iovec; hdr.msg_iovlen = 1; - hdr.msg_control = ctrl.0.as_mut_ptr() as _; - hdr.msg_controllen = cmsg::LEN as _; + hdr.msg_control = ctrl.as_mut_ptr() as _; + hdr.msg_controllen = ctrl.len() as _; hdr.msg_flags = 0; hdr.msg_datalen = buf.len(); } @@ -248,4 +248,8 @@ impl MsgHdr for msghdr_x { fn control_len(&self) -> usize { self.msg_controllen as _ } + + fn recv_flags(&self) -> libc::c_int { + self.msg_flags + } } diff --git a/noq-udp/src/cmsg/mod.rs b/noq-udp/src/cmsg/mod.rs index 3e82cbabf7..58c6c0934d 100644 --- a/noq-udp/src/cmsg/mod.rs +++ b/noq-udp/src/cmsg/mod.rs @@ -1,6 +1,7 @@ use std::{ ffi::{c_int, c_uchar}, ptr, + sync::atomic::{AtomicBool, Ordering}, }; #[cfg(unix)] @@ -11,7 +12,7 @@ mod imp; #[path = "windows.rs"] mod imp; -pub(crate) use imp::Aligned; +pub(crate) use imp::{PAYLOAD_ALIGN, RecvBuf, SendBuf}; /// Helper to encode a series of control messages (native "cmsgs") to a buffer for use in `sendmsg` // like API. @@ -39,11 +40,17 @@ impl<'a, M: MsgHdr> Encoder<'a, M> { /// Append a control message to the buffer. /// + /// Private: each message we send has its own method, next to the size covering it. + /// /// # Panics /// - If insufficient buffer space remains. - /// - If `T` has stricter alignment requirements than `M::ControlMessage` - pub(crate) fn push(&mut self, level: c_int, ty: c_int, value: T) { - assert!(align_of::() <= align_of::()); + fn push(&mut self, level: c_int, ty: c_int, value: T) { + const { + assert!( + align_of::() <= PAYLOAD_ALIGN, + "control message payload is more aligned than a control message buffer can be", + ); + } let space = M::ControlMessage::cmsg_space(size_of_val(&value)); assert!( self.hdr.control_len() >= self.len + space, @@ -74,11 +81,31 @@ impl Drop for Encoder<'_, M> { } } +/// Warns once if the kernel had more to say about a datagram than the buffer could hold. +/// +/// The dropped messages cost us metadata, at worst the GRO segment size. `RECV_LEN` covers +/// every option we enable, so one is unaccounted for, or the caller enabled their own on +/// the socket they gave us. +pub(crate) fn warn_if_control_truncated(hdr: &impl MsgHdr) { + static WARNED: AtomicBool = AtomicBool::new(false); + + if hdr.recv_flags() & imp::MSG_CTRUNC != 0 && !WARNED.swap(true, Ordering::Relaxed) { + crate::log::warn!( + "control messages truncated on receive, some datagram metadata was dropped" + ); + } +} + /// # Safety /// /// `cmsg` must refer to a native cmsg containing a payload of type `T` pub(crate) unsafe fn decode(cmsg: &impl CMsgHdr) -> T { - assert!(align_of::() <= align_of::()); + const { + assert!( + align_of::() <= PAYLOAD_ALIGN, + "control message payload is more aligned than a control message buffer can be", + ); + } debug_assert_eq!(cmsg.len(), C::cmsg_len(size_of::())); unsafe { ptr::read(cmsg.cmsg_data() as *const T) } } @@ -138,6 +165,9 @@ pub(crate) trait MsgHdr { fn set_control_len(&mut self, len: usize); fn control_len(&self) -> usize; + + /// The flags the kernel set on a received message, i.e. `msg_flags`. + fn recv_flags(&self) -> c_int; } pub(crate) trait CMsgHdr { @@ -151,6 +181,3 @@ pub(crate) trait CMsgHdr { fn len(&self) -> usize; } - -#[cfg(unix)] -pub(crate) const LEN: usize = 96; diff --git a/noq-udp/src/cmsg/unix.rs b/noq-udp/src/cmsg/unix.rs index 112bd5ebef..26d28894e2 100644 --- a/noq-udp/src/cmsg/unix.rs +++ b/noq-udp/src/cmsg/unix.rs @@ -1,10 +1,162 @@ -use std::ffi::{c_int, c_uchar}; +use std::{ + ffi::{c_int, c_uchar}, + mem::MaybeUninit, +}; -use super::{CMsgHdr, MsgHdr}; +use super::{CMsgHdr, Encoder, MsgHdr}; +// netbsd sends no IP_TOS control message, so it has no payload type for one. +#[cfg(not(target_os = "netbsd"))] +use crate::imp::IpTosTy; +/// Every payload we put into, or read out of, a control message on this platform. +/// +/// A payload slot holds any one of these, so the largest of them sizes a message. #[derive(Copy, Clone)] -#[repr(align(8))] // Conservative bound for align_of -pub(crate) struct Aligned(pub(crate) T); +#[repr(C)] +#[allow(dead_code)] // the fields are here for their size, nothing reads them +pub(crate) union Payload { + #[cfg(not(target_os = "netbsd"))] + ecn_v4: IpTosTy, + ecn_v6: c_int, + /// `IP_TOS` and, on Darwin, `IPV6_TCLASS` come back as a single byte. + ecn_byte: u8, + segment_size: u16, + #[cfg(not(target_os = "redox"))] + pktinfo_v6: libc::in6_pktinfo, + #[cfg(any(target_os = "linux", target_os = "android"))] + pktinfo_v4: libc::in_pktinfo, + #[cfg(any(bsd, apple, solarish))] + dst_addr_v4: libc::in_addr, + #[cfg(any(target_os = "linux", target_os = "android"))] + timestamp: libc::timespec, +} + +/// Set in `msg_flags` when control messages did not fit in the buffer. +pub(crate) const MSG_CTRUNC: c_int = libc::MSG_CTRUNC; + +/// The buffer space one control message with a payload of this size takes up. +/// +/// +const fn cmsg_space(payload_len: usize) -> usize { + unsafe { libc::CMSG_SPACE(payload_len as _) as usize } +} + +/// The weaker of two alignments, i.e. the largest power of two dividing both. +const fn common_align(a: usize, b: usize) -> usize { + // The lower of the two lowest set bits decides the trailing zeros of the OR. + 1 << (a | b).trailing_zeros() +} + +/// The alignment a control message payload is guaranteed to have. +/// +/// Payloads sit `CMSG_LEN(0)` into their message and messages a sum of `CMSG_SPACE`s into +/// the buffer, so it is what those offsets and [`ControlBuf`]'s alignment share. +pub(crate) const PAYLOAD_ALIGN: usize = common_align( + common_align(unsafe { libc::CMSG_LEN(0) } as usize, cmsg_space(1)), + align_of::>(), +); + +/// Space for one control message carrying any of our payloads. +const MESSAGE_LEN: usize = cmsg_space(size_of::()); + +/// Space for the control messages one `sendmsg` can carry. +/// +/// ECN, GSO segment size and source address, one each; the v4 and v6 forms are exclusive. +pub(crate) const SEND_LEN: usize = 3 * MESSAGE_LEN; + +/// Space for the control messages the kernel can attach to one received datagram. +/// +/// TOS or traffic class, packet info, GRO segment size and receive timestamp, one each, +/// matching the options `UdpSocketState::new` enables. +pub(crate) const RECV_LEN: usize = 4 * MESSAGE_LEN; + +/// A control message buffer of `N` bytes. +#[derive(Copy, Clone)] +#[repr(C)] +pub(crate) struct ControlBuf { + /// Aligns the buffer like the `size_t` the `CMSG_*` macros round offsets to, which + /// covers the headers too: no platform aligns `cmsghdr` more strictly than that. + /// Zero sized: `repr(align)` takes a literal, not an expression. + _align: [usize; 0], + bytes: [MaybeUninit; N], +} + +/// Control message buffer for one `sendmsg`. +pub(crate) type SendBuf = ControlBuf; + +/// Control message buffer for one `recvmsg`. +pub(crate) type RecvBuf = ControlBuf; + +impl ControlBuf { + /// A zeroed buffer, for sending. + pub(crate) const fn zeroed() -> Self { + Self { + _align: [], + bytes: [MaybeUninit::new(0); N], + } + } + + /// An uninitialised buffer, for receiving: the kernel initialises what it uses. + pub(crate) const fn uninit() -> Self { + Self { + _align: [], + bytes: [MaybeUninit::uninit(); N], + } + } + + pub(crate) fn as_mut_ptr(&mut self) -> *mut u8 { + self.bytes.as_mut_ptr().cast() + } + + /// The size of the buffer, for `msg_controllen`. + pub(crate) const fn len(&self) -> usize { + N + } +} + +/// The control messages we send. +/// +/// One method each rather than a generic `push`, keeping the set next to the [`SEND_LEN`] +/// covering it. +impl> Encoder<'_, M> { + /// Sets the ECN codepoint of an IPv4 or IPv4-mapped datagram. + #[cfg(not(target_os = "netbsd"))] + pub(crate) fn push_ecn_v4(&mut self, ecn: IpTosTy) { + self.push(libc::IPPROTO_IP, libc::IP_TOS, ecn); + } + + /// Sets the IPv6 traffic class, which carries the ECN codepoint. + #[cfg(not(target_os = "redox"))] + pub(crate) fn push_ecn_v6(&mut self, ecn: c_int) { + self.push(libc::IPPROTO_IPV6, libc::IPV6_TCLASS, ecn); + } + + /// Sets the GSO segment size the kernel splits an oversized datagram into. + #[cfg(any(target_os = "linux", target_os = "android"))] + pub(crate) fn push_segment_size(&mut self, segment_size: u16) { + self.push(libc::SOL_UDP, libc::UDP_SEGMENT, segment_size); + } + + /// Sets the source address of an IPv4 datagram. + #[cfg(any(target_os = "linux", target_os = "android"))] + pub(crate) fn push_pktinfo_v4(&mut self, pktinfo: libc::in_pktinfo) { + self.push(libc::IPPROTO_IP, libc::IP_PKTINFO, pktinfo); + } + + /// Sets the source address of an IPv4 datagram. + /// + /// `IP_RECVDSTADDR` is `IP_SENDSRCADDR` on FreeBSD, the two have the same value. + #[cfg(any(bsd, apple, solarish))] + pub(crate) fn push_src_addr_v4(&mut self, addr: libc::in_addr) { + self.push(libc::IPPROTO_IP, libc::IP_RECVDSTADDR, addr); + } + + /// Sets the source address of an IPv6 datagram. + #[cfg(not(target_os = "redox"))] + pub(crate) fn push_pktinfo_v6(&mut self, pktinfo: libc::in6_pktinfo) { + self.push(libc::IPPROTO_IPV6, libc::IPV6_PKTINFO, pktinfo); + } +} /// Helpers for [`libc::msghdr`] impl MsgHdr for libc::msghdr { @@ -30,6 +182,10 @@ impl MsgHdr for libc::msghdr { fn control_len(&self) -> usize { self.msg_controllen as _ } + + fn recv_flags(&self) -> c_int { + self.msg_flags + } } /// Helpers for [`libc::cmsghdr`] @@ -56,3 +212,90 @@ impl CMsgHdr for libc::cmsghdr { self.cmsg_len as _ } } + +#[cfg(test)] +mod tests { + use std::mem; + + use super::*; + + /// The payload of every control message we can send in one `sendmsg`. + /// + /// `IpTosTy` is `c_int` or smaller everywhere it exists, so `c_int` stands in for it. + fn sent_payload_lens() -> Vec { + vec![ + size_of::(), // IP_TOS or IPV6_TCLASS + size_of::(), // UDP_SEGMENT + // IP_PKTINFO, IP_RECVDSTADDR or IPV6_PKTINFO + size_of::(), + ] + } + + /// The payload of every control message the kernel can attach to one datagram. + fn received_payload_lens() -> Vec { + vec![ + size_of::(), // IP_TOS or IPV6_TCLASS + size_of::(), // IP_PKTINFO or IPV6_PKTINFO + size_of::(), // UDP_GRO + #[cfg(any(target_os = "linux", target_os = "android"))] + size_of::(), // SCM_TIMESTAMPNS + ] + } + + fn libc_cmsg_space(payload_lens: &[usize]) -> usize { + payload_lens + .iter() + .map(|len| unsafe { libc::CMSG_SPACE(*len as _) as usize }) + .sum() + } + + /// The buffers hold every control message they have to. + /// + /// The constants count messages and assume the largest payload; this adds up the real + /// ones, so a message we failed to count shows up here, not as a truncated datagram. + #[test] + fn control_len_covers_libc() { + let sent = libc_cmsg_space(&sent_payload_lens()); + assert!(SEND_LEN >= sent, "SEND_LEN is {SEND_LEN}, need {sent}"); + + let received = libc_cmsg_space(&received_payload_lens()); + assert!( + RECV_LEN >= received, + "RECV_LEN is {RECV_LEN}, need {received}" + ); + } + + /// Every payload in a full buffer is aligned for the type read out of it. + /// + /// What `cmsg::decode` relies on. musl aligns `cmsghdr` to 4 where glibc aligns it to + /// 8, so aligning the buffer for it rather than for the macros breaks there. + /// + /// + /// + #[test] + fn payloads_are_aligned() { + let mut buf = RecvBuf::zeroed(); + let mut hdr: libc::msghdr = unsafe { mem::zeroed() }; + hdr.msg_control = buf.as_mut_ptr().cast(); + hdr.msg_controllen = buf.len() as _; + + // The largest payload we use, so the messages after the first sit where a real + // receive would put them. + let mut encoder = unsafe { Encoder::new(&mut hdr) }; + for _ in 0..received_payload_lens().len() { + encoder.push(libc::SOL_SOCKET, 0, Payload { ecn_v6: 0 }); + } + encoder.finish(); + + let mut count = 0; + for cmsg in unsafe { super::super::Iter::new(&hdr) } { + assert_eq!( + cmsg.cmsg_data() as usize % PAYLOAD_ALIGN, + 0, + "payload {count} is not aligned to {PAYLOAD_ALIGN}", + ); + count += 1; + } + assert_eq!(count, received_payload_lens().len()); + } +} diff --git a/noq-udp/src/cmsg/windows.rs b/noq-udp/src/cmsg/windows.rs index f4fecca439..3dea5cb07e 100644 --- a/noq-udp/src/cmsg/windows.rs +++ b/noq-udp/src/cmsg/windows.rs @@ -1,19 +1,152 @@ use std::{ ffi::{c_int, c_uchar}, - mem, ptr, + mem::{self, MaybeUninit}, + ptr, }; use windows_sys::Win32::Networking::WinSock; -use super::{CMsgHdr, MsgHdr}; +use super::{CMsgHdr, Encoder, MsgHdr}; +/// Every payload we put into, or read out of, a control message on this platform. +/// +/// A payload slot holds any one of these, so the largest of them sizes a message. +/// +/// +/// #[derive(Copy, Clone)] -#[repr(align(8))] // Conservative bound for align_of -pub(crate) struct Aligned(pub(crate) T); +#[repr(C)] +#[allow(dead_code)] // the fields are here for their size, nothing reads them +pub(crate) union Payload { + ecn: c_int, + segment_size: u32, + pktinfo_v4: WinSock::IN_PKTINFO, + pktinfo_v6: WinSock::IN6_PKTINFO, +} + +/// The alignment a control message payload is guaranteed to have. +/// +/// `WSA_CMSG_DATA` rounds the header size up to it and `WSA_CMSG_SPACE` keeps every +/// following header at a multiple of it, [`ControlBuf`] having at least as much. +pub(crate) const PAYLOAD_ALIGN: usize = mem::align_of::(); + +/// Set in `dwFlags` when control messages did not fit in the buffer. +pub(crate) const MSG_CTRUNC: c_int = WinSock::MSG_CTRUNC as c_int; + +// The four functions below follow the C macros in +// + +/// `WSA_CMSG_ALIGN`, which control message headers are aligned to. +const fn cmsghdr_align(len: usize) -> usize { + (len + mem::align_of::() - 1) & !(mem::align_of::() - 1) +} + +/// `WSA_CMSGDATA_ALIGN`, which control message payloads are aligned to. +const fn cmsgdata_align(len: usize) -> usize { + (len + PAYLOAD_ALIGN - 1) & !(PAYLOAD_ALIGN - 1) +} + +/// `WSA_CMSG_LEN`, the value of `cmsg_len` for a payload of `payload_len` bytes. +const fn cmsg_len(payload_len: usize) -> usize { + cmsgdata_align(mem::size_of::()) + payload_len +} + +/// `WSA_CMSG_SPACE`, the buffer space one control message with this payload takes up. +const fn cmsg_space(payload_len: usize) -> usize { + cmsgdata_align(mem::size_of::() + cmsghdr_align(payload_len)) +} + +/// Space for one control message carrying any of our payloads. +const MESSAGE_LEN: usize = cmsg_space(mem::size_of::()); + +/// Space for the control messages one `WSASendMsg` can carry. +/// +/// The ECN codepoint, the source address and the segment size, one each: the IPv4 and +/// IPv6 forms are mutually exclusive. +pub(crate) const SEND_LEN: usize = 3 * MESSAGE_LEN; + +/// Space for the control messages `WSARecvMsg` can return for one datagram. +/// +/// The ECN codepoint, the packet info and the URO coalesced size, one each. +pub(crate) const RECV_LEN: usize = 3 * MESSAGE_LEN; + +/// A control message buffer of `N` bytes. +#[derive(Copy, Clone)] +#[repr(C)] +pub(crate) struct ControlBuf { + /// Aligns the buffer like the `usize` `WSA_CMSGDATA_ALIGN` rounds to, which covers + /// the headers too: `CMSGHDR` is a `SIZE_T` and two `INT`s. + /// Zero sized: `repr(align)` takes a literal, not an expression. + _align: [usize; 0], + bytes: [MaybeUninit; N], +} + +/// Control message buffer for one `WSASendMsg`. +pub(crate) type SendBuf = ControlBuf; + +/// Control message buffer for one `WSARecvMsg`. +pub(crate) type RecvBuf = ControlBuf; + +impl ControlBuf { + /// A zeroed buffer. + pub(crate) const fn zeroed() -> Self { + Self { + _align: [], + bytes: [MaybeUninit::new(0); N], + } + } + + pub(crate) fn as_mut_ptr(&mut self) -> *mut u8 { + self.bytes.as_mut_ptr().cast() + } + + /// The size of the buffer, for `Control.len`. + pub(crate) const fn len(&self) -> usize { + N + } +} + +/// The control messages we send. +/// +/// One method each rather than a generic `push`, keeping the set next to the [`SEND_LEN`] +/// covering it. +impl> Encoder<'_, M> { + /// Sets the ECN codepoint of an IPv4 datagram. + pub(crate) fn push_ecn_v4(&mut self, ecn: c_int) { + self.push(WinSock::IPPROTO_IP, WinSock::IP_ECN, ecn); + } + + /// Sets the ECN codepoint of an IPv6 datagram. + pub(crate) fn push_ecn_v6(&mut self, ecn: c_int) { + self.push(WinSock::IPPROTO_IPV6, WinSock::IPV6_ECN, ecn); + } + + /// Sets the segment size the stack splits an oversized datagram into. + /// + /// + pub(crate) fn push_segment_size(&mut self, segment_size: u32) { + self.push( + WinSock::IPPROTO_UDP, + WinSock::UDP_SEND_MSG_SIZE, + segment_size, + ); + } + + /// Sets the source address of an IPv4 datagram. + pub(crate) fn push_pktinfo_v4(&mut self, pktinfo: WinSock::IN_PKTINFO) { + self.push(WinSock::IPPROTO_IP, WinSock::IP_PKTINFO, pktinfo); + } + + /// Sets the source address of an IPv6 datagram. + pub(crate) fn push_pktinfo_v6(&mut self, pktinfo: WinSock::IN6_PKTINFO) { + self.push(WinSock::IPPROTO_IPV6, WinSock::IPV6_PKTINFO, pktinfo); + } +} /// Helpers for [`WinSock::WSAMSG`] -// https://learn.microsoft.com/en-us/windows/win32/api/ws2def/ns-ws2def-wsamsg -// https://microsoft.github.io/windows-docs-rs/doc/windows/Win32/Networking/WinSock/struct.WSAMSG.html +/// +/// +/// impl MsgHdr for WinSock::WSAMSG { type ControlMessage = WinSock::CMSGHDR; @@ -43,18 +176,23 @@ impl MsgHdr for WinSock::WSAMSG { fn control_len(&self) -> usize { self.Control.len as _ } + + fn recv_flags(&self) -> c_int { + self.dwFlags as _ + } } /// Helpers for [`WinSock::CMSGHDR`] -// https://learn.microsoft.com/en-us/windows/win32/api/ws2def/ns-ws2def-wsacmsghdr -// https://microsoft.github.io/windows-docs-rs/doc/windows/Win32/Networking/WinSock/struct.CMSGHDR.html +/// +/// +/// impl CMsgHdr for WinSock::CMSGHDR { fn cmsg_len(length: usize) -> usize { - cmsgdata_align(mem::size_of::()) + length + cmsg_len(length) } fn cmsg_space(length: usize) -> usize { - cmsgdata_align(mem::size_of::() + cmsghdr_align(length)) + cmsg_space(length) } fn cmsg_data(&self) -> *mut c_uchar { @@ -72,12 +210,38 @@ impl CMsgHdr for WinSock::CMSGHDR { } } -// Helpers functions for `WinSock::WSAMSG` and `WinSock::CMSGHDR` are based on C macros from -// https://github.com/microsoft/win32metadata/blob/main/generation/WinSDK/RecompiledIdlHeaders/shared/ws2def.h#L741 -fn cmsghdr_align(length: usize) -> usize { - (length + mem::align_of::() - 1) & !(mem::align_of::() - 1) -} +#[cfg(test)] +mod tests { + use super::*; + + /// Every payload in a full buffer is aligned for the type read out of it. + /// + /// Encoding the whole send set also proves [`SEND_LEN`] covers it, `push` panicking + /// rather than overrunning. + #[test] + fn payloads_are_aligned() { + let mut buf = SendBuf::zeroed(); + let mut msg: WinSock::WSAMSG = unsafe { mem::zeroed() }; + msg.Control = WinSock::WSABUF { + buf: buf.as_mut_ptr(), + len: buf.len() as _, + }; + + let mut encoder = unsafe { Encoder::new(&mut msg) }; + encoder.push_pktinfo_v6(unsafe { mem::zeroed() }); + encoder.push_ecn_v6(0); + encoder.push_segment_size(1200); + encoder.finish(); -fn cmsgdata_align(length: usize) -> usize { - (length + mem::align_of::() - 1) & !(mem::align_of::() - 1) + let mut count = 0; + for cmsg in unsafe { super::super::Iter::new(&msg) } { + assert_eq!( + cmsg.cmsg_data() as usize % PAYLOAD_ALIGN, + 0, + "payload {count} is not aligned to {PAYLOAD_ALIGN}", + ); + count += 1; + } + assert_eq!(count, 3); + } } diff --git a/noq-udp/src/linux.rs b/noq-udp/src/linux.rs index 7d72a14aeb..13f8f3deb4 100644 --- a/noq-udp/src/linux.rs +++ b/noq-udp/src/linux.rs @@ -49,7 +49,7 @@ pub(super) mod gso { encoder: &mut cmsg::Encoder<'_, libc::msghdr>, segment_size: u16, ) { - encoder.push(libc::SOL_UDP, libc::UDP_SEGMENT, segment_size); + encoder.push_segment_size(segment_size); } // Avoid calling `supported_by_current_kernel` for each socket by using `OnceLock`. diff --git a/noq-udp/src/unix.rs b/noq-udp/src/unix.rs index c53e3448c3..3e17dd8bbc 100644 --- a/noq-udp/src/unix.rs +++ b/noq-udp/src/unix.rs @@ -53,28 +53,6 @@ pub struct UdpSocketState { impl UdpSocketState { pub fn new(sock: UdpSockRef<'_>) -> io::Result { let io = sock.0; - let mut cmsg_platform_space = 0; - #[cfg(not(target_os = "redox"))] - if cfg!(target_os = "linux") - || cfg!(bsd) - || cfg!(apple) - || cfg!(target_os = "android") - || cfg!(solarish) - { - cmsg_platform_space += - unsafe { libc::CMSG_SPACE(size_of::() as _) as usize }; - } - - assert!( - cmsg::LEN - >= unsafe { libc::CMSG_SPACE(size_of::() as _) as usize } - + cmsg_platform_space - ); - assert!( - align_of::() <= align_of::>(), - "control message buffers will be misaligned" - ); - io.set_nonblocking(true)?; let addr = io.local_addr()?; @@ -402,7 +380,7 @@ fn send( } let mut msg_hdr: libc::msghdr = unsafe { mem::zeroed() }; let mut iovec: libc::iovec = unsafe { mem::zeroed() }; - let mut cmsgs = cmsg::Aligned([0u8; cmsg::LEN]); + let mut cmsgs = cmsg::SendBuf::zeroed(); let dst_addr = socket2::SockAddr::from(transmit.destination); prepare_msg( transmit, @@ -478,7 +456,7 @@ pub(crate) fn send_single( ) -> io::Result<()> { let mut hdr: libc::msghdr = unsafe { mem::zeroed() }; let mut iov: libc::iovec = unsafe { mem::zeroed() }; - let mut ctrl = cmsg::Aligned([0u8; cmsg::LEN]); + let mut ctrl = cmsg::SendBuf::zeroed(); let addr = socket2::SockAddr::from(transmit.destination); prepare_msg( transmit, @@ -508,7 +486,7 @@ fn recv_via_recvmmsg( meta: &mut [RecvMeta], ) -> io::Result { let mut names = [MaybeUninit::::uninit(); BATCH_SIZE]; - let mut ctrls = [cmsg::Aligned(MaybeUninit::<[u8; cmsg::LEN]>::uninit()); BATCH_SIZE]; + let mut ctrls = [cmsg::RecvBuf::uninit(); BATCH_SIZE]; let mut hdrs = unsafe { mem::zeroed::<[libc::mmsghdr; BATCH_SIZE]>() }; let max_msg_count = bufs.len().min(BATCH_SIZE); for i in 0..max_msg_count { @@ -549,7 +527,7 @@ pub(crate) fn recv_single( meta: &mut [RecvMeta], ) -> io::Result { let mut name = MaybeUninit::::uninit(); - let mut ctrl = cmsg::Aligned(MaybeUninit::<[u8; cmsg::LEN]>::uninit()); + let mut ctrl = cmsg::RecvBuf::uninit(); let mut hdr = unsafe { mem::zeroed::() }; prepare_recv(&mut bufs[0], &mut name, &mut ctrl, &mut hdr); let n = loop { @@ -580,7 +558,7 @@ fn prepare_msg( dst_addr: &socket2::SockAddr, hdr: &mut libc::msghdr, iov: &mut libc::iovec, - ctrl: &mut cmsg::Aligned<[u8; cmsg::LEN]>, + ctrl: &mut cmsg::SendBuf, #[allow(unused_variables)] // only used on FreeBSD & macOS encode_src_ip: bool, sendmsg_einval: bool, @@ -600,8 +578,8 @@ fn prepare_msg( hdr.msg_iov = iov; hdr.msg_iovlen = 1; - hdr.msg_control = ctrl.0.as_mut_ptr() as _; - hdr.msg_controllen = cmsg::LEN as _; + hdr.msg_control = ctrl.as_mut_ptr() as _; + hdr.msg_controllen = ctrl.len() as _; let mut encoder = unsafe { cmsg::Encoder::new(hdr) }; let ecn = transmit.ecn.map_or(0, |x| x as libc::c_int); // True for IPv4 or IPv4-Mapped IPv6 @@ -611,12 +589,12 @@ fn prepare_msg( if !sendmsg_einval { #[cfg(not(target_os = "netbsd"))] { - encoder.push(libc::IPPROTO_IP, libc::IP_TOS, ecn as IpTosTy); + encoder.push_ecn_v4(ecn as IpTosTy); } } } else { #[cfg(not(target_os = "redox"))] - encoder.push(libc::IPPROTO_IPV6, libc::IPV6_TCLASS, ecn); + encoder.push_ecn_v6(ecn); } // On apple_fast, prepare_msg is only compiled for send_single (fallback path), while the main @@ -639,7 +617,7 @@ fn prepare_msg( }, ipi_addr: libc::in_addr { s_addr: 0 }, }; - encoder.push(libc::IPPROTO_IP, libc::IP_PKTINFO, pktinfo); + encoder.push_pktinfo_v4(pktinfo); } #[cfg(any(bsd, apple, solarish))] { @@ -647,7 +625,7 @@ fn prepare_msg( let addr = libc::in_addr { s_addr: u32::from_ne_bytes(v4.octets()), }; - encoder.push(libc::IPPROTO_IP, libc::IP_RECVDSTADDR, addr); + encoder.push_src_addr_v4(addr); } } } @@ -661,7 +639,7 @@ fn prepare_msg( s6_addr: v6.octets(), }, }; - encoder.push(libc::IPPROTO_IPV6, libc::IPV6_PKTINFO, pktinfo); + encoder.push_pktinfo_v6(pktinfo); } } } @@ -673,15 +651,15 @@ fn prepare_msg( fn prepare_recv( buf: &mut IoSliceMut<'_>, name: &mut MaybeUninit, - ctrl: &mut cmsg::Aligned>, + ctrl: &mut cmsg::RecvBuf, hdr: &mut libc::msghdr, ) { hdr.msg_name = name.as_mut_ptr() as _; hdr.msg_namelen = size_of::() as _; hdr.msg_iov = buf as *mut IoSliceMut<'_> as *mut libc::iovec; hdr.msg_iovlen = 1; - hdr.msg_control = ctrl.0.as_mut_ptr() as _; - hdr.msg_controllen = cmsg::LEN as _; + hdr.msg_control = ctrl.as_mut_ptr() as _; + hdr.msg_controllen = ctrl.len() as _; hdr.msg_flags = 0; } @@ -699,6 +677,8 @@ pub(crate) fn decode_recv>( timestamp: None, }; + cmsg::warn_if_control_truncated(hdr); + let cmsg_iter = unsafe { cmsg::Iter::new(hdr) }; for cmsg in cmsg_iter { ctrl.decode(cmsg); @@ -851,7 +831,7 @@ mod gso { } #[cfg(target_os = "freebsd")] -type IpTosTy = libc::c_uchar; +pub(crate) type IpTosTy = libc::c_uchar; #[cfg(not(any(target_os = "freebsd", target_os = "netbsd")))] pub(crate) type IpTosTy = libc::c_int; diff --git a/noq-udp/src/windows.rs b/noq-udp/src/windows.rs index d96eacad7b..c333d309d3 100644 --- a/noq-udp/src/windows.rs +++ b/noq-udp/src/windows.rs @@ -16,9 +16,7 @@ use libc::{c_int, c_uint}; use windows_sys::Win32::Networking::WinSock; use crate::{ - EcnCodepoint, IO_ERROR_LOG_INTERVAL, RecvMeta, Transmit, UdpSockRef, - cmsg::{self, CMsgHdr}, - log::debug, + EcnCodepoint, IO_ERROR_LOG_INTERVAL, RecvMeta, Transmit, UdpSockRef, cmsg, log::debug, log_sendmsg_error, }; @@ -38,17 +36,6 @@ pub struct UdpSocketState { impl UdpSocketState { pub fn new(socket: UdpSockRef<'_>) -> io::Result { - assert!( - CMSG_LEN - >= WinSock::CMSGHDR::cmsg_space(size_of::()) - + WinSock::CMSGHDR::cmsg_space(size_of::()) - + WinSock::CMSGHDR::cmsg_space(size_of::()) - ); - assert!( - align_of::() <= align_of::>(), - "control message buffers will be misaligned" - ); - socket.0.set_nonblocking(true)?; // Stop Windows from failing the next recv with WSAECONNRESET or WSAENETRESET when a @@ -277,7 +264,7 @@ impl UdpSocketState { let wsa_recvmsg_ptr = WSARECVMSG_PTR.expect("valid function pointer for WSARecvMsg"); // we cannot use [`socket2::MsgHdrMut`] as we do not have access to inner field which holds the WSAMSG - let mut ctrl_buf = cmsg::Aligned([0; CMSG_LEN]); + let mut ctrl_buf = cmsg::RecvBuf::zeroed(); let mut source: WinSock::SOCKADDR_INET = unsafe { mem::zeroed() }; let mut data = WinSock::WSABUF { buf: bufs[0].as_mut_ptr(), @@ -285,8 +272,8 @@ impl UdpSocketState { }; let ctrl = WinSock::WSABUF { - buf: ctrl_buf.0.as_mut_ptr(), - len: ctrl_buf.0.len() as _, + buf: ctrl_buf.as_mut_ptr(), + len: ctrl_buf.len() as _, }; let mut wsa_msg = WinSock::WSAMSG { @@ -312,6 +299,8 @@ impl UdpSocketState { } } + cmsg::warn_if_control_truncated(&wsa_msg); + let addr = unsafe { let (_, addr) = socket2::SockAddr::try_init(|addr_storage, len| { *len = size_of_val(&source) as _; @@ -447,7 +436,7 @@ fn is_unsupported_error(e: &io::Error) -> bool { fn send(state: &UdpSocketState, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> { // we cannot use [`socket2::sendmsg()`] and [`socket2::MsgHdr`] as we do not have access // to the inner field which holds the WSAMSG - let mut ctrl_buf = cmsg::Aligned([0; CMSG_LEN]); + let mut ctrl_buf = cmsg::SendBuf::zeroed(); let daddr = socket2::SockAddr::from(transmit.destination); let mut data = WinSock::WSABUF { @@ -456,8 +445,8 @@ fn send(state: &UdpSocketState, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) }; let ctrl = WinSock::WSABUF { - buf: ctrl_buf.0.as_mut_ptr(), - len: ctrl_buf.0.len() as _, + buf: ctrl_buf.as_mut_ptr(), + len: ctrl_buf.len() as _, }; let mut wsa_msg = WinSock::WSAMSG { @@ -485,7 +474,7 @@ fn send(state: &UdpSocketState, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) ipi_addr: src_ip.sin_addr, ipi_ifindex: 0, }; - encoder.push(WinSock::IPPROTO_IP, WinSock::IP_PKTINFO, pktinfo); + encoder.push_pktinfo_v4(pktinfo); } WinSock::AF_INET6 if state.pktinfo_v6_enabled.load(Ordering::Relaxed) => { let src_ip = unsafe { ptr::read(ip.as_ptr() as *const WinSock::SOCKADDR_IN6) }; @@ -493,7 +482,7 @@ fn send(state: &UdpSocketState, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) ipi6_addr: src_ip.sin6_addr, ipi6_ifindex: unsafe { src_ip.Anonymous.sin6_scope_id }, }; - encoder.push(WinSock::IPPROTO_IPV6, WinSock::IPV6_PKTINFO, pktinfo); + encoder.push_pktinfo_v6(pktinfo); } WinSock::AF_INET | WinSock::AF_INET6 => {} _ => { @@ -505,21 +494,17 @@ fn send(state: &UdpSocketState, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) let ecn = transmit.ecn.map_or(0, |x| x as c_int); if is_ipv4 { if state.ecn_v4_enabled.load(Ordering::Relaxed) { - encoder.push(WinSock::IPPROTO_IP, WinSock::IP_ECN, ecn); + encoder.push_ecn_v4(ecn); } } else { if state.ecn_v6_enabled.load(Ordering::Relaxed) { - encoder.push(WinSock::IPPROTO_IPV6, WinSock::IPV6_ECN, ecn); + encoder.push_ecn_v6(ecn); } } // Segment size is a u32 https://learn.microsoft.com/en-us/windows/win32/api/ws2tcpip/nf-ws2tcpip-wsasetudpsendmessagesize if let Some(segment_size) = transmit.effective_segment_size() { - encoder.push( - WinSock::IPPROTO_UDP, - WinSock::UDP_SEND_MSG_SIZE, - segment_size as u32, - ); + encoder.push_segment_size(segment_size as u32); } encoder.finish(); @@ -607,8 +592,6 @@ fn set_socket_option( } pub(crate) const BATCH_SIZE: usize = 1; -// Enough to store max(IP_PKTINFO + IP_ECN, IPV6_PKTINFO + IPV6_ECN) + max(UDP_SEND_MSG_SIZE, UDP_COALESCED_INFO) bytes (header + data) and some extra margin -const CMSG_LEN: usize = 128; const OPTION_ON: u32 = 1; static WSARECVMSG_PTR: LazyLock = LazyLock::new(|| { diff --git a/noq-udp/tests/tests.rs b/noq-udp/tests/tests.rs index 6d8601bf1f..67e1097257 100644 --- a/noq-udp/tests/tests.rs +++ b/noq-udp/tests/tests.rs @@ -244,6 +244,35 @@ fn gso() { ); } +/// A datagram that is both segmented and ECN marked. +/// +/// The traffic class arrives after the GRO segment size and the timestamp, so too small a +/// receive buffer drops it and ECN feedback disappears. `gso` sends no ECN and the `ecn_*` +/// tests send one segment, so neither puts enough control messages on a datagram to notice. +#[test] +#[cfg_attr(not(any(target_os = "linux", target_os = "android")), ignore)] +fn gso_with_ecn() { + let send = UdpSocket::bind((Ipv6Addr::LOCALHOST, 0)).unwrap(); + let recv = UdpSocket::bind((Ipv6Addr::LOCALHOST, 0)).unwrap(); + let max_segments = UdpSocketState::new((&send).into()) + .unwrap() + .max_gso_segments(); + let dst_addr = recv.local_addr().unwrap(); + const SEGMENT_SIZE: usize = 128; + let msg = vec![0xAB; SEGMENT_SIZE * max_segments.get()]; + test_send_recv( + &send.into(), + &recv.into(), + Transmit { + destination: dst_addr, + ecn: Some(EcnCodepoint::Ect0), + contents: &msg, + segment_size: Some(SEGMENT_SIZE), + src_ip: None, + }, + ); +} + #[test] fn socket_buffers() { const BUFFER_SIZE: usize = 123456;