refactor(quic): rewrite quic using quinn

Rewrite quic using quinn instead of quinn-proto. libp2p-quic::endpoint::Driver is eliminated (and that hard quinn-proto machinery). Also:
- ECN bits are handled
- Support Generic Send Offload (GSO)

Pull-Request: #3454.

Author: kpp

Cargo.lock

@@ -83,7 +83,7 @@ libp2p-perf = { version = "0.2.0", path = "protocols/perf" }
 libp2p-ping = { version = "0.43.0", path = "protocols/ping" }
 libp2p-plaintext = { version = "0.40.0", path = "transports/plaintext" }
 libp2p-pnet = { version = "0.23.0", path = "transports/pnet" }
-libp2p-quic = { version = "0.8.0-alpha", path = "transports/quic" }
+libp2p-quic = { version = "0.9.0-alpha", path = "transports/quic" }
 libp2p-relay = { version = "0.16.1", path = "protocols/relay" }
 libp2p-rendezvous = { version = "0.13.0", path = "protocols/rendezvous" }
 libp2p-request-response = { version = "0.25.1", path = "protocols/request-response" }

Cargo.toml

@@ -1,3 +1,10 @@
+## 0.9.0-alpha - unreleased
+
+- Use `quinn` instead of `quinn-proto`.
+  See [PR 3454].
+
+[PR 3454]: https://github.com/libp2p/rust-libp2p/pull/3454
+
 ## 0.8.0-alpha
 
 - Raise MSRV to 1.65.

transports/quic/CHANGELOG.md

@@ -1,6 +1,6 @@
 [package]
 name = "libp2p-quic"
-version = "0.8.0-alpha"
+version = "0.9.0-alpha"
 authors = ["Parity Technologies <admin@parity.io>"]
 edition = "2021"
 rust-version = { workspace = true }
@@ -19,15 +19,15 @@ libp2p-tls = { workspace = true }
 libp2p-identity = { workspace = true }
 log = "0.4"
 parking_lot = "0.12.0"
-quinn-proto = { version = "0.10.1", default-features = false, features = ["tls-rustls"] }
+quinn = { version = "0.10.1", default-features = false, features = ["tls-rustls", "futures-io"] }
 rand = "0.8.5"
 rustls = { version = "0.21.2", default-features = false }
 thiserror = "1.0.44"
 tokio = { version = "1.29.1", default-features = false, features = ["net", "rt", "time"], optional = true }
 
 [features]
-tokio = ["dep:tokio", "if-watch/tokio"]
-async-std = ["dep:async-std", "if-watch/smol"]
+tokio = ["dep:tokio", "if-watch/tokio", "quinn/runtime-tokio"]
+async-std = ["dep:async-std", "if-watch/smol", "quinn/runtime-async-std"]
 
 # Passing arguments to the docsrs builder in order to properly document cfg's.
 # More information: https://docs.rs/about/builds#cross-compiling

transports/quic/Cargo.toml

