laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-12 09:33:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into next

Browse Source 
Carries the macOS batched connected-peer receive and the UDP io
reorganization up from master.

Only CHANGELOG.md conflicted, both sides having added to the head of
`### Changed`; resolved as the union, master's entry first.

The restructure looked risky to merge and was not: at `3cdd529` this
branch was byte-identical to the merge base on every file it touches
(`io.rs`, `sockopts_macos.rs`, the whole of `peer/connected_udp/`,
`peer/mod.rs`, `transport/udp/mod.rs`), so master's side carries with
nothing here to lose. `peer/active.rs` is the one file where both lines
had moved; it auto-merged, and the result carries all five of master's
path rewrites alongside this branch's own XX-side changes.

Quartet green on the merged tree at 1794 lib tests.
This commit is contained in:
Johnathan Corgan
2026-08-10 01:52:49 +00:00
parent 3cdd5290de 0c458b380a
commit c80c93f89e
19 changed files with 1505 additions and 1338 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+4
View File
@@ -165,6 +165,10 @@ with v0.4.x or earlier peers.
### Changed
- Connected UDP peer drains now batch macOS receives with `recvmsg_x(2)`,
matching the wildcard UDP receive path instead of issuing one `recv(2)`
syscall per queued datagram.
- Inbound msg1 is classified before it is rate limited, and rekey or restart
msg1 arriving on a link belonging to a promoted peer now draws on its own
token bucket instead of competing with stranger admission for a single
src/node/dataplane/connected_udp.rs
+2 -2
View File
@@ -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,
src/node/encrypt_worker.rs
+4 -4
View File
@@ -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<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
pub connected_socket: Option<std::sync::Arc<crate::transport::udp::ConnectedPeerSocket>>,
/// 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<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
connected_socket: Option<std::sync::Arc<crate::transport::udp::ConnectedPeerSocket>>,
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<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
connected_socket: Option<std::sync::Arc<crate::transport::udp::ConnectedPeerSocket>>,
dest_addr: SocketAddr,
now_ms: u64,
) -> Arc<Self> {
@@ -1020,7 +1020,7 @@ fn flush_batch_sync(
struct EncryptedGroup {
socket: AsyncUdpSocket,
#[cfg(any(target_os = "linux", target_os = "macos"))]
connected_socket: Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
connected_socket: Option<std::sync::Arc<crate::transport::udp::ConnectedPeerSocket>>,
dest_addr: SocketAddr,
wire_packets: Vec<Vec<u8>>,
drop_on_backpressure: bool,
src/peer/active.rs
+5 -5
View File
@@ -147,14 +147,14 @@ struct PeerSendState {
/// per-packet sockaddr handling + route lookup. Behind an `Arc` so
/// in-flight worker jobs survive rekey/address-change rotations.
#[cfg(any(target_os = "linux", target_os = "macos"))]
connected_udp: Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
connected_udp: Option<std::sync::Arc<crate::transport::udp::ConnectedPeerSocket>>,
/// 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<crate::peer::connected_udp::PeerRecvDrain>,
peer_recv_drain: Option<crate::transport::udp::PeerRecvDrain>,
// === Hot counters ===
/// Link statistics.
@@ -475,7 +475,7 @@ impl ActivePeer {
#[cfg(any(target_os = "linux", target_os = "macos"))]
pub(crate) fn connected_udp(
&self,
) -> Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>> {
) -> Option<std::sync::Arc<crate::transport::udp::ConnectedPeerSocket>> {
self.send.connected_udp.clone()
}
@@ -485,8 +485,8 @@ impl ActivePeer {
#[cfg(any(target_os = "linux", target_os = "macos"))]
pub(crate) fn set_connected_udp(
&mut self,
socket: std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>,
drain: crate::peer::connected_udp::PeerRecvDrain,
socket: std::sync::Arc<crate::transport::udp::ConnectedPeerSocket>,
drain: crate::transport::udp::PeerRecvDrain,
) {
// Drop the old drain BEFORE the old socket so its last fd
// reference is released cleanly.
src/peer/connected_udp/mod.rs
-16
View File
@@ -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;
src/peer/mod.rs
-2
View File
@@ -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};
src/transport/udp/io.rs
-1270
View File
File diff suppressed because it is too large Load Diff
src/peer/connected_udp/drain.rs → src/transport/udp/io/connected/drain.rs
+57 -36
View File
@@ -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<u8>], 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<u8>], lens: &mut [usize]) -> io::Result<usize> {
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<u8>], lens: &mut [usize]) -> io:
Ok(count)
}
#[cfg(not(target_os = "linux"))]
fn recv_drain(fd: RawFd, backing: &mut [Vec<u8>], lens: &mut [usize]) -> io::Result<usize> {
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<Vec<u8>> = (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.
src/transport/udp/io/connected/fd.rs
+180
View File
@@ -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<OwnedFd> {
// 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::<libc::c_int>() 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}");
}
}
src/transport/udp/io/connected/mod.rs
+33
View File
@@ -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;
src/peer/connected_udp/socket.rs → src/transport/udp/io/connected/socket.rs
View File
src/transport/udp/io/linux.rs
+181
View File
@@ -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::<libc::c_int>() 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<SocketAddr>],
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::<libc::sockaddr_storage>() 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::<libc::c_int>() 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::<libc::c_int>() as libc::socklen_t,
)
};
}
}
src/transport/udp/io/macos.rs
+199
View File
@@ -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<u8>],
lens: &mut [usize],
) -> io::Result<usize> {
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<SocketAddr>],
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::<libc::sockaddr_storage>() 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::<libc::c_int>() as libc::socklen_t,
)
};
}
src/transport/udp/sockopts_macos.rs → src/transport/udp/io/macos_sockopts.rs
View File
src/transport/udp/io/mod.rs
+266
View File
@@ -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<Vec<u8>> =
(0..BATCH).map(|_| vec![0u8; PAYLOAD_LEN + 64]).collect();
let mut addrs: [Option<SocketAddr>; 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
);
}
}
}
src/transport/udp/io/unix.rs
+387
View File
@@ -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<Self, TransportError> {
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<Self, TransportError> {
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<usize, TransportError> {
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<usize, TransportError> {
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<usize> {
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::<libc::sockaddr_storage>() 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<SocketAddr>],
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<AsyncUdpSocket, TransportError> {
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<T>` is
/// `Sync` when `T: Send`, which `socket2::Socket` satisfies.
#[derive(Clone)]
pub struct AsyncUdpSocket {
inner: Arc<AsyncFd<UdpRawSocket>>,
}
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<usize, TransportError> {
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<SocketAddr>],
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<SocketAddr> {
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),
)),
}
}
src/transport/udp/io/unix_other.rs
+24
View File
@@ -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
}
src/transport/udp/io/windows.rs
+162
View File
@@ -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<Self, TransportError> {
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<Self, TransportError> {
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<usize, TransportError> {
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<usize, TransportError> {
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<AsyncUdpSocket, TransportError> {
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<tokio::net::UdpSocket>,
}
impl AsyncUdpSocket {
/// Send a payload to a destination address.
pub async fn send_to(&self, data: &[u8], dest: &SocketAddr) -> Result<usize, TransportError> {
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))
}
}
src/transport/udp/mod.rs
+1 -3
View File
@@ -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;