Explore k6, oha, h2load etc.

Justfile

@@ -63,10 +63,10 @@ httpwg-over-tcp *args='':
     cargo build --release \
         --package httpwg-loona \
         --package httpwg-cli
-    export TEST_PROTO=h2
+    export PROTO=h2
     export PORT=8001
     export RUST_LOG=${RUST_LOG:-info}
-    ./target/release/httpwg --address localhost:8001 "$@" -- ./target/release/httpwg-loona
+    ./target/release/httpwg --frame-timeout 2000 --connect-timeout 2000 --address localhost:8001 "$@" -- ./target/release/httpwg-loona
 
 instruments:
     #!/usr/bin/env -S bash -eux
@@ -84,7 +84,7 @@ samply:
         --package httpwg-loona \
         --profile profiling \
         --features tracing/release_max_level_info
-    export TEST_PROTO=h2
+    export PROTO=h2
     export PORT=8002
     target/profiling/httpwg-loona
 

crates/buffet/src/bufpool.rs

@@ -7,19 +7,38 @@ mod privatepool;
 
 pub type BufResult<T, B> = (std::io::Result<T>, B);
 
-pub use privatepool::{initialize_allocator_with_num_bufs, num_free, Error, Result, BUF_SIZE};
+pub use privatepool::{
+    initialize_allocator_with_num_bufs, is_allocator_initialized, num_free, Error, Result, BUF_SIZE,
+};
 
 /// Initialize the allocator. Must be called before any other
 /// allocation function.
 pub fn initialize_allocator() -> Result<()> {
+    if is_allocator_initialized() {
+        return Ok(());
+    }
+
     // 64 * 1024 * 4096 bytes = 256 MiB
     #[cfg(not(feature = "miri"))]
     let default_num_bufs = 64 * 1024;
 
     #[cfg(feature = "miri")]
     let default_num_bufs = 1024;
 
-    initialize_allocator_with_num_bufs(default_num_bufs)
+    let mut num_bufs = default_num_bufs;
+
+    if let Ok(env_num_bufs) = std::env::var("BUFFET_NUM_BUFS") {
+        if let Ok(parsed_num_bufs) = env_num_bufs.parse::<usize>() {
+            num_bufs = parsed_num_bufs;
+        }
+    }
+
+    let mem_usage_in_mb: f64 = num_bufs as f64 * (BUF_SIZE as usize) as f64 / 1024.0 / 1024.0;
+    eprintln!(
+        "==== buffet will use {} buffers, for a constant {:.2} MiB usage (override with $BUFFET_NUM_BUFS)",
+        num_bufs, mem_usage_in_mb
+    );
+    initialize_allocator_with_num_bufs(default_num_bufs as _)
 }
 
 impl BufMut {

crates/buffet/src/bufpool/privatepool.rs

@@ -73,6 +73,10 @@ fn with<T>(f: impl FnOnce(&mut Inner) -> T) -> T {
 /// The size of a buffer, in bytes (4 KiB)
 pub const BUF_SIZE: u16 = 4096;
 
+pub fn is_allocator_initialized() -> bool {
+    POOL.with(|pool| unsafe { (*pool.inner.get()).is_some() })
+}
+
 /// Initializes the allocator with the given number of buffers
 pub fn initialize_allocator_with_num_bufs(num_bufs: u32) -> Result<()> {
     POOL.with(|pool| {

crates/buffet/src/io.rs

@@ -44,7 +44,15 @@ pub trait WriteOwned {
 
         for buf in list.pieces.iter().cloned() {
             let buf_len = buf.len();
+
+            let before_write = std::time::Instant::now();
+            tracing::trace!("doing write with buf of len {}", buf_len);
             let (res, _) = self.write_owned(buf).await;
+            tracing::trace!(
+                "doing write with buf of len {}... done in {:?}",
+                buf_len,
+                before_write.elapsed()
+            );
 
             match res {
                 Ok(0) => {
@@ -72,8 +80,15 @@ pub trait WriteOwned {
     /// Write a list of buffers, re-trying the write if the kernel does a
     /// partial write.
     async fn writev_all_owned(&mut self, mut list: PieceList) -> std::io::Result<()> {
+        tracing::trace!("writev_all_owned starts...");
+        let start = std::time::Instant::now();
+
         while !list.is_empty() {
+            tracing::trace!("doing writev_owned with {} pieces", list.len());
+            let before_writev = std::time::Instant::now();
             let n = self.writev_owned(&list).await?;
+            tracing::trace!("writev_owned took {:?}", before_writev.elapsed());
+
             if n == 0 {
                 return Err(std::io::Error::new(
                     std::io::ErrorKind::WriteZero,
@@ -101,6 +116,10 @@ pub trait WriteOwned {
                 }
             }
         }
+        tracing::trace!(
+            "writev_all_owned starts... and succeeds! took {:?}",
+            start.elapsed()
+        );
 
         Ok(())
     }

crates/buffet/src/lib.rs

@@ -44,7 +44,13 @@ pub fn start<F: Future>(task: F) -> F::Output {
     let rt = tokio::runtime::Builder::new_current_thread()
         .enable_all()
         .on_thread_park(move || {
+            let start = std::time::Instant::now();
+            tracing::trace!("thread park, submitting...");
             u.submit().unwrap();
+            tracing::trace!(
+                "thread park, submitting... done! (took {:?})",
+                start.elapsed()
+            );
         })
         .build()
         .unwrap();
@@ -73,7 +79,7 @@ pub fn start<F: Future>(task: F) -> F::Output {
         if (tokio::time::timeout(cancel_submit_timeout, &mut lset).await).is_err() {
             drop(cancel_tx);
 
-            // during this second poll, the async cancellations hopefuly finish
+            // during this second poll, the async cancellations hopefully finish
             let cleanup_timeout = std::time::Duration::from_millis(500);
             if (tokio::time::timeout(cleanup_timeout, lset).await).is_err() {
                 tracing::warn!(

crates/buffet/src/net/net_uring.rs

@@ -1,7 +1,7 @@
 use std::{
     mem::ManuallyDrop,
     net::SocketAddr,
-    os::fd::{AsRawFd, FromRawFd, RawFd},
+    os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd},
     rc::Rc,
 };
 
@@ -53,6 +53,14 @@ impl Drop for TcpStream {
     }
 }
 
+impl IntoRawFd for TcpStream {
+    fn into_raw_fd(self) -> RawFd {
+        let fd = self.fd;
+        std::mem::forget(self);
+        fd
+    }
+}
+
 pub struct TcpListener {
     fd: i32,
 }
@@ -66,7 +74,8 @@ impl TcpListener {
 
         socket.set_nodelay(true)?;
 
-        // FIXME: don't hardcode
+        // FIXME: don't hardcode, but we get test failures on Linux otherwise for some
+        // reason
         socket.set_reuse_port(true)?;
         socket.set_reuse_address(true)?;
         socket.bind(&addr)?;
@@ -129,6 +138,12 @@ impl ReadOwned for TcpReadHalf {
             buf.io_buf_mut_capacity() as u32,
         )
         .build();
+        tracing::trace!(
+            "submitting read_owned, reading from fd {} to {:p} with capacity {}",
+            self.0.fd,
+            buf.io_buf_mut_stable_mut_ptr(),
+            buf.io_buf_mut_capacity()
+        );
         let cqe = get_ring().push(sqe).await;
         let ret = match cqe.error_for_errno() {
             Ok(ret) => ret,
@@ -149,6 +164,7 @@ impl WriteOwned for TcpWriteHalf {
             buf.len().try_into().expect("usize -> u32"),
         )
         .build();
+
         let cqe = get_ring().push(sqe).await;
         let ret = match cqe.error_for_errno() {
             Ok(ret) => ret,
@@ -159,6 +175,35 @@ impl WriteOwned for TcpWriteHalf {
 
     // TODO: implement writev
 
+    // async fn writev_owned(&mut self, list: &PieceList) -> std::io::Result<usize>
+    // {     let mut iovecs: Vec<iovec> = vec![];
+    //     for piece in list.pieces.iter() {
+    //         let iov = iovec {
+    //             iov_base: piece.as_ref().as_ptr() as *mut libc::c_void,
+    //             iov_len: piece.len(),
+    //         };
+    //         iovecs.push(iov);
+    //     }
+    //     let iovecs = iovecs.into_boxed_slice();
+    //     let iov_cnt = iovecs.len();
+    //     // FIXME: don't leak, duh
+    //     let iovecs = Box::leak(iovecs);
+
+    //     let sqe = Writev::new(
+    //         io_uring::types::Fd(self.0.fd),
+    //         iovecs.as_ptr() as *const _,
+    //         iov_cnt as u32,
+    //     )
+    //     .build();
+
+    //     let cqe = get_ring().push(sqe).await;
+    //     let ret = match cqe.error_for_errno() {
+    //         Ok(ret) => ret,
+    //         Err(e) => return Err(std::io::Error::from(e)),
+    //     };
+    //     Ok(ret as usize)
+    // }
+
     async fn shutdown(&mut self) -> std::io::Result<()> {
         tracing::debug!("requesting shutdown");
         let sqe =

crates/buffet/src/roll.rs

@@ -19,7 +19,7 @@ use nom::{
 #[macro_export]
 macro_rules! trace {
     ($($tt:tt)*) => {
-        // tracing::trace!($($tt)*)
+        tracing::trace!($($tt)*)
     };
 }
 

crates/httpwg-harness/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "httpwg-harness"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+eyre = { version = "0.6.12", default-features = false }
+rcgen = { version = "0.13.1", default-features = false, features = ["aws_lc_rs"] }
+rustls = "0.23.12"