test(s2n-quic-dc): add send tests and benchmarks

.github/workflows/ci.yml

@@ -198,12 +198,12 @@ jobs:
         include:
           - os: windows-latest
             # s2n-tls and s2n-quic-dc don't currently build on windows
-            exclude: --exclude s2n-quic-tls --exclude s2n-quic-dc
+            exclude: --exclude s2n-quic-tls --exclude s2n-quic-dc --exclude s2n-quic-dc-benches
           - rust: stable
             os: ubuntu-latest
             target: aarch64-unknown-linux-gnu
             # s2n-quic-dc doesn't currently build on aarch64
-            exclude: --exclude s2n-quic-dc
+            exclude: --exclude s2n-quic-dc  --exclude s2n-quic-dc-benches
           - rust: stable
             os: ubuntu-latest
             target: i686-unknown-linux-gnu
@@ -217,7 +217,7 @@ jobs:
             target: native
             env: S2N_QUIC_PLATFORM_FEATURES_OVERRIDE=""
             # s2n-quic-dc requires platform features
-            exclude: --exclude s2n-quic-dc
+            exclude: --exclude s2n-quic-dc  --exclude s2n-quic-dc-benches
           - rust: stable
             os: ubuntu-latest
             target: native

dc/s2n-quic-dc-benches/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "s2n-quic-dc-benches"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[dependencies]
+aws-lc-rs = "1"
+criterion = { version = "0.5", features = ["html_reports", "async_tokio"] }
+s2n-codec = { path = "../../common/s2n-codec" }
+s2n-quic-dc = { path = "../s2n-quic-dc", features = ["testing"] }
+tokio = { version = "1", features = ["full"] }
+
+[[bench]]
+name = "bench"
+harness = false

dc/s2n-quic-dc/benches/benches/bench.rs → dc/s2n-quic-dc-benches/benches/bench.rs

@@ -3,5 +3,5 @@
 
 use criterion::{criterion_group, criterion_main};
 
-criterion_group!(benches, ::benches::benchmarks);
+criterion_group!(benches, s2n_quic_dc_benches::benchmarks);
 criterion_main!(benches);

dc/s2n-quic-dc/benches/src/crypto/encrypt.rs → dc/s2n-quic-dc-benches/src/crypto/encrypt.rs

@@ -33,12 +33,7 @@ pub fn benchmarks(c: &mut Criterion) {
                         let mut payload = black_box(payload.to_vec());
                         let mut packet_number = 0u32;
                         b.iter(move || {
-                            let _ = black_box(awslc::encrypt(
-                                &key,
-                                &mut packet_number,
-                                header,
-                                &mut payload,
-                            ));
+                            awslc::encrypt(&key, &mut packet_number, header, &mut payload);
                         });
                     },
                 );
@@ -51,12 +46,7 @@ pub fn benchmarks(c: &mut Criterion) {
                         let mut packet_number = 0u32;
                         b.iter(move || {
                             let key = black_box(awslc::key(algo));
-                            let _ = black_box(awslc::encrypt(
-                                &key,
-                                &mut packet_number,
-                                header,
-                                &mut payload,
-                            ));
+                            awslc::encrypt(&key, &mut packet_number, header, &mut payload);
                         });
                     },
                 );

dc/s2n-quic-dc/benches/src/crypto/hkdf.rs → dc/s2n-quic-dc-benches/src/crypto/hkdf.rs

@@ -36,7 +36,7 @@ fn psk(c: &mut Criterion) {
                             let label = black_box(vec![42u8; label_len]);
                             let mut out = black_box(vec![0u8; key_len]);
                             b.iter(move || {
-                                let _ = black_box(awslc::derive_psk(&prk, &label, &mut out));
+                                awslc::derive_psk(&prk, &label, &mut out);
                             });
                         },
                     );
@@ -52,7 +52,7 @@ fn psk(c: &mut Criterion) {
                             let mut out = black_box(vec![0u8; key_len]);
                             b.iter(move || {
                                 let prk = black_box(awslc::prk(&key, alg));
-                                let _ = black_box(awslc::derive_psk(&prk, &label, &mut out));
+                                awslc::derive_psk(&prk, &label, &mut out);
                             });
                         },
                     );
