Refactor benchmarks

Cargo.toml

@@ -62,9 +62,9 @@ edition = "2021"
 exclude = ["docs", "build", "examples", "image"]
 
 [[bench]]
-name = "throughput"
+name = "read_write"
 harness = false
-test = true
+test = false
 
 [dependencies]
 # Local
@@ -133,13 +133,13 @@ strum_macros = "0.25.2"
 sys-info = "0.9.1"
 
 [dev-dependencies]
-regex = "1.9.6"
-criterion = { version = "0.5.1", features = ["html_reports"] }
+divan = "0.1.2"
 once_cell = "1.18.0"
-tracing-test = "0.2.4"
 pretty_assertions = "1.4.0"
-tempfile = "3.8.0"
 rand = "0.8.5"
+regex = "1.9.6"
+tracing-test = "0.2.4"
+tempfile = "3.8.0"
 
 [build-dependencies]
 tonic-build = { version = "0.10.2", default_features = false, features = [
@@ -153,3 +153,6 @@ protobuf-src = { version = "1.1.0", optional = true }
 default = ["vendor-protoc"]
 instrument = []
 vendor-protoc = ["dep:protobuf-src"]
+
+[profile.release]
+debug = true

benches/read_write.rs

@@ -0,0 +1,116 @@
+mod shared;
+
+use divan::Bencher;
+use shared::*;
+
+fn main() {
+    divan::main();
+}
+
+#[inline]
+fn counter(psize: usize) -> impl divan::counter::Counter {
+    divan::counter::BytesCount::new(psize * NUMBER_OF_PACKETS as usize)
+}
+
+#[divan::bench_group(sample_count = 10)]
+mod read {
+    use super::*;
+
+    #[divan::bench(consts = PACKET_SIZES)]
+    fn direct<const N: usize>(b: Bencher) {
+        let (writer, reader) = socket_pair(None, None);
+        let (tx, rx) = channel();
+        let writer = Writer::<N>::new(writer, reader.local_addr().unwrap(), rx);
+
+        spawn(format!("direct_writer_{N}"), move || loop {
+            if !writer.write_all(NUMBER_OF_PACKETS) {
+                break;
+            }
+        });
+
+        b.counter(counter(N)).bench_local(|| {
+            read_to_end::<N>(&reader, &tx, NUMBER_OF_PACKETS);
+        });
+    }
+
+    #[divan::bench(consts = PACKET_SIZES)]
+    fn quilkin<const N: usize>(b: Bencher) {
+        let (writer, reader) = socket_pair(None, None);
+        let (tx, rx) = channel();
+
+        let quilkin_loop = QuilkinLoop::spinup(READ_QUILKIN_PORT, reader.local_addr().unwrap());
+        let writer = Writer::<N>::new(writer, (Ipv4Addr::LOCALHOST, READ_QUILKIN_PORT).into(), rx);
+        let _quilkin_loop = writer.wait_ready(quilkin_loop, &reader);
+
+        spawn(format!("quilkin_writer_{N}"), move || loop {
+            if !writer.write_all(NUMBER_OF_PACKETS) {
+                break;
+            }
+        });
+
+        b.counter(counter(N)).bench_local(|| {
+            read_to_end::<N>(&reader, &tx, NUMBER_OF_PACKETS);
+        });
+    }
+}
+
+#[divan::bench_group(sample_count = 10)]
+mod write {
+    use super::*;
+
+    #[divan::bench(consts = PACKET_SIZES)]
+    fn direct<const N: usize>(b: Bencher) {
+        let (writer, reader) = socket_pair(None, None);
+        let (tx, rx) = channel();
+
+        let writer = Writer::<N>::new(writer, reader.local_addr().unwrap(), rx);
+
+        let (loop_tx, loop_rx) = mpsc::sync_channel(1);
+
+        spawn(format!("direct_reader_{N}"), move || {
+            while let Ok((num, _size)) = loop_rx.recv() {
+                read_to_end::<N>(&reader, &tx, num);
+            }
+        });
+
+        b.counter(counter(N)).bench_local(|| {
+            // Signal the read loop to run
+            loop_tx.send((NUMBER_OF_PACKETS, N)).unwrap();
+
+            writer.write_all(NUMBER_OF_PACKETS);
+        });
+    }
+
+    #[divan::bench(consts = PACKET_SIZES)]
+    fn quilkin<const N: usize>(b: Bencher) {
+        let (writer, reader) = socket_pair(None, None);
+        let (tx, rx) = channel();
+
+        //quilkin::test::enable_log("quilkin=debug");
+
+        let quilkin_loop = QuilkinLoop::spinup(WRITE_QUILKIN_PORT, reader.local_addr().unwrap());
+        let writer = Writer::<N>::new(writer, (Ipv4Addr::LOCALHOST, WRITE_QUILKIN_PORT).into(), rx);
+        let _quilkin_loop = writer.wait_ready(quilkin_loop, &reader);
+
+        let thread = {
+            let (loop_tx, loop_rx) = mpsc::sync_channel(1);
+
+            let thread = spawn(format!("quilkin_reader_{}", N), move || {
+                while let Ok((num, _size)) = loop_rx.recv() {
+                    read_to_end::<N>(&reader, &tx, num);
+                }
+            });
+
+            b.counter(counter(N)).bench_local(|| {
+                // Signal the read loop to run
+                loop_tx.send((NUMBER_OF_PACKETS, N)).unwrap();
+
+                writer.write_all(NUMBER_OF_PACKETS);
+            });
+
+            thread
+        };
+
+        thread.join().unwrap();
+    }
+}