Batch macOS connected UDP receives

Connected peer drains currently issue one recv syscall for each queued datagram on macOS even though the wildcard UDP path already uses recvmsg_x.

Reuse the Darwin batch ABI for connected sockets, preserving partial-batch and EINTR behavior while requesting up to 32 datagrams per receive call. Add a connected-socket burst test that crosses the batch boundary.

CHANGELOG.md

@@ -60,6 +60,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Connected UDP peer drains now batch macOS receives with `recvmsg_x(2)`,
+  matching the wildcard UDP receive path instead of issuing one `recv(2)`
+  syscall per queued datagram.
+
 - The Ethernet transport's per-interface `discovery` flag was renamed to
   `listen` (`transports.ethernet.*`) to match the symmetric `announce`
   (transmit) / `listen` (receive) neighbor-beacon vocabulary. The old

src/peer/connected_udp/drain.rs

@@ -9,7 +9,7 @@
 //!
 //! This module owns the drain side: spawn one OS thread per connected
 //! socket, drain into a fixed-size batch (`recvmmsg(2)` on Linux,
-//! repeated nonblocking `recv(2)` on Darwin), push each packet into
+//! `recvmsg_x(2)` on Darwin), push each packet into
 //! the existing `packet_tx` (the same channel that the wildcard listen
 //! socket feeds), and exit cleanly when the parent signals shutdown
 //! via a self-pipe.
@@ -334,9 +334,9 @@ fn drain_packets(fd: RawFd, backing: &mut [Vec<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.rs

@@ -64,6 +64,56 @@ mod platform {
         ) -> isize;
     }
 
+    /// Drain a connected Darwin UDP socket with one `recvmsg_x(2)` call.
+    ///
+    /// Connected sockets do not need source-address storage, so this is the
+    /// compact counterpart to [`UdpRawSocket::recv_batch`] used by per-peer
+    /// receive threads.
+    #[cfg(target_os = "macos")]
+    pub(crate) fn recvmsg_x_drain(
+        fd: RawFd,
+        backing: &mut [Vec<u8>],
+        lens: &mut [usize],
+    ) -> std::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 = std::io::Error::last_os_error();
+            if error.kind() != std::io::ErrorKind::Interrupted {
+                return Err(error);
+            }
+        };
+
+        let count = received as usize;
+        if count > n {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "recvmsg_x reported more datagrams than requested",
+            ));
+        }
+        for i in 0..count {
+            lens[i] = msgs[i].msg_datalen;
+        }
+        Ok(count)
+    }
+
     /// Wrapper around a `socket2::Socket` providing sync send/recv with
     /// `SO_RXQ_OVFL` ancillary data parsing.
     pub struct UdpRawSocket {
@@ -782,6 +832,8 @@ mod platform {
     }
 }
 
+#[cfg(target_os = "macos")]
+pub(crate) use platform::recvmsg_x_drain;
 pub use platform::{AsyncUdpSocket, UdpRawSocket};
 
 /// Per-peer connected-UDP fast-path fd construction.