From e106c4d32b9f32b5b49d409ddabd1bee7a21c80e Mon Sep 17 00:00:00 2001
From: Christopher Hunt <huntchr@gmail.com>
Date: Wed, 15 Feb 2023 10:05:10 +1100
Subject: [PATCH] benches: benchmark for things in block_on (#5440)

This additional benchmark exercises a common request/reply pattern using an MPSC for requests along with a oneshot payload as a reply mechanism. When used in a current threaded scenario, the bench is 17 times faster on my machine than when using the multi-threaded runtime and one worker thread. Not only that, but if I increase the number of worker threads to 6, performance degrades further.

Does this suggest a scheduling problem with the multi-threaded runtime?

No matter what, hopefully the benchmarks are a useful addition.
---
 benches/Cargo.toml           |  5 ++++
 benches/sync_mpsc_oneshot.rs | 53 ++++++++++++++++++++++++++++++++++++
 2 files changed, 58 insertions(+)
 create mode 100644 benches/sync_mpsc_oneshot.rs

diff --git a/benches/Cargo.toml b/benches/Cargo.toml
index 78e80fb1d..290173233 100644
--- a/benches/Cargo.toml
+++ b/benches/Cargo.toml
@@ -27,6 +27,11 @@ name = "sync_mpsc"
 path = "sync_mpsc.rs"
 harness = false
 
+[[bench]]
+name = "sync_mpsc_oneshot"
+path = "sync_mpsc_oneshot.rs"
+harness = false
+
 [[bench]]
 name = "rt_multi_threaded"
 path = "rt_multi_threaded.rs"
diff --git a/benches/sync_mpsc_oneshot.rs b/benches/sync_mpsc_oneshot.rs
new file mode 100644
index 000000000..04b783b9d
--- /dev/null
+++ b/benches/sync_mpsc_oneshot.rs
@@ -0,0 +1,53 @@
+use bencher::{benchmark_group, benchmark_main, Bencher};
+use tokio::{
+    runtime::Runtime,
+    sync::{mpsc, oneshot},
+};
+
+fn request_reply_current_thread(b: &mut Bencher) {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+
+    request_reply(b, rt);
+}
+
+fn request_reply_multi_threaded(b: &mut Bencher) {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .build()
+        .unwrap();
+
+    request_reply(b, rt);
+}
+
+fn request_reply(b: &mut Bencher, rt: Runtime) {
+    let tx = rt.block_on(async move {
+        let (tx, mut rx) = mpsc::channel::<oneshot::Sender<()>>(10);
+        tokio::spawn(async move {
+            while let Some(reply) = rx.recv().await {
+                reply.send(()).unwrap();
+            }
+        });
+        tx
+    });
+
+    b.iter(|| {
+        let task_tx = tx.clone();
+        rt.block_on(async move {
+            for _ in 0..1_000 {
+                let (o_tx, o_rx) = oneshot::channel();
+                task_tx.send(o_tx).await.unwrap();
+                let _ = o_rx.await;
+            }
+        })
+    });
+}
+
+benchmark_group!(
+    sync_mpsc_oneshot_group,
+    request_reply_current_thread,
+    request_reply_multi_threaded,
+);
+
+benchmark_main!(sync_mpsc_oneshot_group);