Unable to test s2n-quic with older turmoil version

[mcches]

Security issue notifications

N/A

Problem:

The automatic dependency update strategy makes it difficult for consumers who also use the same crate(s). This is evidenced by: cf77e2b

Need By Date:

N/A, this is a testing issue, but its a bit of a current pain point.

Solution:

Could s2n-quic-platform instead pin a low version and minimize public type exposure? This crate may not even need to expose turmoil types at all.

turmoil = { version = ">=0.6", ... }

I'd be happy to submit a PR if this approach is acceptable.

[WesleyRosenblum]

I think it shouldn't be hard to remove turmoil types from the s2n-quic-platform API, something like the following should work:

diff --git a/quic/s2n-quic-platform/src/io/turmoil.rs b/quic/s2n-quic-platform/src/io/turmoil.rs
index a3045478..7527bb59 100644
--- a/quic/s2n-quic-platform/src/io/turmoil.rs
+++ b/quic/s2n-quic-platform/src/io/turmoil.rs
@@ -39,7 +39,7 @@ impl Io {
         Builder::default()
     }
 
-    pub fn new<A: turmoil::ToSocketAddrs + Send + Sync + 'static>(addr: A) -> io::Result<Self> {
+    pub fn new(addr: impl Into<std::net::SocketAddr>) -> io::Result<Self> {
         let builder = Builder::default().with_address(addr)?;
         Ok(Self { builder })
     }
