feat(s2n-quic-qns): add xdp client

Author: camshaft

quic/s2n-quic-platform/src/io/tokio.rs

@@ -15,7 +15,7 @@ use tokio::runtime::Handle;
 
 mod builder;
 mod clock;
-mod task;
+pub(crate) mod task;
 #[cfg(test)]
 mod tests;
 

quic/s2n-quic-platform/src/io/xdp.rs

@@ -7,7 +7,7 @@ use s2n_quic_core::{
 };
 
 pub use s2n_quic_core::{
-    io::{rx, tx},
+    io::rx,
     sync::{spsc, worker},
     xdp::path::Tuple as PathHandle,
 };
@@ -38,6 +38,54 @@ pub mod socket {
     }
 }
 
+pub mod tx {
+    pub use s2n_quic_core::io::tx::*;
+
+    pub fn channel(
+        socket: ::std::net::UdpSocket,
+    ) -> (
+        impl Tx<PathHandle = crate::message::default::Handle>,
+        impl core::future::Future<Output = ::std::io::Result<()>>,
+    ) {
+        // Initial packets don't need to be bigger than the minimum
+        let max_mtu = s2n_quic_core::path::MaxMtu::MIN;
+
+        // It's unlikely the initial packets will utilize, so just disable it
+        let gso = crate::features::Gso::default();
+        gso.disable();
+
+        // compute the payload size for each message from the number of GSO segments we can
+        // fill
+        let payload_len = {
+            let max_mtu: u16 = max_mtu.into();
+            (max_mtu as u32 * gso.max_segments() as u32).min(u16::MAX as u32)
+        };
+
+        // 512Kb
+        let tx_buffer_size: u32 = 1 << 19;
+        let entries = tx_buffer_size / payload_len;
+        let entries = if entries.is_power_of_two() {
+            entries
+        } else {
+            // round up to the nearest power of two, since the ring buffers require it
+            entries.next_power_of_two()
+        };
+
+        let mut producers = vec![];
+
+        let (producer, consumer) = crate::socket::ring::pair(entries, payload_len);
+        producers.push(producer);
+
+        // spawn a task that actually flushes the ring buffer to the socket
+        let task = crate::io::tokio::task::tx(socket, consumer, gso.clone());
+
+        // construct the TX side for the endpoint event loop
+        let io = crate::socket::io::tx::Tx::new(producers, gso, max_mtu);
+
+        (io, task)
+    }
+}
+
 impl From<PathHandle> for crate::message::msg::Handle {
     #[inline]
     fn from(handle: PathHandle) -> Self {

quic/s2n-quic-qns/src/client/interop.rs

@@ -9,11 +9,7 @@ use crate::{
     Result,
 };
 use core::time::Duration;
-use s2n_quic::{
-    client::Connect,
-    provider::{event, io},
-    Client,
-};
+use s2n_quic::{client::Connect, provider::event, Client};
 use std::{
     collections::{hash_map::Entry, HashMap},
     path::PathBuf,
@@ -40,15 +36,9 @@ pub struct Interop {
     #[structopt(long)]
     download_dir: Option<PathBuf>,
 
-    #[structopt(long)]
-    disable_gso: bool,
-
     #[structopt(long, parse(try_from_str = parse_duration))]
     keep_alive: Option<Duration>,
 
-    #[structopt(short, long, default_value = "::")]
-    local_ip: std::net::IpAddr,
-
     #[structopt(long, env = "TESTCASE", possible_values = &Testcase::supported(is_supported_testcase))]
     testcase: Option<Testcase>,
 
@@ -57,6 +47,9 @@ pub struct Interop {
 
     #[structopt(long, default_value)]
     tls: TlsProviders,
+
+    #[structopt(flatten)]
+    io: crate::io::Client,
 }
 
 impl Interop {
@@ -130,14 +123,7 @@ impl Interop {
     }
 
     fn client(&self) -> Result<Client> {
-        let mut io_builder =
-            io::Default::builder().with_receive_address((self.local_ip, 0u16).into())?;
-
-        if self.disable_gso {
-            io_builder = io_builder.with_gso_disabled()?;
-        }
-
-        let io = io_builder.build()?;
+        let io = self.io.build()?;
 
         let client = Client::builder()
             .with_io(io)?

quic/s2n-quic-qns/src/client/perf.rs

@@ -3,7 +3,7 @@
 
 use crate::{perf, tls, Result};
 use futures::future::try_join_all;
-use s2n_quic::{client, provider::io, Client, Connection};
+use s2n_quic::{client, Client, Connection};
 use std::path::PathBuf;
 use structopt::StructOpt;
 
@@ -29,15 +29,6 @@ pub struct Perf {
     #[structopt(long)]
     connections: Option<usize>,
 
-    #[structopt(long)]
-    disable_gso: bool,
-
-    #[structopt(long, default_value = "9000")]
-    max_mtu: u16,
-
-    #[structopt(short, long, default_value = "::")]
-    local_ip: std::net::IpAddr,
-
     #[structopt(long, default_value)]
     send: u64,
 
@@ -50,6 +41,9 @@ pub struct Perf {
     /// Logs statistics for the endpoint
     #[structopt(long)]
     stats: bool,
+
+    #[structopt(flatten)]
+    io: crate::io::Client,
 }
 
 impl Perf {
@@ -106,15 +100,7 @@ impl Perf {
     }
 
     fn client(&self) -> Result<Client> {
-        let mut io_builder = io::Default::builder()
-            .with_receive_address((self.local_ip, 0u16).into())?
-            .with_max_mtu(self.max_mtu)?;
-
-        if self.disable_gso {
-            io_builder = io_builder.with_gso_disabled()?;
-        }
-
-        let io = io_builder.build()?;
+        let io = self.io.build()?;
 
         let tls = s2n_quic::provider::tls::default::Client::builder()
             .with_certificate(tls::default::ca(self.ca.as_ref())?)?

quic/s2n-quic-qns/src/io.rs

@@ -0,0 +1,93 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::Result;
+use s2n_quic::provider::io;
+use structopt::StructOpt;
+
+#[derive(Debug, StructOpt)]
+pub struct Server {
+    #[structopt(short, long, default_value = "::")]
+    pub ip: std::net::IpAddr,
+
+    #[structopt(short, long, default_value = "443")]
+    pub port: u16,
+
+    #[structopt(long)]
+    pub disable_gso: bool,
+
+    #[structopt(long, default_value = "9000")]
+    pub max_mtu: u16,
+
+    #[cfg(feature = "xdp")]
+    #[structopt(flatten)]
+    xdp: crate::xdp::Xdp,
+}
+
+impl Server {
+    #[cfg(feature = "xdp")]
+    pub fn build(&self) -> Result<impl io::Provider> {
+        // GSO isn't currently supported for XDP so just read it to avoid a dead_code warning
+        let _ = self.disable_gso;
+        let _ = self.max_mtu;
+
+        let addr = (self.ip, self.port).into();
+
+        self.xdp.server(addr)
+    }
+
+    #[cfg(not(feature = "xdp"))]
+    pub fn build(&self) -> Result<impl io::Provider> {
+        let mut io_builder = io::Default::builder()
+            .with_receive_address((self.ip, self.port).into())?
+            .with_max_mtu(self.max_mtu)?;
+
+        if self.disable_gso {
+            io_builder = io_builder.with_gso_disabled()?;
+        }
+
+        Ok(io_builder.build()?)
+    }
+}
+
+#[derive(Debug, StructOpt)]
+pub struct Client {
+    #[structopt(long)]
+    pub disable_gso: bool,
+
+    #[structopt(long, default_value = "9000")]
+    pub max_mtu: u16,
+
+    #[structopt(short, long, default_value = "::")]
+    pub local_ip: std::net::IpAddr,
+
+    #[cfg(feature = "xdp")]
+    #[structopt(flatten)]
+    xdp: crate::xdp::Xdp,
+}
+
+impl Client {
+    #[cfg(feature = "xdp")]
+    pub fn build(&self) -> Result<impl io::Provider> {
+        // GSO isn't currently supported for XDP so just read it to avoid a dead_code warning
+        let _ = self.disable_gso;
+        let _ = self.max_mtu;
+
+        let addr = (self.local_ip, 0u16).into();
+
+        self.xdp.client(addr)
+    }
+
+    #[cfg(not(feature = "xdp"))]
+    pub fn build(&self) -> Result<impl io::Provider> {
+        let mut io_builder = io::Default::builder()
+            .with_receive_address((self.local_ip, 0u16).into())?
+            .with_max_mtu(self.max_mtu)?;
+
+        if self.disable_gso {
+            io_builder = io_builder.with_gso_disabled()?;
+        }
+
+        Ok(io_builder.build()?)
+    }
+}

quic/s2n-quic-qns/src/main.rs

@@ -9,6 +9,7 @@ pub type Result<T, E = Error> = core::result::Result<T, E>;
 mod client;
 mod file;
 mod interop;
+mod io;
 mod perf;
 mod server;
 mod tls;

quic/s2n-quic-qns/src/server/interop.rs

@@ -14,7 +14,6 @@ use s2n_quic::{
     provider::{
         endpoint_limits,
         event::{events, Subscriber},
-        io,
     },
     Server,
 };
@@ -27,12 +26,6 @@ use tokio::spawn;
 
 #[derive(Debug, StructOpt)]
 pub struct Interop {
-    #[structopt(short, long, default_value = "::")]
-    ip: std::net::IpAddr,
-
-    #[structopt(short, long, default_value = "443")]
-    port: u16,
-
     #[structopt(long)]
     certificate: Option<PathBuf>,
 
@@ -45,18 +38,14 @@ pub struct Interop {
     #[structopt(long, default_value = ".")]
     www_dir: PathBuf,
 
-    #[structopt(long)]
-    disable_gso: bool,
-
     #[structopt(long, env = "TESTCASE", possible_values = &Testcase::supported(is_supported_testcase))]
     testcase: Option<Testcase>,
 
     #[structopt(long, default_value)]
     tls: TlsProviders,
 
-    #[cfg(feature = "xdp")]
     #[structopt(flatten)]
-    xdp: crate::xdp::Xdp,
+    io: crate::io::Server,
 }
 
 impl Interop {
@@ -89,28 +78,6 @@ impl Interop {
         Err(crate::CRASH_ERROR_MESSAGE.into())
     }
 
-    #[cfg(feature = "xdp")]
-    fn io(&self) -> Result<impl io::Provider> {
-        // GSO isn't currently supported for XDP so just read it to avoid a dead_code warning
-        let _ = self.disable_gso;
-
-        let addr = (self.ip, self.port).into();
-
-        self.xdp.server(addr)
-    }
-
-    #[cfg(not(feature = "xdp"))]
-    fn io(&self) -> Result<impl io::Provider> {
-        let mut io_builder =
-            io::Default::builder().with_receive_address((self.ip, self.port).into())?;
-
-        if self.disable_gso {
-            io_builder = io_builder.with_gso_disabled()?;
-        }
-
-        Ok(io_builder.build()?)
-    }
-
     fn server(&self) -> Result<Server> {
         let mut max_handshakes = 100;
         if let Some(Testcase::Retry) = self.testcase {
@@ -121,7 +88,7 @@ impl Interop {
             .with_inflight_handshake_limit(max_handshakes)?
             .build()?;
 
-        let io = self.io()?;
+        let io = self.io.build()?;
 
         let server = Server::builder()
             .with_io(io)?
@@ -157,7 +124,7 @@ impl Interop {
             }
         };
 
-        eprintln!("Server listening on port {}", self.port);
+        eprintln!("Server listening on port {}", self.io.port);
 
         Ok(server)
     }