diff --git a/CHANGELOG.md b/CHANGELOG.md index 89037c7..9daff2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -165,6 +165,10 @@ with v0.4.x or earlier peers. ### Changed +- Connected UDP peer drains now batch macOS receives with `recvmsg_x(2)`, + matching the wildcard UDP receive path instead of issuing one `recv(2)` + syscall per queued datagram. + - Inbound msg1 is classified before it is rate limited, and rekey or restart msg1 arriving on a link belonging to a promoted peer now draws on its own token bucket instead of competing with stranger admission for a single diff --git a/src/node/dataplane/connected_udp.rs b/src/node/dataplane/connected_udp.rs index 0bd2335..b22419f 100644 --- a/src/node/dataplane/connected_udp.rs +++ b/src/node/dataplane/connected_udp.rs @@ -149,7 +149,7 @@ impl Node { send_buf, ) .map_err(|e| format!("open_connected_fd: {e}"))?; - let socket = std::sync::Arc::new(crate::peer::connected_udp::ConnectedPeerSocket::from_fd( + let socket = std::sync::Arc::new(crate::transport::udp::ConnectedPeerSocket::from_fd( owned, peer_socket_addr, local_addr, @@ -157,7 +157,7 @@ impl Node { // Spawn the drain thread. It feeds `packet_tx` exactly like // the wildcard listen socket — rx_loop dispatches identically. - let drain = crate::peer::connected_udp::PeerRecvDrain::spawn( + let drain = crate::transport::udp::PeerRecvDrain::spawn( socket.clone(), transport_id, peer_socket_addr, diff --git a/src/node/encrypt_worker.rs b/src/node/encrypt_worker.rs index 677debb..7dec1d7 100644 --- a/src/node/encrypt_worker.rs +++ b/src/node/encrypt_worker.rs @@ -132,7 +132,7 @@ pub(crate) struct FmpSendJob { /// the job completes and the worker drops it, only the peer's /// strong ref remains. #[cfg(any(target_os = "linux", target_os = "macos"))] - pub connected_socket: Option>, + pub connected_socket: Option>, /// Bulk endpoint data may be dropped when the kernel reports UDP /// send-queue exhaustion. Control/rekey frames keep retrying so /// congestion cannot strand the session. @@ -715,7 +715,7 @@ fn mac_now_ms() -> u64 { struct MacSequencedSendFlow { key: MacSendFlowKey, socket: AsyncUdpSocket, - connected_socket: Option>, + connected_socket: Option>, dest_addr: SocketAddr, next_seq: std::sync::atomic::AtomicU64, last_used_ms: std::sync::atomic::AtomicU64, @@ -752,7 +752,7 @@ impl MacSequencedSendFlow { fn spawn( key: MacSendFlowKey, socket: AsyncUdpSocket, - connected_socket: Option>, + connected_socket: Option>, dest_addr: SocketAddr, now_ms: u64, ) -> Arc { @@ -1020,7 +1020,7 @@ fn flush_batch_sync( struct EncryptedGroup { socket: AsyncUdpSocket, #[cfg(any(target_os = "linux", target_os = "macos"))] - connected_socket: Option>, + connected_socket: Option>, dest_addr: SocketAddr, wire_packets: Vec>, drop_on_backpressure: bool, diff --git a/src/peer/active.rs b/src/peer/active.rs index 384cb66..2525768 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -147,14 +147,14 @@ struct PeerSendState { /// per-packet sockaddr handling + route lookup. Behind an `Arc` so /// in-flight worker jobs survive rekey/address-change rotations. #[cfg(any(target_os = "linux", target_os = "macos"))] - connected_udp: Option>, + connected_udp: Option>, /// Per-peer recv drain thread. Always paired with `connected_udp`: /// the kernel routes inbound packets from this peer to the /// connected socket, so it *must* be drained or the kernel recv /// buffer fills. Drop signals shutdown via self-pipe. #[cfg(any(target_os = "linux", target_os = "macos"))] - peer_recv_drain: Option, + peer_recv_drain: Option, // === Hot counters === /// Link statistics. @@ -475,7 +475,7 @@ impl ActivePeer { #[cfg(any(target_os = "linux", target_os = "macos"))] pub(crate) fn connected_udp( &self, - ) -> Option> { + ) -> Option> { self.send.connected_udp.clone() } @@ -485,8 +485,8 @@ impl ActivePeer { #[cfg(any(target_os = "linux", target_os = "macos"))] pub(crate) fn set_connected_udp( &mut self, - socket: std::sync::Arc, - drain: crate::peer::connected_udp::PeerRecvDrain, + socket: std::sync::Arc, + drain: crate::transport::udp::PeerRecvDrain, ) { // Drop the old drain BEFORE the old socket so its last fd // reference is released cleanly. diff --git a/src/peer/connected_udp/mod.rs b/src/peer/connected_udp/mod.rs deleted file mode 100644 index bb8d58a..0000000 --- a/src/peer/connected_udp/mod.rs +++ /dev/null @@ -1,16 +0,0 @@ -//! Per-peer connected-UDP fast-path handles. -//! -//! The connected-socket rationale and the fd-construction syscall -//! sequence live in `crate::transport::udp::open_connected_fd`. This -//! module owns the runtime handle types that adopt the resulting fd: -//! -//! - [`socket::ConnectedPeerSocket`] — the owning fd wrapper. -//! - [`drain::PeerRecvDrain`] — the recv-side drain thread that must -//! accompany every connected socket (the kernel routes the peer's -//! inbound packets to it, so it has to be drained). - -pub(crate) mod drain; -pub(crate) mod socket; - -pub(crate) use drain::PeerRecvDrain; -pub(crate) use socket::ConnectedPeerSocket; diff --git a/src/peer/mod.rs b/src/peer/mod.rs index 1443fe3..bd18617 100644 --- a/src/peer/mod.rs +++ b/src/peer/mod.rs @@ -6,8 +6,6 @@ //! 2. **ActivePeer** - Authenticated phase, after successful Noise handshake mod active; -#[cfg(any(target_os = "linux", target_os = "macos"))] -pub(crate) mod connected_udp; pub(crate) mod machine; pub use active::{ActivePeer, ConnectivityState}; diff --git a/src/transport/udp/io.rs b/src/transport/udp/io.rs deleted file mode 100644 index 6a2842a..0000000 --- a/src/transport/udp/io.rs +++ /dev/null @@ -1,1270 +0,0 @@ -//! UDP socket wrapper with platform-specific receive implementations. -//! -//! On Linux, provides `SO_RXQ_OVFL` kernel drop counter support via -//! `recvmsg()` ancillary data parsing. The async wrapper uses -//! `tokio::io::unix::AsyncFd` for integration with the tokio runtime. -//! -//! On macOS, uses the same `recvmsg()` path but without `SO_RXQ_OVFL` -//! (kernel drop counting is not available; the drops field returns 0). -//! -//! On Windows, uses `tokio::net::UdpSocket` directly (kernel drop -//! counting is not available; the drops field always returns 0). -//! -//! Follows the pattern established by `transport/ethernet/socket.rs`. - -use crate::transport::TransportError; -use socket2::{Domain, Protocol, Socket, Type}; -use std::net::SocketAddr; -use std::sync::Arc; -#[cfg(unix)] -use tracing::warn; - -// ============================================================================ -// Unix implementation -// ============================================================================ - -#[cfg(unix)] -mod platform { - use super::*; - use std::os::unix::io::{AsRawFd, RawFd}; - use tokio::io::unix::AsyncFd; - - /// Maximum number of datagrams a single recvmmsg / recvmsg_x / sendmmsg - /// syscall will pull from / push to the kernel. Tuned to amortise syscall + - /// per-task-wakeup overhead across a useful burst without blowing the - /// stack (each slot owns an mmsghdr/msghdr_x + sockaddr_storage + iovec). - #[cfg(any(target_os = "linux", target_os = "macos"))] - const BATCH_SIZE: usize = 32; - - /// Darwin-private `msghdr_x` for the `recvmsg_x` / `sendmsg_x` syscalls. - /// Layout matches `bsd/sys/socket_private.h` in xnu — same as `msghdr` plus - /// a trailing `msg_datalen` (per-message bytes-received output, in lieu of - /// the `msg_len` field that `mmsghdr` uses on Linux). - #[cfg(target_os = "macos")] - #[repr(C)] - #[allow(non_camel_case_types)] - struct msghdr_x { - msg_name: *mut libc::c_void, - msg_namelen: libc::socklen_t, - msg_iov: *mut libc::iovec, - msg_iovlen: libc::c_int, - msg_control: *mut libc::c_void, - msg_controllen: libc::socklen_t, - msg_flags: libc::c_int, - msg_datalen: usize, - } - - #[cfg(target_os = "macos")] - unsafe extern "C" { - fn recvmsg_x( - s: libc::c_int, - msgp: *const msghdr_x, - cnt: libc::c_uint, - flags: libc::c_int, - ) -> isize; - } - - /// Wrapper around a `socket2::Socket` providing sync send/recv with - /// `SO_RXQ_OVFL` ancillary data parsing. - pub struct UdpRawSocket { - inner: Socket, - local_addr: SocketAddr, - } - - impl UdpRawSocket { - /// Create, bind, and configure a UDP socket. - /// - /// Enables `SO_RXQ_OVFL` for kernel drop counting (non-fatal if - /// unsupported). Sets non-blocking mode for async integration. - pub fn open( - bind_addr: SocketAddr, - recv_buf_size: usize, - send_buf_size: usize, - ) -> Result { - let domain = if bind_addr.is_ipv4() { - Domain::IPV4 - } else { - Domain::IPV6 - }; - let sock = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP)) - .map_err(|e| TransportError::StartFailed(format!("socket create failed: {}", e)))?; - - sock.set_nonblocking(true).map_err(|e| { - TransportError::StartFailed(format!("set nonblocking failed: {}", e)) - })?; - - // SO_REUSEPORT lets per-peer `ConnectedPeerSocket`s bind - // to the same wildcard port the listen socket holds. Must - // be set BEFORE bind. Without this, the connected-UDP - // activation handler fails with EADDRINUSE on Linux and - // every outbound packet falls back to the wildcard listen - // socket — losing the kernel 5-tuple cache benefit and - // most of the multihop forwarding throughput gain. - let _ = sock.set_reuse_port(true); - let _ = sock.set_reuse_address(true); - - sock.bind(&bind_addr.into()) - .map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?; - - // Set socket buffer sizes - sock.set_recv_buffer_size(recv_buf_size) - .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; - sock.set_send_buffer_size(send_buf_size) - .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; - - let actual_recv = sock - .recv_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?; - let actual_send = sock - .send_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?; - - if actual_recv < recv_buf_size { - warn!( - requested = recv_buf_size, - actual = actual_recv, - "UDP recv buffer clamped by kernel (increase net.core.rmem_max)" - ); - } - if actual_send < send_buf_size { - warn!( - requested = send_buf_size, - actual = actual_send, - "UDP send buffer clamped by kernel (increase net.core.wmem_max)" - ); - } - - // Enable SO_RXQ_OVFL for kernel drop counter in recvmsg ancillary data. - // Non-fatal: older kernels or non-Linux platforms may not support it. - #[cfg(target_os = "linux")] - { - let enable: libc::c_int = 1; - let ret = unsafe { - libc::setsockopt( - sock.as_raw_fd(), - libc::SOL_SOCKET, - libc::SO_RXQ_OVFL, - &enable as *const _ as *const libc::c_void, - std::mem::size_of::() as libc::socklen_t, - ) - }; - if ret < 0 { - warn!( - "setsockopt(SO_RXQ_OVFL) failed: {}", - std::io::Error::last_os_error() - ); - } - } - - let local_addr = sock - .local_addr() - .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? - .as_socket() - .ok_or_else(|| { - TransportError::StartFailed("local address is not an IP socket".into()) - })?; - - Ok(Self { - inner: sock, - local_addr, - }) - } - - /// Adopt an existing bound UDP socket. - /// - /// This preserves socket identity/NAT mapping created by bootstrap code. - pub fn adopt( - socket: std::net::UdpSocket, - recv_buf_size: usize, - send_buf_size: usize, - ) -> Result { - let sock = Socket::from(socket); - - sock.set_nonblocking(true).map_err(|e| { - TransportError::StartFailed(format!("set nonblocking failed: {}", e)) - })?; - - sock.set_recv_buffer_size(recv_buf_size) - .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; - sock.set_send_buffer_size(send_buf_size) - .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; - - let actual_recv = sock - .recv_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?; - let actual_send = sock - .send_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?; - - if actual_recv < recv_buf_size { - warn!( - requested = recv_buf_size, - actual = actual_recv, - "UDP recv buffer clamped by kernel (increase net.core.rmem_max)" - ); - } - if actual_send < send_buf_size { - warn!( - requested = send_buf_size, - actual = actual_send, - "UDP send buffer clamped by kernel (increase net.core.wmem_max)" - ); - } - - #[cfg(target_os = "linux")] - { - let enable: libc::c_int = 1; - let ret = unsafe { - libc::setsockopt( - sock.as_raw_fd(), - libc::SOL_SOCKET, - libc::SO_RXQ_OVFL, - &enable as *const _ as *const libc::c_void, - std::mem::size_of::() as libc::socklen_t, - ) - }; - if ret < 0 { - warn!( - "setsockopt(SO_RXQ_OVFL) failed: {}", - std::io::Error::last_os_error() - ); - } - } - - let local_addr = sock - .local_addr() - .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? - .as_socket() - .ok_or_else(|| { - TransportError::StartFailed("local address is not an IP socket".into()) - })?; - - Ok(Self { - inner: sock, - local_addr, - }) - } - - /// Get the local bound address. - pub fn local_addr(&self) -> SocketAddr { - self.local_addr - } - - /// Get the actual receive buffer size granted by the kernel. - pub fn recv_buffer_size(&self) -> Result { - self.inner - .recv_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e))) - } - - /// Get the actual send buffer size granted by the kernel. - pub fn send_buffer_size(&self) -> Result { - self.inner - .send_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e))) - } - - /// Synchronous send to a destination address. - /// - /// Returns the number of bytes sent, or an `io::Error`. - pub fn send_to(&self, data: &[u8], dest: &SocketAddr) -> std::io::Result { - let dest: socket2::SockAddr = (*dest).into(); - self.inner.send_to(data, &dest) - } - - /// Synchronous receive with `SO_RXQ_OVFL` ancillary data parsing. - /// - /// Returns `(bytes_read, source_addr, kernel_drops)`. The `kernel_drops` - /// value is a cumulative counter since socket creation; it is 0 if - /// `SO_RXQ_OVFL` is not supported. - /// - /// The production receive path on Linux/macOS uses `recv_batch` - /// (recvmmsg / recvmsg_x); this single-packet variant remains for - /// other unix targets and for the local `tests` module. - #[cfg_attr(any(target_os = "linux", target_os = "macos"), allow(dead_code))] - pub fn recv_from(&self, buf: &mut [u8]) -> std::io::Result<(usize, SocketAddr, u32)> { - let fd = self.inner.as_raw_fd(); - - let mut iov = libc::iovec { - iov_base: buf.as_mut_ptr() as *mut libc::c_void, - iov_len: buf.len(), - }; - - // Control message buffer sized for SO_RXQ_OVFL (u32). - // CMSG_SPACE computes the aligned size including header. - #[cfg(target_os = "linux")] - const CMSG_BUF_SIZE: usize = unsafe { libc::CMSG_SPACE(4) } as usize; - #[cfg(not(target_os = "linux"))] - const CMSG_BUF_SIZE: usize = 64; - let mut cmsg_buf = [0u8; CMSG_BUF_SIZE]; - - let mut src_addr: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; - let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; - msg.msg_name = &mut src_addr as *mut _ as *mut libc::c_void; - msg.msg_namelen = std::mem::size_of::() as libc::socklen_t; - msg.msg_iov = &mut iov; - msg.msg_iovlen = 1 as _; - msg.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void; - msg.msg_controllen = cmsg_buf.len() as _; - - let n = unsafe { libc::recvmsg(fd, &mut msg, 0) }; - if n < 0 { - return Err(std::io::Error::last_os_error()); - } - - // Parse source address from sockaddr_storage - let addr = sockaddr_to_socket_addr(&src_addr)?; - - // Walk cmsg chain for SO_RXQ_OVFL drop counter - #[cfg(target_os = "linux")] - let mut drops: u32 = 0; - #[cfg(not(target_os = "linux"))] - let drops: u32 = 0; - #[cfg(target_os = "linux")] - unsafe { - let mut cmsg = libc::CMSG_FIRSTHDR(&msg); - while !cmsg.is_null() { - if (*cmsg).cmsg_level == libc::SOL_SOCKET - && (*cmsg).cmsg_type == libc::SO_RXQ_OVFL - { - let data = libc::CMSG_DATA(cmsg); - drops = std::ptr::read_unaligned(data as *const u32); - } - cmsg = libc::CMSG_NXTHDR(&msg, cmsg); - } - } - - Ok((n as usize, addr, drops)) - } - - /// Receive up to `BATCH_SIZE` datagrams in a single recvmmsg syscall - /// (Linux only — macOS falls through to per-packet recvmsg). - /// - /// Returns `(count, kernel_drops)`. Caller pre-sizes `bufs` (each - /// must be at least the configured MTU) and the matching `addrs` / - /// `lens` slices; on return, slots `[0..count)` are valid. - /// - /// `kernel_drops` is the `SO_RXQ_OVFL` cumulative counter sampled - /// from the cmsg chain of the FIRST datagram in the batch. The - /// counter is monotonic per-socket since `SO_RXQ_OVFL` was enabled, - /// so a single sample per batch is sufficient to feed the 1Hz - /// congestion detector in `sample_transport_congestion()`. Returns - /// `(0, 0)` on a spurious wakeup with no datagrams ready. - #[cfg(target_os = "linux")] - pub fn recv_batch( - &self, - bufs: &mut [&mut [u8]], - addrs: &mut [Option], - lens: &mut [usize], - ) -> std::io::Result<(usize, u32)> { - let n = bufs.len().min(addrs.len()).min(lens.len()).min(BATCH_SIZE); - if n == 0 { - return Ok((0, 0)); - } - let fd = self.inner.as_raw_fd(); - - // CMSG buffer wired to msgs[0] only. SO_RXQ_OVFL delivers a - // monotonic u32 drop counter; sampling once per batch gives - // the 1Hz congestion detector ample fresh values under load - // (one batch = up to 32 datagrams). - const CMSG_BUF_SIZE: usize = unsafe { libc::CMSG_SPACE(4) } as usize; - let mut cmsg_buf = [0u8; CMSG_BUF_SIZE]; - - // Stack-allocated parallel arrays; lifetime tied to this call. - let mut iovs: [libc::iovec; BATCH_SIZE] = unsafe { std::mem::zeroed() }; - let mut storages: [libc::sockaddr_storage; BATCH_SIZE] = unsafe { std::mem::zeroed() }; - let mut msgs: [libc::mmsghdr; BATCH_SIZE] = unsafe { std::mem::zeroed() }; - - for i in 0..n { - iovs[i].iov_base = bufs[i].as_mut_ptr() as *mut libc::c_void; - iovs[i].iov_len = bufs[i].len(); - msgs[i].msg_hdr.msg_name = &mut storages[i] as *mut _ as *mut libc::c_void; - msgs[i].msg_hdr.msg_namelen = - std::mem::size_of::() as libc::socklen_t; - msgs[i].msg_hdr.msg_iov = &mut iovs[i]; - msgs[i].msg_hdr.msg_iovlen = 1; - msgs[i].msg_len = 0; - } - // Only msgs[0] carries a cmsg buffer — sampling the OVFL counter - // there is enough since it is socket-wide and monotonic. - msgs[0].msg_hdr.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void; - msgs[0].msg_hdr.msg_controllen = cmsg_buf.len() as _; - - let r = unsafe { - libc::recvmmsg( - fd, - msgs.as_mut_ptr(), - n as libc::c_uint, - 0, - std::ptr::null_mut(), - ) - }; - if r < 0 { - return Err(std::io::Error::last_os_error()); - } - let count = r as usize; - for i in 0..count { - lens[i] = msgs[i].msg_len as usize; - addrs[i] = sockaddr_to_socket_addr(&storages[i]).ok(); - } - - // Walk msgs[0] cmsg chain for SO_RXQ_OVFL. Skip when no - // datagram landed (cmsg buffer is undefined in that case). - let mut drops: u32 = 0; - if count > 0 { - unsafe { - let mut cmsg = libc::CMSG_FIRSTHDR(&msgs[0].msg_hdr); - while !cmsg.is_null() { - if (*cmsg).cmsg_level == libc::SOL_SOCKET - && (*cmsg).cmsg_type == libc::SO_RXQ_OVFL - { - let data = libc::CMSG_DATA(cmsg); - drops = std::ptr::read_unaligned(data as *const u32); - } - cmsg = libc::CMSG_NXTHDR(&msgs[0].msg_hdr, cmsg); - } - } - } - - Ok((count, drops)) - } - - /// Receive up to `BATCH_SIZE` datagrams in a single `recvmsg_x` syscall - /// (macOS). Same `(count, drops)` contract as the Linux `recv_batch`, - /// except `drops` is always 0 — Darwin has no `SO_RXQ_OVFL` equivalent. - /// - /// `recvmsg_x` is a Darwin-private syscall (not in the public SDK) but - /// is shipped in production xnu and is used by quinn-udp for the same - /// per-syscall-amortisation reason as our Linux `recvmmsg` path. - #[cfg(target_os = "macos")] - pub fn recv_batch( - &self, - bufs: &mut [&mut [u8]], - addrs: &mut [Option], - lens: &mut [usize], - ) -> std::io::Result<(usize, u32)> { - let n = bufs.len().min(addrs.len()).min(lens.len()).min(BATCH_SIZE); - if n == 0 { - return Ok((0, 0)); - } - let fd = self.inner.as_raw_fd(); - - let mut iovs: [libc::iovec; BATCH_SIZE] = unsafe { std::mem::zeroed() }; - let mut storages: [libc::sockaddr_storage; BATCH_SIZE] = unsafe { std::mem::zeroed() }; - let mut msgs: [msghdr_x; BATCH_SIZE] = unsafe { std::mem::zeroed() }; - - for i in 0..n { - iovs[i].iov_base = bufs[i].as_mut_ptr() as *mut libc::c_void; - iovs[i].iov_len = bufs[i].len(); - msgs[i].msg_name = &mut storages[i] as *mut _ as *mut libc::c_void; - msgs[i].msg_namelen = - std::mem::size_of::() as libc::socklen_t; - msgs[i].msg_iov = &mut iovs[i]; - msgs[i].msg_iovlen = 1; - // No cmsg consumption — leave msg_control null. (msg_controllen - // is documented as not overwritten by macOS recvmsg_x; zeroed - // init keeps it sane.) - } - - let r = unsafe { recvmsg_x(fd, msgs.as_ptr(), n as libc::c_uint, 0) }; - if r < 0 { - return Err(std::io::Error::last_os_error()); - } - let count = r as usize; - for i in 0..count { - lens[i] = msgs[i].msg_datalen; - addrs[i] = sockaddr_to_socket_addr(&storages[i]).ok(); - } - - Ok((count, 0)) - } - - /// Wrap this socket in a tokio `AsyncFd` for async I/O. - pub fn into_async(self) -> Result { - let async_fd = AsyncFd::new(self) - .map_err(|e| TransportError::StartFailed(format!("AsyncFd::new failed: {}", e)))?; - Ok(AsyncUdpSocket { - inner: Arc::new(async_fd), - }) - } - } - - impl AsRawFd for UdpRawSocket { - fn as_raw_fd(&self) -> RawFd { - self.inner.as_raw_fd() - } - } - - /// Async wrapper around `UdpRawSocket` using tokio's `AsyncFd`. - /// - /// `Arc`-shareable between send and receive tasks. `AsyncFd` is - /// `Sync` when `T: Send`, which `socket2::Socket` satisfies. - #[derive(Clone)] - pub struct AsyncUdpSocket { - inner: Arc>, - } - - impl AsRawFd for AsyncUdpSocket { - fn as_raw_fd(&self) -> RawFd { - self.inner.get_ref().as_raw_fd() - } - } - - impl AsyncUdpSocket { - /// Send a payload to a destination address. - pub async fn send_to( - &self, - data: &[u8], - dest: &SocketAddr, - ) -> Result { - loop { - let mut guard = self - .inner - .writable() - .await - .map_err(|e| TransportError::SendFailed(format!("writable wait: {}", e)))?; - - match guard.try_io(|inner| inner.get_ref().send_to(data, dest)) { - Ok(Ok(n)) => return Ok(n), - Ok(Err(e)) => return Err(TransportError::SendFailed(format!("{}", e))), - Err(_would_block) => continue, - } - } - } - - /// Receive a payload, source address, and kernel drop counter. - /// - /// Returns `(bytes_read, source_addr, kernel_drops)`. On Linux/macOS - /// the production receive path uses `recv_batch`; this single-packet - /// variant remains for other unix targets and for the local `tests` - /// module. - #[cfg_attr(any(target_os = "linux", target_os = "macos"), allow(dead_code))] - pub async fn recv_from( - &self, - buf: &mut [u8], - ) -> Result<(usize, SocketAddr, u32), TransportError> { - loop { - let mut guard = self - .inner - .readable() - .await - .map_err(|e| TransportError::RecvFailed(format!("readable wait: {}", e)))?; - - match guard.try_io(|inner| inner.get_ref().recv_from(buf)) { - Ok(Ok(result)) => return Ok(result), - Ok(Err(e)) => return Err(TransportError::RecvFailed(format!("{}", e))), - Err(_would_block) => continue, - } - } - } - - /// Drain up to `BATCH_SIZE` datagrams from the kernel via - /// `recvmmsg` (Linux) or `recvmsg_x` (macOS). Returns - /// `(count, kernel_drops)`; same buffer / addr / len contract as - /// `UdpRawSocket::recv_batch`. `kernel_drops` is always 0 on macOS. - #[cfg(any(target_os = "linux", target_os = "macos"))] - pub async fn recv_batch( - &self, - bufs: &mut [&mut [u8]], - addrs: &mut [Option], - lens: &mut [usize], - ) -> Result<(usize, u32), TransportError> { - loop { - let mut guard = self - .inner - .readable() - .await - .map_err(|e| TransportError::RecvFailed(format!("readable wait: {}", e)))?; - - match guard.try_io(|inner| inner.get_ref().recv_batch(bufs, addrs, lens)) { - Ok(Ok((0, _))) => { - // Spurious wakeup or no datagrams ready — yield - // back to the reactor instead of busy-looping. - guard.clear_ready(); - continue; - } - Ok(Ok(result)) => return Ok(result), - Ok(Err(e)) => return Err(TransportError::RecvFailed(format!("{}", e))), - Err(_would_block) => continue, - } - } - } - } - - /// Convert a `libc::sockaddr_storage` to `std::net::SocketAddr`. - fn sockaddr_to_socket_addr(storage: &libc::sockaddr_storage) -> std::io::Result { - match storage.ss_family as libc::c_int { - libc::AF_INET => { - let addr: &libc::sockaddr_in = - unsafe { &*(storage as *const _ as *const libc::sockaddr_in) }; - let ip = std::net::Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr)); - let port = u16::from_be(addr.sin_port); - Ok(SocketAddr::from((ip, port))) - } - libc::AF_INET6 => { - let addr: &libc::sockaddr_in6 = - unsafe { &*(storage as *const _ as *const libc::sockaddr_in6) }; - let ip = std::net::Ipv6Addr::from(addr.sin6_addr.s6_addr); - let port = u16::from_be(addr.sin6_port); - Ok(SocketAddr::from((ip, port))) - } - family => Err(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!("unsupported address family: {}", family), - )), - } - } -} - -// ============================================================================ -// Windows implementation -// ============================================================================ - -#[cfg(windows)] -mod platform { - use super::*; - - /// UDP socket wrapper (Windows). - /// - /// Uses `socket2::Socket` for configuration and `tokio::net::UdpSocket` - /// for async I/O. Kernel drop counting is not available on Windows; - /// the drops field always returns 0. - pub struct UdpRawSocket { - inner: Socket, - local_addr: SocketAddr, - } - - impl UdpRawSocket { - /// Create, bind, and configure a UDP socket. - /// - /// Sets non-blocking mode and configures buffer sizes. The socket - /// is bound immediately so `local_addr()` returns the actual - /// assigned address (important when binding to port 0). - pub fn open( - bind_addr: SocketAddr, - recv_buf_size: usize, - send_buf_size: usize, - ) -> Result { - let domain = if bind_addr.is_ipv4() { - Domain::IPV4 - } else { - Domain::IPV6 - }; - let sock = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP)) - .map_err(|e| TransportError::StartFailed(format!("socket create failed: {}", e)))?; - - sock.set_nonblocking(true).map_err(|e| { - TransportError::StartFailed(format!("set nonblocking failed: {}", e)) - })?; - - sock.bind(&bind_addr.into()) - .map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?; - - // Set socket buffer sizes - sock.set_recv_buffer_size(recv_buf_size) - .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; - sock.set_send_buffer_size(send_buf_size) - .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; - - let local_addr = sock - .local_addr() - .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? - .as_socket() - .ok_or_else(|| { - TransportError::StartFailed("local address is not an IP socket".into()) - })?; - - Ok(Self { - inner: sock, - local_addr, - }) - } - - /// Adopt an existing bound UDP socket. - pub fn adopt( - socket: std::net::UdpSocket, - recv_buf_size: usize, - send_buf_size: usize, - ) -> Result { - let sock = Socket::from(socket); - - sock.set_nonblocking(true).map_err(|e| { - TransportError::StartFailed(format!("set nonblocking failed: {}", e)) - })?; - - sock.set_recv_buffer_size(recv_buf_size) - .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; - sock.set_send_buffer_size(send_buf_size) - .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; - - let local_addr = sock - .local_addr() - .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? - .as_socket() - .ok_or_else(|| { - TransportError::StartFailed("local address is not an IP socket".into()) - })?; - - Ok(Self { - inner: sock, - local_addr, - }) - } - - /// Get the local bound address. - pub fn local_addr(&self) -> SocketAddr { - self.local_addr - } - - /// Get the actual receive buffer size. - pub fn recv_buffer_size(&self) -> Result { - self.inner - .recv_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e))) - } - - /// Get the actual send buffer size. - pub fn send_buffer_size(&self) -> Result { - self.inner - .send_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e))) - } - - /// Wrap this socket in an async wrapper for tokio I/O. - pub fn into_async(self) -> Result { - let std_socket: std::net::UdpSocket = self.inner.into(); - let tokio_socket = tokio::net::UdpSocket::from_std(std_socket) - .map_err(|e| TransportError::StartFailed(format!("tokio socket failed: {}", e)))?; - - Ok(AsyncUdpSocket { - inner: Arc::new(tokio_socket), - }) - } - } - - /// Async UDP socket wrapper (Windows). - /// - /// Uses `tokio::net::UdpSocket` directly. Kernel drop counting - /// is not available; the drops field always returns 0. - #[derive(Clone)] - pub struct AsyncUdpSocket { - inner: Arc, - } - - impl AsyncUdpSocket { - /// Send a payload to a destination address. - pub async fn send_to( - &self, - data: &[u8], - dest: &SocketAddr, - ) -> Result { - self.inner - .send_to(data, dest) - .await - .map_err(|e| TransportError::SendFailed(format!("{}", e))) - } - - /// Receive a payload, source address, and kernel drop counter. - /// - /// Returns `(bytes_read, source_addr, 0)`. The drops field is always 0 - /// on Windows since kernel drop counting is not available. - pub async fn recv_from( - &self, - buf: &mut [u8], - ) -> Result<(usize, SocketAddr, u32), TransportError> { - let (n, addr) = self - .inner - .recv_from(buf) - .await - .map_err(|e| TransportError::RecvFailed(format!("{}", e)))?; - Ok((n, addr, 0)) - } - } -} - -pub use platform::{AsyncUdpSocket, UdpRawSocket}; - -/// Per-peer connected-UDP fast-path fd construction. -/// -/// One of the levers boringtun uses to hit 2.5–3.2 Gbps on a real -/// NIC: after a peer is established, give them their **own UDP socket -/// `connect()`-ed to their address**. The kernel then routes inbound -/// packets from that peer directly to the connected socket -/// (most-specific-match wins over the wildcard listen socket under -/// `SO_REUSEPORT`), and lets us `send(2)` with `msg_name = NULL` — -/// skipping the per-packet sockaddr copy + route lookup + neighbor -/// resolve. This module owns the fd-construction syscall sequence only -/// (socket / sockopt / bind / connect + buffer sizing); the owning -/// handle type that adopts the returned fd lives in -/// `crate::peer::connected_udp`. -/// -/// Gated to Linux/macOS: the rest of `io.rs` compiles more broadly -/// (Windows uses `tokio::net::UdpSocket`), but the connected fast path -/// is libc-syscall + `sockopts_macos` specific. -#[cfg(any(target_os = "linux", target_os = "macos"))] -mod connected { - use std::io; - use std::net::SocketAddr; - use std::os::unix::io::{AsRawFd, FromRawFd, OwnedFd, RawFd}; - - /// Open a `connect()`-ed UDP socket for one peer and return the owning - /// fd. Performs the full socket / sockopt / bind / connect syscall - /// sequence. On any mid-construction failure the fd is closed (via the - /// `OwnedFd` RAII guard) before the error is returned; on success - /// ownership of the fd transfers to the returned `OwnedFd`. Callers - /// adopt it into a `crate::peer::connected_udp::ConnectedPeerSocket` - /// via `ConnectedPeerSocket::from_fd`. - /// - /// `local_addr` is the wildcard bind address (e.g. `0.0.0.0:51820` - /// or `[::]:51820`) — the same address the listen socket bound to. - /// `peer_addr` is the kernel `SocketAddr` of the established peer's - /// UDP endpoint. `recv_buf` / `send_buf` are the requested buffer - /// sizes; they're applied with `SO_*BUFFORCE` first and fall back to - /// the normal `SO_*BUF` if the process can't bypass the kernel - /// ceiling. - pub(crate) fn open_connected_fd( - local_addr: SocketAddr, - peer_addr: SocketAddr, - recv_buf: usize, - send_buf: usize, - ) -> io::Result { - // Family must match between local and peer. - if local_addr.is_ipv4() != peer_addr.is_ipv4() { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "ConnectedPeerSocket: local + peer address families differ", - )); - } - - let domain = if local_addr.is_ipv4() { - libc::AF_INET - } else { - libc::AF_INET6 - }; - // Linux accepts SOCK_NONBLOCK | SOCK_CLOEXEC directly. Darwin - // does not, so we set the equivalent fd flags with fcntl below. - #[cfg(target_os = "linux")] - let typ = libc::SOCK_DGRAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC; - #[cfg(not(target_os = "linux"))] - let typ = libc::SOCK_DGRAM; - let fd = unsafe { libc::socket(domain, typ, libc::IPPROTO_UDP) }; - if fd < 0 { - return Err(io::Error::last_os_error()); - } - // Adopt the fd into an OwnedFd immediately: from here its Drop - // closes the fd on any early `return Err` / `?` below. Ownership - // transfers to the caller only via the final `Ok(owned)`. - let owned = unsafe { OwnedFd::from_raw_fd(fd) }; - let raw = owned.as_raw_fd(); - - #[cfg(not(target_os = "linux"))] - set_nonblocking_cloexec(raw)?; - - // SO_REUSEADDR lets us bind to the same local port the listen - // socket already holds. SO_REUSEPORT lets the UDP demux permit - // several sockets bound to the same address and route the peer - // 5-tuple to the connected sibling. - set_sockopt_int(raw, libc::SOL_SOCKET, libc::SO_REUSEADDR, 1)?; - set_sockopt_int(raw, libc::SOL_SOCKET, libc::SO_REUSEPORT, 1)?; - - #[cfg(target_os = "macos")] - crate::transport::udp::sockopts_macos::apply_udp_socket_tuning(raw, "connected-udp-peer"); - - // Buffer sizes — try the FORCE variants first (succeed if we - // have CAP_NET_ADMIN), then fall back to the ceiling-clamped - // normal variants. The ceiling-clamped path always succeeds - // even if it gives us less than we asked for. - #[cfg(target_os = "linux")] - { - set_buf_size(raw, libc::SO_RCVBUFFORCE, libc::SO_RCVBUF, recv_buf); - set_buf_size(raw, libc::SO_SNDBUFFORCE, libc::SO_SNDBUF, send_buf); - } - #[cfg(not(target_os = "linux"))] - { - set_buf_size(raw, libc::SO_RCVBUF, recv_buf); - set_buf_size(raw, libc::SO_SNDBUF, send_buf); - } - - // Bind to the wildcard local address (same port as listen socket). - let local_sa: socket2::SockAddr = local_addr.into(); - let bind_r = unsafe { - libc::bind( - raw, - local_sa.as_ptr() as *const libc::sockaddr, - local_sa.len(), - ) - }; - if bind_r < 0 { - return Err(syscall_err("bind", local_addr)); - } - - // Connect to the peer — locks in the per-packet kernel route. - let peer_sa: socket2::SockAddr = peer_addr.into(); - let conn_r = unsafe { - libc::connect( - raw, - peer_sa.as_ptr() as *const libc::sockaddr, - peer_sa.len(), - ) - }; - if conn_r < 0 { - return Err(syscall_err("connect", peer_addr)); - } - - Ok(owned) - } - - fn syscall_err(syscall: &str, addr: SocketAddr) -> io::Error { - let err = io::Error::last_os_error(); - io::Error::new(err.kind(), format!("{syscall} {addr}: {err}")) - } - - #[cfg(not(target_os = "linux"))] - fn set_nonblocking_cloexec(fd: RawFd) -> io::Result<()> { - let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; - if flags < 0 { - return Err(io::Error::last_os_error()); - } - if unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } < 0 { - return Err(io::Error::last_os_error()); - } - - let fd_flags = unsafe { libc::fcntl(fd, libc::F_GETFD) }; - if fd_flags < 0 { - return Err(io::Error::last_os_error()); - } - if unsafe { libc::fcntl(fd, libc::F_SETFD, fd_flags | libc::FD_CLOEXEC) } < 0 { - return Err(io::Error::last_os_error()); - } - Ok(()) - } - - /// Set an integer-valued socket option on `fd`. Returns the kernel - /// error on failure so the caller can `?`-propagate. - fn set_sockopt_int( - fd: RawFd, - level: libc::c_int, - name: libc::c_int, - value: libc::c_int, - ) -> io::Result<()> { - let r = unsafe { - libc::setsockopt( - fd, - level, - name, - &value as *const _ as *const libc::c_void, - std::mem::size_of::() as libc::socklen_t, - ) - }; - if r < 0 { - Err(io::Error::last_os_error()) - } else { - Ok(()) - } - } - - /// Try `SO_*BUFFORCE` first (bypasses the rmem/wmem ceiling) and - /// fall back to `SO_*BUF` if that fails. Returns silently — buffer - /// sizing is best-effort. - #[cfg(target_os = "linux")] - fn set_buf_size(fd: RawFd, force_name: libc::c_int, normal_name: libc::c_int, size: usize) { - let value: libc::c_int = size as libc::c_int; - let r = unsafe { - libc::setsockopt( - fd, - libc::SOL_SOCKET, - force_name, - &value as *const _ as *const libc::c_void, - std::mem::size_of::() as libc::socklen_t, - ) - }; - if r < 0 { - // Fall back to non-force — kernel may clamp. - let _ = unsafe { - libc::setsockopt( - fd, - libc::SOL_SOCKET, - normal_name, - &value as *const _ as *const libc::c_void, - std::mem::size_of::() as libc::socklen_t, - ) - }; - } - } - - #[cfg(not(target_os = "linux"))] - fn set_buf_size(fd: RawFd, normal_name: libc::c_int, size: usize) { - let value: libc::c_int = size as libc::c_int; - let _ = unsafe { - libc::setsockopt( - fd, - libc::SOL_SOCKET, - normal_name, - &value as *const _ as *const libc::c_void, - std::mem::size_of::() as libc::socklen_t, - ) - }; - } - - #[cfg(all(test, target_os = "linux"))] - mod tests { - use super::*; - use std::net::UdpSocket; - - const BUF: usize = 1 << 20; - - #[test] - fn bind_failure_names_bind_and_local_addr() { - // A plain socket without SO_REUSEPORT holds the address, so the - // SO_REUSEPORT bind below is refused with EADDRINUSE. - let holder = UdpSocket::bind("127.0.0.1:0").expect("holder bind"); - let holder_addr = holder.local_addr().expect("holder addr"); - - let err = open_connected_fd(holder_addr, "127.0.0.1:9".parse().unwrap(), BUF, BUF) - .expect_err("bind must fail against a non-reuseport holder"); - - assert_eq!(err.kind(), io::ErrorKind::AddrInUse, "{err}"); - let msg = err.to_string(); - assert!(msg.starts_with("bind "), "{msg}"); - assert!(msg.contains(&holder_addr.to_string()), "{msg}"); - assert!(!msg.contains("connect"), "{msg}"); - } - - #[test] - fn connect_failure_names_connect_and_peer_addr() { - // connect(2) to the broadcast address without SO_BROADCAST fails - // synchronously with EACCES; the bind before it succeeds. - let err = open_connected_fd( - "127.0.0.1:0".parse().unwrap(), - "255.255.255.255:9999".parse().unwrap(), - BUF, - BUF, - ) - .expect_err("connect to broadcast without SO_BROADCAST must fail"); - - assert_eq!(err.kind(), io::ErrorKind::PermissionDenied, "{err}"); - let msg = err.to_string(); - assert!(msg.starts_with("connect "), "{msg}"); - assert!(msg.contains("255.255.255.255:9999"), "{msg}"); - assert!(!msg.contains("bind"), "{msg}"); - } - } -} - -#[cfg(any(target_os = "linux", target_os = "macos"))] -pub(crate) use connected::open_connected_fd; - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_udp_socket_bind() { - // Bind to an ephemeral port - let sock = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) - .expect("failed to bind UDP socket"); - - let addr = sock.local_addr(); - assert!(addr.port() > 0, "should be assigned an ephemeral port"); - assert!(addr.ip().is_loopback()); - } - - #[test] - fn test_udp_socket_buffer_sizes() { - let sock = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) - .expect("failed to bind UDP socket"); - - let recv_buf = sock.recv_buffer_size().expect("get recv buffer"); - let send_buf = sock.send_buffer_size().expect("get send buffer"); - assert!(recv_buf > 0, "recv buffer should be non-zero"); - assert!(send_buf > 0, "send buffer should be non-zero"); - } - - #[tokio::test] - async fn test_async_udp_socket_send_recv() { - let sock1 = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) - .expect("failed to bind socket 1"); - let addr1 = sock1.local_addr(); - let async1 = sock1.into_async().expect("into_async 1"); - - let sock2 = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) - .expect("failed to bind socket 2"); - let addr2 = sock2.local_addr(); - let async2 = sock2.into_async().expect("into_async 2"); - - // Send from socket 1 to socket 2 - let payload = b"hello fips"; - let sent = async1.send_to(payload, &addr2).await.expect("send_to"); - assert_eq!(sent, payload.len()); - - // Receive on socket 2 - let mut buf = [0u8; 1024]; - let (n, src, _drops) = async2.recv_from(&mut buf).await.expect("recv_from"); - assert_eq!(n, payload.len()); - assert_eq!(&buf[..n], payload); - assert_eq!(src, addr1); - } - - /// Microbench: compare per-packet `recv_from` (single recvmsg syscall + - /// task wakeup per datagram — the macOS pre-recvmsg_x baseline) vs - /// `recv_batch` (the new recvmsg_x path, up to 32 datagrams per syscall). - /// Both modes run back-to-back in this binary on loopback so the only - /// thing that varies is the receive-syscall strategy. Sender is a tight - /// `socket.send_to()` loop in a separate task; receiver counts datagrams - /// drained over a fixed wall-clock window per mode. - /// - /// Run with: - /// cargo test --release -p fips --lib transport::udp::io::tests::bench_udp_recv_amortization -- --ignored --nocapture - /// - /// Sender runs on a dedicated *blocking* OS thread (std::net::UdpSocket - /// in default blocking mode) so it always saturates the kernel rx queue - /// regardless of how the tokio receiver schedules. That's the scenario - /// where recvmmsg / recvmsg_x is meant to win: the receiver wakes up to - /// find N packets already buffered, and one syscall reaps the burst. - #[cfg(any(target_os = "linux", target_os = "macos"))] - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - #[ignore = "microbenchmark; run explicitly with --ignored --nocapture"] - async fn bench_udp_recv_amortization() { - use std::sync::Arc; - use std::sync::atomic::{AtomicBool, Ordering}; - use std::time::{Duration, Instant}; - - const RECV_BUF: usize = 4 * 1024 * 1024; - const SEND_BUF: usize = 1024 * 1024; - const PAYLOAD_LEN: usize = 100; - const WINDOW: Duration = Duration::from_secs(3); - const WARMUP: Duration = Duration::from_millis(500); - - async fn run_mode( - label: &str, - batched: bool, - sender_threads: usize, - ) -> (u64, u64, Duration) { - let rx_sock = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), RECV_BUF, SEND_BUF) - .expect("rx bind"); - let rx_addr = rx_sock.local_addr(); - let rx = rx_sock.into_async().expect("rx into_async"); - - // Senders: N dedicated blocking std threads. More threads → deeper - // kernel rx queue → larger amortization opportunity for recv_batch. - // ENOBUFS / EAGAIN just yield and retry; we want saturation, not - // perfect accounting. Sent count is best-effort. - let stop = Arc::new(AtomicBool::new(false)); - let mut sender_handles = Vec::with_capacity(sender_threads); - for _ in 0..sender_threads { - let stop_tx = stop.clone(); - sender_handles.push(std::thread::spawn(move || { - let sock = std::net::UdpSocket::bind("127.0.0.1:0").expect("tx bind"); - sock.connect(rx_addr).expect("tx connect"); - let payload = vec![0xABu8; PAYLOAD_LEN]; - let mut sent: u64 = 0; - while !stop_tx.load(Ordering::Relaxed) { - match sock.send(&payload) { - Ok(_) => sent += 1, - Err(_) => std::thread::yield_now(), - } - } - sent - })); - } - - // Warm-up: let the sender thread reach steady state and the - // kernel rx queue start filling. - tokio::time::sleep(WARMUP).await; - - let start = Instant::now(); - let deadline = start + WINDOW; - let mut recv_count: u64 = 0; - let mut last_drops: u32 = 0; - - if batched { - const BATCH: usize = 32; - let mut backing: Vec> = - (0..BATCH).map(|_| vec![0u8; PAYLOAD_LEN + 64]).collect(); - let mut addrs: [Option; BATCH] = std::array::from_fn(|_| None); - let mut lens: [usize; BATCH] = [0; BATCH]; - let mut batch_sum: u64 = 0; - let mut batch_calls: u64 = 0; - - while Instant::now() < deadline { - let mut bufs: [&mut [u8]; BATCH] = { - let mut iter = backing.iter_mut(); - std::array::from_fn(|_| iter.next().unwrap().as_mut_slice()) - }; - match rx.recv_batch(&mut bufs, &mut addrs, &mut lens).await { - Ok((n, drops)) => { - recv_count += n as u64; - batch_sum += n as u64; - batch_calls += 1; - last_drops = drops; - } - Err(_) => break, - } - } - let avg_batch = if batch_calls > 0 { - batch_sum as f64 / batch_calls as f64 - } else { - 0.0 - }; - eprintln!( - "[{:>10}] avg_batch_per_call={:.2} ({} calls)", - label, avg_batch, batch_calls - ); - } else { - let mut buf = vec![0u8; PAYLOAD_LEN + 64]; - while Instant::now() < deadline { - match rx.recv_from(&mut buf).await { - Ok((_n, _src, drops)) => { - recv_count += 1; - last_drops = drops; - } - Err(_) => break, - } - } - } - let elapsed = start.elapsed(); - - stop.store(true, Ordering::Relaxed); - drop(rx); - let sent: u64 = sender_handles - .into_iter() - .map(|h| h.join().unwrap_or(0)) - .sum(); - - let pps = (recv_count as f64) / elapsed.as_secs_f64(); - let mbps = - (recv_count as f64) * (PAYLOAD_LEN as f64) * 8.0 / 1e6 / elapsed.as_secs_f64(); - eprintln!( - "[{:>10}] recv={:>10} sent={:>10} elapsed={:?} pps={:>12.0} mbps={:>7.1} kdrops={}", - label, recv_count, sent, elapsed, pps, mbps, last_drops - ); - (recv_count, sent, elapsed) - } - - eprintln!("--- udp recv amortization bench ---"); - eprintln!( - "payload={}B window={:?} warmup={:?} runtime=multi_thread(2)", - PAYLOAD_LEN, WINDOW, WARMUP - ); - - // Sweep sender concurrency. Each level shows how the win scales as - // the rx queue gets deeper (more amortization opportunity). - for senders in [1usize, 2, 4, 8] { - eprintln!("\n=== sender_threads = {} ===", senders); - let (b_recv, _, b_el) = run_mode(" recv_from", false, senders).await; - let (x_recv, _, x_el) = run_mode("recv_batch", true, senders).await; - let (x_recv2, _, x_el2) = run_mode("recv_batch", true, senders).await; - let (b_recv2, _, b_el2) = run_mode(" recv_from", false, senders).await; - - let baseline_pps = - (b_recv as f64 / b_el.as_secs_f64() + b_recv2 as f64 / b_el2.as_secs_f64()) / 2.0; - let batched_pps = - (x_recv as f64 / x_el.as_secs_f64() + x_recv2 as f64 / x_el2.as_secs_f64()) / 2.0; - let speedup = batched_pps / baseline_pps; - eprintln!( - "--- senders={}: baseline={:.0} pps batched={:.0} pps speedup={:.2}x ---", - senders, baseline_pps, batched_pps, speedup - ); - } - } -} diff --git a/src/peer/connected_udp/drain.rs b/src/transport/udp/io/connected/drain.rs similarity index 90% rename from src/peer/connected_udp/drain.rs rename to src/transport/udp/io/connected/drain.rs index 46e06a5..de27650 100644 --- a/src/peer/connected_udp/drain.rs +++ b/src/transport/udp/io/connected/drain.rs @@ -9,7 +9,7 @@ //! //! This module owns the drain side: spawn one OS thread per connected //! socket, drain into a fixed-size batch (`recvmmsg(2)` on Linux, -//! repeated nonblocking `recv(2)` on Darwin), push each packet into +//! `recvmsg_x(2)` on Darwin), push each packet into //! the existing `packet_tx` (the same channel that the wildcard listen //! socket feeds), and exit cleanly when the parent signals shutdown //! via a self-pipe. @@ -334,9 +334,9 @@ fn drain_packets(fd: RawFd, backing: &mut [Vec], lens: &mut [usize]) -> io:: recvmmsg_drain(fd, backing, lens) } -#[cfg(not(target_os = "linux"))] +#[cfg(target_os = "macos")] fn drain_packets(fd: RawFd, backing: &mut [Vec], lens: &mut [usize]) -> io::Result { - recv_drain(fd, backing, lens) + crate::transport::udp::io::recvmsg_x_drain(fd, backing, lens) } /// One-shot `recvmmsg(2)` on a non-blocking fd. Returns the number of @@ -389,39 +389,6 @@ fn recvmmsg_drain(fd: RawFd, backing: &mut [Vec], lens: &mut [usize]) -> io: Ok(count) } -#[cfg(not(target_os = "linux"))] -fn recv_drain(fd: RawFd, backing: &mut [Vec], lens: &mut [usize]) -> io::Result { - let n = backing.len().min(lens.len()); - if n == 0 { - return Ok(0); - } - - let mut count = 0usize; - while count < n { - let r = unsafe { - libc::recv( - fd, - backing[count].as_mut_ptr() as *mut libc::c_void, - backing[count].len(), - 0, - ) - }; - if r < 0 { - let err = io::Error::last_os_error(); - if err.kind() == io::ErrorKind::Interrupted { - continue; - } - if err.kind() == io::ErrorKind::WouldBlock && count > 0 { - return Ok(count); - } - return Err(err); - } - lens[count] = r as usize; - count += 1; - } - Ok(count) -} - #[cfg(test)] mod tests { use super::*; @@ -429,6 +396,60 @@ mod tests { use std::time::Duration; use tokio::sync::mpsc; + #[cfg(target_os = "macos")] + #[test] + fn recvmsg_x_drain_receives_connected_socket_burst() { + let receiver = UdpSocket::bind("127.0.0.1:0").expect("bind receiver"); + let sender = UdpSocket::bind("127.0.0.1:0").expect("bind sender"); + receiver + .connect(sender.local_addr().expect("sender address")) + .expect("connect receiver"); + sender + .connect(receiver.local_addr().expect("receiver address")) + .expect("connect sender"); + receiver + .set_nonblocking(true) + .expect("set receiver nonblocking"); + + const PACKETS: usize = 40; + for sequence in 0..PACKETS as u8 { + sender + .send(&[sequence, 0xAA, 0xBB, 0xCC]) + .expect("send burst packet"); + } + + let mut backing: Vec> = (0..PACKETS).map(|_| vec![0u8; 64]).collect(); + let mut lens = [0usize; PACKETS]; + let deadline = std::time::Instant::now() + Duration::from_secs(1); + let mut count = 0; + while count < PACKETS { + match crate::transport::udp::io::recvmsg_x_drain( + receiver.as_raw_fd(), + &mut backing[count..], + &mut lens[count..], + ) { + Ok(received) => count += received, + Err(error) + if error.kind() == io::ErrorKind::WouldBlock + && std::time::Instant::now() < deadline => + { + std::thread::yield_now(); + } + Err(error) => panic!("recvmsg_x burst failed after {count} packets: {error}"), + } + assert!( + std::time::Instant::now() < deadline || count == PACKETS, + "timed out after receiving {count} of {PACKETS} packets" + ); + } + + assert_eq!(count, PACKETS); + for sequence in 0..PACKETS { + assert_eq!(lens[sequence], 4); + assert_eq!(backing[sequence][0], sequence as u8); + } + } + /// End-to-end: open a ConnectedPeerSocket, spawn a drain thread /// on it, send packets at it from a remote, verify they land in /// the packet_tx mpsc with the correct transport_id + peer_addr. diff --git a/src/transport/udp/io/connected/fd.rs b/src/transport/udp/io/connected/fd.rs new file mode 100644 index 0000000..c8f6a81 --- /dev/null +++ b/src/transport/udp/io/connected/fd.rs @@ -0,0 +1,180 @@ +//! Connected-socket fd construction: the socket / sockopt / bind / +//! connect syscall sequence only. Why established peers get their own +//! socket at all is in the parent module's docs; the handle that adopts +//! the fd is `super::socket::ConnectedPeerSocket`. +//! +//! The four points where the sequence differs between Linux and macOS — +//! the `socket(2)` type flags, the follow-up fd-flag call, the +//! service-type tuning and the buffer-size strategy — come from the `sys` +//! module selected below, which is the same seam the shared receive code +//! uses. + +use std::io; +use std::net::SocketAddr; +use std::os::unix::io::{AsRawFd, FromRawFd, OwnedFd, RawFd}; + +#[cfg(target_os = "linux")] +use super::super::linux as sys; +#[cfg(target_os = "macos")] +use super::super::macos as sys; + +/// Open a `connect()`-ed UDP socket for one peer and return the owning +/// fd. Performs the full socket / sockopt / bind / connect syscall +/// sequence. On any mid-construction failure the fd is closed (via the +/// `OwnedFd` RAII guard) before the error is returned; on success +/// ownership of the fd transfers to the returned `OwnedFd`. Callers +/// adopt it into a `super::socket::ConnectedPeerSocket` via +/// `ConnectedPeerSocket::from_fd`. +/// +/// `local_addr` is the wildcard bind address (e.g. `0.0.0.0:51820` +/// or `[::]:51820`) — the same address the listen socket bound to. +/// `peer_addr` is the kernel `SocketAddr` of the established peer's +/// UDP endpoint. `recv_buf` / `send_buf` are the requested buffer +/// sizes, applied best-effort: on Linux with `SO_*BUFFORCE` first, +/// falling back to the normal `SO_*BUF` if the process can't bypass the +/// kernel ceiling; on macOS with `SO_*BUF` alone, which has no force +/// variant. +pub(crate) fn open_connected_fd( + local_addr: SocketAddr, + peer_addr: SocketAddr, + recv_buf: usize, + send_buf: usize, +) -> io::Result { + // Family must match between local and peer. + if local_addr.is_ipv4() != peer_addr.is_ipv4() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "ConnectedPeerSocket: local + peer address families differ", + )); + } + + let domain = if local_addr.is_ipv4() { + libc::AF_INET + } else { + libc::AF_INET6 + }; + let fd = unsafe { libc::socket(domain, sys::SOCKET_TYPE, libc::IPPROTO_UDP) }; + if fd < 0 { + return Err(io::Error::last_os_error()); + } + // Adopt the fd into an OwnedFd immediately: from here its Drop + // closes the fd on any early `return Err` / `?` below. Ownership + // transfers to the caller only via the final `Ok(owned)`. + let owned = unsafe { OwnedFd::from_raw_fd(fd) }; + let raw = owned.as_raw_fd(); + + sys::prepare_fd(raw)?; + + // SO_REUSEADDR lets us bind to the same local port the listen + // socket already holds. SO_REUSEPORT lets the UDP demux permit + // several sockets bound to the same address and route the peer + // 5-tuple to the connected sibling. + set_sockopt_int(raw, libc::SOL_SOCKET, libc::SO_REUSEADDR, 1)?; + set_sockopt_int(raw, libc::SOL_SOCKET, libc::SO_REUSEPORT, 1)?; + + sys::tune_connected_socket(raw); + + // Buffer sizes — best effort; see the per-platform implementation. + sys::set_buf_sizes(raw, recv_buf, send_buf); + + // Bind to the wildcard local address (same port as listen socket). + let local_sa: socket2::SockAddr = local_addr.into(); + let bind_r = unsafe { + libc::bind( + raw, + local_sa.as_ptr() as *const libc::sockaddr, + local_sa.len(), + ) + }; + if bind_r < 0 { + return Err(syscall_err("bind", local_addr)); + } + + // Connect to the peer — locks in the per-packet kernel route. + let peer_sa: socket2::SockAddr = peer_addr.into(); + let conn_r = unsafe { + libc::connect( + raw, + peer_sa.as_ptr() as *const libc::sockaddr, + peer_sa.len(), + ) + }; + if conn_r < 0 { + return Err(syscall_err("connect", peer_addr)); + } + + Ok(owned) +} + +fn syscall_err(syscall: &str, addr: SocketAddr) -> io::Error { + let err = io::Error::last_os_error(); + io::Error::new(err.kind(), format!("{syscall} {addr}: {err}")) +} + +/// Set an integer-valued socket option on `fd`. Returns the kernel +/// error on failure so the caller can `?`-propagate. +fn set_sockopt_int( + fd: RawFd, + level: libc::c_int, + name: libc::c_int, + value: libc::c_int, +) -> io::Result<()> { + let r = unsafe { + libc::setsockopt( + fd, + level, + name, + &value as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if r < 0 { + Err(io::Error::last_os_error()) + } else { + Ok(()) + } +} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::*; + use std::net::UdpSocket; + + const BUF: usize = 1 << 20; + + #[test] + fn bind_failure_names_bind_and_local_addr() { + // A plain socket without SO_REUSEPORT holds the address, so the + // SO_REUSEPORT bind below is refused with EADDRINUSE. + let holder = UdpSocket::bind("127.0.0.1:0").expect("holder bind"); + let holder_addr = holder.local_addr().expect("holder addr"); + + let err = open_connected_fd(holder_addr, "127.0.0.1:9".parse().unwrap(), BUF, BUF) + .expect_err("bind must fail against a non-reuseport holder"); + + assert_eq!(err.kind(), io::ErrorKind::AddrInUse, "{err}"); + let msg = err.to_string(); + assert!(msg.starts_with("bind "), "{msg}"); + assert!(msg.contains(&holder_addr.to_string()), "{msg}"); + assert!(!msg.contains("connect"), "{msg}"); + } + + #[test] + fn connect_failure_names_connect_and_peer_addr() { + // connect(2) to the broadcast address without SO_BROADCAST fails + // synchronously with EACCES; the bind before it succeeds. + let err = open_connected_fd( + "127.0.0.1:0".parse().unwrap(), + "255.255.255.255:9999".parse().unwrap(), + BUF, + BUF, + ) + .expect_err("connect to broadcast without SO_BROADCAST must fail"); + + assert_eq!(err.kind(), io::ErrorKind::PermissionDenied, "{err}"); + let msg = err.to_string(); + assert!(msg.starts_with("connect "), "{msg}"); + assert!(msg.contains("255.255.255.255:9999"), "{msg}"); + assert!(!msg.contains("bind"), "{msg}"); + } +} diff --git a/src/transport/udp/io/connected/mod.rs b/src/transport/udp/io/connected/mod.rs new file mode 100644 index 0000000..ac272e8 --- /dev/null +++ b/src/transport/udp/io/connected/mod.rs @@ -0,0 +1,33 @@ +//! Per-peer connected-UDP fast path. +//! +//! One of the levers boringtun uses to hit 2.5–3.2 Gbps on a real NIC: +//! after a peer is established, give them their **own UDP socket +//! `connect()`-ed to their address**. The kernel then routes inbound +//! packets from that peer directly to the connected socket +//! (most-specific-match wins over the wildcard listen socket under +//! `SO_REUSEPORT`), and lets us `send(2)` with `msg_name = NULL` — +//! skipping the per-packet sockaddr copy + route lookup + neighbor +//! resolve. +//! +//! The whole mechanism lives here, in three pieces that only make sense +//! together: +//! +//! - `fd` — the socket / sockopt / bind / connect syscall sequence that +//! constructs the fd. +//! - `socket::ConnectedPeerSocket` — the owning handle that adopts it and +//! closes it on drop. +//! - `drain::PeerRecvDrain` — the recv-side drain thread that must +//! accompany every connected socket, since the kernel routes the peer's +//! inbound packets to it and something has to read them. +//! +//! Gated to Linux and macOS: the rest of `io` compiles more broadly +//! (Windows uses `tokio::net::UdpSocket`), but this path is libc-syscall +//! and Darwin-sockopt specific. + +mod drain; +mod fd; +mod socket; + +pub(crate) use drain::PeerRecvDrain; +pub(crate) use fd::open_connected_fd; +pub(crate) use socket::ConnectedPeerSocket; diff --git a/src/peer/connected_udp/socket.rs b/src/transport/udp/io/connected/socket.rs similarity index 100% rename from src/peer/connected_udp/socket.rs rename to src/transport/udp/io/connected/socket.rs diff --git a/src/transport/udp/io/linux.rs b/src/transport/udp/io/linux.rs new file mode 100644 index 0000000..2a26c4a --- /dev/null +++ b/src/transport/udp/io/linux.rs @@ -0,0 +1,181 @@ +//! Linux-specific UDP receive and connected-socket construction. +//! +//! Holds the three seam points the shared Unix code delegates to — +//! `SO_RXQ_OVFL` drop counting, the ancillary-data control buffer, and +//! `recvmmsg(2)` batching — plus the Linux half of the per-peer +//! connected-fd syscall sequence. + +use super::unix::{BATCH_SIZE, sockaddr_to_socket_addr}; +use std::io; +use std::net::SocketAddr; +use std::os::unix::io::RawFd; +use tracing::warn; + +/// Control message buffer size for `SO_RXQ_OVFL` (u32). `CMSG_SPACE` +/// computes the aligned size including header. +pub(super) const CMSG_BUF_SIZE: usize = unsafe { libc::CMSG_SPACE(4) } as usize; + +/// Enable `SO_RXQ_OVFL` so `recvmsg` ancillary data carries the kernel +/// drop counter. Non-fatal: older kernels may not support it. +pub(super) fn enable_drop_counting(fd: RawFd) { + let enable: libc::c_int = 1; + let ret = unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_RXQ_OVFL, + &enable as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + warn!( + "setsockopt(SO_RXQ_OVFL) failed: {}", + std::io::Error::last_os_error() + ); + } +} + +/// Walk the cmsg chain for the `SO_RXQ_OVFL` drop counter. +pub(super) fn parse_drops(msg: &libc::msghdr) -> u32 { + let mut drops: u32 = 0; + unsafe { + let mut cmsg = libc::CMSG_FIRSTHDR(msg); + while !cmsg.is_null() { + if (*cmsg).cmsg_level == libc::SOL_SOCKET && (*cmsg).cmsg_type == libc::SO_RXQ_OVFL { + let data = libc::CMSG_DATA(cmsg); + drops = std::ptr::read_unaligned(data as *const u32); + } + cmsg = libc::CMSG_NXTHDR(msg, cmsg); + } + } + drops +} + +/// Receive up to `BATCH_SIZE` datagrams in a single `recvmmsg` syscall. +/// +/// Returns `(count, kernel_drops)`. Caller pre-sizes `bufs` (each must be +/// at least the configured MTU) and the matching `addrs` / `lens` slices; +/// on return, slots `[0..count)` are valid. +/// +/// `kernel_drops` is the `SO_RXQ_OVFL` cumulative counter sampled from the +/// cmsg chain of the FIRST datagram in the batch. The counter is monotonic +/// per-socket since `SO_RXQ_OVFL` was enabled, so a single sample per batch +/// is sufficient to feed the 1Hz congestion detector in +/// `sample_transport_congestion()`. Returns `(0, 0)` on a spurious wakeup +/// with no datagrams ready. +pub(super) fn recv_batch( + fd: RawFd, + bufs: &mut [&mut [u8]], + addrs: &mut [Option], + lens: &mut [usize], +) -> io::Result<(usize, u32)> { + let n = bufs.len().min(addrs.len()).min(lens.len()).min(BATCH_SIZE); + if n == 0 { + return Ok((0, 0)); + } + + // CMSG buffer wired to msgs[0] only. SO_RXQ_OVFL delivers a + // monotonic u32 drop counter; sampling once per batch gives + // the 1Hz congestion detector ample fresh values under load + // (one batch = up to 32 datagrams). + let mut cmsg_buf = [0u8; CMSG_BUF_SIZE]; + + // Stack-allocated parallel arrays; lifetime tied to this call. + let mut iovs: [libc::iovec; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + let mut storages: [libc::sockaddr_storage; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + let mut msgs: [libc::mmsghdr; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + + for i in 0..n { + iovs[i].iov_base = bufs[i].as_mut_ptr() as *mut libc::c_void; + iovs[i].iov_len = bufs[i].len(); + msgs[i].msg_hdr.msg_name = &mut storages[i] as *mut _ as *mut libc::c_void; + msgs[i].msg_hdr.msg_namelen = + std::mem::size_of::() as libc::socklen_t; + msgs[i].msg_hdr.msg_iov = &mut iovs[i]; + msgs[i].msg_hdr.msg_iovlen = 1; + msgs[i].msg_len = 0; + } + // Only msgs[0] carries a cmsg buffer — sampling the OVFL counter + // there is enough since it is socket-wide and monotonic. + msgs[0].msg_hdr.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void; + msgs[0].msg_hdr.msg_controllen = cmsg_buf.len() as _; + + let r = unsafe { + libc::recvmmsg( + fd, + msgs.as_mut_ptr(), + n as libc::c_uint, + 0, + std::ptr::null_mut(), + ) + }; + if r < 0 { + return Err(io::Error::last_os_error()); + } + let count = r as usize; + for i in 0..count { + lens[i] = msgs[i].msg_len as usize; + addrs[i] = sockaddr_to_socket_addr(&storages[i]).ok(); + } + + // Walk msgs[0] cmsg chain for SO_RXQ_OVFL. Skip when no + // datagram landed (cmsg buffer is undefined in that case). + let drops = if count > 0 { + parse_drops(&msgs[0].msg_hdr) + } else { + 0 + }; + + Ok((count, drops)) +} + +// ============================================================================ +// Connected-socket construction seam (see `super::connected`) +// ============================================================================ + +/// Linux accepts `SOCK_NONBLOCK | SOCK_CLOEXEC` directly in `socket(2)`. +pub(super) const SOCKET_TYPE: libc::c_int = + libc::SOCK_DGRAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC; + +/// Nothing further to do: `SOCKET_TYPE` already carried the fd flags. +pub(super) fn prepare_fd(_fd: RawFd) -> io::Result<()> { + Ok(()) +} + +/// No Darwin-style service-type tuning applies here. +pub(super) fn tune_connected_socket(_fd: RawFd) {} + +/// Apply receive and send buffer sizes to a connected peer socket. +pub(super) fn set_buf_sizes(fd: RawFd, recv_buf: usize, send_buf: usize) { + set_buf_size(fd, libc::SO_RCVBUFFORCE, libc::SO_RCVBUF, recv_buf); + set_buf_size(fd, libc::SO_SNDBUFFORCE, libc::SO_SNDBUF, send_buf); +} + +/// Try `SO_*BUFFORCE` first (bypasses the rmem/wmem ceiling) and +/// fall back to `SO_*BUF` if that fails. Returns silently — buffer +/// sizing is best-effort. +fn set_buf_size(fd: RawFd, force_name: libc::c_int, normal_name: libc::c_int, size: usize) { + let value: libc::c_int = size as libc::c_int; + let r = unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + force_name, + &value as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if r < 0 { + // Fall back to non-force — kernel may clamp. + let _ = unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + normal_name, + &value as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + } +} diff --git a/src/transport/udp/io/macos.rs b/src/transport/udp/io/macos.rs new file mode 100644 index 0000000..3d081f0 --- /dev/null +++ b/src/transport/udp/io/macos.rs @@ -0,0 +1,199 @@ +//! Darwin-specific UDP receive and connected-socket construction. +//! +//! Holds the `recvmsg_x` FFI and the seam points the shared Unix code +//! delegates to. Darwin has no `SO_RXQ_OVFL` equivalent, so the drop +//! counter is always 0 here; the batching win comes from `recvmsg_x` +//! instead. + +use super::unix::{BATCH_SIZE, sockaddr_to_socket_addr}; +use std::io; +use std::net::SocketAddr; +use std::os::unix::io::RawFd; + +/// Darwin-private `msghdr_x` for the `recvmsg_x` / `sendmsg_x` syscalls. +/// Layout matches `bsd/sys/socket_private.h` in xnu — same as `msghdr` plus +/// a trailing `msg_datalen` (per-message bytes-received output, in lieu of +/// the `msg_len` field that `mmsghdr` uses on Linux). +#[repr(C)] +#[allow(non_camel_case_types)] +struct msghdr_x { + msg_name: *mut libc::c_void, + msg_namelen: libc::socklen_t, + msg_iov: *mut libc::iovec, + msg_iovlen: libc::c_int, + msg_control: *mut libc::c_void, + msg_controllen: libc::socklen_t, + msg_flags: libc::c_int, + msg_datalen: usize, +} + +unsafe extern "C" { + fn recvmsg_x( + s: libc::c_int, + msgp: *const msghdr_x, + cnt: libc::c_uint, + flags: libc::c_int, + ) -> isize; +} + +/// Darwin has no `SO_RXQ_OVFL`, so there is no ancillary data to read; +/// the buffer only has to be large enough to be harmless. +pub(super) const CMSG_BUF_SIZE: usize = 64; + +/// No kernel drop counter is available on Darwin. +pub(super) fn enable_drop_counting(_fd: RawFd) {} + +/// Always 0 — see [`enable_drop_counting`]. +pub(super) fn parse_drops(_msg: &libc::msghdr) -> u32 { + 0 +} + +/// Drain a connected Darwin UDP socket with one `recvmsg_x(2)` call. +/// +/// Connected sockets do not need source-address storage, so this is the +/// compact counterpart to [`recv_batch`] used by per-peer receive threads. +pub(crate) fn recvmsg_x_drain( + fd: RawFd, + backing: &mut [Vec], + lens: &mut [usize], +) -> io::Result { + let n = backing.len().min(lens.len()).min(BATCH_SIZE); + if n == 0 { + return Ok(0); + } + + let mut iovs: [libc::iovec; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + let mut msgs: [msghdr_x; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + + for i in 0..n { + iovs[i].iov_base = backing[i].as_mut_ptr() as *mut libc::c_void; + iovs[i].iov_len = backing[i].len(); + msgs[i].msg_iov = &mut iovs[i]; + msgs[i].msg_iovlen = 1; + } + + let received = loop { + let received = unsafe { recvmsg_x(fd, msgs.as_ptr(), n as libc::c_uint, 0) }; + if received >= 0 { + break received; + } + let error = io::Error::last_os_error(); + if error.kind() != io::ErrorKind::Interrupted { + return Err(error); + } + }; + + let count = received as usize; + if count > n { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "recvmsg_x reported more datagrams than requested", + )); + } + for i in 0..count { + lens[i] = msgs[i].msg_datalen; + } + Ok(count) +} + +/// Receive up to `BATCH_SIZE` datagrams in a single `recvmsg_x` syscall. +/// Same `(count, drops)` contract as the Linux `recv_batch`, except +/// `drops` is always 0 — Darwin has no `SO_RXQ_OVFL` equivalent. +/// +/// `recvmsg_x` is a Darwin-private syscall (not in the public SDK) but +/// is shipped in production xnu and is used by quinn-udp for the same +/// per-syscall-amortisation reason as our Linux `recvmmsg` path. +pub(super) fn recv_batch( + fd: RawFd, + bufs: &mut [&mut [u8]], + addrs: &mut [Option], + lens: &mut [usize], +) -> io::Result<(usize, u32)> { + let n = bufs.len().min(addrs.len()).min(lens.len()).min(BATCH_SIZE); + if n == 0 { + return Ok((0, 0)); + } + + let mut iovs: [libc::iovec; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + let mut storages: [libc::sockaddr_storage; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + let mut msgs: [msghdr_x; BATCH_SIZE] = unsafe { std::mem::zeroed() }; + + for i in 0..n { + iovs[i].iov_base = bufs[i].as_mut_ptr() as *mut libc::c_void; + iovs[i].iov_len = bufs[i].len(); + msgs[i].msg_name = &mut storages[i] as *mut _ as *mut libc::c_void; + msgs[i].msg_namelen = std::mem::size_of::() as libc::socklen_t; + msgs[i].msg_iov = &mut iovs[i]; + msgs[i].msg_iovlen = 1; + // No cmsg consumption — leave msg_control null. (msg_controllen + // is documented as not overwritten by macOS recvmsg_x; zeroed + // init keeps it sane.) + } + + let r = unsafe { recvmsg_x(fd, msgs.as_ptr(), n as libc::c_uint, 0) }; + if r < 0 { + return Err(io::Error::last_os_error()); + } + let count = r as usize; + for i in 0..count { + lens[i] = msgs[i].msg_datalen; + addrs[i] = sockaddr_to_socket_addr(&storages[i]).ok(); + } + + Ok((count, 0)) +} + +// ============================================================================ +// Connected-socket construction seam (see `super::connected`) +// ============================================================================ + +/// Darwin does not accept `SOCK_NONBLOCK` / `SOCK_CLOEXEC` in `socket(2)`; +/// [`prepare_fd`] sets the equivalent fd flags instead. +pub(super) const SOCKET_TYPE: libc::c_int = libc::SOCK_DGRAM; + +/// Set `O_NONBLOCK` and `FD_CLOEXEC` with `fcntl`, standing in for the +/// socket-type flags Linux passes to `socket(2)`. +pub(super) fn prepare_fd(fd: RawFd) -> io::Result<()> { + let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; + if flags < 0 { + return Err(io::Error::last_os_error()); + } + if unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } < 0 { + return Err(io::Error::last_os_error()); + } + + let fd_flags = unsafe { libc::fcntl(fd, libc::F_GETFD) }; + if fd_flags < 0 { + return Err(io::Error::last_os_error()); + } + if unsafe { libc::fcntl(fd, libc::F_SETFD, fd_flags | libc::FD_CLOEXEC) } < 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +/// Apply the Darwin service-type hint to a connected peer socket. +pub(super) fn tune_connected_socket(fd: RawFd) { + super::macos_sockopts::apply_udp_socket_tuning(fd, "connected-udp-peer"); +} + +/// Apply receive and send buffer sizes to a connected peer socket. +/// Darwin has no `SO_*BUFFORCE`, so this is the ceiling-clamped path only. +pub(super) fn set_buf_sizes(fd: RawFd, recv_buf: usize, send_buf: usize) { + set_buf_size(fd, libc::SO_RCVBUF, recv_buf); + set_buf_size(fd, libc::SO_SNDBUF, send_buf); +} + +/// Returns silently — buffer sizing is best-effort. +fn set_buf_size(fd: RawFd, normal_name: libc::c_int, size: usize) { + let value: libc::c_int = size as libc::c_int; + let _ = unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + normal_name, + &value as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; +} diff --git a/src/transport/udp/sockopts_macos.rs b/src/transport/udp/io/macos_sockopts.rs similarity index 100% rename from src/transport/udp/sockopts_macos.rs rename to src/transport/udp/io/macos_sockopts.rs diff --git a/src/transport/udp/io/mod.rs b/src/transport/udp/io/mod.rs new file mode 100644 index 0000000..e372afe --- /dev/null +++ b/src/transport/udp/io/mod.rs @@ -0,0 +1,266 @@ +//! UDP socket wrapper with platform-specific receive implementations. +//! +//! The public surface is two types, `UdpRawSocket` and `AsyncUdpSocket`, +//! supplied by whichever backend this target compiles: +//! +//! - `unix` is shared by every Unix target and holds the socket +//! lifecycle, the synchronous calls and the `AsyncFd` wrapper. Where +//! behaviour genuinely differs per OS it defers to `linux`, `macos` +//! or `unix_other`: `SO_RXQ_OVFL` drop counting on Linux, `recvmsg_x` +//! batching on Darwin, neither elsewhere. +//! - `windows` is a separate backend over `tokio::net::UdpSocket`. It +//! shares no implementation with the Unix side — there is no fd, no +//! `AsyncFd`, no drop counting and no batching — only the type names. +//! +//! `connected` holds the per-peer connected-socket fast path — the fd +//! construction, the owning handle and its drain thread — which exists on +//! Linux and macOS only, and `macos_sockopts` the Darwin service-type +//! tuning it applies. Module names are written plainly rather than as +//! intra-doc links because each one is `cfg`-gated out on some target, so +//! no single configuration can resolve them all. +//! +//! Follows the pattern established by `transport/ethernet/socket.rs`. + +#[cfg(target_os = "linux")] +mod linux; +#[cfg(target_os = "macos")] +mod macos; +#[cfg(target_os = "macos")] +mod macos_sockopts; +#[cfg(unix)] +mod unix; +#[cfg(all(unix, not(any(target_os = "linux", target_os = "macos"))))] +mod unix_other; +#[cfg(windows)] +mod windows; + +#[cfg(any(target_os = "linux", target_os = "macos"))] +mod connected; + +#[cfg(unix)] +pub use unix::{AsyncUdpSocket, UdpRawSocket}; +#[cfg(windows)] +pub use windows::{AsyncUdpSocket, UdpRawSocket}; + +#[cfg(target_os = "macos")] +pub(crate) use macos::recvmsg_x_drain; + +#[cfg(any(target_os = "linux", target_os = "macos"))] +pub(crate) use connected::{ConnectedPeerSocket, PeerRecvDrain, open_connected_fd}; + +#[cfg(test)] +mod tests { + use super::UdpRawSocket; + #[cfg(any(target_os = "linux", target_os = "macos"))] + use std::net::SocketAddr; + + #[test] + fn test_udp_socket_bind() { + // Bind to an ephemeral port + let sock = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) + .expect("failed to bind UDP socket"); + + let addr = sock.local_addr(); + assert!(addr.port() > 0, "should be assigned an ephemeral port"); + assert!(addr.ip().is_loopback()); + } + + #[test] + fn test_udp_socket_buffer_sizes() { + let sock = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) + .expect("failed to bind UDP socket"); + + let recv_buf = sock.recv_buffer_size().expect("get recv buffer"); + let send_buf = sock.send_buffer_size().expect("get send buffer"); + assert!(recv_buf > 0, "recv buffer should be non-zero"); + assert!(send_buf > 0, "send buffer should be non-zero"); + } + + #[tokio::test] + async fn test_async_udp_socket_send_recv() { + let sock1 = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) + .expect("failed to bind socket 1"); + let addr1 = sock1.local_addr(); + let async1 = sock1.into_async().expect("into_async 1"); + + let sock2 = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) + .expect("failed to bind socket 2"); + let addr2 = sock2.local_addr(); + let async2 = sock2.into_async().expect("into_async 2"); + + // Send from socket 1 to socket 2 + let payload = b"hello fips"; + let sent = async1.send_to(payload, &addr2).await.expect("send_to"); + assert_eq!(sent, payload.len()); + + // Receive on socket 2 + let mut buf = [0u8; 1024]; + let (n, src, _drops) = async2.recv_from(&mut buf).await.expect("recv_from"); + assert_eq!(n, payload.len()); + assert_eq!(&buf[..n], payload); + assert_eq!(src, addr1); + } + + /// Microbench: compare per-packet `recv_from` (single recvmsg syscall + + /// task wakeup per datagram — the macOS pre-recvmsg_x baseline) vs + /// `recv_batch` (the new recvmsg_x path, up to 32 datagrams per syscall). + /// Both modes run back-to-back in this binary on loopback so the only + /// thing that varies is the receive-syscall strategy. Sender is a tight + /// `socket.send_to()` loop in a separate task; receiver counts datagrams + /// drained over a fixed wall-clock window per mode. + /// + /// Run with: + /// cargo test --release -p fips --lib transport::udp::io::tests::bench_udp_recv_amortization -- --ignored --nocapture + /// + /// Sender runs on a dedicated *blocking* OS thread (std::net::UdpSocket + /// in default blocking mode) so it always saturates the kernel rx queue + /// regardless of how the tokio receiver schedules. That's the scenario + /// where recvmmsg / recvmsg_x is meant to win: the receiver wakes up to + /// find N packets already buffered, and one syscall reaps the burst. + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[ignore = "microbenchmark; run explicitly with --ignored --nocapture"] + async fn bench_udp_recv_amortization() { + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::time::{Duration, Instant}; + + const RECV_BUF: usize = 4 * 1024 * 1024; + const SEND_BUF: usize = 1024 * 1024; + const PAYLOAD_LEN: usize = 100; + const WINDOW: Duration = Duration::from_secs(3); + const WARMUP: Duration = Duration::from_millis(500); + + async fn run_mode( + label: &str, + batched: bool, + sender_threads: usize, + ) -> (u64, u64, Duration) { + let rx_sock = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), RECV_BUF, SEND_BUF) + .expect("rx bind"); + let rx_addr = rx_sock.local_addr(); + let rx = rx_sock.into_async().expect("rx into_async"); + + // Senders: N dedicated blocking std threads. More threads → deeper + // kernel rx queue → larger amortization opportunity for recv_batch. + // ENOBUFS / EAGAIN just yield and retry; we want saturation, not + // perfect accounting. Sent count is best-effort. + let stop = Arc::new(AtomicBool::new(false)); + let mut sender_handles = Vec::with_capacity(sender_threads); + for _ in 0..sender_threads { + let stop_tx = stop.clone(); + sender_handles.push(std::thread::spawn(move || { + let sock = std::net::UdpSocket::bind("127.0.0.1:0").expect("tx bind"); + sock.connect(rx_addr).expect("tx connect"); + let payload = vec![0xABu8; PAYLOAD_LEN]; + let mut sent: u64 = 0; + while !stop_tx.load(Ordering::Relaxed) { + match sock.send(&payload) { + Ok(_) => sent += 1, + Err(_) => std::thread::yield_now(), + } + } + sent + })); + } + + // Warm-up: let the sender thread reach steady state and the + // kernel rx queue start filling. + tokio::time::sleep(WARMUP).await; + + let start = Instant::now(); + let deadline = start + WINDOW; + let mut recv_count: u64 = 0; + let mut last_drops: u32 = 0; + + if batched { + const BATCH: usize = 32; + let mut backing: Vec> = + (0..BATCH).map(|_| vec![0u8; PAYLOAD_LEN + 64]).collect(); + let mut addrs: [Option; BATCH] = std::array::from_fn(|_| None); + let mut lens: [usize; BATCH] = [0; BATCH]; + let mut batch_sum: u64 = 0; + let mut batch_calls: u64 = 0; + + while Instant::now() < deadline { + let mut bufs: [&mut [u8]; BATCH] = { + let mut iter = backing.iter_mut(); + std::array::from_fn(|_| iter.next().unwrap().as_mut_slice()) + }; + match rx.recv_batch(&mut bufs, &mut addrs, &mut lens).await { + Ok((n, drops)) => { + recv_count += n as u64; + batch_sum += n as u64; + batch_calls += 1; + last_drops = drops; + } + Err(_) => break, + } + } + let avg_batch = if batch_calls > 0 { + batch_sum as f64 / batch_calls as f64 + } else { + 0.0 + }; + eprintln!( + "[{:>10}] avg_batch_per_call={:.2} ({} calls)", + label, avg_batch, batch_calls + ); + } else { + let mut buf = vec![0u8; PAYLOAD_LEN + 64]; + while Instant::now() < deadline { + match rx.recv_from(&mut buf).await { + Ok((_n, _src, drops)) => { + recv_count += 1; + last_drops = drops; + } + Err(_) => break, + } + } + } + let elapsed = start.elapsed(); + + stop.store(true, Ordering::Relaxed); + drop(rx); + let sent: u64 = sender_handles + .into_iter() + .map(|h| h.join().unwrap_or(0)) + .sum(); + + let pps = (recv_count as f64) / elapsed.as_secs_f64(); + let mbps = + (recv_count as f64) * (PAYLOAD_LEN as f64) * 8.0 / 1e6 / elapsed.as_secs_f64(); + eprintln!( + "[{:>10}] recv={:>10} sent={:>10} elapsed={:?} pps={:>12.0} mbps={:>7.1} kdrops={}", + label, recv_count, sent, elapsed, pps, mbps, last_drops + ); + (recv_count, sent, elapsed) + } + + eprintln!("--- udp recv amortization bench ---"); + eprintln!( + "payload={}B window={:?} warmup={:?} runtime=multi_thread(2)", + PAYLOAD_LEN, WINDOW, WARMUP + ); + + // Sweep sender concurrency. Each level shows how the win scales as + // the rx queue gets deeper (more amortization opportunity). + for senders in [1usize, 2, 4, 8] { + eprintln!("\n=== sender_threads = {} ===", senders); + let (b_recv, _, b_el) = run_mode(" recv_from", false, senders).await; + let (x_recv, _, x_el) = run_mode("recv_batch", true, senders).await; + let (x_recv2, _, x_el2) = run_mode("recv_batch", true, senders).await; + let (b_recv2, _, b_el2) = run_mode(" recv_from", false, senders).await; + + let baseline_pps = + (b_recv as f64 / b_el.as_secs_f64() + b_recv2 as f64 / b_el2.as_secs_f64()) / 2.0; + let batched_pps = + (x_recv as f64 / x_el.as_secs_f64() + x_recv2 as f64 / x_el2.as_secs_f64()) / 2.0; + let speedup = batched_pps / baseline_pps; + eprintln!( + "--- senders={}: baseline={:.0} pps batched={:.0} pps speedup={:.2}x ---", + senders, baseline_pps, batched_pps, speedup + ); + } + } +} diff --git a/src/transport/udp/io/unix.rs b/src/transport/udp/io/unix.rs new file mode 100644 index 0000000..0b4c70f --- /dev/null +++ b/src/transport/udp/io/unix.rs @@ -0,0 +1,387 @@ +//! UDP socket implementation shared by every Unix target. +//! +//! Everything here is platform-independent within Unix: socket creation, +//! adoption, buffer sizing, the synchronous send/receive calls, and the +//! `AsyncFd`-based async wrapper. The three points where behaviour genuinely +//! differs per OS — enabling the kernel drop counter, sizing and parsing the +//! ancillary control buffer, and batched receive — are delegated to the `sys` +//! module selected below. + +use crate::transport::TransportError; +use socket2::{Domain, Protocol, Socket, Type}; +use std::net::SocketAddr; +use std::os::unix::io::{AsRawFd, RawFd}; +use std::sync::Arc; +use tokio::io::unix::AsyncFd; +use tracing::warn; + +#[cfg(target_os = "linux")] +use super::linux as sys; +#[cfg(target_os = "macos")] +use super::macos as sys; +#[cfg(not(any(target_os = "linux", target_os = "macos")))] +use super::unix_other as sys; + +/// Maximum number of datagrams a single recvmmsg / recvmsg_x / sendmmsg +/// syscall will pull from / push to the kernel. Tuned to amortise syscall + +/// per-task-wakeup overhead across a useful burst without blowing the +/// stack (each slot owns an mmsghdr/msghdr_x + sockaddr_storage + iovec). +#[cfg(any(target_os = "linux", target_os = "macos"))] +pub(super) const BATCH_SIZE: usize = 32; + +/// Wrapper around a `socket2::Socket` providing sync send/recv with +/// `SO_RXQ_OVFL` ancillary data parsing. +pub struct UdpRawSocket { + inner: Socket, + local_addr: SocketAddr, +} + +impl UdpRawSocket { + /// Create, bind, and configure a UDP socket. + /// + /// Enables `SO_RXQ_OVFL` for kernel drop counting (non-fatal if + /// unsupported). Sets non-blocking mode for async integration. + pub fn open( + bind_addr: SocketAddr, + recv_buf_size: usize, + send_buf_size: usize, + ) -> Result { + let domain = if bind_addr.is_ipv4() { + Domain::IPV4 + } else { + Domain::IPV6 + }; + let sock = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP)) + .map_err(|e| TransportError::StartFailed(format!("socket create failed: {}", e)))?; + + sock.set_nonblocking(true) + .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; + + // SO_REUSEPORT lets per-peer `ConnectedPeerSocket`s bind + // to the same wildcard port the listen socket holds. Must + // be set BEFORE bind. Without this, the connected-UDP + // activation handler fails with EADDRINUSE on Linux and + // every outbound packet falls back to the wildcard listen + // socket — losing the kernel 5-tuple cache benefit and + // most of the multihop forwarding throughput gain. + let _ = sock.set_reuse_port(true); + let _ = sock.set_reuse_address(true); + + sock.bind(&bind_addr.into()) + .map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?; + + // Set socket buffer sizes + sock.set_recv_buffer_size(recv_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; + sock.set_send_buffer_size(send_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; + + let actual_recv = sock + .recv_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?; + let actual_send = sock + .send_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?; + + if actual_recv < recv_buf_size { + warn!( + requested = recv_buf_size, + actual = actual_recv, + "UDP recv buffer clamped by kernel (increase net.core.rmem_max)" + ); + } + if actual_send < send_buf_size { + warn!( + requested = send_buf_size, + actual = actual_send, + "UDP send buffer clamped by kernel (increase net.core.wmem_max)" + ); + } + + sys::enable_drop_counting(sock.as_raw_fd()); + + let local_addr = sock + .local_addr() + .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? + .as_socket() + .ok_or_else(|| { + TransportError::StartFailed("local address is not an IP socket".into()) + })?; + + Ok(Self { + inner: sock, + local_addr, + }) + } + + /// Adopt an existing bound UDP socket. + /// + /// This preserves socket identity/NAT mapping created by bootstrap code. + pub fn adopt( + socket: std::net::UdpSocket, + recv_buf_size: usize, + send_buf_size: usize, + ) -> Result { + let sock = Socket::from(socket); + + sock.set_nonblocking(true) + .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; + + sock.set_recv_buffer_size(recv_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; + sock.set_send_buffer_size(send_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; + + let actual_recv = sock + .recv_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?; + let actual_send = sock + .send_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?; + + if actual_recv < recv_buf_size { + warn!( + requested = recv_buf_size, + actual = actual_recv, + "UDP recv buffer clamped by kernel (increase net.core.rmem_max)" + ); + } + if actual_send < send_buf_size { + warn!( + requested = send_buf_size, + actual = actual_send, + "UDP send buffer clamped by kernel (increase net.core.wmem_max)" + ); + } + + sys::enable_drop_counting(sock.as_raw_fd()); + + let local_addr = sock + .local_addr() + .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? + .as_socket() + .ok_or_else(|| { + TransportError::StartFailed("local address is not an IP socket".into()) + })?; + + Ok(Self { + inner: sock, + local_addr, + }) + } + + /// Get the local bound address. + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } + + /// Get the actual receive buffer size granted by the kernel. + pub fn recv_buffer_size(&self) -> Result { + self.inner + .recv_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e))) + } + + /// Get the actual send buffer size granted by the kernel. + pub fn send_buffer_size(&self) -> Result { + self.inner + .send_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e))) + } + + /// Synchronous send to a destination address. + /// + /// Returns the number of bytes sent, or an `io::Error`. + pub fn send_to(&self, data: &[u8], dest: &SocketAddr) -> std::io::Result { + let dest: socket2::SockAddr = (*dest).into(); + self.inner.send_to(data, &dest) + } + + /// Synchronous receive with `SO_RXQ_OVFL` ancillary data parsing. + /// + /// Returns `(bytes_read, source_addr, kernel_drops)`. The `kernel_drops` + /// value is a cumulative counter since socket creation; it is 0 if + /// `SO_RXQ_OVFL` is not supported. + /// + /// The production receive path on Linux/macOS uses `recv_batch` + /// (recvmmsg / recvmsg_x); this single-packet variant remains for + /// other unix targets and for the local `tests` module. + #[cfg_attr(any(target_os = "linux", target_os = "macos"), allow(dead_code))] + pub fn recv_from(&self, buf: &mut [u8]) -> std::io::Result<(usize, SocketAddr, u32)> { + let fd = self.inner.as_raw_fd(); + + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr() as *mut libc::c_void, + iov_len: buf.len(), + }; + + let mut cmsg_buf = [0u8; sys::CMSG_BUF_SIZE]; + + let mut src_addr: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; + let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; + msg.msg_name = &mut src_addr as *mut _ as *mut libc::c_void; + msg.msg_namelen = std::mem::size_of::() as libc::socklen_t; + msg.msg_iov = &mut iov; + msg.msg_iovlen = 1 as _; + msg.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void; + msg.msg_controllen = cmsg_buf.len() as _; + + let n = unsafe { libc::recvmsg(fd, &mut msg, 0) }; + if n < 0 { + return Err(std::io::Error::last_os_error()); + } + + // Parse source address from sockaddr_storage + let addr = sockaddr_to_socket_addr(&src_addr)?; + + let drops = sys::parse_drops(&msg); + + Ok((n as usize, addr, drops)) + } + + /// Receive up to `BATCH_SIZE` datagrams in a single syscall — + /// `recvmmsg` on Linux, `recvmsg_x` on macOS. See the per-platform + /// implementations for the full `(count, drops)` contract. + #[cfg(any(target_os = "linux", target_os = "macos"))] + pub fn recv_batch( + &self, + bufs: &mut [&mut [u8]], + addrs: &mut [Option], + lens: &mut [usize], + ) -> std::io::Result<(usize, u32)> { + sys::recv_batch(self.inner.as_raw_fd(), bufs, addrs, lens) + } + + /// Wrap this socket in a tokio `AsyncFd` for async I/O. + pub fn into_async(self) -> Result { + let async_fd = AsyncFd::new(self) + .map_err(|e| TransportError::StartFailed(format!("AsyncFd::new failed: {}", e)))?; + Ok(AsyncUdpSocket { + inner: Arc::new(async_fd), + }) + } +} + +impl AsRawFd for UdpRawSocket { + fn as_raw_fd(&self) -> RawFd { + self.inner.as_raw_fd() + } +} + +/// Async wrapper around `UdpRawSocket` using tokio's `AsyncFd`. +/// +/// `Arc`-shareable between send and receive tasks. `AsyncFd` is +/// `Sync` when `T: Send`, which `socket2::Socket` satisfies. +#[derive(Clone)] +pub struct AsyncUdpSocket { + inner: Arc>, +} + +impl AsRawFd for AsyncUdpSocket { + fn as_raw_fd(&self) -> RawFd { + self.inner.get_ref().as_raw_fd() + } +} + +impl AsyncUdpSocket { + /// Send a payload to a destination address. + pub async fn send_to(&self, data: &[u8], dest: &SocketAddr) -> Result { + loop { + let mut guard = self + .inner + .writable() + .await + .map_err(|e| TransportError::SendFailed(format!("writable wait: {}", e)))?; + + match guard.try_io(|inner| inner.get_ref().send_to(data, dest)) { + Ok(Ok(n)) => return Ok(n), + Ok(Err(e)) => return Err(TransportError::SendFailed(format!("{}", e))), + Err(_would_block) => continue, + } + } + } + + /// Receive a payload, source address, and kernel drop counter. + /// + /// Returns `(bytes_read, source_addr, kernel_drops)`. On Linux/macOS + /// the production receive path uses `recv_batch`; this single-packet + /// variant remains for other unix targets and for the local `tests` + /// module. + #[cfg_attr(any(target_os = "linux", target_os = "macos"), allow(dead_code))] + pub async fn recv_from( + &self, + buf: &mut [u8], + ) -> Result<(usize, SocketAddr, u32), TransportError> { + loop { + let mut guard = self + .inner + .readable() + .await + .map_err(|e| TransportError::RecvFailed(format!("readable wait: {}", e)))?; + + match guard.try_io(|inner| inner.get_ref().recv_from(buf)) { + Ok(Ok(result)) => return Ok(result), + Ok(Err(e)) => return Err(TransportError::RecvFailed(format!("{}", e))), + Err(_would_block) => continue, + } + } + } + + /// Drain up to `BATCH_SIZE` datagrams from the kernel via + /// `recvmmsg` (Linux) or `recvmsg_x` (macOS). Returns + /// `(count, kernel_drops)`; same buffer / addr / len contract as + /// `UdpRawSocket::recv_batch`. `kernel_drops` is always 0 on macOS. + #[cfg(any(target_os = "linux", target_os = "macos"))] + pub async fn recv_batch( + &self, + bufs: &mut [&mut [u8]], + addrs: &mut [Option], + lens: &mut [usize], + ) -> Result<(usize, u32), TransportError> { + loop { + let mut guard = self + .inner + .readable() + .await + .map_err(|e| TransportError::RecvFailed(format!("readable wait: {}", e)))?; + + match guard.try_io(|inner| inner.get_ref().recv_batch(bufs, addrs, lens)) { + Ok(Ok((0, _))) => { + // Spurious wakeup or no datagrams ready — yield + // back to the reactor instead of busy-looping. + guard.clear_ready(); + continue; + } + Ok(Ok(result)) => return Ok(result), + Ok(Err(e)) => return Err(TransportError::RecvFailed(format!("{}", e))), + Err(_would_block) => continue, + } + } + } +} + +/// Convert a `libc::sockaddr_storage` to `std::net::SocketAddr`. +pub(super) fn sockaddr_to_socket_addr( + storage: &libc::sockaddr_storage, +) -> std::io::Result { + match storage.ss_family as libc::c_int { + libc::AF_INET => { + let addr: &libc::sockaddr_in = + unsafe { &*(storage as *const _ as *const libc::sockaddr_in) }; + let ip = std::net::Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr)); + let port = u16::from_be(addr.sin_port); + Ok(SocketAddr::from((ip, port))) + } + libc::AF_INET6 => { + let addr: &libc::sockaddr_in6 = + unsafe { &*(storage as *const _ as *const libc::sockaddr_in6) }; + let ip = std::net::Ipv6Addr::from(addr.sin6_addr.s6_addr); + let port = u16::from_be(addr.sin6_port); + Ok(SocketAddr::from((ip, port))) + } + family => Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("unsupported address family: {}", family), + )), + } +} diff --git a/src/transport/udp/io/unix_other.rs b/src/transport/udp/io/unix_other.rs new file mode 100644 index 0000000..e14abbf --- /dev/null +++ b/src/transport/udp/io/unix_other.rs @@ -0,0 +1,24 @@ +//! Seam implementation for Unix targets that are neither Linux nor macOS. +//! +//! Two supported targets land here: FreeBSD, and Android — which is +//! `target_os = "android"` rather than `"linux"`, and is the one of the two +//! that CI actually lints. +//! +//! These targets get the portable `recvmsg(2)` path only: no kernel drop +//! counter and no receive batching, so the shared code's `recv_batch` and +//! the connected-socket fast path are both compiled out for them. What +//! remains is the small set of hooks `super::unix` calls unconditionally. + +use std::os::unix::io::RawFd; + +/// No ancillary data is consumed on these targets; the buffer only has to +/// be large enough to be harmless. +pub(super) const CMSG_BUF_SIZE: usize = 64; + +/// No kernel drop counter is available. +pub(super) fn enable_drop_counting(_fd: RawFd) {} + +/// Always 0 — see [`enable_drop_counting`]. +pub(super) fn parse_drops(_msg: &libc::msghdr) -> u32 { + 0 +} diff --git a/src/transport/udp/io/windows.rs b/src/transport/udp/io/windows.rs new file mode 100644 index 0000000..bf71de9 --- /dev/null +++ b/src/transport/udp/io/windows.rs @@ -0,0 +1,162 @@ +//! UDP socket implementation for Windows. +//! +//! Uses `socket2::Socket` for configuration and `tokio::net::UdpSocket` +//! for async I/O. Neither kernel drop counting nor receive batching is +//! available, so this backend shares no code with the Unix one beyond the +//! type names it exports. + +use crate::transport::TransportError; +use socket2::{Domain, Protocol, Socket, Type}; +use std::net::SocketAddr; +use std::sync::Arc; + +/// UDP socket wrapper (Windows). +/// +/// Uses `socket2::Socket` for configuration and `tokio::net::UdpSocket` +/// for async I/O. Kernel drop counting is not available on Windows; +/// the drops field always returns 0. +pub struct UdpRawSocket { + inner: Socket, + local_addr: SocketAddr, +} + +impl UdpRawSocket { + /// Create, bind, and configure a UDP socket. + /// + /// Sets non-blocking mode and configures buffer sizes. The socket + /// is bound immediately so `local_addr()` returns the actual + /// assigned address (important when binding to port 0). + pub fn open( + bind_addr: SocketAddr, + recv_buf_size: usize, + send_buf_size: usize, + ) -> Result { + let domain = if bind_addr.is_ipv4() { + Domain::IPV4 + } else { + Domain::IPV6 + }; + let sock = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP)) + .map_err(|e| TransportError::StartFailed(format!("socket create failed: {}", e)))?; + + sock.set_nonblocking(true) + .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; + + sock.bind(&bind_addr.into()) + .map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?; + + // Set socket buffer sizes + sock.set_recv_buffer_size(recv_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; + sock.set_send_buffer_size(send_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; + + let local_addr = sock + .local_addr() + .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? + .as_socket() + .ok_or_else(|| { + TransportError::StartFailed("local address is not an IP socket".into()) + })?; + + Ok(Self { + inner: sock, + local_addr, + }) + } + + /// Adopt an existing bound UDP socket. + pub fn adopt( + socket: std::net::UdpSocket, + recv_buf_size: usize, + send_buf_size: usize, + ) -> Result { + let sock = Socket::from(socket); + + sock.set_nonblocking(true) + .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; + + sock.set_recv_buffer_size(recv_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; + sock.set_send_buffer_size(send_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; + + let local_addr = sock + .local_addr() + .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? + .as_socket() + .ok_or_else(|| { + TransportError::StartFailed("local address is not an IP socket".into()) + })?; + + Ok(Self { + inner: sock, + local_addr, + }) + } + + /// Get the local bound address. + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } + + /// Get the actual receive buffer size. + pub fn recv_buffer_size(&self) -> Result { + self.inner + .recv_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e))) + } + + /// Get the actual send buffer size. + pub fn send_buffer_size(&self) -> Result { + self.inner + .send_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e))) + } + + /// Wrap this socket in an async wrapper for tokio I/O. + pub fn into_async(self) -> Result { + let std_socket: std::net::UdpSocket = self.inner.into(); + let tokio_socket = tokio::net::UdpSocket::from_std(std_socket) + .map_err(|e| TransportError::StartFailed(format!("tokio socket failed: {}", e)))?; + + Ok(AsyncUdpSocket { + inner: Arc::new(tokio_socket), + }) + } +} + +/// Async UDP socket wrapper (Windows). +/// +/// Uses `tokio::net::UdpSocket` directly. Kernel drop counting +/// is not available; the drops field always returns 0. +#[derive(Clone)] +pub struct AsyncUdpSocket { + inner: Arc, +} + +impl AsyncUdpSocket { + /// Send a payload to a destination address. + pub async fn send_to(&self, data: &[u8], dest: &SocketAddr) -> Result { + self.inner + .send_to(data, dest) + .await + .map_err(|e| TransportError::SendFailed(format!("{}", e))) + } + + /// Receive a payload, source address, and kernel drop counter. + /// + /// Returns `(bytes_read, source_addr, 0)`. The drops field is always 0 + /// on Windows since kernel drop counting is not available. + pub async fn recv_from( + &self, + buf: &mut [u8], + ) -> Result<(usize, SocketAddr, u32), TransportError> { + let (n, addr) = self + .inner + .recv_from(buf) + .await + .map_err(|e| TransportError::RecvFailed(format!("{}", e)))?; + Ok((n, addr, 0)) + } +} diff --git a/src/transport/udp/mod.rs b/src/transport/udp/mod.rs index d7a4fbe..e24c535 100644 --- a/src/transport/udp/mod.rs +++ b/src/transport/udp/mod.rs @@ -7,10 +7,8 @@ use super::{ TransportId, TransportState, TransportType, }; pub(crate) mod io; -#[cfg(target_os = "macos")] -pub(crate) mod sockopts_macos; #[cfg(any(target_os = "linux", target_os = "macos"))] -pub(crate) use io::open_connected_fd; +pub(crate) use io::{ConnectedPeerSocket, PeerRecvDrain, open_connected_fd}; mod stats; use super::resolve_socket_addr; use crate::config::UdpConfig;