diff --git a/src/node/encrypt_worker.rs b/src/node/encrypt_worker.rs index 80219da..677debb 100644 --- a/src/node/encrypt_worker.rs +++ b/src/node/encrypt_worker.rs @@ -132,8 +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. @@ -716,8 +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, @@ -754,9 +752,7 @@ impl MacSequencedSendFlow { fn spawn( key: MacSendFlowKey, socket: AsyncUdpSocket, - connected_socket: Option< - std::sync::Arc, - >, + connected_socket: Option>, dest_addr: SocketAddr, now_ms: u64, ) -> Arc { @@ -1024,8 +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/node/handlers/connected_udp.rs b/src/node/handlers/connected_udp.rs index e125b1b..fc9d851 100644 --- a/src/node/handlers/connected_udp.rs +++ b/src/node/handlers/connected_udp.rs @@ -142,20 +142,24 @@ impl Node { (peer_sa, local, recv_buf, send_buf, tx) }; - // Open the connected socket on the kernel side. - let socket = std::sync::Arc::new( - crate::transport::udp::connected_peer::ConnectedPeerSocket::open( - local_addr, - peer_socket_addr, - recv_buf, - send_buf, - ) - .map_err(|e| format!("ConnectedPeerSocket::open: {e}"))?, - ); + // Open the connected socket on the kernel side, then adopt the + // fd into the owning handle. + let owned = crate::transport::udp::open_connected_fd( + local_addr, + peer_socket_addr, + recv_buf, + send_buf, + ) + .map_err(|e| format!("open_connected_fd: {e}"))?; + let socket = std::sync::Arc::new(crate::peer::connected_udp::ConnectedPeerSocket::from_fd( + owned, + peer_socket_addr, + local_addr, + )); // Spawn the drain thread. It feeds `packet_tx` exactly like // the wildcard listen socket — rx_loop dispatches identically. - let drain = crate::transport::udp::peer_drain::PeerRecvDrain::spawn( + let drain = crate::peer::connected_udp::PeerRecvDrain::spawn( socket.clone(), transport_id, peer_socket_addr, diff --git a/src/peer/active.rs b/src/peer/active.rs index d727f72..5c29c25 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -206,15 +206,14 @@ pub struct ActivePeer { /// 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, } impl ActivePeer { @@ -374,7 +373,7 @@ impl ActivePeer { #[cfg(any(target_os = "linux", target_os = "macos"))] pub(crate) fn connected_udp( &self, - ) -> Option> { + ) -> Option> { self.connected_udp.clone() } @@ -384,8 +383,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::transport::udp::peer_drain::PeerRecvDrain, + socket: std::sync::Arc, + drain: crate::peer::connected_udp::PeerRecvDrain, ) { // Drop the old drain BEFORE the old socket so its last fd // reference is released cleanly. diff --git a/src/transport/udp/peer_drain.rs b/src/peer/connected_udp/drain.rs similarity index 97% rename from src/transport/udp/peer_drain.rs rename to src/peer/connected_udp/drain.rs index 5e12442..af83c94 100644 --- a/src/transport/udp/peer_drain.rs +++ b/src/peer/connected_udp/drain.rs @@ -1,5 +1,6 @@ -// Paired with `connected_peer.rs`: dormant in this PR until the -// activation handler is wired into the node tick (follow-up). +// Paired with the connected-socket opener (`io::open_connected_fd`): +// dormant in this PR until the activation handler is wired into the +// node tick (follow-up). #![allow(dead_code)] //! Recv-side drain thread for a per-peer connected UDP socket. @@ -23,11 +24,8 @@ //! of a dedicated OS thread. The drain *function* `drain_loop` stays //! useful in either shape; only the wakeup mechanism differs. -#![cfg(any(target_os = "linux", target_os = "macos"))] - -use super::super::{ReceivedPacket, TransportAddr, TransportId}; -use super::PacketTx; -use super::connected_peer::ConnectedPeerSocket; +use super::socket::ConnectedPeerSocket; +use crate::transport::{PacketTx, ReceivedPacket, TransportAddr, TransportId}; use std::io; use std::net::SocketAddr; use std::os::unix::io::{AsRawFd, RawFd}; @@ -448,10 +446,10 @@ mod tests { // Our connected socket. Use an ephemeral local port so we // don't conflict with anything else on the test host. let local_addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); - let socket = Arc::new( - ConnectedPeerSocket::open(local_addr, peer_addr, 1 << 20, 1 << 20) - .expect("ConnectedPeerSocket::open"), - ); + let owned = + crate::transport::udp::open_connected_fd(local_addr, peer_addr, 1 << 20, 1 << 20) + .expect("open_connected_fd"); + let socket = Arc::new(ConnectedPeerSocket::from_fd(owned, peer_addr, local_addr)); // packet_tx for the drain thread to push into. let (tx, mut rx) = mpsc::channel::(64); diff --git a/src/peer/connected_udp/mod.rs b/src/peer/connected_udp/mod.rs new file mode 100644 index 0000000..bb8d58a --- /dev/null +++ b/src/peer/connected_udp/mod.rs @@ -0,0 +1,16 @@ +//! 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/connected_udp/socket.rs b/src/peer/connected_udp/socket.rs new file mode 100644 index 0000000..85b709a --- /dev/null +++ b/src/peer/connected_udp/socket.rs @@ -0,0 +1,235 @@ +//! Owning handle for a per-peer `connect()`-ed UDP socket. +//! +//! Adopts an fd produced by `crate::transport::udp::open_connected_fd` +//! and closes it on drop. See that function's docs for why established +//! peers get their own connected socket. +#![allow(dead_code)] + +use std::net::SocketAddr; +use std::os::unix::io::{AsRawFd, IntoRawFd, OwnedFd, RawFd}; + +/// A `connect()`-ed UDP socket for one established peer. +/// +/// Owns the raw fd and closes it on drop. Configured (by +/// `crate::transport::udp::open_connected_fd`) with: +/// - `SO_REUSEADDR` and `SO_REUSEPORT` so it can share the listen port +/// with the wildcard socket and any other peers' connected sockets. +/// - The receive / send buffer sizes inherited from the configured +/// UDP transport (best-effort via `*BUFFORCE` variants — the kernel +/// silently falls back to the normal `*BUF` ceiling if our process +/// lacks `CAP_NET_ADMIN`). +/// - `O_NONBLOCK` so callers that drive it from an OS-thread shard +/// loop don't accidentally block the entire shard on a single +/// recv / send. +/// - `connect()`-ed to the peer's `SocketAddr`, locking in the +/// per-packet kernel-side route + ARP / neighbor cache so neither +/// needs to be redone on the data path. +#[derive(Debug)] +pub(crate) struct ConnectedPeerSocket { + fd: RawFd, + peer_addr: SocketAddr, + local_addr: SocketAddr, +} + +impl ConnectedPeerSocket { + /// Adopt an already-opened, bound, and `connect()`-ed fd (from + /// `crate::transport::udp::open_connected_fd`) into an owning + /// handle. Takes ownership of the fd; it is closed on drop. + pub(crate) fn from_fd(fd: OwnedFd, peer_addr: SocketAddr, local_addr: SocketAddr) -> Self { + Self { + fd: fd.into_raw_fd(), + peer_addr, + local_addr, + } + } + + pub fn peer_addr(&self) -> SocketAddr { + self.peer_addr + } + + #[allow(dead_code)] // wired up by future per-peer recv loops + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } +} + +impl AsRawFd for ConnectedPeerSocket { + fn as_raw_fd(&self) -> RawFd { + self.fd + } +} + +impl Drop for ConnectedPeerSocket { + fn drop(&mut self) { + // Best-effort close. Ignore the result — if close fails the + // kernel has already done what it can; we don't want to panic + // in Drop. + unsafe { + libc::close(self.fd); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::UdpSocket; + + /// Open a connected peer socket the way production does: build the + /// fd via the transport opener, then adopt it into the handle. + fn open( + local_addr: SocketAddr, + peer_addr: SocketAddr, + recv_buf: usize, + send_buf: usize, + ) -> std::io::Result { + let fd = + crate::transport::udp::open_connected_fd(local_addr, peer_addr, recv_buf, send_buf)?; + Ok(ConnectedPeerSocket::from_fd(fd, peer_addr, local_addr)) + } + + /// Open a connected peer socket against a fresh loopback UDP + /// listener and exercise the round-trip: connected socket sends + /// without msg_name → listener receives → listener replies → + /// connected socket receives without parsing msg_name. Validates + /// reuse flags + `bind` + `connect` + `O_NONBLOCK`. + #[test] + fn open_send_recv_loopback() { + // Peer (the "remote") side: a regular blocking UDP socket on + // loopback. We'll have our connected socket send to it. + let peer = UdpSocket::bind("127.0.0.1:0").expect("bind peer"); + let peer_addr = peer.local_addr().expect("peer local_addr"); + peer.set_read_timeout(Some(std::time::Duration::from_millis(500))) + .expect("set_read_timeout"); + + // Our side: a wildcard listen address (use 127.0.0.1:0 to + // avoid colliding with any real local service). Connect to the + // peer. Linux requires that we bind before connect — the + // ConnectedPeerSocket constructor does both. + let local_addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let sock = open( + local_addr, + peer_addr, + /* recv_buf */ 1 << 20, + /* send_buf */ 1 << 20, + ) + .expect("ConnectedPeerSocket::open"); + + // Confirm the socket is in fact connected: `send(2)` should + // succeed without specifying a destination. + let payload = b"hello-from-connected-socket"; + let r = unsafe { + libc::send( + sock.as_raw_fd(), + payload.as_ptr() as *const libc::c_void, + payload.len(), + 0, + ) + }; + assert!(r >= 0, "send failed: {}", std::io::Error::last_os_error()); + assert_eq!(r as usize, payload.len()); + + let mut recv_buf = [0u8; 64]; + let (len, from) = peer.recv_from(&mut recv_buf).expect("peer recv"); + assert_eq!(len, payload.len()); + assert_eq!(&recv_buf[..len], payload); + + // Reply back from the peer. Since our socket is connected to + // peer_addr, the kernel UDP demux should route this packet to + // our connected socket (most-specific-match) and `recv(2)` + // without sockaddr should pick it up. + let reply = b"hello-back"; + peer.send_to(reply, from).expect("peer send_to"); + + // Drain on the connected socket. Spin briefly because + // O_NONBLOCK + a tiny one-shot recv would race with the + // kernel's veth-less loopback delivery. + let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500); + loop { + let mut buf = [0u8; 64]; + let r = unsafe { + libc::recv( + sock.as_raw_fd(), + buf.as_mut_ptr() as *mut libc::c_void, + buf.len(), + 0, + ) + }; + if r >= 0 { + assert_eq!(r as usize, reply.len()); + assert_eq!(&buf[..r as usize], reply); + break; + } + let err = std::io::Error::last_os_error(); + if err.kind() == std::io::ErrorKind::WouldBlock { + if std::time::Instant::now() >= deadline { + panic!("connected socket never received reply"); + } + std::thread::sleep(std::time::Duration::from_millis(2)); + continue; + } + panic!("recv failed: {err}"); + } + } + + /// Two connected sockets coexisting on the same local port via + /// `SO_REUSEPORT`, each connected to a different peer. + #[test] + fn two_connected_sockets_share_listen_port() { + let peer_a = UdpSocket::bind("127.0.0.1:0").expect("bind peer_a"); + let peer_b = UdpSocket::bind("127.0.0.1:0").expect("bind peer_b"); + let peer_a_addr = peer_a.local_addr().expect("peer_a local_addr"); + let peer_b_addr = peer_b.local_addr().expect("peer_b local_addr"); + + // Anchor a shared local port via a wildcard socket on a + // non-zero ephemeral port, then open two connected sockets + // bound to the same port. + let anchor = UdpSocket::bind("127.0.0.1:0").expect("bind anchor"); + let shared_port = anchor.local_addr().expect("anchor local_addr").port(); + let shared_local: SocketAddr = format!("127.0.0.1:{shared_port}").parse().unwrap(); + // Drop the anchor so the only thing holding the port is the + // connected sockets' reuse semantics. + drop(anchor); + + let sock_a = open(shared_local, peer_a_addr, 1 << 20, 1 << 20).expect("open sock_a"); + let sock_b = open(shared_local, peer_b_addr, 1 << 20, 1 << 20).expect("open sock_b"); + + assert_eq!(sock_a.peer_addr(), peer_a_addr); + assert_eq!(sock_b.peer_addr(), peer_b_addr); + } + + /// The production fast path keeps the wildcard UDP listener bound + /// while opening a sibling socket connected to a peer. This catches + /// the Darwin regression where the adopted traversal socket used a + /// different reuse mode than the connected-peer socket and every + /// activation failed with EADDRINUSE. + #[test] + fn connected_socket_shares_live_listener_port() { + let peer = UdpSocket::bind("127.0.0.1:0").expect("bind peer"); + let peer_addr = peer.local_addr().expect("peer local_addr"); + + let listener = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::DGRAM, + Some(socket2::Protocol::UDP), + ) + .expect("create listener"); + listener + .set_reuse_address(true) + .expect("listener reuseaddr"); + listener.set_reuse_port(true).expect("listener reuseport"); + listener + .bind(&"0.0.0.0:0".parse::().unwrap().into()) + .expect("bind listener"); + let local = listener + .local_addr() + .expect("listener local addr") + .as_socket() + .expect("ip socket"); + + let sock = open(local, peer_addr, 1 << 20, 1 << 20).expect("open connected sibling"); + + assert_eq!(sock.local_addr(), local); + assert_eq!(sock.peer_addr(), peer_addr); + } +} diff --git a/src/peer/mod.rs b/src/peer/mod.rs index 07bcea1..95c2a56 100644 --- a/src/peer/mod.rs +++ b/src/peer/mod.rs @@ -8,6 +8,8 @@ //! while maintaining type safety for phase-specific operations. mod active; +#[cfg(any(target_os = "linux", target_os = "macos"))] +pub(crate) mod connected_udp; mod connection; pub use active::{ActivePeer, ConnectivityState}; diff --git a/src/transport/udp/connected_peer.rs b/src/transport/udp/connected_peer.rs deleted file mode 100644 index fa8adb8..0000000 --- a/src/transport/udp/connected_peer.rs +++ /dev/null @@ -1,462 +0,0 @@ -// The connected-UDP fast path is infra-ready but not yet wired into the -// encrypt-worker dispatch site (a follow-up PR will refcount-clone the -// socket into each FmpSendJob). Keep the API surface in tree. -#![allow(dead_code)] - -//! Connected per-peer UDP socket. -//! -//! 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), so the demux happens once at socket-receive -//! time instead of repeatedly at the application layer. -//! - Lets us `sendmsg(2)` with `msg_name = NULL` (or `send(2)`), -//! skipping the per-packet sockaddr copy + route lookup + neighbor -//! resolve that the kernel otherwise repeats for every datagram on -//! an unconnected socket. -//! - Combines cleanly with UDP_GSO: the connected socket sends one -//! super-skb to one cached destination, and the kernel skips -//! per-segment route lookups. -//! -//! Multiple connected sockets coexist with the wildcard listen socket -//! via `SO_REUSEPORT` plus `SO_REUSEADDR`. The UDP demux picks the -//! most specific match (5-tuple of connected sockets beats the -//! wildcard), so traffic from new / unknown peers continues to land on -//! the listen socket (handshakes / discovery), while established peers' -//! steady-state traffic goes directly to their dedicated socket. -//! -//! **Scope of this module:** infrastructure only — the FD lifecycle -//! (open / close), buffer sizing (matches the listen socket's -//! `SO_RCVBUF` / `SO_SNDBUF` via `FORCE` variants where possible), -//! and unit tests exercising the open + bind + connect path. - -#![cfg(any(target_os = "linux", target_os = "macos"))] - -use std::io; -use std::net::SocketAddr; -use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd}; - -/// A `connect()`-ed UDP socket for one established peer. -/// -/// Owns the raw fd and closes it on drop. Configured with: -/// - `SO_REUSEADDR` and `SO_REUSEPORT` so it can share the listen port -/// with the wildcard socket and any other peers' connected sockets. -/// - The receive / send buffer sizes inherited from the configured -/// UDP transport (best-effort via `*BUFFORCE` variants — the kernel -/// silently falls back to the normal `*BUF` ceiling if our process -/// lacks `CAP_NET_ADMIN`). -/// - `O_NONBLOCK` so callers that drive it from an OS-thread shard -/// loop don't accidentally block the entire shard on a single -/// recv / send. -/// - `connect()`-ed to the peer's `SocketAddr`, locking in the -/// per-packet kernel-side route + ARP / neighbor cache so neither -/// needs to be redone on the data path. -#[derive(Debug)] -pub(crate) struct ConnectedPeerSocket { - fd: RawFd, - peer_addr: SocketAddr, - local_addr: SocketAddr, -} - -impl ConnectedPeerSocket { - /// Open a new peer-connected UDP socket. - /// - /// `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 fn open( - local_addr: SocketAddr, - peer_addr: SocketAddr, - recv_buf: usize, - send_buf: usize, - ) -> io::Result { - let fd = open_connected_fd(local_addr, peer_addr, recv_buf, send_buf)?; - Ok(Self { - fd: fd.into_raw_fd(), - peer_addr, - local_addr, - }) - } - - #[cfg(not(target_os = "linux"))] - fn set_nonblocking_cloexec(&self) -> io::Result<()> { - let flags = unsafe { libc::fcntl(self.fd, libc::F_GETFL) }; - if flags < 0 { - return Err(io::Error::last_os_error()); - } - if unsafe { libc::fcntl(self.fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } < 0 { - return Err(io::Error::last_os_error()); - } - - let fd_flags = unsafe { libc::fcntl(self.fd, libc::F_GETFD) }; - if fd_flags < 0 { - return Err(io::Error::last_os_error()); - } - if unsafe { libc::fcntl(self.fd, libc::F_SETFD, fd_flags | libc::FD_CLOEXEC) } < 0 { - return Err(io::Error::last_os_error()); - } - Ok(()) - } - - /// Set an integer-valued socket option. Returns the kernel error - /// on failure but doesn't `?`-propagate caller-side because most - /// callers want to log + continue rather than fail the whole open. - fn set_sockopt_int( - &self, - level: libc::c_int, - name: libc::c_int, - value: libc::c_int, - ) -> io::Result<()> { - let r = unsafe { - libc::setsockopt( - self.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(&self, 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( - self.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( - self.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(&self, normal_name: libc::c_int, size: usize) { - let value: libc::c_int = size as libc::c_int; - let _ = unsafe { - libc::setsockopt( - self.fd, - libc::SOL_SOCKET, - normal_name, - &value as *const _ as *const libc::c_void, - std::mem::size_of::() as libc::socklen_t, - ) - }; - } - - pub fn peer_addr(&self) -> SocketAddr { - self.peer_addr - } - - #[allow(dead_code)] // wired up by future per-peer recv loops - pub fn local_addr(&self) -> SocketAddr { - self.local_addr - } -} - -/// Open a `connect()`-ed UDP socket for one peer and return the owning -/// fd. Performs the full socket / sockopt / bind / connect syscall -/// sequence (see [`ConnectedPeerSocket`] for the rationale). On any -/// mid-construction failure the fd is closed before the error is -/// returned; on success ownership of the fd transfers to the returned -/// [`OwnedFd`]. `ConnectedPeerSocket::open` is a thin wrapper that -/// adopts this fd into the struct. -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()); - } - // Take ownership of the fd in a temporary wrapper so its Drop - // closes the fd on any error path below. Ownership is transferred - // out (via mem::forget + OwnedFd) only after connect() succeeds. - let sock = ConnectedPeerSocket { - fd, - peer_addr, - local_addr, - }; - #[cfg(not(target_os = "linux"))] - sock.set_nonblocking_cloexec()?; - - // 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. - sock.set_sockopt_int(libc::SOL_SOCKET, libc::SO_REUSEADDR, 1)?; - sock.set_sockopt_int(libc::SOL_SOCKET, libc::SO_REUSEPORT, 1)?; - - #[cfg(target_os = "macos")] - crate::transport::udp::darwin_sockopts::apply_udp_socket_tuning(sock.fd, "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")] - { - sock.set_buf_size(libc::SO_RCVBUFFORCE, libc::SO_RCVBUF, recv_buf); - sock.set_buf_size(libc::SO_SNDBUFFORCE, libc::SO_SNDBUF, send_buf); - } - #[cfg(not(target_os = "linux"))] - { - sock.set_buf_size(libc::SO_RCVBUF, recv_buf); - sock.set_buf_size(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( - sock.fd, - local_sa.as_ptr() as *const libc::sockaddr, - local_sa.len(), - ) - }; - if bind_r < 0 { - return Err(io::Error::last_os_error()); - } - - // 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( - sock.fd, - peer_sa.as_ptr() as *const libc::sockaddr, - peer_sa.len(), - ) - }; - if conn_r < 0 { - return Err(io::Error::last_os_error()); - } - - // Success: transfer ownership of the fd to an OwnedFd. Neutralize - // the temporary wrapper's Drop so the fd is not closed here — its - // close-on-error role is done. Every early error above still drops - // `sock` and closes the fd, unchanged from before. - let raw = sock.fd; - std::mem::forget(sock); - Ok(unsafe { OwnedFd::from_raw_fd(raw) }) -} - -impl AsRawFd for ConnectedPeerSocket { - fn as_raw_fd(&self) -> RawFd { - self.fd - } -} - -impl Drop for ConnectedPeerSocket { - fn drop(&mut self) { - // Best-effort close. Ignore the result — if close fails the - // kernel has already done what it can; we don't want to panic - // in Drop. - unsafe { - libc::close(self.fd); - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::net::UdpSocket; - - /// Open a connected peer socket against a fresh loopback UDP - /// listener and exercise the round-trip: connected socket sends - /// without msg_name → listener receives → listener replies → - /// connected socket receives without parsing msg_name. Validates - /// reuse flags + `bind` + `connect` + `O_NONBLOCK`. - #[test] - fn open_send_recv_loopback() { - // Peer (the "remote") side: a regular blocking UDP socket on - // loopback. We'll have our connected socket send to it. - let peer = UdpSocket::bind("127.0.0.1:0").expect("bind peer"); - let peer_addr = peer.local_addr().expect("peer local_addr"); - peer.set_read_timeout(Some(std::time::Duration::from_millis(500))) - .expect("set_read_timeout"); - - // Our side: a wildcard listen address (use 127.0.0.1:0 to - // avoid colliding with any real local service). Connect to the - // peer. Linux requires that we bind before connect — the - // ConnectedPeerSocket constructor does both. - let local_addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); - let sock = ConnectedPeerSocket::open( - local_addr, - peer_addr, - /* recv_buf */ 1 << 20, - /* send_buf */ 1 << 20, - ) - .expect("ConnectedPeerSocket::open"); - - // Confirm the socket is in fact connected: `send(2)` should - // succeed without specifying a destination. - let payload = b"hello-from-connected-socket"; - let r = unsafe { - libc::send( - sock.as_raw_fd(), - payload.as_ptr() as *const libc::c_void, - payload.len(), - 0, - ) - }; - assert!(r >= 0, "send failed: {}", std::io::Error::last_os_error()); - assert_eq!(r as usize, payload.len()); - - let mut recv_buf = [0u8; 64]; - let (len, from) = peer.recv_from(&mut recv_buf).expect("peer recv"); - assert_eq!(len, payload.len()); - assert_eq!(&recv_buf[..len], payload); - - // Reply back from the peer. Since our socket is connected to - // peer_addr, the kernel UDP demux should route this packet to - // our connected socket (most-specific-match) and `recv(2)` - // without sockaddr should pick it up. - let reply = b"hello-back"; - peer.send_to(reply, from).expect("peer send_to"); - - // Drain on the connected socket. Spin briefly because - // O_NONBLOCK + a tiny one-shot recv would race with the - // kernel's veth-less loopback delivery. - let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500); - loop { - let mut buf = [0u8; 64]; - let r = unsafe { - libc::recv( - sock.as_raw_fd(), - buf.as_mut_ptr() as *mut libc::c_void, - buf.len(), - 0, - ) - }; - if r >= 0 { - assert_eq!(r as usize, reply.len()); - assert_eq!(&buf[..r as usize], reply); - break; - } - let err = std::io::Error::last_os_error(); - if err.kind() == std::io::ErrorKind::WouldBlock { - if std::time::Instant::now() >= deadline { - panic!("connected socket never received reply"); - } - std::thread::sleep(std::time::Duration::from_millis(2)); - continue; - } - panic!("recv failed: {err}"); - } - } - - /// Two connected sockets coexisting on the same local port via - /// `SO_REUSEPORT`, each connected to a different peer. - #[test] - fn two_connected_sockets_share_listen_port() { - let peer_a = UdpSocket::bind("127.0.0.1:0").expect("bind peer_a"); - let peer_b = UdpSocket::bind("127.0.0.1:0").expect("bind peer_b"); - let peer_a_addr = peer_a.local_addr().expect("peer_a local_addr"); - let peer_b_addr = peer_b.local_addr().expect("peer_b local_addr"); - - // Anchor a shared local port via a wildcard socket on a - // non-zero ephemeral port, then open two connected sockets - // bound to the same port. - let anchor = UdpSocket::bind("127.0.0.1:0").expect("bind anchor"); - let shared_port = anchor.local_addr().expect("anchor local_addr").port(); - let shared_local: SocketAddr = format!("127.0.0.1:{shared_port}").parse().unwrap(); - // Drop the anchor so the only thing holding the port is the - // connected sockets' reuse semantics. - drop(anchor); - - let sock_a = ConnectedPeerSocket::open(shared_local, peer_a_addr, 1 << 20, 1 << 20) - .expect("open sock_a"); - let sock_b = ConnectedPeerSocket::open(shared_local, peer_b_addr, 1 << 20, 1 << 20) - .expect("open sock_b"); - - assert_eq!(sock_a.peer_addr(), peer_a_addr); - assert_eq!(sock_b.peer_addr(), peer_b_addr); - } - - /// The production fast path keeps the wildcard UDP listener bound - /// while opening a sibling socket connected to a peer. This catches - /// the Darwin regression where the adopted traversal socket used a - /// different reuse mode than the connected-peer socket and every - /// activation failed with EADDRINUSE. - #[test] - fn connected_socket_shares_live_listener_port() { - let peer = UdpSocket::bind("127.0.0.1:0").expect("bind peer"); - let peer_addr = peer.local_addr().expect("peer local_addr"); - - let listener = socket2::Socket::new( - socket2::Domain::IPV4, - socket2::Type::DGRAM, - Some(socket2::Protocol::UDP), - ) - .expect("create listener"); - listener - .set_reuse_address(true) - .expect("listener reuseaddr"); - listener.set_reuse_port(true).expect("listener reuseport"); - listener - .bind(&"0.0.0.0:0".parse::().unwrap().into()) - .expect("bind listener"); - let local = listener - .local_addr() - .expect("listener local addr") - .as_socket() - .expect("ip socket"); - - let sock = ConnectedPeerSocket::open(local, peer_addr, 1 << 20, 1 << 20) - .expect("open connected sibling"); - - assert_eq!(sock.local_addr(), local); - assert_eq!(sock.peer_addr(), peer_addr); - } -} diff --git a/src/transport/udp/darwin_sockopts.rs b/src/transport/udp/darwin_sockopts.rs index 3d6c2f6..c54fc71 100644 --- a/src/transport/udp/darwin_sockopts.rs +++ b/src/transport/udp/darwin_sockopts.rs @@ -1,6 +1,6 @@ -// Applied at `ConnectedPeerSocket::open` (dormant in this PR; see -// `connected_peer.rs`). Linux toolchain only checks gates — keep -// the module visible on Linux so clippy doesn't lose track. +// Applied inside `io::open_connected_fd` (dormant in this PR; see +// `io.rs`). Linux toolchain only checks gates — keep the module +// visible on Linux so clippy doesn't lose track. #![allow(dead_code)] //! Darwin UDP socket tuning. diff --git a/src/transport/udp/io.rs b/src/transport/udp/io.rs index a4e6460..979f5af 100644 --- a/src/transport/udp/io.rs +++ b/src/transport/udp/io.rs @@ -784,6 +784,232 @@ mod platform { 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 + `darwin_sockopts` specific. +#[cfg(any(target_os = "linux", target_os = "macos"))] +mod connected { + // The connected-UDP fast path is infra-ready but not yet wired into + // the encrypt-worker dispatch site (a follow-up PR will refcount-clone + // the socket into each FmpSendJob). Keep the API surface in tree. + #![allow(dead_code)] + + 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::darwin_sockopts::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(io::Error::last_os_error()); + } + + // 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(io::Error::last_os_error()); + } + + Ok(owned) + } + + #[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(any(target_os = "linux", target_os = "macos"))] +pub(crate) use connected::open_connected_fd; + #[cfg(test)] mod tests { use super::*; diff --git a/src/transport/udp/mod.rs b/src/transport/udp/mod.rs index a8ad380..1d8a64e 100644 --- a/src/transport/udp/mod.rs +++ b/src/transport/udp/mod.rs @@ -6,13 +6,11 @@ use super::{ DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, TransportError, TransportId, TransportState, TransportType, }; -#[cfg(unix)] -pub(crate) mod connected_peer; #[cfg(target_os = "macos")] pub(crate) mod darwin_sockopts; pub(crate) mod io; -#[cfg(unix)] -pub(crate) mod peer_drain; +#[cfg(any(target_os = "linux", target_os = "macos"))] +pub(crate) use io::open_connected_fd; mod stats; use super::resolve_socket_addr; use crate::config::UdpConfig;