@@ -0,0 +1,142 @@
+// Copyright 2017-2020 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use quinn::VarInt;
+use std::{sync::Arc, time::Duration};
+
+/// Config for the transport.
+#[derive(Clone)]
+pub struct Config {
+    /// Timeout for the initial handshake when establishing a connection.
+    /// The actual timeout is the minimum of this and the [`Config::max_idle_timeout`].
+    pub handshake_timeout: Duration,
+    /// Maximum duration of inactivity in ms to accept before timing out the connection.
+    pub max_idle_timeout: u32,
+    /// Period of inactivity before sending a keep-alive packet.
+    /// Must be set lower than the idle_timeout of both
+    /// peers to be effective.
+    ///
+    /// See [`quinn::TransportConfig::keep_alive_interval`] for more
+    /// info.
+    pub keep_alive_interval: Duration,
+    /// Maximum number of incoming bidirectional streams that may be open
+    /// concurrently by the remote peer.
+    pub max_concurrent_stream_limit: u32,
+
+    /// Max unacknowledged data in bytes that may be send on a single stream.
+    pub max_stream_data: u32,
+
+    /// Max unacknowledged data in bytes that may be send in total on all streams
+    /// of a connection.
+    pub max_connection_data: u32,
+
+    /// Support QUIC version draft-29 for dialing and listening.
+    ///
+    /// Per default only QUIC Version 1 / [`libp2p_core::multiaddr::Protocol::QuicV1`]
+    /// is supported.
+    ///
+    /// If support for draft-29 is enabled servers support draft-29 and version 1 on all
+    /// QUIC listening addresses.
+    /// As client the version is chosen based on the remote's address.
+    pub support_draft_29: bool,
+
+    /// TLS client config for the inner [`quinn::ClientConfig`].
+    client_tls_config: Arc<rustls::ClientConfig>,
+    /// TLS server config for the inner [`quinn::ServerConfig`].
+    server_tls_config: Arc<rustls::ServerConfig>,
+}
+
+impl Config {
+    /// Creates a new configuration object with default values.
+    pub fn new(keypair: &libp2p_identity::Keypair) -> Self {
+        let client_tls_config = Arc::new(libp2p_tls::make_client_config(keypair, None).unwrap());
+        let server_tls_config = Arc::new(libp2p_tls::make_server_config(keypair).unwrap());
+        Self {
+            client_tls_config,
+            server_tls_config,
+            support_draft_29: false,
+            handshake_timeout: Duration::from_secs(5),
+            max_idle_timeout: 30 * 1000,
+            max_concurrent_stream_limit: 256,
+            keep_alive_interval: Duration::from_secs(15),
+            max_connection_data: 15_000_000,
+
+            // Ensure that one stream is not consuming the whole connection.
+            max_stream_data: 10_000_000,
+        }
+    }
+}
+
+/// Represents the inner configuration for [`quinn`].
+#[derive(Debug, Clone)]
+pub(crate) struct QuinnConfig {
+    pub(crate) client_config: quinn::ClientConfig,
+    pub(crate) server_config: quinn::ServerConfig,
+    pub(crate) endpoint_config: quinn::EndpointConfig,
+}
+
+impl From<Config> for QuinnConfig {
+    fn from(config: Config) -> QuinnConfig {
+        let Config {
+            client_tls_config,
+            server_tls_config,
+            max_idle_timeout,
+            max_concurrent_stream_limit,
+            keep_alive_interval,
+            max_connection_data,
+            max_stream_data,
+            support_draft_29,
+            handshake_timeout: _,
+        } = config;
+        let mut transport = quinn::TransportConfig::default();
+        // Disable uni-directional streams.
+        transport.max_concurrent_uni_streams(0u32.into());
+        transport.max_concurrent_bidi_streams(max_concurrent_stream_limit.into());
+        // Disable datagrams.
+        transport.datagram_receive_buffer_size(None);
+        transport.keep_alive_interval(Some(keep_alive_interval));
+        transport.max_idle_timeout(Some(VarInt::from_u32(max_idle_timeout).into()));
+        transport.allow_spin(false);
+        transport.stream_receive_window(max_stream_data.into());
+        transport.receive_window(max_connection_data.into());
+        let transport = Arc::new(transport);
+
+        let mut server_config = quinn::ServerConfig::with_crypto(server_tls_config);
+        server_config.transport = Arc::clone(&transport);
+        // Disables connection migration.
+        // Long-term this should be enabled, however we then need to handle address change
+        // on connections in the `Connection`.
+        server_config.migration(false);
+
+        let mut client_config = quinn::ClientConfig::new(client_tls_config);
+        client_config.transport_config(transport);
+
+        let mut endpoint_config = quinn::EndpointConfig::default();
+        if !support_draft_29 {
+            endpoint_config.supported_versions(vec![1]);
+        }
+
+        QuinnConfig {
+            client_config,
+            server_config,
+            endpoint_config,
+        }
+    }
+}

transports/quic/src/config.rs

