Add dgram crate

.github/workflows/stable.yml

@@ -104,7 +104,7 @@ jobs:
     # duplicate builds for PRs created from internal branches.
     if: github.event_name == 'push' || github.event.pull_request.head.repo.full_name != github.repository
     env:
-      IPHONEOS_DEPLOYMENT_TARGET: "10.0"
+      IPHONEOS_DEPLOYMENT_TARGET: "17.5"
     steps:
       - name: Checkout sources
         uses: actions/checkout@v4

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = [ "apps", "h3i", "octets", "qlog", "quiche" ]
+members = [ "apps", "dgram", "h3i", "octets", "qlog", "quiche" ]
 exclude = [ "fuzz", "tools/http3_test" ]
 
 [profile.bench]

Dockerfile

@@ -6,6 +6,7 @@ COPY apps/ ./apps/
 COPY octets/ ./octets/
 COPY qlog/ ./qlog/
 COPY quiche/ ./quiche/
+COPY dgram ./dgram/
 
 RUN apt-get update && apt-get install -y cmake && rm -rf /var/lib/apt/lists/*
 

apps/Cargo.toml

@@ -21,6 +21,7 @@ sfv = ["quiche/sfv"]
 default = ["qlog", "sfv"]
 
 [dependencies]
+dgram = { path = "../dgram" }
 docopt = "1"
 env_logger = "0.10"
 mio = { version = "0.8", features = ["net", "os-poll"] }

apps/src/args.rs

@@ -461,6 +461,7 @@ Options:
   --max-field-section-size BYTES    Max size of uncompressed HTTP/3 field section. Default is unlimited.
   --qpack-max-table-capacity BYTES  Max capacity of QPACK dynamic table decoding. Any value other that 0 is currently unsupported.
   --qpack-blocked-streams STREAMS   Limit of streams that can be blocked while decoding. Any value other that 0 is currently unsupported.
+  --disable-gro               Disable GRO (linux only).
   --disable-gso               Disable GSO (linux only).
   --disable-pacing            Disable pacing (linux only).
   --initial-cwnd-packets PACKETS      The initial congestion window size in terms of packet count [default: 10].
@@ -476,6 +477,7 @@ pub struct ServerArgs {
     pub cert: String,
     pub key: String,
     pub disable_gso: bool,
+    pub disable_gro: bool,
     pub disable_pacing: bool,
     pub enable_pmtud: bool,
 }
@@ -490,6 +492,7 @@ impl Args for ServerArgs {
         let index = args.get_str("--index").to_string();
         let cert = args.get_str("--cert").to_string();
         let key = args.get_str("--key").to_string();
+        let disable_gro = args.get_bool("--disable-gro");
         let disable_gso = args.get_bool("--disable-gso");
         let disable_pacing = args.get_bool("--disable-pacing");
         let enable_pmtud = args.get_bool("--enable-pmtud");
@@ -502,6 +505,7 @@ impl Args for ServerArgs {
             cert,
             key,
             disable_gso,
+            disable_gro,
             disable_pacing,
             enable_pmtud,
         }

apps/src/bin/quiche-server.rs

@@ -47,6 +47,7 @@ use quiche_apps::args::*;
 
 use quiche_apps::common::*;
 
+use quiche_apps::recvfrom::*;
 use quiche_apps::sendto::*;
 
 const MAX_BUF_SIZE: usize = 65507;
@@ -91,13 +92,20 @@ fn main() {
         .register(&mut socket, mio::Token(0), mio::Interest::READABLE)
         .unwrap();
 
+    let enable_gro = if args.disable_gro {
+        false
+    } else {
+        detect_gro(&socket)
+    };
+
     let max_datagram_size = MAX_DATAGRAM_SIZE;
     let enable_gso = if args.disable_gso {
         false
     } else {
         detect_gso(&socket, max_datagram_size)
     };
 
+    trace!("GRO detected: {}", enable_gro);
     trace!("GSO detected: {}", enable_gso);
 
     // Create the configuration for the QUIC connections.
@@ -211,7 +219,7 @@ fn main() {
                 break 'read;
             }
 
-            let (len, from) = match socket.recv_from(&mut buf) {
+            let recv_data = match recv_from(&socket, &mut buf) {
                 Ok(v) => v,
 
                 Err(e) => {
@@ -226,6 +234,8 @@ fn main() {
                 },
             };
 
+            let from = recv_data.peer_addr.expect("Invalid peer IP address");
+            let len = recv_data.bytes;
             trace!("got {} bytes", len);
 
             let pkt_buf = &mut buf[..len];

apps/src/lib.rs

@@ -30,4 +30,5 @@ extern crate log;
 pub mod args;
 pub mod client;
 pub mod common;
+pub mod recvfrom;
 pub mod sendto;

apps/src/recvfrom.rs

@@ -0,0 +1,63 @@
+use dgram::RecvData;
+
+#[cfg(target_os = "linux")]
+use std::io;
+
+/// For Linux, try to detect if GRO is available. If it is, the
+/// [`UdpGroSegment`] socket option will be set on the passed socket.
+///
+/// [`UdpGroSegment`]: https://docs.rs/nix/latest/nix/sys/socket/sockopt/struct.UdpGroSegment.html
+#[cfg(target_os = "linux")]
+pub fn detect_gro(socket: &mio::net::UdpSocket) -> bool {
+    use nix::sys::socket::setsockopt;
+    use nix::sys::socket::sockopt::UdpGroSegment;
+    use std::os::unix::io::AsRawFd;
+
+    // mio::net::UdpSocket doesn't implement AsFd (yet?).
+    let fd = unsafe { std::os::fd::BorrowedFd::borrow_raw(socket.as_raw_fd()) };
+
+    match setsockopt(&fd, UdpGroSegment, &true) {
+        Ok(_) => {
+            debug!("Successfully set UDP_GRO socket option");
+            true
+        },
+        Err(e) => {
+            debug!("Setting UDP_GRO failed: {:?}", e);
+            false
+        },
+    }
+}
+
+#[cfg(not(target_os = "linux"))]
+pub fn detect_gro(_socket: &mio::net::UdpSocket) -> bool {
+    false
+}
+
+#[cfg(target_os = "linux")]
+pub fn recv_from(
+    socket: &mio::net::UdpSocket, buf: &mut [u8],
+) -> io::Result<RecvData> {
+    use dgram::RecvMsgSettings;
+    use std::os::unix::io::AsRawFd;
+
+    let mut recvmsg_cmsg_settings = RecvMsgSettings {
+        store_cmsgs: false,
+        cmsg_space: &mut vec![],
+    };
+    socket.try_io(|| {
+        let fd =
+            unsafe { std::os::fd::BorrowedFd::borrow_raw(socket.as_raw_fd()) };
+
+        dgram::sync::recv_from(&fd, buf, &mut recvmsg_cmsg_settings)
+    })
+}
+
+#[cfg(not(target_os = "linux"))]
+pub fn recv_from(
+    socket: &mio::net::UdpSocket, buf: &mut [u8],
+) -> std::io::Result<RecvData> {
+    match socket.recv_from(buf) {
+        Ok((read, from)) => Ok(RecvData::new(Some(from), read, 0)),
+        Err(e) => Err(e),
+    }
+}

apps/src/sendto.rs

@@ -38,7 +38,16 @@ pub fn detect_gso(socket: &mio::net::UdpSocket, segment_size: usize) -> bool {
     // mio::net::UdpSocket doesn't implement AsFd (yet?).
     let fd = unsafe { std::os::fd::BorrowedFd::borrow_raw(socket.as_raw_fd()) };
 
-    setsockopt(&fd, UdpGsoSegment, &(segment_size as i32)).is_ok()
+    match setsockopt(&fd, UdpGsoSegment, &(segment_size as i32)) {
+        Ok(_) => {
+            debug!("Successfully set UDP_SEGMENT socket option");
+            true
+        },
+        Err(e) => {
+            debug!("Setting UDP_SEGMENT failed: {:?}", e);
+            false
+        },
+    }
 }
 
 /// For non-Linux, there is no GSO support.
@@ -53,35 +62,25 @@ fn send_to_gso_pacing(
     socket: &mio::net::UdpSocket, buf: &[u8], send_info: &quiche::SendInfo,
     segment_size: usize,
 ) -> io::Result<usize> {
-    use nix::sys::socket::sendmsg;
-    use nix::sys::socket::ControlMessage;
-    use nix::sys::socket::MsgFlags;
-    use nix::sys::socket::SockaddrStorage;
-    use std::io::IoSlice;
+    use dgram::SendMsgSettings;
     use std::os::unix::io::AsRawFd;
 
-    let iov = [IoSlice::new(buf)];
-    let segment_size = segment_size as u16;
-    let dst = SockaddrStorage::from(send_info.to);
-    let sockfd = socket.as_raw_fd();
-
-    // GSO option.
-    let cmsg_gso = ControlMessage::UdpGsoSegments(&segment_size);
-
-    // Pacing option.
-    let send_time = std_time_to_u64(&send_info.at);
-    let cmsg_txtime = ControlMessage::TxTime(&send_time);
-
-    match sendmsg(
-        sockfd,
-        &iov,
-        &[cmsg_gso, cmsg_txtime],
-        MsgFlags::empty(),
-        Some(&dst),
-    ) {
-        Ok(v) => Ok(v),
-        Err(e) => Err(e.into()),
-    }
+    let sendmsg_settings = SendMsgSettings {
+        segment_size: Some(segment_size as u16),
+        tx_time: Some(send_info.at),
+        dst: Some(send_info.to),
+        ..Default::default()
+    };
+
+    // Important to use try_io so events keep coming even if we see
+    // EAGAIN/EWOULDBLOCK
+    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(&fd, buf, sendmsg_settings)
+    })
 }
 
 /// For non-Linux platforms.
@@ -90,7 +89,7 @@ fn send_to_gso_pacing(
     _socket: &mio::net::UdpSocket, _buf: &[u8], _send_info: &quiche::SendInfo,
     _segment_size: usize,
 ) -> io::Result<usize> {
-    panic!("send_to_gso() should not be called on non-linux platforms");
+    panic!("send_to_gso_pacing() should not be called on non-linux platforms");
 }
 
 /// A wrapper function of send_to().
@@ -132,18 +131,3 @@ pub fn send_to(
 
     Ok(written)
 }
-
-#[cfg(target_os = "linux")]
-fn std_time_to_u64(time: &std::time::Instant) -> u64 {
-    const NANOS_PER_SEC: u64 = 1_000_000_000;
-
-    const INSTANT_ZERO: std::time::Instant =
-        unsafe { std::mem::transmute(std::time::UNIX_EPOCH) };
-
-    let raw_time = time.duration_since(INSTANT_ZERO);
-
-    let sec = raw_time.as_secs();
-    let nsec = raw_time.subsec_nanos();
-
-    sec * NANOS_PER_SEC + nsec as u64
-}

dgram/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "dgram"
+version = "0.1.0"
+edition = "2021"
+
+ [features]
+ async = ["dep:tokio"]
+
+[dependencies]
+libc = "0.2.76"
+nix = { version = "0.27", features = ["net", "socket", "uio"] }
+smallvec = { version = "1.10", features = ["union"] }
+tokio = { version = "1.29", features = ["full", "test-util"], optional = true }