@@ -75,7 +75,7 @@ mod awslc {
         let out_len = out.len();
         let out_len = OutLen(out_len);
 
-        prk.expand(&[&label], out_len)
+        prk.expand(&[label], out_len)
             .unwrap()
             .fill(&mut out[..out_len.0])
             .unwrap();

dc/s2n-quic-dc/benches/src/crypto/hmac.rs → dc/s2n-quic-dc-benches/src/crypto/hmac.rs

@@ -25,7 +25,7 @@ fn init(c: &mut Criterion) {
             aws_lc_rs::rand::fill(&mut key).unwrap();
             let key = black_box(&key);
             b.iter(move || {
-                let _ = black_box(aws_lc_rs::hmac::Key::new(alg, &key));
+                let _ = black_box(aws_lc_rs::hmac::Key::new(alg, key));
             });
         });
     }

dc/s2n-quic-dc/benches/src/lib.rs → dc/s2n-quic-dc-benches/src/lib.rs

@@ -5,8 +5,10 @@ use criterion::Criterion;
 
 pub mod crypto;
 pub mod datagram;
+pub mod streams;
 
 pub fn benchmarks(c: &mut Criterion) {
     crypto::benchmarks(c);
     datagram::benchmarks(c);
+    streams::benchmarks(c);
 }

dc/s2n-quic-dc-benches/src/streams.rs

@@ -0,0 +1,95 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use criterion::Criterion;
+use s2n_quic_dc::stream::{self, server::tokio::accept, socket::Protocol};
+use tokio::{
+    io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
+    net::{TcpListener, TcpStream},
+};
+
+async fn copy_data(
+    input: &'static [u8],
+    a: impl AsyncWrite + Send + 'static,
+    b: impl AsyncRead + Send + 'static,
+) {
+    let a = tokio::spawn(async move {
+        tokio::pin!(a);
+        for _ in 0..30 {
+            a.write_all(input).await.unwrap();
+        }
+        a.shutdown().await.unwrap();
+    });
+
+    let b = tokio::spawn(async move {
+        tokio::pin!(b);
+        let mut void = vec![0; 1024 * 1024];
+        while b.read(&mut void[..]).await.unwrap() != 0 {
+            // Read until EOF
+        }
+    });
+
+    tokio::try_join!(a, b).unwrap();
+}
+
+fn pair(
+    protocol: Protocol,
+    accept_flavor: accept::Flavor,
+) -> (stream::testing::Client, stream::testing::Server) {
+    let client = stream::testing::Client::default();
+    let server = stream::testing::Server::new(protocol, accept_flavor);
+    client.handshake_with(&server).unwrap();
+    (client, server)
+}
+
+pub fn benchmarks(c: &mut Criterion) {
+    let mut group = c.benchmark_group("streams/throughput");
+
+    group.throughput(criterion::Throughput::Bytes(1024 * 1024 * 30));
+
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+
+    let buffer = &*vec![0x0; 1024 * 1024].leak();
+
+    group.bench_function(criterion::BenchmarkId::new("duplex", ""), |b| {
+        b.to_async(&rt).iter(move || async move {
+            let (a, b) = tokio::io::duplex(1024 * 1024);
+            copy_data(buffer, a, b).await;
+        });
+    });
+
+    group.bench_function(criterion::BenchmarkId::new("tcp", ""), |b| {
+        b.to_async(&rt).iter(move || async move {
+            let server = TcpListener::bind("localhost:0").await.unwrap();
+            let server_addr = server.local_addr().unwrap();
+            let (a, b) = tokio::join!(TcpStream::connect(server_addr), async move {
+                server.accept().await.unwrap().0
+            });
+            copy_data(buffer, a.unwrap(), b).await;
+        });
+    });
+
+    for protocol in [Protocol::Udp, Protocol::Tcp] {
+        let _rt = rt.enter();
+        let (client, server) = pair(protocol, accept::Flavor::Fifo);
+        let name = format!("{protocol:?}").to_lowercase();
+        group.bench_function(criterion::BenchmarkId::new("dcquic", name), |b| {
+            b.to_async(&rt).iter(|| {
+                let client = &client;
+                let server = &server;
+                async move {
+                    let (a, b) =
+                        tokio::join!(async { client.connect_to(server).await.unwrap() }, async {
+                            let (b, _addr) = server.accept().await.unwrap();
+                            b
+                        });
+
+                    copy_data(buffer, a, b).await;
+                }
+            });
+        });
+    }
+}