@@ -20,9 +20,12 @@
 
 //! Future that drives a QUIC connection until is has performed its TLS handshake.
 
-use crate::{Connection, Error};
+use crate::{Connection, ConnectionError, Error};
 
-use futures::prelude::*;
+use futures::{
+    future::{select, Either, FutureExt, Select},
+    prelude::*,
+};
 use futures_timer::Delay;
 use libp2p_identity::PeerId;
 use std::{
@@ -34,64 +37,46 @@ use std::{
 /// A QUIC connection currently being negotiated.
 #[derive(Debug)]
 pub struct Connecting {
-    connection: Option<Connection>,
-    timeout: Delay,
+    connecting: Select<quinn::Connecting, Delay>,
 }
 
 impl Connecting {
-    pub(crate) fn new(connection: Connection, timeout: Duration) -> Self {
+    pub(crate) fn new(connection: quinn::Connecting, timeout: Duration) -> Self {
         Connecting {
-            connection: Some(connection),
-            timeout: Delay::new(timeout),
+            connecting: select(connection, Delay::new(timeout)),
         }
     }
 }
 
+impl Connecting {
+    /// Returns the address of the node we're connected to.
+    /// Panics if the connection is still handshaking.
+    fn remote_peer_id(connection: &quinn::Connection) -> PeerId {
+        let identity = connection
+            .peer_identity()
+            .expect("connection got identity because it passed TLS handshake; qed");
+        let certificates: Box<Vec<rustls::Certificate>> =
+            identity.downcast().expect("we rely on rustls feature; qed");
+        let end_entity = certificates
+            .get(0)
+            .expect("there should be exactly one certificate; qed");
+        let p2p_cert = libp2p_tls::certificate::parse(end_entity)
+            .expect("the certificate was validated during TLS handshake; qed");
+        p2p_cert.peer_id()
+    }
+}
+
 impl Future for Connecting {
     type Output = Result<(PeerId, Connection), Error>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let connection = self
-            .connection
-            .as_mut()
-            .expect("Future polled after it has completed");
-
-        loop {
-            let event = match connection.poll_event(cx) {
-                Poll::Ready(Some(event)) => event,
-                Poll::Ready(None) => return Poll::Ready(Err(Error::EndpointDriverCrashed)),
-                Poll::Pending => {
-                    return self
-                        .timeout
-                        .poll_unpin(cx)
-                        .map(|()| Err(Error::HandshakeTimedOut));
-                }
-            };
-            match event {
-                quinn_proto::Event::Connected => {
-                    // Parse the remote's Id identity from the certificate.
-                    let identity = connection
-                        .peer_identity()
-                        .expect("connection got identity because it passed TLS handshake; qed");
-                    let certificates: Box<Vec<rustls::Certificate>> =
-                        identity.downcast().expect("we rely on rustls feature; qed");
-                    let end_entity = certificates
-                        .get(0)
-                        .expect("there should be exactly one certificate; qed");
-                    let p2p_cert = libp2p_tls::certificate::parse(end_entity)
-                        .expect("the certificate was validated during TLS handshake; qed");
-                    let peer_id = p2p_cert.peer_id();
+        let connection = match futures::ready!(self.connecting.poll_unpin(cx)) {
+            Either::Right(_) => return Poll::Ready(Err(Error::HandshakeTimedOut)),
+            Either::Left((connection, _)) => connection.map_err(ConnectionError)?,
+        };
 
-                    return Poll::Ready(Ok((peer_id, self.connection.take().unwrap())));
-                }
-                quinn_proto::Event::ConnectionLost { reason } => {
-                    return Poll::Ready(Err(Error::Connection(reason.into())))
-                }
-                quinn_proto::Event::HandshakeDataReady | quinn_proto::Event::Stream(_) => {}
-                quinn_proto::Event::DatagramReceived => {
-                    debug_assert!(false, "Datagrams are not supported")
-                }
-            }
-        }
+        let peer_id = Self::remote_peer_id(&connection);
+        let muxer = Connection::new(connection);
+        Poll::Ready(Ok((peer_id, muxer)))
     }
 }

transports/quic/src/connection.rs

@@ -0,0 +1,86 @@
+// Copyright 2022 Protocol Labs.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use std::{
+    io::{self},
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures::{AsyncRead, AsyncWrite};
+
+/// A single stream on a connection
+pub struct Stream {
+    /// A send part of the stream
+    send: quinn::SendStream,
+    /// A receive part of the stream
+    recv: quinn::RecvStream,
+    /// Whether the stream is closed or not
+    close_result: Option<Result<(), io::ErrorKind>>,
+}
+
+impl Stream {
+    pub(super) fn new(send: quinn::SendStream, recv: quinn::RecvStream) -> Self {
+        Self {
+            send,
+            recv,
+            close_result: None,
+        }
+    }
+}
+
+impl AsyncRead for Stream {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        if let Some(close_result) = self.close_result {
+            if close_result.is_err() {
+                return Poll::Ready(Ok(0));
+            }
+        }
+        Pin::new(&mut self.recv).poll_read(cx, buf)
+    }
+}
+
+impl AsyncWrite for Stream {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.send).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.send).poll_flush(cx)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
+        if let Some(close_result) = self.close_result {
+            // For some reason poll_close needs to be 'fuse'able
+            return Poll::Ready(close_result.map_err(Into::into));
+        }
+        let close_result = futures::ready!(Pin::new(&mut self.send).poll_close(cx));
+        self.close_result = Some(close_result.as_ref().map_err(|e| e.kind()).copied());
+        Poll::Ready(close_result)
+    }
+}

