bench(bin): add 100MB upload benchmark (#2199)

* refactor(bin/client/h3): inline StreamHandlerType

`StreamHandlerType` does not carry state. In addition, its sole purpose is its
attached function `StreamHandlerType::make_handler`.

* feat(bin/client/h3): support non-test upload

Previously one could only trigger an upload when specifying a test.

Instead, match on the method whether to up- or download.

* bench(bin): add 100MB upload benchmark

neqo-bin/benches/main.rs

@@ -12,7 +12,8 @@ use tokio::runtime::Runtime;
 
 struct Benchmark {
     name: String,
-    requests: Vec<u64>,
+    requests: Vec<usize>,
+    upload: bool,
 }
 
 fn transfer(c: &mut Criterion) {
@@ -21,30 +22,42 @@ fn transfer(c: &mut Criterion) {
 
     let done_sender = spawn_server();
     let mtu = env::var("MTU").map_or_else(|_| String::new(), |mtu| format!("/mtu-{mtu}"));
-    for Benchmark { name, requests } in [
+    for Benchmark {
+        name,
+        requests,
+        upload,
+    } in [
         Benchmark {
             name: format!("1-conn/1-100mb-resp{mtu} (aka. Download)"),
             requests: vec![100 * 1024 * 1024],
+            upload: false,
         },
         Benchmark {
             name: format!("1-conn/10_000-parallel-1b-resp{mtu} (aka. RPS)"),
             requests: vec![1; 10_000],
+            upload: false,
         },
         Benchmark {
             name: format!("1-conn/1-1b-resp{mtu} (aka. HPS)"),
             requests: vec![1; 1],
+            upload: false,
+        },
+        Benchmark {
+            name: format!("1-conn/1-100mb-resp{mtu} (aka. Upload)"),
+            requests: vec![100 * 1024 * 1024],
+            upload: true,
         },
     ] {
         let mut group = c.benchmark_group(name);
         group.throughput(if requests[0] > 1 {
             assert_eq!(requests.len(), 1);
-            Throughput::Bytes(requests[0])
+            Throughput::Bytes(requests[0] as u64)
         } else {
             Throughput::Elements(requests.len() as u64)
         });
         group.bench_function("client", |b| {
             b.to_async(Runtime::new().unwrap()).iter_batched(
-                || client::client(client::Args::new(&requests)),
+                || client::client(client::Args::new(&requests, upload)),
                 |client| async move {
                     client.await.unwrap();
                 },

neqo-bin/src/client/http3.rs

@@ -45,11 +45,6 @@ impl<'a> Handler<'a> {
             handled_urls: Vec::new(),
             stream_handlers: HashMap::new(),
             all_paths: Vec::new(),
-            handler_type: if args.test.is_some() {
-                StreamHandlerType::Upload
-            } else {
-                StreamHandlerType::Download
-            },
             args,
         };
 
@@ -271,36 +266,6 @@ trait StreamHandler {
     fn process_data_writable(&mut self, client: &mut Http3Client, stream_id: StreamId);
 }
 
-enum StreamHandlerType {
-    Download,
-    Upload,
-}
-
-impl StreamHandlerType {
-    fn make_handler(
-        handler_type: &Self,
-        url: &Url,
-        args: &Args,
-        all_paths: &mut Vec<PathBuf>,
-        client: &mut Http3Client,
-        client_stream_id: StreamId,
-    ) -> Box<dyn StreamHandler> {
-        match handler_type {
-            Self::Download => {
-                let out_file = get_output_file(url, args.output_dir.as_ref(), all_paths);
-                client.stream_close_send(client_stream_id).unwrap();
-                Box::new(DownloadStreamHandler { out_file })
-            }
-            Self::Upload => Box::new(UploadStreamHandler {
-                data: vec![42; args.upload_size],
-                offset: 0,
-                chunk_size: STREAM_IO_BUFFER_SIZE,
-                start: Instant::now(),
-            }),
-        }
-    }
-}
-
 struct DownloadStreamHandler {
     out_file: Option<BufWriter<File>>,
 }
@@ -403,7 +368,6 @@ struct UrlHandler<'a> {
     handled_urls: Vec<Url>,
     stream_handlers: HashMap<StreamId, Box<dyn StreamHandler>>,
     all_paths: Vec<PathBuf>,
-    handler_type: StreamHandlerType,
     args: &'a Args,
 }
 
@@ -441,14 +405,25 @@ impl UrlHandler<'_> {
             Ok(client_stream_id) => {
                 qdebug!("Successfully created stream id {client_stream_id} for {url}");
 
-                let handler: Box<dyn StreamHandler> = StreamHandlerType::make_handler(
-                    &self.handler_type,
-                    &url,
-                    self.args,
-                    &mut self.all_paths,
-                    client,
-                    client_stream_id,
-                );
+                let handler: Box<dyn StreamHandler> = match self.args.method.as_str() {
+                    "GET" => {
+                        let out_file = get_output_file(
+                            &url,
+                            self.args.output_dir.as_ref(),
+                            &mut self.all_paths,
+                        );
+                        client.stream_close_send(client_stream_id).unwrap();
+                        Box::new(DownloadStreamHandler { out_file })
+                    }
+                    "POST" => Box::new(UploadStreamHandler {
+                        data: vec![42; self.args.upload_size],
+                        offset: 0,
+                        chunk_size: STREAM_IO_BUFFER_SIZE,
+                        start: Instant::now(),
+                    }),
+                    _ => unimplemented!(),
+                };
+
                 self.stream_handlers.insert(client_stream_id, handler);
                 self.handled_urls.push(url);
                 true

neqo-bin/src/client/mod.rs

@@ -175,15 +175,15 @@ impl Args {
     #[must_use]
     #[cfg(any(test, feature = "bench"))]
     #[allow(clippy::missing_panics_doc)]
-    pub fn new(requests: &[u64]) -> Self {
+    pub fn new(requests: &[usize], upload: bool) -> Self {
         use std::str::FromStr;
         Self {
             shared: crate::SharedArgs::default(),
             urls: requests
                 .iter()
                 .map(|r| Url::from_str(&format!("http://[::1]:12345/{r}")).unwrap())
                 .collect(),
-            method: "GET".into(),
+            method: if upload { "POST".into() } else { "GET".into() },
             header: vec![],
             max_concurrent_push_streams: 10,
             download_in_series: false,
@@ -196,7 +196,7 @@ impl Args {
             ipv4_only: false,
             ipv6_only: false,
             test: None,
-            upload_size: 100,
+            upload_size: if upload { requests[0] } else { 100 },
             stats: false,
         }
     }

neqo-bin/src/lib.rs

@@ -309,7 +309,7 @@ mod tests {
 
         let temp_dir = TempDir::new();
 
-        let mut client_args = client::Args::new(&[1]);
+        let mut client_args = client::Args::new(&[1], false);
         client_args.set_qlog_dir(temp_dir.path());
         let mut server_args = server::Args::default();
         server_args.set_qlog_dir(temp_dir.path());