diff --git a/quinn-udp/Cargo.toml b/quinn-udp/Cargo.toml index 5e7f47ae5..96c9aceda 100644 --- a/quinn-udp/Cargo.toml +++ b/quinn-udp/Cargo.toml @@ -17,7 +17,7 @@ default = ["tracing", "tracing-log"] # Configure `tracing` to log events via `log` if no `tracing` subscriber exists. tracing-log = ["tracing/log"] log = ["dep:log"] -# Use private Apple APIs to send multiple packets in a single syscall. +# Support private Apple APIs to send multiple packets in a single syscall. fast-apple-datapath = [] [dependencies] diff --git a/quinn-udp/src/unix.rs b/quinn-udp/src/unix.rs index 7b7818f1f..12cf02af9 100644 --- a/quinn-udp/src/unix.rs +++ b/quinn-udp/src/unix.rs @@ -72,6 +72,13 @@ pub struct UdpSocketState { /// In particular, we do not use IP_TOS cmsg_type in this case, /// which is not supported on Linux <3.13 and results in not sending the UDP packet at all. sendmsg_einval: AtomicBool, + + /// Whether to use Apple's fast `sendmsg_x`/`recvmsg_x` APIs. + /// + /// These private APIs provide better performance but may not be available on all + /// Apple OS versions. Callers must verify availability before enabling. + #[cfg(apple_fast)] + apple_fast_path: AtomicBool, } impl UdpSocketState { @@ -191,6 +198,8 @@ impl UdpSocketState { gro_segments: gro::gro_segments(), may_fragment, sendmsg_einval: AtomicBool::new(false), + #[cfg(apple_fast)] + apple_fast_path: AtomicBool::new(false), }) } @@ -225,13 +234,50 @@ impl UdpSocketState { send(self, socket.0, transmit) } + #[cfg(not(any( + apple, + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly", + solarish + )))] pub fn recv( &self, socket: UdpSockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta], ) -> io::Result { - recv(socket.0, bufs, meta) + recv_via_recvmmsg(socket.0, bufs, meta) + } + + #[cfg(apple_fast)] + pub fn recv( + &self, + socket: UdpSockRef<'_>, + bufs: &mut [IoSliceMut<'_>], + meta: &mut [RecvMeta], + ) -> io::Result { + if self.is_apple_fast_path_enabled() { + recv_via_recvmsg_x(socket.0, bufs, meta) + } else { + recv_single(socket.0, bufs, meta) + } + } + + #[cfg(any( + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly", + solarish, + apple_slow + ))] + pub fn recv( + &self, + socket: UdpSockRef<'_>, + bufs: &mut [IoSliceMut<'_>], + meta: &mut [RecvMeta], + ) -> io::Result { + recv_single(socket.0, bufs, meta) } /// The maximum amount of segments which can be transmitted if a platform @@ -295,6 +341,27 @@ impl UdpSocketState { fn set_sendmsg_einval(&self) { self.sendmsg_einval.store(true, Ordering::Relaxed) } + + /// Enables Apple's fast UDP datapath using private `sendmsg_x`/`recvmsg_x` APIs. + /// Once enabled, this also updates [`max_gso_segments`] to allow batched sends. + /// + /// # Safety + /// + /// These APIs may crash on unsupported OS versions, so callers must verify + /// availability before enabling. + /// + /// [`max_gso_segments`]: Self::max_gso_segments + #[cfg(apple_fast)] + pub unsafe fn set_apple_fast_path(&self) { + self.apple_fast_path.store(true, Ordering::Relaxed); + self.max_gso_segments.store(BATCH_SIZE, Ordering::Relaxed); + } + + /// Returns whether Apple's fast UDP datapath is enabled for this socket. + #[cfg(apple_fast)] + pub fn is_apple_fast_path_enabled(&self) -> bool { + self.apple_fast_path.load(Ordering::Relaxed) + } } #[cfg(not(any(apple, target_os = "openbsd", target_os = "netbsd")))] @@ -384,6 +451,20 @@ fn send( #[cfg(apple_fast)] fn send(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> { + if state.is_apple_fast_path_enabled() { + send_via_sendmsg_x(state, io, transmit) + } else { + send_single(state, io, transmit) + } +} + +/// Send using the fast `sendmsg_x` API. +#[cfg(apple_fast)] +fn send_via_sendmsg_x( + state: &UdpSocketState, + io: SockRef<'_>, + transmit: &Transmit<'_>, +) -> io::Result<()> { let mut hdrs = unsafe { mem::zeroed::<[msghdr_x; BATCH_SIZE]>() }; let mut iovs = unsafe { mem::zeroed::<[libc::iovec; BATCH_SIZE]>() }; let mut ctrls = [cmsg::Aligned([0u8; CMSG_LEN]); BATCH_SIZE]; @@ -468,6 +549,7 @@ fn send_single(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit<'_>) } } +/// Receive using the batched `recvmmsg` syscall. #[cfg(not(any( apple, target_os = "openbsd", @@ -475,7 +557,11 @@ fn send_single(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit<'_>) target_os = "dragonfly", solarish )))] -fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> io::Result { +fn recv_via_recvmmsg( + io: SockRef<'_>, + bufs: &mut [IoSliceMut<'_>], + meta: &mut [RecvMeta], +) -> io::Result { let mut names = [MaybeUninit::::uninit(); BATCH_SIZE]; let mut ctrls = [cmsg::Aligned(MaybeUninit::<[u8; CMSG_LEN]>::uninit()); BATCH_SIZE]; let mut hdrs = unsafe { mem::zeroed::<[libc::mmsghdr; BATCH_SIZE]>() }; @@ -516,8 +602,13 @@ fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> Ok(msg_count as usize) } +/// Receive using the fast `recvmsg_x` API. #[cfg(apple_fast)] -fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> io::Result { +fn recv_via_recvmsg_x( + io: SockRef<'_>, + bufs: &mut [IoSliceMut<'_>], + meta: &mut [RecvMeta], +) -> io::Result { let mut names = [MaybeUninit::::uninit(); BATCH_SIZE]; // MacOS 10.15 `recvmsg_x` does not override the `msghdr_x` // `msg_controllen`. Thus, after the call to `recvmsg_x`, one does not know @@ -551,17 +642,6 @@ fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> Ok(msg_count as usize) } -#[cfg(any( - target_os = "openbsd", - target_os = "netbsd", - target_os = "dragonfly", - solarish, - apple_slow -))] -fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> io::Result { - recv_single(io, bufs, meta) -} - #[cfg(any( target_os = "openbsd", target_os = "netbsd", @@ -1078,19 +1158,14 @@ mod gso { // On Apple platforms using the `sendmsg_x` call, UDP datagram segmentation is not // offloaded to the NIC or even the kernel, but instead done here in user space in // [`send`]) and then passed to the OS as individual `iovec`s (up to `BATCH_SIZE`). +// The initial value is 1 (no batching); callers can enable batching via +// `UdpSocketState::set_apple_fast_path()` which updates `max_gso_segments`. #[cfg(not(any(target_os = "linux", target_os = "android")))] mod gso { use super::*; pub(super) fn max_gso_segments() -> usize { - #[cfg(apple_fast)] - { - BATCH_SIZE - } - #[cfg(not(apple_fast))] - { - 1 - } + 1 } #[cfg_attr(apple_fast, allow(dead_code))] // Unused when apple_fast is enabled diff --git a/quinn-udp/tests/tests.rs b/quinn-udp/tests/tests.rs index 41e707f4f..0862f1835 100644 --- a/quinn-udp/tests/tests.rs +++ b/quinn-udp/tests/tests.rs @@ -369,3 +369,95 @@ fn ip_to_v6_mapped(x: IpAddr) -> IpAddr { IpAddr::V6(_) => x, } } + +/// Test Apple fast datapath enable/disable functionality. +/// +/// This test verifies that: +/// 1. `max_gso_segments()` returns 1 by default (fast path disabled) +/// 2. After calling `set_apple_fast_path()`, `max_gso_segments()` returns `BATCH_SIZE` +/// 3. Send/recv still works correctly with the fast path enabled +#[test] +#[cfg(apple_fast)] +fn apple_fast_datapath() { + let send = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let recv = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let dst_addr = recv.local_addr().unwrap(); + + let send_state = UdpSocketState::new((&send).into()).unwrap(); + let recv_state = UdpSocketState::new((&recv).into()).unwrap(); + + // Initially, fast path should be disabled and max_gso_segments should be 1 + assert!( + !send_state.is_apple_fast_path_enabled(), + "fast path should be disabled initially" + ); + assert_eq!( + send_state.max_gso_segments(), + 1, + "max_gso_segments should be 1 before enabling fast path" + ); + + // Enable the fast path + // SAFETY: Assume that sendmsg_x/recvmsg_x are available on the macOS test host. + unsafe { + send_state.set_apple_fast_path(); + recv_state.set_apple_fast_path(); + } + + // After enabling, fast path should be enabled and max_gso_segments should be BATCH_SIZE + assert!( + send_state.is_apple_fast_path_enabled(), + "fast path should be enabled after calling set_apple_fast_path()" + ); + assert_eq!( + send_state.max_gso_segments(), + quinn_udp::BATCH_SIZE, + "max_gso_segments should be BATCH_SIZE after enabling fast path" + ); + + // Verify send/recv still works with fast path enabled + recv.set_nonblocking(false).unwrap(); + + const SEGMENT_SIZE: usize = 128; + let segments = send_state.max_gso_segments(); + let msg = vec![0xAB; SEGMENT_SIZE * segments]; + + send_state + .try_send( + (&send).into(), + &Transmit { + destination: dst_addr, + ecn: None, + contents: &msg, + segment_size: Some(SEGMENT_SIZE), + src_ip: None, + }, + ) + .unwrap(); + + // Receive all segments + let mut buf = [0u8; u16::MAX as usize]; + let mut total_received = 0; + while total_received < segments { + let mut meta = RecvMeta::default(); + let n = recv_state + .recv( + (&recv).into(), + &mut [IoSliceMut::new(&mut buf)], + slice::from_mut(&mut meta), + ) + .unwrap(); + assert_eq!(n, 1); + let received_segments = meta.len / meta.stride; + for i in 0..received_segments { + assert_eq!( + &buf[i * meta.stride..(i + 1) * meta.stride], + &msg[(total_received + i) * SEGMENT_SIZE..(total_received + i + 1) * SEGMENT_SIZE], + "segment {} content mismatch", + total_received + i + ); + } + total_received += received_segments; + } + assert_eq!(total_received, segments, "should receive all segments"); +}