API changes

1) Remove Mio layer and replace it with a sync FD layer
2) Move RecvData, RecvMetrics, and StoreCmsgSettings to `lib`, since
   syscalls is private
3) Rework the signature of `recv_msg` to take a `StoreCmsgSettings` to
   allow applications to handle custom cmsgs

apps/src/sendto.rs

@@ -55,15 +55,30 @@ fn send_to_gso_pacing(
 ) -> io::Result<usize> {
     use dgram::GsoSettings;
 
-    dgram::sync::send_to(
-        socket,
-        buf,
-        Some(GsoSettings {
-            segment_size: segment_size as u16,
-        }),
-        Some(send_info.at),
-        &send_info.to,
-    )
+    loop {
+        // Important to use try_io so events keep coming even if we see
+        // EAGAIN/EWOULDBLOCK
+        let res = socket.try_io(|| {
+            // mio::net::UdpSocket doesn't implement AsFd (yet?).
+            let fd = unsafe {
+                std::os::fd::BorrowedFd::borrow_raw(socket.as_raw_fd())
+            };
+
+            dgram::sync::send_to(
+                socket,
+                buf,
+                Some(GsoSettings {
+                    segment_size: segment_size as u16,
+                }),
+                Some(send_info.at),
+                &send_info.to,
+            )
+        });
+
+        if let Ok(sent) = res {
+            return sent.map_err(Into::into);
+        }
+    }
 }
 
 /// For non-Linux platforms.

dgram/src/lib.rs

@@ -4,16 +4,91 @@ mod syscalls;
 #[cfg(feature = "async")]
 pub mod tokio;
 
+use std::net::SocketAddr;
+use std::time::SystemTime;
+
+use nix::sys::socket::ControlMessageOwned;
+
 #[derive(Default, Copy, Clone)]
 pub struct GsoSettings {
     pub segment_size: u16,
 }
 
