feat(quiche): add keep-alive and a GSO opt-out

rs/web-transport-quiche/Cargo.toml

@@ -33,6 +33,7 @@ tokio = { version = "1", default-features = false, features = [
 
 tokio-quiche = "0.19"
 tracing = "0.1"
+url = "2"
 web-transport-proto = { workspace = true }
 web-transport-trait = { workspace = true }
 
@@ -44,4 +45,3 @@ rustls-pemfile = "2"
 sha2 = "0.10"
 tokio = { version = "1", features = ["full"] }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
-url = "2"

rs/web-transport-quiche/src/client.rs

@@ -104,6 +104,26 @@ impl ClientBuilder {
         Self(self.0.with_server_certificate_hashes(hashes))
     }
 
+    /// Send a PING on this interval, keeping an idle connection alive.
+    ///
+    /// Disabled by default. This must be shorter than the peer's
+    /// [Settings::max_idle_timeout] to have any effect; a third of it is a
+    /// reasonable choice.
+    pub fn with_keep_alive(self, interval: std::time::Duration) -> Self {
+        Self(self.0.with_keep_alive(interval))
+    }
+
+    /// Enable UDP generic segmentation offload (GSO), on by default.
+    ///
+    /// GSO cuts syscall overhead at high throughput by handing the kernel
+    /// several packets at once, but some NICs and virtual network stacks
+    /// mishandle it. Turn it off if large sends are being dropped.
+    ///
+    /// Only Linux supports GSO; elsewhere this does nothing.
+    pub fn with_gso(self, enabled: bool) -> Self {
+        Self(self.0.with_gso(enabled))
+    }
+
     /// Connect to the WebTransport server at the given URL.
     ///
     /// DNS resolution and socket setup happen eagerly. The returned [Connecting]
@@ -116,13 +136,7 @@ impl ClientBuilder {
         request: impl Into<ConnectRequest>,
     ) -> Result<Connecting, ClientError> {
         let request = request.into();
-
-        let port = request.url.port().unwrap_or(443);
-
-        let host = match request.url.host() {
-            Some(host) => host.to_string(),
-            None => return Err(ClientError::InvalidUrl(request.url.to_string())),
-        };
+        let (host, port) = Self::target(&request)?;
 
         let connecting = self.0.connect(&host, port).await?;
 
@@ -131,6 +145,20 @@ impl ClientBuilder {
             request,
         })
     }
+
+    /// The host and port to dial for a request.
+    fn target(request: &ConnectRequest) -> Result<(String, u16), ClientError> {
+        // `Host` renders IPv6 in URL form, bracketed, which is not what a
+        // resolver or a TLS server name wants.
+        let host = match request.url.host() {
+            Some(url::Host::Domain(host)) => host.to_string(),
+            Some(url::Host::Ipv4(ip)) => ip.to_string(),
+            Some(url::Host::Ipv6(ip)) => ip.to_string(),
+            None => return Err(ClientError::InvalidUrl(request.url.to_string())),
+        };
+
+        Ok((host, request.url.port().unwrap_or(443)))
+    }
 }
 
 /// A WebTransport connection that is still completing the handshake.

rs/web-transport-quiche/src/ez/client.rs

@@ -1,9 +1,11 @@
 use std::io;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio_quiche::settings::{CertificateKind, Hooks, TlsCertificatePaths};
 
 use rustls_pki_types::{CertificateDer, PrivateKeyDer};
 
+use crate::ez::socket::capabilities;
 use crate::ez::tls::{ClientHook, ClientVerify};
 use crate::ez::DriverState;
 
@@ -30,6 +32,8 @@ pub struct ClientBuilder {
     tls: Option<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>)>,
     verify: ClientVerify,
     server_name: Option<String>,
+    keep_alive: Option<Duration>,
+    gso: bool,
 }
 
 impl Default for ClientBuilder {
@@ -50,9 +54,33 @@ impl ClientBuilder {
             tls: None,
             verify: ClientVerify::Default,
             server_name: None,
+            keep_alive: None,
+            gso: true,
         }
     }
 
+    /// Send a PING on this interval, keeping an idle connection alive.
+    ///
+    /// Disabled by default. This must be shorter than the peer's
+    /// [Settings::max_idle_timeout] to have any effect; a third of it is a
+    /// reasonable choice.
+    pub fn with_keep_alive(mut self, interval: Duration) -> Self {
+        self.keep_alive = Some(interval);
+        self
+    }
+
+    /// Enable UDP generic segmentation offload (GSO), on by default.
+    ///
+    /// GSO cuts syscall overhead at high throughput by handing the kernel
+    /// several packets at once, but some NICs and virtual network stacks
+    /// mishandle it. Turn it off if large sends are being dropped.
+    ///
+    /// Only Linux supports GSO; elsewhere this does nothing.
+    pub fn with_gso(mut self, enabled: bool) -> Self {
+        self.gso = enabled;
+        self
+    }
+
     /// Listen for incoming packets on the given socket.
     ///
     /// Defaults to an ephemeral port if not specified.
