perf(bin): add criterion benchmarks (#1758)

* perf(bin): add criterion benchmarks

Wraps the `neqo-client` and `neqo-server` code, starts the server and runs
various benchmarks through the client.

Benchmarks:
- single-request-1gb
- single-request-1mb
- requests-per-second
- handshakes-per-second

* Rename benchmark instances

* Turn off logging

* Remove 100mb sample size restriction

* Use v6

* Remove 1gb

It just takes too long on the bench machine

* Use /dev/null

* rework imports

* Fix import

* Test without CPU pinning

* Revert "Test without CPU pinning"

This reverts commit a0ef46a.

* Pin all but neqo-bin to CPU 0

* Quote package

* Add rational for neqo-bin handling

* Rework tuples

* Pin first

* Just duplicate the two calls

* Add --workspace flag

* Remove taskset from neqo-bin

---------

Co-authored-by: Martin Thomson <[email protected]>

.github/workflows/bench.yml

@@ -70,11 +70,15 @@ jobs:
       - name: Prepare machine
         run: sudo /root/bin/prep.sh
 
-      # Pin the benchmark run to core 0 and run all benchmarks at elevated priority.
       - name: Run cargo bench
         run: |
-          taskset -c 0 nice -n -20 \
-            cargo "+$TOOLCHAIN" bench --features bench -- --noplot | tee results.txt
+          # Pin all but neqo-bin benchmarks to CPU 0. neqo-bin benchmarks run
+          # both a client and a server, thus benefiting from multiple CPU cores.
+          #
+          # Run all benchmarks at elevated priority.
+          taskset -c 0 nice -n -20 cargo "+$TOOLCHAIN" bench --workspace --exclude neqo-bin --features bench -- --noplot | tee results.txt
+          nice -n -20 cargo "+$TOOLCHAIN" bench --package neqo-bin --features bench -- --noplot | tee -a results.txt
+
 
       # Compare various configurations of neqo against msquic, and gather perf data
       # during the hyperfine runs.

neqo-bin/Cargo.toml

@@ -11,12 +11,12 @@ license.workspace = true
 
 [[bin]]
 name = "neqo-client"
-path = "src/bin/client/main.rs"
+path = "src/bin/client.rs"
 bench = false
 
 [[bin]]
 name = "neqo-server"
-path = "src/bin/server/main.rs"
+path = "src/bin/server.rs"
 bench = false
 
 [lints]
@@ -40,6 +40,18 @@ regex = { version = "1.9", default-features = false, features = ["unicode-perl"]
 tokio = { version = "1", default-features = false, features = ["net", "time", "macros", "rt", "rt-multi-thread"] }
 url = { version = "2.5", default-features = false }
 
+[dev-dependencies]
+criterion = { version = "0.5", default-features = false, features = ["html_reports", "async_tokio"] }
+tokio = { version = "1", default-features = false, features = ["sync"] }
+
+[features]
+bench = []
+
 [lib]
 # See https://github.com/bheisler/criterion.rs/blob/master/book/src/faq.md#cargo-bench-gives-unrecognized-option-errors-for-valid-command-line-options
 bench = false
+
+[[bench]]
+name = "main"
+harness = false
+required-features = ["bench"]

neqo-bin/benches/main.rs

@@ -0,0 +1,92 @@
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use std::{path::PathBuf, str::FromStr};
+
+use criterion::{criterion_group, criterion_main, BatchSize, Criterion, Throughput};
+use neqo_bin::{client, server};
+use tokio::runtime::Runtime;
+
+struct Benchmark {
+    name: String,
+    requests: Vec<u64>,
+    /// Download resources in series using separate connections.
+    download_in_series: bool,
+    sample_size: Option<usize>,
+}
+
+fn transfer(c: &mut Criterion) {
+    neqo_common::log::init(Some(log::LevelFilter::Off));
+    neqo_crypto::init_db(PathBuf::from_str("../test-fixture/db").unwrap());
+
+    let done_sender = spawn_server();
+
+    for Benchmark {
+        name,
+        requests,
+        download_in_series,
+        sample_size,
+    } in [
+        Benchmark {
+            name: "1-conn/1-100mb-resp (aka. Download)".to_string(),
+            requests: vec![100 * 1024 * 1024],
+            download_in_series: false,
+            sample_size: Some(10),
+        },
+        Benchmark {
+            name: "1-conn/10_000-1b-seq-resp (aka. RPS)".to_string(),
+            requests: vec![1; 10_000],
+            download_in_series: false,
+            sample_size: None,
+        },
+        Benchmark {
+            name: "100-seq-conn/1-1b-resp (aka. HPS)".to_string(),
+            requests: vec![1; 100],
+            download_in_series: true,
+            sample_size: None,
+        },
+    ] {
+        let mut group = c.benchmark_group(name);
+        group.throughput(if requests[0] > 1 {
+            assert_eq!(requests.len(), 1);
+            Throughput::Bytes(requests[0])
+        } else {
+            Throughput::Elements(requests.len() as u64)
+        });
+        if let Some(size) = sample_size {
+            group.sample_size(size);
+        }
+        group.bench_function("client", |b| {
+            b.to_async(Runtime::new().unwrap()).iter_batched(
+                || client::client(client::Args::new(&requests, download_in_series)),
+                |client| async move {
+                    client.await.unwrap();
+                },
+                BatchSize::PerIteration,
+            );
+        });
+        group.finish();
+    }
+
+    done_sender.send(()).unwrap();
+}
+
+fn spawn_server() -> tokio::sync::oneshot::Sender<()> {
+    let (done_sender, mut done_receiver) = tokio::sync::oneshot::channel();
+    std::thread::spawn(move || {
+        Runtime::new().unwrap().block_on(async {
+            let mut server = Box::pin(server::server(server::Args::default()));
+            tokio::select! {
+                _ = &mut done_receiver => {}
+                _ = &mut server  => {}
+            }
+        });
+    });
+    done_sender
+}
+
+criterion_group!(benches, transfer);
+criterion_main!(benches);

neqo-bin/src/bin/client.rs

@@ -0,0 +1,14 @@
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use clap::Parser;
+
+#[tokio::main]
+async fn main() -> Result<(), neqo_bin::client::Error> {
+    let args = neqo_bin::client::Args::parse();
+
+    neqo_bin::client::client(args).await
+}

neqo-bin/src/bin/server.rs

@@ -0,0 +1,14 @@
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use clap::Parser;
+
+#[tokio::main]
+async fn main() -> Result<(), std::io::Error> {
+    let args = neqo_bin::server::Args::parse();
+
+    neqo_bin::server::server(args).await
+}

neqo-bin/src/bin/client/http09.rs → neqo-bin/src/client/http09.rs

@@ -25,8 +25,7 @@ use neqo_transport::{
 };
 use url::Url;
 
-use super::{get_output_file, Args, KeyUpdateState, Res};
-use crate::qlog_new;
+use super::{get_output_file, qlog_new, Args, KeyUpdateState, Res};
 
 pub struct Handler<'a> {
     streams: HashMap<StreamId, Option<BufWriter<File>>>,

neqo-bin/src/bin/client/http3.rs → neqo-bin/src/client/http3.rs

@@ -26,7 +26,7 @@ use neqo_transport::{
 };
 use url::Url;
 
-use crate::{get_output_file, qlog_new, Args, KeyUpdateState, Res};
+use super::{get_output_file, qlog_new, Args, KeyUpdateState, Res};
 
 pub(crate) struct Handler<'a> {
     #[allow(

neqo-bin/src/bin/client/main.rs → neqo-bin/src/client/mod.rs

@@ -21,66 +21,67 @@ use futures::{
     future::{select, Either},
     FutureExt, TryFutureExt,
 };
-use neqo_bin::udp;
 use neqo_common::{self as common, qdebug, qerror, qinfo, qlog::NeqoQlog, qwarn, Datagram, Role};
 use neqo_crypto::{
     constants::{TLS_AES_128_GCM_SHA256, TLS_AES_256_GCM_SHA384, TLS_CHACHA20_POLY1305_SHA256},
     init, Cipher, ResumptionToken,
 };
-use neqo_http3::{Error, Output};
+use neqo_http3::Output;
 use neqo_transport::{AppError, ConnectionId, Error as TransportError, Version};
 use qlog::{events::EventImportance, streamer::QlogStreamer};
 use tokio::time::Sleep;
 use url::{Origin, Url};
 
+use crate::{udp, SharedArgs};
+
 mod http09;
 mod http3;
 
 const BUFWRITER_BUFFER_SIZE: usize = 64 * 1024;
 
 #[derive(Debug)]
-pub enum ClientError {
+pub enum Error {
     ArgumentError(&'static str),
     Http3Error(neqo_http3::Error),
     IoError(io::Error),
     QlogError,
     TransportError(neqo_transport::Error),
 }
 
-impl From<io::Error> for ClientError {
+impl From<io::Error> for Error {
     fn from(err: io::Error) -> Self {
         Self::IoError(err)
     }
 }
 
-impl From<neqo_http3::Error> for ClientError {
+impl From<neqo_http3::Error> for Error {
     fn from(err: neqo_http3::Error) -> Self {
         Self::Http3Error(err)
     }
 }
 
-impl From<qlog::Error> for ClientError {
+impl From<qlog::Error> for Error {
     fn from(_err: qlog::Error) -> Self {
         Self::QlogError
     }
 }
 
-impl From<neqo_transport::Error> for ClientError {
+impl From<neqo_transport::Error> for Error {
     fn from(err: neqo_transport::Error) -> Self {
         Self::TransportError(err)
     }
 }
 
-impl Display for ClientError {
+impl Display for Error {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(f, "Error: {self:?}")?;
         Ok(())
     }
 }
 
-impl std::error::Error for ClientError {}
+impl std::error::Error for Error {}
 
-type Res<T> = Result<T, ClientError>;
+type Res<T> = Result<T, Error>;
 
 /// Track whether a key update is needed.
 #[derive(Debug, PartialEq, Eq)]
@@ -90,14 +91,14 @@ impl KeyUpdateState {
     pub fn maybe_update<F, E>(&mut self, update_fn: F) -> Res<()>
     where
         F: FnOnce() -> Result<(), E>,
-        E: Into<ClientError>,
+        E: Into<Error>,
     {
         if self.0 {
             if let Err(e) = update_fn() {
                 let e = e.into();
                 match e {
-                    ClientError::TransportError(TransportError::KeyUpdateBlocked)
-                    | ClientError::Http3Error(Error::TransportError(
+                    Error::TransportError(TransportError::KeyUpdateBlocked)
+                    | Error::Http3Error(neqo_http3::Error::TransportError(
                         TransportError::KeyUpdateBlocked,
                     )) => (),
                     _ => return Err(e),
@@ -123,7 +124,7 @@ pub struct Args {
     verbose: clap_verbosity_flag::Verbosity<clap_verbosity_flag::InfoLevel>,
 
     #[command(flatten)]
-    shared: neqo_bin::SharedArgs,
+    shared: SharedArgs,
 
     urls: Vec<Url>,
 
@@ -189,6 +190,36 @@ pub struct Args {
 }
 
 impl Args {
+    #[must_use]
+    #[cfg(feature = "bench")]
+    #[allow(clippy::missing_panics_doc)]
+    pub fn new(requests: &[u64], download_in_series: bool) -> Self {
+        use std::str::FromStr;
+        Self {
+            verbose: clap_verbosity_flag::Verbosity::<clap_verbosity_flag::InfoLevel>::default(),
+            shared: crate::SharedArgs::default(),
+            urls: requests
+                .iter()
+                .map(|r| Url::from_str(&format!("http://[::1]:12345/{r}")).unwrap())
+                .collect(),
+            method: "GET".into(),
+            header: vec![],
+            max_concurrent_push_streams: 10,
+            download_in_series,
+            concurrency: 100,
+            output_read_data: false,
+            output_dir: Some("/dev/null".into()),
+            resume: false,
+            key_update: false,
+            ech: None,
+            ipv4_only: false,
+            ipv6_only: false,
+            test: None,
+            upload_size: 100,
+            stats: false,
+        }
+    }
+
     fn get_ciphers(&self) -> Vec<Cipher> {
         self.shared
             .ciphers
@@ -445,10 +476,10 @@ fn qlog_new(args: &Args, hostname: &str, cid: &ConnectionId) -> Res<NeqoQlog> {
     }
 }
 
-#[tokio::main]
-async fn main() -> Res<()> {
-    let mut args = Args::parse();
+pub async fn client(mut args: Args) -> Res<()> {
     neqo_common::log::init(Some(args.verbose.log_level_filter()));
+    init();
+
     args.update_for_tests();
 
     init();