+#[cfg(target_os = "linux")]
+#[derive(Default)]
+pub struct StoreCmsgSettings {
+    store_cmsgs: bool,
+    cmsg_space: Vec<u8>,
+}
+
+/// Output of a `recvmsg` call.
+#[derive(Default)]
+pub struct RecvData {
+    /// The number of bytes which `recvmsg` returned.
+    pub bytes: usize,
+    /// The peer address for this message.
+    pub peer_addr: Option<SocketAddr>,
+    /// Metrics for this `recvmsg` call.
+    ///
+    /// If no valid metrics exist - for example, when the RXQOVFL sockopt is not
+    /// set - this will be `None`.
+    pub metrics: Option<RecvMetrics>,
+    /// The `UDP_GRO_SEGMENTS` control message data from the result of
+    /// `recvmsg`, if it exist.
+    pub gro: Option<u16>,
+    /// The RX_TIME control message data from the result of `recvmsg`, if it
+    /// exists.
+    pub rx_time: Option<SystemTime>,
+    cmsgs: Vec<ControlMessageOwned>,
+}
+
+impl RecvData {
+    pub fn new(
+        peer_addr: Option<SocketAddr>, bytes: usize, cmsg_space_len: usize,
+    ) -> Self {
+        Self {
+            peer_addr,
+            bytes,
+            metrics: None,
+            gro: None,
+            rx_time: None,
+            cmsgs: Vec::with_capacity(cmsg_space_len),
+        }
+    }
+
+    pub fn from_bytes(bytes: usize) -> Self {
+        Self {
+            bytes,
+            ..Default::default()
+        }
+    }
+
+    /// Returns the list of cmsgs which were returned from calling `recvmsg`. If
+    /// `recvmsg` was called with `recv_cmsgs` set to to `false`, this will
+    /// return an empty slice.
+    pub fn cmsgs(&self) -> &[ControlMessageOwned] {
+        &self.cmsgs
+    }
+}
+
+/// Metrics for `recvmsg` calls.
+#[derive(Default)]
+pub struct RecvMetrics {
+    /// The number of packets dropped between the last received packet and this
+    /// one.
+    ///
+    /// See SO_RXQOVFL for more.
+    pub udp_packets_dropped: u64,
+}
+
 #[cfg(target_os = "linux")]
 mod linux_imports {
     pub(super) use crate::syscalls::recv_msg;
     pub(super) use crate::syscalls::send_msg;
     pub(super) use crate::GsoSettings;
+    pub(super) use crate::RecvData;
+    pub(super) use crate::RecvMetrics;
+    pub(super) use crate::StoreCmsgSettings;
     pub(super) use nix::errno::Errno;
     pub(super) use nix::sys::socket::getsockopt;
     pub(super) use nix::sys::socket::recvmsg;

dgram/src/sync.rs

@@ -1,76 +1,47 @@
-use crate::syscalls::RecvData;
-use mio::net::UdpSocket;
+use crate::RecvData;
+use crate::StoreCmsgSettings;
 use std::io::Result;
 use std::net::SocketAddr;
+use std::os::fd::AsFd;
 use std::time::Instant;
 
 #[cfg(target_os = "linux")]
 use super::linux_imports::*;
 
 #[cfg(target_os = "linux")]
 pub fn send_to(
-    socket: &UdpSocket, send_buf: &[u8], gso_settings: Option<GsoSettings>,
+    fd: &impl AsFd, send_buf: &[u8], gso_settings: Option<GsoSettings>,
     tx_time: Option<Instant>, client_addr: &SocketAddr,
 ) -> Result<usize> {
-    loop {
-        // Important to use try_io so events keep coming even if we see
-        // EAGAIN/EWOULDBLOCK
-        let res = socket.try_io(|| {
-            // mio::net::UdpSocket doesn't implement AsFd (yet?).
-            let fd = unsafe {
-                std::os::fd::BorrowedFd::borrow_raw(socket.as_raw_fd())
-            };
+    let sent = send_msg(
+        fd,
+        send_buf,
+        gso_settings,
+        tx_time,
+        &SockaddrStorage::from(*client_addr),
+    );
 
-            let sent = send_msg(
-                fd,
-                send_buf,
-                gso_settings,
-                tx_time,
-                &SockaddrStorage::from(*client_addr),
-            );
-
-            match sent {
-                Err(Errno::EAGAIN) => Err(std::io::Error::last_os_error()),
-                _ => Ok(sent),
-            }
-        });
-
-        if let Ok(sent) = res {
-            return sent.map_err(Into::into);
-        }
+    match sent {
+        Err(Errno::EAGAIN) => Err(std::io::Error::last_os_error()),
+        _ => Ok(sent?),
     }
 }
 
 #[cfg(target_os = "linux")]
-pub async fn recv_from(
-    socket: &UdpSocket, read_buf: &mut [u8], cmsg_space: &mut Vec<u8>,
-    msg_flags: Option<MsgFlags>,
+pub fn recv_from(
+    fd: &impl AsFd, read_buf: &mut [u8], msg_flags: Option<MsgFlags>,
+    store_cmsg_settings: &mut StoreCmsgSettings,
 ) -> Result<RecvData> {
-    loop {
-        // Important to use try_io so events keep coming even if we see
-        // EAGAIN/EWOULDBLOCK
-        let res = socket.try_io(|| {
-            // mio::net::UdpSocket doesn't implement AsFd (yet?).
-            let fd = unsafe {
-                std::os::fd::BorrowedFd::borrow_raw(socket.as_raw_fd())
-            };
-
-            let recvd = recv_msg(
-                fd,
-                read_buf,
-                cmsg_space,
-                msg_flags.unwrap_or(MsgFlags::empty()),
-            );
-
-            match recvd {
-                Err(Errno::EAGAIN) => Err(std::io::Error::last_os_error()),
-                _ => Ok(recvd),
-            }
-        });
+    let recvd = recv_msg(
+        fd,
+        read_buf,
+        msg_flags.unwrap_or(MsgFlags::empty()),
+        store_cmsg_settings,
+    );
 
-        if let Ok(recvd) = res {
-            return recvd.map_err(Into::into);
-        }
+    match recvd {
+        Err(Errno::EAGAIN) => Err(std::io::Error::last_os_error()),
+        _ => Ok(recvd?),
     }
 }