transports/quic/src/connection/connecting.rs

@@ -1,47 +1,39 @@
-use std::{net::SocketAddr, time::Duration};
+use crate::{provider::Provider, Error};
 
 use futures::future::Either;
+
 use rand::{distributions, Rng};
 
-use crate::{
-    endpoint::{self, ToEndpoint},
-    Error, Provider,
+use std::{
+    net::{SocketAddr, UdpSocket},
+    time::Duration,
 };
 
 pub(crate) async fn hole_puncher<P: Provider>(
-    endpoint_channel: endpoint::Channel,
+    socket: UdpSocket,
     remote_addr: SocketAddr,
     timeout_duration: Duration,
 ) -> Error {
-    let punch_holes_future = punch_holes::<P>(endpoint_channel, remote_addr);
+    let punch_holes_future = punch_holes::<P>(socket, remote_addr);
     futures::pin_mut!(punch_holes_future);
     match futures::future::select(P::sleep(timeout_duration), punch_holes_future).await {
         Either::Left(_) => Error::HandshakeTimedOut,
         Either::Right((hole_punch_err, _)) => hole_punch_err,
     }
 }
 
-async fn punch_holes<P: Provider>(
-    mut endpoint_channel: endpoint::Channel,
-    remote_addr: SocketAddr,
-) -> Error {
+async fn punch_holes<P: Provider>(socket: UdpSocket, remote_addr: SocketAddr) -> Error {
     loop {
         let sleep_duration = Duration::from_millis(rand::thread_rng().gen_range(10..=200));
         P::sleep(sleep_duration).await;
 
-        let random_udp_packet = ToEndpoint::SendUdpPacket(quinn_proto::Transmit {
-            destination: remote_addr,
-            ecn: None,
-            contents: rand::thread_rng()
-                .sample_iter(distributions::Standard)
-                .take(64)
-                .collect(),
-            segment_size: None,
-            src_ip: None,
-        });
+        let contents: Vec<u8> = rand::thread_rng()
+            .sample_iter(distributions::Standard)
+            .take(64)
+            .collect();
 
-        if endpoint_channel.send(random_udp_packet).await.is_err() {
-            return Error::EndpointDriverCrashed;
+        if let Err(e) = P::send_to(&socket, &contents, remote_addr).await {
+            return Error::Io(e);
         }
     }
 }

transports/quic/src/connection/stream.rs

@@ -57,16 +57,17 @@
 
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 
+mod config;
 mod connection;
-mod endpoint;
 mod hole_punching;
 mod provider;
 mod transport;
 
 use std::net::SocketAddr;
 
-pub use connection::{Connecting, Connection, Substream};
-pub use endpoint::Config;
+pub use config::Config;
+pub use connection::{Connecting, Connection, Stream};
+
 #[cfg(feature = "async-std")]
 pub use provider::async_std;
 #[cfg(feature = "tokio")]
@@ -89,8 +90,7 @@ pub enum Error {
     #[error(transparent)]
     Io(#[from] std::io::Error),
 
-    /// The task spawned in [`Provider::spawn`] to drive
-    /// the quic endpoint has crashed.
+    /// The task to drive a quic endpoint has crashed.
     #[error("Endpoint driver crashed")]
     EndpointDriverCrashed,
 
@@ -110,9 +110,9 @@ pub enum Error {
 /// Dialing a remote peer failed.
 #[derive(Debug, thiserror::Error)]
 #[error(transparent)]
-pub struct ConnectError(#[from] quinn_proto::ConnectError);
+pub struct ConnectError(quinn::ConnectError);
 
 /// Error on an established [`Connection`].
 #[derive(Debug, thiserror::Error)]
 #[error(transparent)]
-pub struct ConnectionError(#[from] quinn_proto::ConnectionError);
+pub struct ConnectionError(quinn::ConnectionError);

transports/quic/src/connection/substream.rs

@@ -18,11 +18,11 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use futures::{future::BoxFuture, Future};
+use futures::future::BoxFuture;
 use if_watch::IfEvent;
 use std::{
     io,
-    net::SocketAddr,
+    net::{SocketAddr, UdpSocket},
     task::{Context, Poll},
     time::Duration,
 };
@@ -32,40 +32,20 @@ pub mod async_std;
 #[cfg(feature = "tokio")]
 pub mod tokio;
 
-/// Size of the buffer for reading data 0x10000.
-#[cfg(any(feature = "async-std", feature = "tokio"))]
-const RECEIVE_BUFFER_SIZE: usize = 65536;
+pub enum Runtime {
+    #[cfg(feature = "tokio")]
+    Tokio,
+    #[cfg(feature = "async-std")]
+    AsyncStd,
+    Dummy,
+}
 
-/// Provider for non-blocking receiving and sending on a [`std::net::UdpSocket`]
-/// and spawning tasks.
+/// Provider for a corresponding quinn runtime and spawning tasks.
 pub trait Provider: Unpin + Send + Sized + 'static {
     type IfWatcher: Unpin + Send;
 
-    /// Create a new providing that is wrapping the socket.
-    ///
-    /// Note: The socket must be set to non-blocking.
-    fn from_socket(socket: std::net::UdpSocket) -> io::Result<Self>;
-
-    /// Receive a single packet.
-    ///
-    /// Returns the message and the address the message came from.
-    fn poll_recv_from(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<(Vec<u8>, SocketAddr)>>;
-
-    /// Set sending a packet on the socket.
-    ///
-    /// Since only one packet can be sent at a time, this may only be called if a preceding
-    /// call to [`Provider::poll_send_flush`] returned [`Poll::Ready`].
-    fn start_send(&mut self, data: Vec<u8>, addr: SocketAddr);
-
-    /// Flush a packet send in [`Provider::start_send`].
-    ///
-    /// If [`Poll::Ready`] is returned the socket is ready for sending a new packet.
-    fn poll_send_flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>>;
-
-    /// Run the given future in the background until it ends.
-    ///
-    /// This is used to spawn the task that is driving the endpoint.
-    fn spawn(future: impl Future<Output = ()> + Send + 'static);
+    /// Run the corresponding runtime
+    fn runtime() -> Runtime;
 
     /// Create a new [`if_watch`] watcher that reports [`IfEvent`]s for network interface changes.
     fn new_if_watcher() -> io::Result<Self::IfWatcher>;
@@ -78,4 +58,11 @@ pub trait Provider: Unpin + Send + Sized + 'static {
 
     /// Sleep for specified amount of time.
     fn sleep(duration: Duration) -> BoxFuture<'static, ()>;
+
+    /// Sends data on the socket to the given address. On success, returns the number of bytes written.
+    fn send_to<'a>(
+        udp_socket: &'a UdpSocket,
+        buf: &'a [u8],
+        target: SocketAddr,
+    ) -> BoxFuture<'a, io::Result<usize>>;
 }

transports/quic/src/endpoint.rs

@@ -18,13 +18,10 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use async_std::{net::UdpSocket, task::spawn};
-use futures::{future::BoxFuture, ready, Future, FutureExt, Stream, StreamExt};
+use futures::{future::BoxFuture, FutureExt};
 use std::{
     io,
-    net::SocketAddr,
-    pin::Pin,
-    sync::Arc,
+    net::UdpSocket,
     task::{Context, Poll},
     time::Duration,
 };
@@ -34,65 +31,14 @@ use crate::GenTransport;
 /// Transport with [`async-std`] runtime.
 pub type Transport = GenTransport<Provider>;
 
-/// Provider for reading / writing to a sockets and spawning
-/// tasks using [`async-std`].
-pub struct Provider {
-    socket: Arc<UdpSocket>,
-    // Future for sending a packet.
-    // This is needed since [`async_Std::net::UdpSocket`] does not
-    // provide a poll-style interface for sending a packet.
-    send_packet: Option<BoxFuture<'static, Result<(), io::Error>>>,
-    recv_stream: ReceiveStream,
-}
+/// Provider for quinn runtime and spawning tasks using [`async-std`].
+pub struct Provider;
 
 impl super::Provider for Provider {
     type IfWatcher = if_watch::smol::IfWatcher;
 
-    fn from_socket(socket: std::net::UdpSocket) -> io::Result<Self> {
-        let socket = Arc::new(socket.into());
-        let recv_stream = ReceiveStream::new(Arc::clone(&socket));
-        Ok(Provider {
-            socket,
-            send_packet: None,
-            recv_stream,
-        })
-    }
-
-    fn poll_recv_from(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<(Vec<u8>, SocketAddr)>> {
-        match self.recv_stream.poll_next_unpin(cx) {
-            Poll::Ready(ready) => {
-                Poll::Ready(ready.expect("ReceiveStream::poll_next never returns None."))
-            }
-            Poll::Pending => Poll::Pending,
-        }
-    }
-
-    fn start_send(&mut self, data: Vec<u8>, addr: SocketAddr) {
-        let socket = self.socket.clone();
-        let send = async move {
-            socket.send_to(&data, addr).await?;
-            Ok(())
-        }
-        .boxed();
-        self.send_packet = Some(send)
-    }
-
-    fn poll_send_flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
-        let pending = match self.send_packet.as_mut() {
-            Some(pending) => pending,
-            None => return Poll::Ready(Ok(())),
-        };
-        match pending.poll_unpin(cx) {
-            Poll::Ready(result) => {
-                self.send_packet = None;
-                Poll::Ready(result)
-            }
-            Poll::Pending => Poll::Pending,
-        }
-    }
-
-    fn spawn(future: impl Future<Output = ()> + Send + 'static) {
-        spawn(future);
+    fn runtime() -> super::Runtime {
+        super::Runtime::AsyncStd
     }
 
     fn new_if_watcher() -> io::Result<Self::IfWatcher> {
@@ -109,48 +55,16 @@ impl super::Provider for Provider {
     fn sleep(duration: Duration) -> BoxFuture<'static, ()> {
         async_std::task::sleep(duration).boxed()
     }
-}
-
-type ReceiveStreamItem = (
-    Result<(usize, SocketAddr), io::Error>,
-    Arc<UdpSocket>,
-    Vec<u8>,
-);
 
-/// Wrapper around the socket to implement `Stream` on it.
-struct ReceiveStream {
-    /// Future for receiving a packet on the socket.
-    // This is needed since [`async_Std::net::UdpSocket`] does not
-    // provide a poll-style interface for receiving packets.
-    fut: BoxFuture<'static, ReceiveStreamItem>,
-}
-
-impl ReceiveStream {
-    fn new(socket: Arc<UdpSocket>) -> Self {
-        let fut = ReceiveStream::next(socket, vec![0; super::RECEIVE_BUFFER_SIZE]).boxed();
-        Self { fut: fut.boxed() }
-    }
-
-    async fn next(socket: Arc<UdpSocket>, mut socket_recv_buffer: Vec<u8>) -> ReceiveStreamItem {
-        let recv = socket.recv_from(&mut socket_recv_buffer).await;
-        (recv, socket, socket_recv_buffer)
-    }
-}
-
-impl Stream for ReceiveStream {
-    type Item = Result<(Vec<u8>, SocketAddr), io::Error>;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let (result, socket, buffer) = ready!(self.fut.poll_unpin(cx));
-
-        let result = result.map(|(packet_len, packet_src)| {
-            debug_assert!(packet_len <= buffer.len());
-            // Copies the bytes from the `socket_recv_buffer` they were written into.
-            (buffer[..packet_len].into(), packet_src)
-        });
-        // Set the future for receiving the next packet on the stream.
-        self.fut = ReceiveStream::next(socket, buffer).boxed();
-
-        Poll::Ready(Some(result))
+    fn send_to<'a>(
+        udp_socket: &'a UdpSocket,
+        buf: &'a [u8],
+        target: std::net::SocketAddr,
+    ) -> BoxFuture<'a, io::Result<usize>> {
+        Box::pin(async move {
+            async_std::net::UdpSocket::from(udp_socket.try_clone()?)
+                .send_to(buf, target)
+                .await
+        })
     }
 }

transports/quic/src/hole_punching.rs

@@ -18,72 +18,27 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use futures::{future::BoxFuture, ready, Future, FutureExt};
+use futures::{future::BoxFuture, FutureExt};
 use std::{
     io,
-    net::SocketAddr,
+    net::{SocketAddr, UdpSocket},
     task::{Context, Poll},
     time::Duration,
 };
-use tokio::{io::ReadBuf, net::UdpSocket};
 
 use crate::GenTransport;
 
 /// Transport with [`tokio`] runtime.
 pub type Transport = GenTransport<Provider>;
 
-/// Provider for reading / writing to a sockets and spawning
-/// tasks using [`tokio`].
-pub struct Provider {
-    socket: UdpSocket,
-    socket_recv_buffer: Vec<u8>,
-    next_packet_out: Option<(Vec<u8>, SocketAddr)>,
-}
+/// Provider for quinn runtime and spawning tasks using [`tokio`].
+pub struct Provider;
 
 impl super::Provider for Provider {
     type IfWatcher = if_watch::tokio::IfWatcher;
 
-    fn from_socket(socket: std::net::UdpSocket) -> std::io::Result<Self> {
-        let socket = UdpSocket::from_std(socket)?;
-        Ok(Provider {
-            socket,
-            socket_recv_buffer: vec![0; super::RECEIVE_BUFFER_SIZE],
-            next_packet_out: None,
-        })
-    }
-
-    fn poll_send_flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
-        let (data, addr) = match self.next_packet_out.as_ref() {
-            Some(pending) => pending,
-            None => return Poll::Ready(Ok(())),
-        };
-        match self.socket.poll_send_to(cx, data.as_slice(), *addr) {
-            Poll::Ready(result) => {
-                self.next_packet_out = None;
-                Poll::Ready(result.map(|_| ()))
-            }
-            Poll::Pending => Poll::Pending,
-        }
-    }
-
-    fn poll_recv_from(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<(Vec<u8>, SocketAddr)>> {
-        let Self {
-            socket,
-            socket_recv_buffer,
-            ..
-        } = self;
-        let mut read_buf = ReadBuf::new(socket_recv_buffer.as_mut_slice());
-        let packet_src = ready!(socket.poll_recv_from(cx, &mut read_buf)?);
-        let bytes = read_buf.filled().to_vec();
-        Poll::Ready(Ok((bytes, packet_src)))
-    }
-
-    fn start_send(&mut self, data: Vec<u8>, addr: SocketAddr) {
-        self.next_packet_out = Some((data, addr));
-    }
-
-    fn spawn(future: impl Future<Output = ()> + Send + 'static) {
-        tokio::spawn(future);
+    fn runtime() -> super::Runtime {
+        super::Runtime::Tokio
     }
 
     fn new_if_watcher() -> io::Result<Self::IfWatcher> {
@@ -100,4 +55,16 @@ impl super::Provider for Provider {
     fn sleep(duration: Duration) -> BoxFuture<'static, ()> {
         tokio::time::sleep(duration).boxed()
     }
+
+    fn send_to<'a>(
+        udp_socket: &'a UdpSocket,
+        buf: &'a [u8],
+        target: SocketAddr,
+    ) -> BoxFuture<'a, io::Result<usize>> {
+        Box::pin(async move {
+            tokio::net::UdpSocket::from_std(udp_socket.try_clone()?)?
+                .send_to(buf, target)
+                .await
+        })
+    }
 }

transports/quic/src/lib.rs

@@ -428,7 +428,7 @@ async fn smoke<P: Provider>() {
     assert_eq!(b_connected, a_peer_id);
 }
 
-async fn build_streams<P: Provider>() -> (SubstreamBox, SubstreamBox) {
+async fn build_streams<P: Provider + Spawn>() -> (SubstreamBox, SubstreamBox) {
     let (_, mut a_transport) = create_default_transport::<P>();
     let (_, mut b_transport) = create_default_transport::<P>();
 
@@ -522,7 +522,7 @@ async fn start_listening(transport: &mut Boxed<(PeerId, StreamMuxerBox)>, addr:
     }
 }
 
-fn prop<P: Provider + BlockOn>(
+fn prop<P: Provider + BlockOn + Spawn>(
     number_listeners: NonZeroU8,
     number_streams: NonZeroU8,
 ) -> quickcheck::TestResult {
@@ -599,7 +599,7 @@ fn prop<P: Provider + BlockOn>(
     quickcheck::TestResult::passed()
 }
 
-async fn answer_inbound_streams<P: Provider, const BUFFER_SIZE: usize>(
+async fn answer_inbound_streams<P: Provider + Spawn, const BUFFER_SIZE: usize>(
     mut connection: StreamMuxerBox,
 ) {
     loop {
@@ -634,7 +634,7 @@ async fn answer_inbound_streams<P: Provider, const BUFFER_SIZE: usize>(
     }
 }
 
-async fn open_outbound_streams<P: Provider, const BUFFER_SIZE: usize>(
+async fn open_outbound_streams<P: Provider + Spawn, const BUFFER_SIZE: usize>(
     mut connection: StreamMuxerBox,
     number_streams: usize,
     completed_streams_tx: mpsc::Sender<()>,
@@ -740,3 +740,22 @@ impl BlockOn for libp2p_quic::tokio::Provider {
             .unwrap()
     }
 }
+
+trait Spawn {
+    /// Run the given future in the background until it ends.
+    fn spawn(future: impl Future<Output = ()> + Send + 'static);
+}
+
+#[cfg(feature = "async-std")]
+impl Spawn for libp2p_quic::async_std::Provider {
+    fn spawn(future: impl Future<Output = ()> + Send + 'static) {
+        async_std::task::spawn(future);
+    }
+}
+
+#[cfg(feature = "tokio")]
+impl Spawn for libp2p_quic::tokio::Provider {
+    fn spawn(future: impl Future<Output = ()> + Send + 'static) {
+        tokio::spawn(future);
+    }
+}