Move 'read loop to recv_from in quiche-server

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

@@ -31,6 +31,8 @@ use std::io;
 
 use std::net;
 
+use std::os::fd::AsRawFd;
+
 use std::io::prelude::*;
 
 use std::collections::HashMap;
@@ -47,6 +49,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 +94,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 +221,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 +236,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,57 @@
+use dgram::RecvData;
+
+use std::io;
+
+// TODO: migrate setup code to use [dgram::SocketCapabilities].
+
+/// 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()) };
+
+    setsockopt(&fd, UdpGroSegment, &true).is_ok()
+}
+
+#[cfg(not(target_os = "linux"))]
+pub fn detect_gro(socket: &mio::net::UdpSocket, _segment_size: usize) -> bool {
+    false
+}
+
+#[cfg(target_os = "linux")]
+pub fn recv_from(
+    socket: &mio::net::UdpSocket, buf: &mut [u8],
+) -> io::Result<RecvData> {
+    use dgram::RecvMsgCmsgSettings;
+    use std::os::unix::io::AsRawFd;
+
+    let mut recvmsg_cmsg_settings = RecvMsgCmsgSettings {
+        store_cmsgs: false,
+        cmsg_space: vec![],
+    };
+
+    socket.try_io(|| {
+        let fd =
+            unsafe { std::os::fd::BorrowedFd::borrow_raw(socket.as_raw_fd()) };
+
+        dgram::sync::recv_from(&fd, buf, None, &mut recvmsg_cmsg_settings)
+    })
+}
+
+// TODO: shoould it be up to the user to handle blocking errors?
+#[cfg(not(target_os = "linux"))]
+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(_) => {},
+    }
+}

apps/src/sendto.rs

@@ -63,23 +63,17 @@ fn send_to_gso_pacing(
         ..Default::default()
     };
 
-    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(&fd, buf, sendmsg_settings)
-        });
-
-        // TODO: make sure error cases are the same
-        if res.is_ok() {
-            return res;
-        }
-    }
+    // 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()) };
+
+        // TODO: make sure this actually errors properly, pretty sure there
+        // was some weirdness with the `Blocked` error
+        dgram::sync::send_to(&fd, buf, sendmsg_settings)
+    })
 }
 
 /// For non-Linux platforms.
@@ -88,7 +82,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().

dgram/src/lib.rs

@@ -23,10 +23,10 @@ pub struct SendMsgCmsgSettings {
 
 /// Settings for handling control messages when receiving data.
 #[cfg(target_os = "linux")]
-#[derive(Default)]
+#[derive(Clone, Default)]
 pub struct RecvMsgCmsgSettings {
-    store_cmsgs: bool,
-    cmsg_space: Vec<u8>,
+    pub store_cmsgs: bool,
+    pub cmsg_space: Vec<u8>,
 }
 
 /// Output of a `recvmsg` call.

dgram/src/sync.rs

@@ -14,8 +14,10 @@ pub fn send_to(
     let sent = send_msg(fd, send_buf, sendmsg_settings);
 
     match sent {
+        Ok(s) => Ok(s),
+        // TODO: propagate or transform?
         Err(Errno::EAGAIN) => Err(std::io::Error::last_os_error()),
-        _ => Ok(sent?),
+        Err(e) => Err(e.into()),
     }
 }
 
@@ -32,11 +34,13 @@ pub fn recv_from(
     );
 
     match recvd {
+        Ok(r) => Ok(r),
         Err(Errno::EAGAIN) => Err(std::io::Error::last_os_error()),
-        _ => Ok(recvd?),
+        Err(e) => Err(e.into()),
     }
 }
 
+// TODO: these async functions shouldn't be here
 #[cfg(not(target_os = "linux"))]
 pub async fn send_to(
     socket: &UdpSocket, client_addr: SocketAddr, send_buf: &[u8],