@@ -66,7 +66,7 @@ impl Io {
         let socket = if let Some(socket) = socket {
             socket
         } else if let Some(addr) = addr {
-            UdpSocket::bind(&*addr).await?
+            UdpSocket::bind(addr).await?
         } else {
             return Err(io::Error::new(
                 io::ErrorKind::InvalidInput,
diff --git a/quic/s2n-quic-platform/src/io/turmoil/builder.rs b/quic/s2n-quic-platform/src/io/turmoil/builder.rs
index 800772aa..38471337 100644
--- a/quic/s2n-quic-platform/src/io/turmoil/builder.rs
+++ b/quic/s2n-quic-platform/src/io/turmoil/builder.rs
@@ -7,7 +7,7 @@ use super::*;
 pub struct Builder {
     pub(super) handle: Option<Handle>,
     pub(super) socket: Option<UdpSocket>,
-    pub(super) addr: Option<Box<dyn turmoil::ToSocketAddrs + Send + Sync + 'static>>,
+    pub(super) addr: Option<std::net::SocketAddr>,
     pub(super) mtu_config_builder: mtu::Builder,
 }
 
@@ -21,12 +21,9 @@ impl Builder {
     /// Sets the local address for the runtime to listen on.
     ///
     /// NOTE: this method is mutually exclusive with `with_socket`
-    pub fn with_address<A: turmoil::ToSocketAddrs + Send + Sync + 'static>(
-        mut self,
-        addr: A,
-    ) -> io::Result<Self> {
+    pub fn with_address(mut self, addr: impl Into<std::net::SocketAddr>) -> io::Result<Self> {
         debug_assert!(self.socket.is_none(), "socket has already been set");
-        self.addr = Some(Box::new(addr));
+        self.addr = Some(addr.into());
         Ok(self)
     }

[mcches]

Nice, turmoil::ToSocketAddrs is also stable so I'm not too worried about it, but the less the better.

[mcches]

I did a bit of experimentation with different strategies here and I don't think >=0.6 is the right way to go. The resolver will still choose the latest available version for this crate, which can differ if the consumer has a different version of turmoil pinned. That only works if the consumer also has >=0.6 or the latest minor version pinned. Maybe thats acceptable given the "unstable" nature of this feature.

Is there a better way where we just use a "bring your own Io" path and avoid the turmoil dependency altogether?

[maddeleine]

@mcches Would @WesleyRosenblum's solution work for your usecase?

[mcches]

It's an improvement, but I don't think it solves the problem as we still need a turmoil dependency in this crate. I think being able to provide an Io/Socket impl is the best path forward here.

[mcches]

I was able to implement my own io externally following the same technique that exists in the repo. All the pieces were public. It's a bit involved, so perhaps there is a way to simplify things without taking on the turmoil dependency.

//! Custom s2n-quic IO provider using turmoil 0.6 UDP sockets.
//!
//! This crate provides an [`Io`] type that implements the `s2n_quic::provider::io::Provider`
//! trait using turmoil's simulated [`UdpSocket`](turmoil::net::UdpSocket). This allows
//! s2n-quic Server and Client instances to operate within a turmoil simulation.

mod clock;

use clock::Clock;
use core::future::Future;
use futures::future::poll_fn;
use s2n_quic_core::{
    endpoint::Endpoint,
    inet::{self, SocketAddress},
    io::event_loop::{select::Select, EventLoop},
    path::{self, mtu},
};
use s2n_quic_platform::{
    features::Gso,
    message::{simple::Message, Message as _},
    socket::{
        io::{rx, tx},
        ring::{self, Consumer, Producer},
        stats,
    },
};
use std::io::{self, ErrorKind};
use tokio::runtime::Handle;
use turmoil::net::UdpSocket;

pub type PathHandle = path::Tuple;

/// Builder for configuring the turmoil IO provider.
pub struct Builder {
    handle: Option<Handle>,
    socket: Option<UdpSocket>,
    addr: Option<Box<dyn turmoil::ToSocketAddrs + Send + Sync + 'static>>,
    mtu_config_builder: mtu::Builder,
}

impl Default for Builder {
    fn default() -> Self {
        Self {
            handle: None,
            socket: None,
            addr: None,
            mtu_config_builder: mtu::Builder::default(),
        }
    }
}

impl Builder {
    /// Sets the tokio runtime handle.
    #[must_use]
    pub fn with_handle(mut self, handle: Handle) -> Self {
        self.handle = Some(handle);
        self
    }

    /// Sets the local address for the runtime to listen on.
    ///
    /// NOTE: this method is mutually exclusive with `with_socket`
    pub fn with_address<A: turmoil::ToSocketAddrs + Send + Sync + 'static>(
        mut self,
        addr: A,
    ) -> io::Result<Self> {
        debug_assert!(self.socket.is_none(), "socket has already been set");
        self.addr = Some(Box::new(addr));
        Ok(self)
    }

    /// Sets the socket used for sending and receiving.
    ///
    /// NOTE: this method is mutually exclusive with `with_address`
    pub fn with_socket(mut self, socket: UdpSocket) -> io::Result<Self> {
        debug_assert!(self.addr.is_none(), "address has already been set");
        self.socket = Some(socket);
        Ok(self)
    }

    /// Sets the largest maximum transmission unit (MTU) that can be sent on a path.
    pub fn with_max_mtu(mut self, max_mtu: u16) -> io::Result<Self> {
        self.mtu_config_builder = self
            .mtu_config_builder
            .with_max_mtu(max_mtu)
            .map_err(|err| io::Error::new(ErrorKind::InvalidInput, format!("{err}")))?;
        Ok(self)
    }

    /// Builds the IO provider.
    pub fn build(self) -> io::Result<Io> {
        Ok(Io { builder: self })
    }
}

/// A turmoil-based IO provider for s2n-quic.
///
/// This type implements `s2n_quic::provider::io::Provider` so it can be passed
/// directly to `Server::builder().with_io(io)` or `Client::builder().with_io(io)`.
pub struct Io {
    builder: Builder,
}

impl Io {
    /// Creates a new IO provider that will bind to the given address.
    pub fn new<A: turmoil::ToSocketAddrs + Send + Sync + 'static>(addr: A) -> io::Result<Self> {
        let builder = Builder::default().with_address(addr)?;
        Ok(Self { builder })
    }

    /// Returns a new [`Builder`] for configuring the IO provider.
    pub fn builder() -> Builder {
        Builder::default()
    }

    async fn setup<E: Endpoint<PathHandle = PathHandle>>(
        self,
        mut endpoint: E,
    ) -> io::Result<(impl Future<Output = ()>, SocketAddress)> {
        let Builder {
            handle: _,
            socket,
            addr,
            mtu_config_builder,
        } = self.builder;

        let mtu_config = mtu_config_builder
            .build()
            .map_err(|err| io::Error::new(ErrorKind::InvalidInput, format!("{err}")))?;

        endpoint.set_mtu_config(mtu_config);

        let clock = Clock::default();

        let socket = if let Some(socket) = socket {
            socket
        } else if let Some(addr) = addr {
            UdpSocket::bind(&*addr).await?
        } else {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "missing bind address",
            ));
        };

        let local_addr = socket.local_addr()?;
        let local_addr: inet::SocketAddress = local_addr.into();
        let payload_len: usize = mtu_config.max_mtu().into();
        let payload_len = payload_len as u32;

        // Number of entries in each ring buffer. Matches the value used in
        // s2n-quic-platform's turmoil module.
        let entries = 1024;

        let (rx, rx_producer) = {
            let mut consumers = vec![];
            let (producer, consumer) = ring::pair(entries, payload_len);
            consumers.push(consumer);
            let rx = rx::Rx::new(consumers, mtu_config.max_mtu(), local_addr.into());
            (rx, producer)
        };

        let (tx, tx_consumer) = {
            let mut producers = vec![];
            let (producer, consumer) = ring::pair(entries, payload_len);
            producers.push(producer);

            let gso = Gso::default();
            // GSO is not supported by turmoil so disable it
            gso.disable();

            let tx = tx::Tx::new(producers, gso, mtu_config.max_mtu());
            (tx, consumer)
        };

        let (stats_sender, stats_recv) = stats::channel();

        // Spawn a task that does the actual socket calls and coordinates with the
        // event loop through the ring buffers
        tokio::spawn(run_io(socket, rx_producer, tx_consumer, stats_sender));

        let event_loop = EventLoop {
            clock,
            rx,
            tx,
            endpoint,
            cooldown: Default::default(),
            stats: stats_recv,
        }
        .start(local_addr);

        Ok((event_loop, local_addr))
    }

    /// Starts the IO provider with the given endpoint.
    ///
    /// Returns a join handle for the spawned task and the local address.
    pub fn start<E: Endpoint<PathHandle = PathHandle>>(
        mut self,
        endpoint: E,
    ) -> io::Result<(tokio::task::JoinHandle<()>, SocketAddress)> {
        let handle = if let Some(handle) = self.builder.handle.take() {
            handle
        } else {
            Handle::try_current().map_err(io::Error::other)?
        };

        let guard = handle.enter();

        let task = handle.spawn(async move {
            let (instance, _local_addr) = self.setup(endpoint).await.unwrap();
            instance.await;
        });

        drop(guard);

        // TODO: this is a potentially async operation - can we get this here?
        let local_addr = Default::default();

        Ok((task, local_addr))
    }
}

impl s2n_quic::provider::io::Provider for Io {
    type PathHandle = PathHandle;
    type Error = io::Error;

    fn start<E: Endpoint<PathHandle = Self::PathHandle>>(
        self,
        endpoint: E,
    ) -> Result<SocketAddress, Self::Error> {
        let (_join_handle, local_addr) = Io::start(self, endpoint)?;
        Ok(local_addr)
    }
}

/// Turmoil doesn't allow splitting sockets for Tx and Rx so we spawn a single
/// task to handle both directions via ring buffers.
async fn run_io(
    socket: UdpSocket,
    mut producer: Producer<Message>,
    mut consumer: Consumer<Message>,
    stats: stats::Sender,
) -> io::Result<()> {
    let mut poll_producer = false;

    loop {
        let socket_ready = socket.readable();
        let consumer_ready = poll_fn(|cx| consumer.poll_acquire(u32::MAX, cx));
        let producer_ready = async {
            // Only poll the producer if we need more capacity — otherwise we would
            // constantly wake up
            if poll_producer {
                poll_fn(|cx| producer.poll_acquire(u32::MAX, cx)).await
            } else {
                core::future::pending().await
            }
        };
        // The socket task doesn't have any application wakeups to handle
        let application_wakeup = core::future::pending();

        // We replace the timer future with `socket_ready` since we don't have a
        // timer here.
        let is_readable = Select::new(
            consumer_ready,
            producer_ready,
            application_wakeup,
            socket_ready,
        )
        .await
        .unwrap()
        .timeout_expired;

        if is_readable {
            let mut count = 0;
            for entry in producer.data() {
                let res = socket.try_recv_from(entry.payload_mut());
                stats.recv().on_operation_result(&res, |_len| 1);
                if let Ok((len, addr)) = res {
                    count += 1;
                    entry.set_remote_address(&(addr.into()));
                    unsafe {
                        entry.set_payload_len(len);
                    }
                } else {
                    break;
                }
            }

            producer.release(count);
            poll_producer = producer.data().is_empty();
        }

        {
            let mut count = 0;
            for entry in consumer.data() {
                let addr = *entry.remote_address();
                let addr: std::net::SocketAddr = addr.into();
                let payload = entry.payload_mut();
                let res = socket.try_send_to(payload, addr);
                stats.recv().on_operation_result(&res, |_len| 1);
                if res.is_ok() {
                    count += 1;
                } else {
                    break;
                }
            }

            consumer.release(count);
        }

        // check to see if the rings are open, otherwise shut down
        if !(producer.is_open() && consumer.is_open()) {
            return Ok(());
        }
    }
}

[WesleyRosenblum]

Would it work if we re-exported turmoil so your code could depend on the same version that s2n-quic-platform does? Or do you actually need to test on older versions of turmoil?

[mcches]

I don't think that is ideal either. Best case is s2n has no turmoil dep and its simple to inject your own Io or socket for turmoil usage.

[boquan-fang]

Hello @mcches, is the existing s2n-quic IO provider API sufficient for you to plug in the turmoil code you mentioned above? I am trying to understand if there are anything else you need from s2n-quic.