dc/s2n-quic-dc/Cargo.toml

@@ -12,7 +12,7 @@ exclude = ["corpus.tar.gz"]
 
 [features]
 default = ["tokio"]
-testing = ["bolero-generator", "s2n-quic-core/testing", "tracing-subscriber"]
+testing = ["bolero-generator", "s2n-quic-core/testing", "s2n-quic-platform/testing", "tracing-subscriber"]
 tokio = ["tokio/io-util", "tokio/net", "tokio/rt-multi-thread", "tokio/time"]
 
 [dependencies]
@@ -51,11 +51,13 @@ bolero-generator = "0.11"
 insta = "1"
 s2n-codec = { path = "../../common/s2n-codec", features = ["testing"] }
 s2n-quic-core = { path = "../../quic/s2n-quic-core", features = ["testing"] }
+s2n-quic-platform = { path = "../../quic/s2n-quic-platform", features = ["testing"] }
 tokio = { version = "1", features = ["full"] }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [lints.rust.unexpected_cfgs]
 level = "warn"
 check-cfg = [
     'cfg(kani)',
+    'cfg(todo)',
 ]

dc/s2n-quic-dc/src/path/secret/map.rs

@@ -19,6 +19,9 @@ mod state;
 mod status;
 mod store;
 
+#[cfg(any(test, feature = "testing"))]
+pub mod testing;
+
 #[cfg(test)]
 mod event_tests;
 
@@ -213,8 +216,8 @@ impl Map {
         self.store.test_insert(entry);
     }
 
-    #[cfg(test)]
-    fn test_insert_pair(
+    #[cfg(any(test, feature = "testing"))]
+    pub(crate) fn test_insert_pair(
         &self,
         local_addr: SocketAddr,
         peer: &Self,

dc/s2n-quic-dc/src/path/secret/map/testing.rs

@@ -0,0 +1,28 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{event, path::secret};
+
+pub fn new(capacity: usize) -> secret::Map {
+    crate::testing::init_tracing();
+
+    let subscriber = event::tracing::Subscriber::default();
+
+    let signer = secret::stateless_reset::Signer::random();
+
+    if s2n_quic_platform::io::testing::is_in_env() {
+        secret::Map::new(
+            signer,
+            capacity,
+            s2n_quic_platform::io::testing::time::Clock::default(),
+            subscriber,
+        )
+    } else {
+        secret::Map::new(
+            signer,
+            capacity,
+            s2n_quic_core::time::StdClock::default(),
+            subscriber,
+        )
+    }
+}

dc/s2n-quic-dc/src/stream.rs

@@ -26,6 +26,9 @@ pub mod server;
 pub mod shared;
 pub mod socket;
 
+#[cfg(any(test, feature = "testing"))]
+pub mod testing;
+
 bitflags::bitflags! {
     #[derive(Clone, Copy, Debug, PartialEq, Eq)]
     pub struct TransportFeatures: u8 {

dc/s2n-quic-dc/src/stream/application.rs

@@ -10,7 +10,7 @@ use crate::{
         socket,
     },
 };
-use core::fmt;
+use core::{fmt, time::Duration};
 use s2n_quic_core::{buffer, time::Timestamp};
 use std::{io, net::SocketAddr};
 
@@ -25,11 +25,11 @@ pub struct Builder {
 impl Builder {
     /// Builds the stream and emits an event indicating that the stream was built
     #[inline]
-    pub(crate) fn build<Pub>(self, publisher: &Pub) -> io::Result<Stream>
+    pub(crate) fn build<Pub>(self, publisher: &Pub) -> io::Result<(Stream, Duration)>
     where
         Pub: event::EndpointPublisher,
     {
-        {
+        let sojourn_time = {
             let remote_address = self.shared.read_remote_addr();
             let remote_address = &remote_address;
             let credential_id = &*self.shared.credentials().id;
@@ -43,9 +43,12 @@ impl Builder {
                 stream_id,
                 sojourn_time,
             });
-        }
+
+            sojourn_time
+        };
 
         self.build_without_event()
+            .map(|stream| (stream, sojourn_time))
     }
 
     #[inline]