refactor(runner): separate into smaller modules

Cargo.lock

@@ -367,6 +367,94 @@ version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
+[[package]]
+name = "futures"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-channel"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+]
+
+[[package]]
+name = "futures-core"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
+
+[[package]]
+name = "futures-executor"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-io"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
+
+[[package]]
+name = "futures-macro"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "futures-sink"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
+
+[[package]]
+name = "futures-task"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393"
+
+[[package]]
+name = "futures-util"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-io",
+ "futures-macro",
+ "futures-sink",
+ "futures-task",
+ "memchr",
+ "pin-project-lite",
+ "slab",
+]
+
 [[package]]
 name = "getrandom"
 version = "0.2.17"
@@ -779,6 +867,7 @@ dependencies = [
  "claims",
  "clap",
  "common",
+ "futures",
  "miette",
  "rustls",
  "serde",
@@ -961,6 +1050,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "slab"
+version = "0.4.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
+
 [[package]]
 name = "smallvec"
 version = "1.15.1"

Cargo.toml

@@ -14,6 +14,7 @@ common = { path = "common" }
 aws-lc-rs = "1"
 base64 = "0.22"
 clap = { version = "4.5", features = ["derive"] }
+futures = "0.3"
 miette = { version = "7", features = ["fancy"] }
 rcgen = "0.14"
 rustls = { version = "0.23", default-features = false, features = [

common/src/lib.rs

@@ -29,10 +29,12 @@ pub struct BenchRecord {
     pub mode: KeyExchangeMode,
     /// Payload size in bytes
     pub payload_bytes: u64,
+    /// TCP connection latency in nanoseconds
+    pub tcp_ns: u128,
     /// Handshake latency in nanoseconds
-    pub handshake_ns: u64,
+    pub handshake_ns: u128,
     /// Time-to-last-byte in nanoseconds (from connection start)
-    pub ttlb_ns: u64,
+    pub ttlb_ns: u128,
 }
 
 impl BenchRecord {
@@ -56,10 +58,9 @@ impl fmt::Display for BenchRecord {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
     use claims::{assert_err, assert_ok};
     use serde_json::Value;
-
-    use super::*;
     use std::str::FromStr;
 
     #[test]
@@ -68,6 +69,7 @@ mod tests {
             iteration: 0,
             mode: KeyExchangeMode::X25519,
             payload_bytes: 1024,
+            tcp_ns: 500_000,
             handshake_ns: 1_000_000,
             ttlb_ns: 2_000_000,
         };
@@ -83,6 +85,7 @@ mod tests {
             iteration: 42,
             mode: KeyExchangeMode::X25519Mlkem768,
             payload_bytes: 4096,
+            tcp_ns: 1_000_000,
             handshake_ns: 5_000_000,
             ttlb_ns: 10_000_000,
         };

justfile

@@ -12,6 +12,7 @@ alias c := check
 alias d := docs
 alias f := fmt
 alias t := test
+alias bench := benchmark
 
 # Run all checks (fmt, clippy, docs, test)
 [group("dev")]

runner/Cargo.toml

@@ -7,6 +7,7 @@ edition.workspace = true
 [dependencies]
 clap.workspace = true
 common.workspace = true
+futures.workspace = true
 miette.workspace = true
 rustls.workspace = true
 serde.workspace = true

runner/src/bench.rs

@@ -0,0 +1,153 @@
+use common::{
+    BenchRecord, KeyExchangeMode,
+    protocol::{read_payload, write_request},
+};
+use futures::{StreamExt, stream::FuturesUnordered};
+use miette::{Context, IntoDiagnostic};
+use rustls::pki_types::ServerName;
+use std::{
+    io::{Write, stdout},
+    net::SocketAddr,
+    time::Instant,
+};
+use tokio::net::TcpStream;
+use tokio_rustls::TlsConnector;
+use tracing::info;
+
+use crate::config::BenchmarkConfig;
+
+/// Result of a single benchmark iteration.
+struct IterationResult {
+    tcp: u128,
+    handshake: u128,
+    ttlb: u128,
+}
+
+/// Run a single benchmark iteration over TLS.
+async fn run_iteration(
+    server: SocketAddr,
+    payload_bytes: u32,
+    tls_connector: &TlsConnector,
+    server_name: &ServerName<'static>,
+) -> miette::Result<IterationResult> {
+    let tcp_start = Instant::now();
+
+    let stream = TcpStream::connect(server)
+        .await
+        .into_diagnostic()
+        .context("TCP connection failed")?;
+
+    let tcp_ns = tcp_start.elapsed().as_nanos();
+
+    let hs_start = Instant::now();
+    let mut tls_stream = tls_connector
+        .connect(server_name.clone(), stream)
+        .await
+        .into_diagnostic()
+        .context("TLS handshake failed")?;
+
+    let handshake_ns = hs_start.elapsed().as_nanos();
+
+    let ttlb_start = Instant::now();
+    write_request(&mut tls_stream, u64::from(payload_bytes))
+        .await
+        .into_diagnostic()
+        .context("write request failed")?;
+
+    read_payload(&mut tls_stream, u64::from(payload_bytes))
+        .await
+        .into_diagnostic()
+        .context("read payload failed")?;
+
+    let ttlb_ns = tcp_ns + handshake_ns + ttlb_start.elapsed().as_nanos();
+
+    Ok(IterationResult {
+        tcp: tcp_ns,
+        handshake: handshake_ns,
+        ttlb: ttlb_ns,
+    })
+}
+
+pub async fn run_benchmark(
+    config: &BenchmarkConfig,
+    tls_connector: &TlsConnector,
+    server_name: &ServerName<'static>,
+) -> miette::Result<()> {
+    let server = config.server;
+
+    info!(
+        warmup = config.warmup,
+        iters = config.iters,
+        concurrency = config.concurrency,
+        "running benchmark iterations"
+    );
+
+    for _ in 0..config.warmup {
+        run_iteration(server, config.payload, tls_connector, server_name).await?;
+    }
+    info!("warmup complete");
+
+    #[allow(clippy::cast_possible_truncation)] // concurrency is limited to reasonable values
+    let mut output = stdout();
+    run_and_write(config, tls_connector, server_name, &mut output).await?;
+    output
+        .flush()
+        .into_diagnostic()
+        .context("failed to flush output")?;
+
+    info!("benchmark complete");
+    Ok(())
+}
+
+async fn run_single_iteration(
+    i: u32,
+    payload_bytes: u32,
+    mode: KeyExchangeMode,
+    server: SocketAddr,
+    tls_connector: TlsConnector,
+    server_name: ServerName<'static>,
+) -> miette::Result<BenchRecord> {
+    let result = run_iteration(server, payload_bytes, &tls_connector, &server_name).await?;
+
+    Ok(BenchRecord {
+        iteration: u64::from(i),
+        mode,
+        payload_bytes: u64::from(payload_bytes),
+        tcp_ns: result.tcp,
+        handshake_ns: result.handshake,
+        ttlb_ns: result.ttlb,
+    })
+}
+
+async fn run_and_write<W: Write + Send>(
+    config: &BenchmarkConfig,
+    tls_connector: &TlsConnector,
+    server_name: &ServerName<'static>,
+    output: &mut W,
+) -> miette::Result<()> {
+    let mut in_flight = FuturesUnordered::new();
+    let mut issued = 0;
+
+    loop {
+        while issued < config.iters && in_flight.len() < config.concurrency as usize {
+            in_flight.push(run_single_iteration(
+                issued,
+                config.payload,
+                config.mode,
+                config.server,
+                tls_connector.clone(),
+                server_name.clone(),
+            ));
+            issued += 1;
+        }
+
+        match in_flight.next().await {
+            Some(record) => writeln!(output, "{}", record?)
+                .into_diagnostic()
+                .context("failed to write record")?,
+            None => break,
+        }
+    }
+
+    Ok(())
+}

runner/src/config/mod.rs

@@ -12,7 +12,7 @@ use std::{fs::read_to_string, net::SocketAddr, path::Path};
 
 #[derive(Debug, Clone, Deserialize)]
 pub struct BenchmarkConfig {
-    pub mode: String,
+    pub mode: KeyExchangeMode,
     pub payload: u32,
     pub iters: u32,
     pub warmup: u32,
@@ -61,7 +61,7 @@ pub fn load_from_file(path: &Path) -> error::Result<Config> {
 pub fn load_from_cli(args: &Args) -> error::Result<Config> {
     Ok(Config {
         benchmarks: vec![BenchmarkConfig {
-            mode: args.mode.to_string(),
+            mode: args.mode,
             payload: args.payload_bytes,
             iters: args.iters,
             warmup: args.warmup,
@@ -73,21 +73,10 @@ pub fn load_from_cli(args: &Args) -> error::Result<Config> {
     })
 }
 
-impl Config {
-    /// Get the key exchange mode from the first benchmark configuration.
-    #[must_use]
-    pub fn server_mode(&self) -> KeyExchangeMode {
-        self.benchmarks
-            .first()
-            .and_then(|b| b.mode.parse().ok())
-            .unwrap_or(KeyExchangeMode::X25519)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
-    use claims::assert_ok;
+    use claims::{assert_err, assert_ok, assert_some};
 
     const VALID_CONFIG: &str = r#"
 [[benchmarks]]
@@ -124,7 +113,7 @@ server = "127.0.0.1:4433"
 "#;
         let config = get_config_from_str(toml);
         assert_eq!(config.benchmarks.len(), 1);
-        assert_eq!(config.benchmarks[0].mode, "x25519");
+        assert_eq!(config.benchmarks[0].mode, KeyExchangeMode::X25519);
         assert_eq!(config.benchmarks[0].payload, 1024);
     }
 
@@ -132,8 +121,8 @@ server = "127.0.0.1:4433"
     fn valid_multiple_benchmarks() {
         let config = get_config_from_str(VALID_CONFIG);
         assert_eq!(config.benchmarks.len(), 2);
-        assert_eq!(config.benchmarks[0].mode, "x25519");
-        assert_eq!(config.benchmarks[1].mode, "x25519mlkem768");
+        assert_eq!(config.benchmarks[0].mode, KeyExchangeMode::X25519);
+        assert_eq!(config.benchmarks[1].mode, KeyExchangeMode::X25519Mlkem768);
     }
 
     #[test]
@@ -147,8 +136,7 @@ warmup = 10
 concurrency = 1
 server = "127.0.0.1:4433"
 "#;
-        let config = get_config_from_str(toml);
-        assert!(config.server_mode() == KeyExchangeMode::X25519); // fallback
+        assert_err!(toml::from_str::<Config>(toml));
     }
 
     #[test]
@@ -163,8 +151,7 @@ concurrency = 1
 server = "127.0.0.1:4433"
 "#;
         let config = get_config_from_str(toml);
-        let result = validate_config(&config, toml, std::path::Path::new("test.toml"));
-        assert!(result.is_err());
+        assert_err!(validate_config(&config, toml, Path::new("test.toml")));
     }
 
     #[test]
@@ -179,8 +166,7 @@ concurrency = 1
 server = "127.0.0.1:4433"
 "#;
         let config = get_config_from_str(toml);
-        let result = validate_config(&config, toml, std::path::Path::new("test.toml"));
-        assert!(result.is_err());
+        assert_err!(validate_config(&config, toml, Path::new("test.toml")));
     }
 
     #[test]
@@ -195,8 +181,7 @@ concurrency = 0
 server = "127.0.0.1:4433"
 "#;
         let config = get_config_from_str(toml);
-        let result = validate_config(&config, toml, std::path::Path::new("test.toml"));
-        assert!(result.is_err());
+        assert_err!(validate_config(&config, toml, Path::new("test.toml")));
     }
 
     #[test]
@@ -218,7 +203,8 @@ concurrency = 1
 server = "127.0.0.1:4433"
 "#;
         let config = get_config_from_str(toml);
-        assert_eq!(config.server_mode(), KeyExchangeMode::X25519);
+        let benchmark = assert_some!(config.benchmarks.first());
+        assert_eq!(benchmark.mode, KeyExchangeMode::X25519);
     }
 
     #[test]
@@ -233,6 +219,7 @@ concurrency = 1
 server = "127.0.0.1:4433"
 "#;
         let config = get_config_from_str(toml);
-        assert_eq!(config.server_mode(), KeyExchangeMode::X25519Mlkem768);
+        let benchmark = assert_some!(config.benchmarks.first());
+        assert_eq!(benchmark.mode, KeyExchangeMode::X25519Mlkem768);
     }
 }

runner/src/config/utils.rs

@@ -2,7 +2,6 @@ use crate::{
     config::{BenchmarkConfig, Config},
     error::{self, ConfigError},
 };
-use common::{self, KeyExchangeMode};
 use miette::{NamedSource, SourceSpan};
 use std::path::Path;
 
@@ -31,21 +30,6 @@ fn validate_benchmark(
 ) -> error::Result<()> {
     let src = NamedSource::new(path.display().to_string(), content.to_string());
 
-    // Validate mode
-    if benchmark.mode.parse::<KeyExchangeMode>().is_err() {
-        return Err(ConfigError::ValidationError {
-            src,
-            span: find_field_span(content, idx, "mode"),
-            field: "mode".into(),
-            idx,
-            message: format!(
-                "Invalid key exchange mode '{}'. Valid values: 'x25519', 'x25519mlkem768'",
-                benchmark.mode
-            ),
-        }
-        .into());
-    }
-
     validate_positive_field(src.clone(), content, idx, "payload", benchmark.payload)?;
     validate_positive_field(src.clone(), content, idx, "iters", benchmark.iters)?;
     validate_positive_field(src, content, idx, "concurrency", benchmark.concurrency)?;

runner/src/error.rs

@@ -1,4 +1,4 @@
-#![allow(unused_assignments)]
+#![allow(unused)]
 //! Error types for the benchmark runner.
 
 use miette::{Diagnostic, NamedSource, SourceSpan};

runner/src/lib.rs

@@ -1,3 +0,0 @@
-pub mod args;
-pub mod config;
-pub mod error;

runner/src/main.rs

@@ -6,285 +6,27 @@
 //!
 //! Outputs NDJSON records to stdout or a file.
 
+mod args;
+mod bench;
+mod config;
+mod error;
+mod tls;
+
 use clap::Parser;
-use common::{
-    BenchRecord, KeyExchangeMode,
-    protocol::{read_payload, write_request},
-};
 use miette::{Context, IntoDiagnostic};
-use runner::{
-    args::Args,
-    config::{load_from_cli, load_from_file},
-};
-use rustls::{
-    ClientConfig, DigitallySignedStruct, SignatureScheme,
-    client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier},
-    crypto::aws_lc_rs::{
-        self,
-        kx_group::{X25519, X25519MLKEM768},
-    },
-    pki_types::{CertificateDer, ServerName, UnixTime},
-    version::TLS13,
-};
-use std::{
-    env,
-    io::{Write, stdout},
-    net::SocketAddr,
-    sync::Arc,
-    time::Instant,
-};
-use tokio::{net::TcpStream, sync::Semaphore, task::JoinHandle};
+use rustls::pki_types::ServerName;
+use std::{env, sync::Arc};
 use tokio_rustls::TlsConnector;
 use tracing::info;
 use tracing_subscriber::EnvFilter;
 use uuid::Uuid;
 
-/// Result of a single benchmark iteration.
-struct IterationResult {
-    handshake_ns: u64,
-    ttlb_ns: u64,
-}
-
-/// Certificate verifier that accepts any certificate.
-/// Used for benchmarking where we don't need to verify the server's identity.
-#[derive(Debug)]
-struct NoVerifier;
-
-impl ServerCertVerifier for NoVerifier {
-    fn verify_server_cert(
-        &self,
-        _end_entity: &CertificateDer<'_>,
-        _intermediates: &[CertificateDer<'_>],
-        _server_name: &ServerName<'_>,
-        _ocsp_response: &[u8],
-        _now: UnixTime,
-    ) -> Result<ServerCertVerified, rustls::Error> {
-        Ok(ServerCertVerified::assertion())
-    }
-
-    fn verify_tls12_signature(
-        &self,
-        _message: &[u8],
-        _cert: &CertificateDer<'_>,
-        _dss: &DigitallySignedStruct,
-    ) -> Result<HandshakeSignatureValid, rustls::Error> {
-        Ok(HandshakeSignatureValid::assertion())
-    }
-
-    fn verify_tls13_signature(
-        &self,
-        _message: &[u8],
-        _cert: &CertificateDer<'_>,
-        _dss: &DigitallySignedStruct,
-    ) -> Result<HandshakeSignatureValid, rustls::Error> {
-        Ok(HandshakeSignatureValid::assertion())
-    }
-
-    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
-        vec![
-            SignatureScheme::ECDSA_NISTP256_SHA256,
-            SignatureScheme::ECDSA_NISTP384_SHA384,
-            SignatureScheme::ECDSA_NISTP521_SHA512,
-            SignatureScheme::ED25519,
-            SignatureScheme::RSA_PSS_SHA256,
-            SignatureScheme::RSA_PSS_SHA384,
-            SignatureScheme::RSA_PSS_SHA512,
-            SignatureScheme::RSA_PKCS1_SHA256,
-            SignatureScheme::RSA_PKCS1_SHA384,
-            SignatureScheme::RSA_PKCS1_SHA512,
-        ]
-    }
-}
-
-/// Build TLS client config for the given key exchange mode.
-fn build_tls_config(mode: KeyExchangeMode) -> miette::Result<Arc<ClientConfig>> {
-    let mut provider = aws_lc_rs::default_provider();
-    provider.kx_groups = match mode {
-        KeyExchangeMode::X25519 => vec![X25519],
-        KeyExchangeMode::X25519Mlkem768 => vec![X25519MLKEM768],
-    };
-
-    let config = ClientConfig::builder_with_provider(Arc::new(provider))
-        .with_protocol_versions(&[&TLS13])
-        .into_diagnostic()
-        .context("failed to set TLS versions")?
-        .dangerous()
-        .with_custom_certificate_verifier(Arc::new(NoVerifier))
-        .with_no_client_auth();
-
-    Ok(Arc::new(config))
-}
-
-/// Run a single benchmark iteration over TLS.
-#[allow(clippy::cast_possible_truncation)] // nanoseconds won't overflow u64 for reasonable durations
-async fn run_iteration(
-    server: SocketAddr,
-    payload_bytes: u32,
-    tls_connector: &TlsConnector,
-    server_name: &ServerName<'static>,
-) -> miette::Result<IterationResult> {
-    let start = Instant::now();
-
-    let stream = TcpStream::connect(server)
-        .await
-        .into_diagnostic()
-        .context("TCP connection failed")?;
-
-    let mut tls_stream = tls_connector
-        .connect(server_name.clone(), stream)
-        .await
-        .into_diagnostic()
-        .context("TLS handshake failed")?;
-
-    let handshake_ns = start.elapsed().as_nanos() as u64;
-
-    write_request(&mut tls_stream, u64::from(payload_bytes))
-        .await
-        .into_diagnostic()
-        .context("write request failed")?;
-
-    read_payload(&mut tls_stream, u64::from(payload_bytes))
-        .await
-        .into_diagnostic()
-        .context("read payload failed")?;
-
-    let ttlb_ns = start.elapsed().as_nanos() as u64;
-
-    Ok(IterationResult {
-        handshake_ns,
-        ttlb_ns,
-    })
-}
-
-#[allow(clippy::future_not_send)] // References held across await points
-async fn run_benchmark(
-    config: &runner::config::BenchmarkConfig,
-    tls_connector: &TlsConnector,
-    server_name: &ServerName<'static>,
-) -> miette::Result<()> {
-    let server = config.server;
-
-    info!(
-        warmup = config.warmup,
-        iters = config.iters,
-        concurrency = config.concurrency,
-        "running benchmark iterations"
-    );
-
-    for _ in 0..config.warmup {
-        run_iteration(server, config.payload, tls_connector, server_name).await?;
-    }
-    info!("warmup complete");
-
-    let test_conn = tls_connector
-        .connect(
-            server_name.clone(),
-            TcpStream::connect(server)
-                .await
-                .into_diagnostic()
-                .context(format!("failed to connect to server {server}"))?,
-        )
-        .await
-        .into_diagnostic()
-        .context("TLS handshake failed")?;
-
-    let cipher = test_conn.get_ref().1.negotiated_cipher_suite();
-    info!(cipher = ?cipher, "TLS handshake complete");
-
-    #[allow(clippy::cast_possible_truncation)] // concurrency is limited to reasonable values
-    let semaphore = Arc::new(Semaphore::new(config.concurrency as usize));
-    let tasks = spawn_benchmark_tasks(config, &semaphore, tls_connector, server_name);
-
-    // Output to stdout for now
-    {
-        let mut output = stdout();
-        write_results(&mut output, tasks).await?;
-        output
-            .flush()
-            .into_diagnostic()
-            .context("failed to flush output")?;
-    }
-
-    info!("benchmark complete");
-    Ok(())
-}
-
-type ReturnHandle = JoinHandle<(IterationResult, Option<BenchRecord>)>;
-
-fn spawn_benchmark_tasks(
-    config: &runner::config::BenchmarkConfig,
-    semaphore: &Arc<Semaphore>,
-    tls_connector: &TlsConnector,
-    server_name: &ServerName<'static>,
-) -> Vec<ReturnHandle> {
-    let server = config.server;
-    let payload_bytes = config.payload;
-    let mode = config
-        .mode
-        .parse::<KeyExchangeMode>()
-        .expect("mode should be valid");
-
-    (0..config.iters)
-        .map(|i| {
-            spawn_single_iteration(
-                i,
-                payload_bytes,
-                mode,
-                server,
-                semaphore.clone(),
-                tls_connector.clone(),
-                server_name.clone(),
-            )
-        })
-        .collect()
-}
-
-fn spawn_single_iteration(
-    i: u32,
-    payload_bytes: u32,
-    mode: KeyExchangeMode,
-    server: SocketAddr,
-    semaphore: Arc<Semaphore>,
-    tls_connector: TlsConnector,
-    server_name: ServerName<'static>,
-) -> ReturnHandle {
-    tokio::spawn(async move {
-        let _permit = semaphore
-            .acquire()
-            .await
-            .expect("semaphore should not be closed");
-
-        let result = run_iteration(server, payload_bytes, &tls_connector, &server_name)
-            .await
-            .expect("iteration should not fail");
-
-        let record = BenchRecord {
-            iteration: u64::from(i),
-            mode,
-            payload_bytes: u64::from(payload_bytes),
-            handshake_ns: result.handshake_ns,
-            ttlb_ns: result.ttlb_ns,
-        };
-
-        (result, Some(record))
-    })
-}
-
-// #[allow(clippy::future_not_send)] // dyn Write is not Send
-async fn write_results<W: Write + Send>(
-    output: &mut W,
-    tasks: Vec<ReturnHandle>,
-) -> miette::Result<()> {
-    for task in tasks {
-        let (_result, record) = task.await.expect("task should not panic");
-        if let Some(record) = record {
-            writeln!(output, "{record}")
-                .into_diagnostic()
-                .context("failed to write record")?;
-        }
-    }
-    Ok(())
-}
+use crate::{
+    args::Args,
+    bench::run_benchmark,
+    config::{load_from_cli, load_from_file},
+    tls::build_tls_config,
+};
 
 #[tokio::main]
 async fn main() -> miette::Result<()> {
@@ -314,9 +56,6 @@ async fn main() -> miette::Result<()> {
         load_from_cli(&args)?
     };
 
-    let tls_config = build_tls_config(config.server_mode())?;
-    let tls_connector = TlsConnector::from(tls_config);
-
     let server_name = ServerName::try_from("localhost".to_string())
         .into_diagnostic()
         .context("invalid server name")?;
@@ -331,6 +70,9 @@ async fn main() -> miette::Result<()> {
             "running benchmark"
         );
 
+        let tls_config = build_tls_config(benchmark.mode)?;
+        let tls_connector = TlsConnector::from(Arc::new(tls_config));
+
         run_benchmark(benchmark, &tls_connector, &server_name).await?;
     }
 

runner/src/tls.rs

@@ -0,0 +1,86 @@
+use common::KeyExchangeMode;
+use miette::{Context, IntoDiagnostic};
+use rustls::{
+    ClientConfig, DigitallySignedStruct, SignatureScheme,
+    client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier},
+    compress::CompressionCache,
+    crypto::aws_lc_rs::{
+        self,
+        kx_group::{X25519, X25519MLKEM768},
+    },
+    pki_types::{CertificateDer, ServerName, UnixTime},
+    version::TLS13,
+};
+use std::sync::Arc;
+
+/// Certificate verifier that accepts any certificate.
+/// Used for benchmarking where we don't need to verify the server's identity.
+#[derive(Debug)]
+pub struct NoVerifier;
+
+impl ServerCertVerifier for NoVerifier {
+    fn verify_server_cert(
+        &self,
+        _end_entity: &CertificateDer<'_>,
+        _intermediates: &[CertificateDer<'_>],
+        _server_name: &ServerName<'_>,
+        _ocsp_response: &[u8],
+        _now: UnixTime,
+    ) -> Result<ServerCertVerified, rustls::Error> {
+        Ok(ServerCertVerified::assertion())
+    }
+
+    fn verify_tls12_signature(
+        &self,
+        _message: &[u8],
+        _cert: &CertificateDer<'_>,
+        _dss: &DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, rustls::Error> {
+        Ok(HandshakeSignatureValid::assertion())
+    }
+
+    fn verify_tls13_signature(
+        &self,
+        _message: &[u8],
+        _cert: &CertificateDer<'_>,
+        _dss: &DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, rustls::Error> {
+        Ok(HandshakeSignatureValid::assertion())
+    }
+
+    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
+        vec![
+            SignatureScheme::ECDSA_NISTP256_SHA256,
+            SignatureScheme::ECDSA_NISTP384_SHA384,
+            SignatureScheme::ECDSA_NISTP521_SHA512,
+            SignatureScheme::ED25519,
+            SignatureScheme::RSA_PSS_SHA256,
+            SignatureScheme::RSA_PSS_SHA384,
+            SignatureScheme::RSA_PSS_SHA512,
+            SignatureScheme::RSA_PKCS1_SHA256,
+            SignatureScheme::RSA_PKCS1_SHA384,
+            SignatureScheme::RSA_PKCS1_SHA512,
+        ]
+    }
+}
+
+/// Build TLS client config for the given key exchange mode.
+pub fn build_tls_config(mode: KeyExchangeMode) -> miette::Result<ClientConfig> {
+    let mut provider = aws_lc_rs::default_provider();
+    provider.kx_groups = match mode {
+        KeyExchangeMode::X25519 => vec![X25519],
+        KeyExchangeMode::X25519Mlkem768 => vec![X25519MLKEM768],
+    };
+
+    let mut config = ClientConfig::builder_with_provider(Arc::new(provider))
+        .with_protocol_versions(&[&TLS13])
+        .into_diagnostic()
+        .context("failed to set TLS versions")?
+        .dangerous()
+        .with_custom_certificate_verifier(Arc::new(NoVerifier))
+        .with_no_client_auth();
+
+    config.cert_compression_cache = Arc::new(CompressionCache::Disabled);
+
+    Ok(config)
+}