From fa736b19a8c5b1517ff04dbcf3c8ec5be8862060 Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Sun, 9 Aug 2026 22:59:10 +0300 Subject: [PATCH 1/2] Batch macOS connected UDP receives Connected peer drains currently issue one recv syscall for each queued datagram on macOS even though the wildcard UDP path already uses recvmsg_x. Reuse the Darwin batch ABI for connected sockets, preserving partial-batch and EINTR behavior while requesting up to 32 datagrams per receive call. Add a connected-socket burst test that crosses the batch boundary. --- CHANGELOG.md | 4 ++ src/peer/connected_udp/drain.rs | 93 ++++++++++++++++++++------------- src/transport/udp/io.rs | 52 ++++++++++++++++++ 3 files changed, 113 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c9cdcb6..6d428bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -60,6 +60,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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. + - The Ethernet transport's per-interface `discovery` flag was renamed to `listen` (`transports.ethernet.*`) to match the symmetric `announce` (transmit) / `listen` (receive) neighbor-beacon vocabulary. The old diff --git a/src/peer/connected_udp/drain.rs b/src/peer/connected_udp/drain.rs index 46e06a5..de27650 100644 --- a/src/peer/connected_udp/drain.rs +++ b/src/peer/connected_udp/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.rs b/src/transport/udp/io.rs index 6a2842a..5fcf6e6 100644 --- a/src/transport/udp/io.rs +++ b/src/transport/udp/io.rs @@ -64,6 +64,56 @@ mod platform { ) -> isize; } + /// 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 [`UdpRawSocket::recv_batch`] used by per-peer + /// receive threads. + #[cfg(target_os = "macos")] + pub(crate) fn recvmsg_x_drain( + fd: RawFd, + backing: &mut [Vec], + lens: &mut [usize], + ) -> std::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 = std::io::Error::last_os_error(); + if error.kind() != std::io::ErrorKind::Interrupted { + return Err(error); + } + }; + + let count = received as usize; + if count > n { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "recvmsg_x reported more datagrams than requested", + )); + } + for i in 0..count { + lens[i] = msgs[i].msg_datalen; + } + Ok(count) + } + /// Wrapper around a `socket2::Socket` providing sync send/recv with /// `SO_RXQ_OVFL` ancillary data parsing. pub struct UdpRawSocket { @@ -782,6 +832,8 @@ mod platform { } } +#[cfg(target_os = "macos")] +pub(crate) use platform::recvmsg_x_drain; pub use platform::{AsyncUdpSocket, UdpRawSocket}; /// Per-peer connected-UDP fast-path fd construction. From 0c458b380a535c6b270417c4c0919b33fbcf9145 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 10 Aug 2026 01:27:03 +0000 Subject: [PATCH 2/2] Give the UDP io module and the connected-socket fast path one home each MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `io.rs` had grown to 1322 lines in which almost nothing was actually generic: two mutually exclusive `platform` modules, plus a third module for the connected-socket fast path, with Linux and macOS diverging repeatedly inside the Unix one rather than at any module boundary. The only genuinely shared items were the doc header, four imports and the re-exports. Turn it into a directory that says which is which. `io/unix.rs` holds everything both Unix targets share: socket creation and adoption, buffer sizing, the synchronous calls, and the `AsyncFd` wrapper. The three points where behaviour really differs move to `io/linux.rs` and `io/macos.rs` behind identical signatures, so the shared file selects a `sys` module once at the top and its bodies carry no `cfg` at all: - `enable_drop_counting` — `SO_RXQ_OVFL` on Linux, nothing on Darwin - `CMSG_BUF_SIZE` / `parse_drops` — the ancillary control buffer and its reader, which only Linux populates - `recv_batch` — `recvmmsg` against `recvmsg_x` `io/unix_other.rs` supplies the same three for the Unix targets that are neither: FreeBSD, and Android, which is `target_os = "android"` and is the one of the two that CI lints. Naming them is the point — this arm was previously spelled `not(target_os = "linux")` and was only correct because of a `cfg` two files away. `io/windows.rs` stays a whole separate backend over `tokio::net::UdpSocket`. It shares no implementation with the Unix side, so it exports the same two type names and nothing else. The connected-socket fast path becomes `io/connected/`, and this is the part that was split across two trees. Constructing the fd lived in `io.rs`; the handle that adopts it and the drain thread that must accompany it lived under `peer/`, which they had no reason to — that module imported nothing from `peer`, reached only into `transport::udp`, and pointed at it three times in its own docs to explain itself. All three pieces are now siblings: `fd.rs`, `socket.rs` and `drain.rs`. The four platform differences in fd construction — the `socket(2)` type flags, the follow-up fd-flag call, the service-type tuning and the buffer-size strategy — join the same `sys` seam as the receive code, which is what makes this the right parent for them. `sockopts_macos.rs` moves in beside them as `io/macos_sockopts.rs`: only this module ever consumed it, so it drops from `pub(crate)` to a private child, and its file-wide `dead_code` allowance for the unreferenced service-type table stays confined to the constants it was written for. No behaviour change. Every call sequence, error path and syscall order is preserved, including the order in which the connected socket applies its options and which of those calls propagate errors rather than being best-effort. The crate-facing names are unchanged except for the two handle types, which move with the module and are re-exported as `transport::udp::{ConnectedPeerSocket, PeerRecvDrain}` beside the `open_connected_fd` that was already there. --- src/node/dataplane/connected_udp.rs | 4 +- src/node/encrypt_worker.rs | 8 +- src/peer/active.rs | 10 +- src/peer/connected_udp/mod.rs | 16 - src/peer/mod.rs | 2 - src/transport/udp/io.rs | 1322 ----------------- .../udp/io/connected}/drain.rs | 0 src/transport/udp/io/connected/fd.rs | 180 +++ src/transport/udp/io/connected/mod.rs | 33 + .../udp/io/connected}/socket.rs | 0 src/transport/udp/io/linux.rs | 181 +++ src/transport/udp/io/macos.rs | 199 +++ .../macos_sockopts.rs} | 0 src/transport/udp/io/mod.rs | 266 ++++ src/transport/udp/io/unix.rs | 387 +++++ src/transport/udp/io/unix_other.rs | 24 + src/transport/udp/io/windows.rs | 162 ++ src/transport/udp/mod.rs | 4 +- 18 files changed, 1444 insertions(+), 1354 deletions(-) delete mode 100644 src/peer/connected_udp/mod.rs delete mode 100644 src/transport/udp/io.rs rename src/{peer/connected_udp => transport/udp/io/connected}/drain.rs (100%) create mode 100644 src/transport/udp/io/connected/fd.rs create mode 100644 src/transport/udp/io/connected/mod.rs rename src/{peer/connected_udp => transport/udp/io/connected}/socket.rs (100%) create mode 100644 src/transport/udp/io/linux.rs create mode 100644 src/transport/udp/io/macos.rs rename src/transport/udp/{sockopts_macos.rs => io/macos_sockopts.rs} (100%) create mode 100644 src/transport/udp/io/mod.rs create mode 100644 src/transport/udp/io/unix.rs create mode 100644 src/transport/udp/io/unix_other.rs create mode 100644 src/transport/udp/io/windows.rs 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 07b1023..36cafb3 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -134,14 +134,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. @@ -412,7 +412,7 @@ impl ActivePeer { #[cfg(any(target_os = "linux", target_os = "macos"))] pub(crate) fn connected_udp( &self, - ) -> Option> { + ) -> Option> { self.send.connected_udp.clone() } @@ -422,8 +422,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 5fcf6e6..0000000 --- a/src/transport/udp/io.rs +++ /dev/null @@ -1,1322 +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; - } - - /// 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 [`UdpRawSocket::recv_batch`] used by per-peer - /// receive threads. - #[cfg(target_os = "macos")] - pub(crate) fn recvmsg_x_drain( - fd: RawFd, - backing: &mut [Vec], - lens: &mut [usize], - ) -> std::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 = std::io::Error::last_os_error(); - if error.kind() != std::io::ErrorKind::Interrupted { - return Err(error); - } - }; - - let count = received as usize; - if count > n { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "recvmsg_x reported more datagrams than requested", - )); - } - for i in 0..count { - lens[i] = msgs[i].msg_datalen; - } - Ok(count) - } - - /// 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)) - } - } -} - -#[cfg(target_os = "macos")] -pub(crate) use platform::recvmsg_x_drain; -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 100% rename from src/peer/connected_udp/drain.rs rename to src/transport/udp/io/connected/drain.rs 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;