@@ -161,18 +189,16 @@ impl ClientBuilder {
 
         socket.connect(remote).await?;
 
+        // Enable the offloads the kernel supports before the socket is wrapped;
+        // `from_udp` starts with everything disabled.
+        let capabilities = capabilities(&socket, self.gso);
+
         // Connect to the server using the addr we just resolved.
-        #[cfg_attr(not(target_os = "linux"), allow(unused_mut))]
         let mut socket = tokio_quiche::socket::Socket::<
             Arc<tokio::net::UdpSocket>,
             Arc<tokio::net::UdpSocket>,
         >::from_udp(socket)?;
-
-        // Enable UDP GSO/GRO offload where the kernel supports it (Linux only).
-        // This mirrors the server listener and cuts syscall overhead at high
-        // throughput; it's a no-op if send/recv don't share one FD.
-        #[cfg(target_os = "linux")]
-        socket.apply_max_capabilities();
+        socket.capabilities = capabilities;
 
         // Only the fully-insecure path (no verification of any kind) deserves a
         // warning; hash- and root-based verification still authenticate the peer.
@@ -222,6 +248,7 @@ impl ClientBuilder {
             dgram_in.0,
             dgram_out.1,
             dgram_max.clone(),
+            self.keep_alive,
         );
 
         let conn = tokio_quiche::quic::connect_with_config(socket, Some(server_name), &params, app)

rs/web-transport-quiche/src/ez/driver.rs

@@ -7,7 +7,8 @@ use std::{
         atomic::{AtomicUsize, Ordering},
         Arc,
     },
-    task::{Poll, Waker},
+    task::{Context, Poll, Waker},
+    time::Duration,
 };
 use tokio_quiche::{
     buf_factory::BufFactory,
@@ -236,6 +237,42 @@ impl DriverState {
     }
 }
 
+/// Periodically asks quiche to make the next packet ack-eliciting, keeping an
+/// otherwise silent connection out of the peer's idle timeout and holding NAT
+/// bindings open.
+struct KeepAlive {
+    period: Duration,
+    /// Created on the first poll so the timer registers with the runtime that
+    /// actually drives the connection, not whoever built the endpoint.
+    ticker: Option<tokio::time::Interval>,
+}
+
+impl KeepAlive {
+    fn new(period: Duration) -> Self {
+        Self {
+            period,
+            ticker: None,
+        }
+    }
+
+    /// Returns true when a keep-alive is due.
+    fn poll(&mut self, cx: &mut Context) -> bool {
+        let period = self.period;
+        let ticker = self.ticker.get_or_insert_with(|| {
+            // The first tick is one period out; `interval` would instead fire
+            // immediately and ping a connection that just finished handshaking.
+            let start = tokio::time::Instant::now() + period;
+            let mut ticker = tokio::time::interval_at(start, period);
+            // A late tick means the connection was busy, which is exactly when a
+            // keep-alive is unnecessary. Don't replay the backlog.
+            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+            ticker
+        });
+
+        ticker.poll_tick(cx).is_ready()
+    }
+}
+
 pub(super) struct Driver {
     state: Lock<DriverState>,
 
@@ -253,6 +290,8 @@ pub(super) struct Driver {
     // Writable datagram size in bytes, published once at handshake. 0 means the
     // peer didn't negotiate the datagram extension.
     dgram_max: Arc<AtomicUsize>,
+
+    keep_alive: Option<KeepAlive>,
 }
 
 impl Driver {
@@ -263,6 +302,7 @@ impl Driver {
         dgram_in: flume::Sender<Bytes>,
         dgram_out: flume::Receiver<Bytes>,
         dgram_max: Arc<AtomicUsize>,
+        keep_alive: Option<Duration>,
     ) -> Self {
         Self {
             state,
@@ -274,6 +314,7 @@ impl Driver {
             dgram_in,
             dgram_out,
             dgram_max,
+            keep_alive: keep_alive.map(KeepAlive::new),
         }
     }
 
@@ -486,6 +527,16 @@ impl Driver {
             return Poll::Pending;
         }
 
+        // quiche only adds a PING if the next packet wouldn't already be
+        // ack-eliciting, so a tick on a busy connection costs nothing.
+        let mut keep_alive = false;
+        if let Some(k) = self.keep_alive.as_mut() {
+            if k.poll(&mut Context::from_waker(waker)) {
+                qconn.send_ack_eliciting()?;
+                keep_alive = true;
+            }
+        }
+
         // Snapshot stats while we hold an immutable view; stored under the lock below.
         let stats = ConnectionStats::from_quiche(qconn);
 
@@ -549,7 +600,9 @@ impl Driver {
             self.flush_send(qconn, stream_id)?;
         }
 
-        if sleep {
+        // Returning Ready hands control back to the io loop, which flushes the
+        // scheduled PING to the socket.
+        if sleep && !keep_alive {
             Poll::Pending
         } else {
             Poll::Ready(Ok(()))

rs/web-transport-quiche/src/ez/mod.rs

@@ -11,6 +11,7 @@ mod lock;
 mod recv;
 mod send;
 mod server;
+mod socket;
 mod stream;
 pub mod tls;