test: benchmark MPMC on Tokio and replace Pollster

AGENTS.md

@@ -18,7 +18,7 @@ Decision: [#257](https://github.com/apache/asyncband/pull/257).
 
 ## Documentation
 
-Keep each Markdown prose paragraph and list item on one source line.
+Keep each Markdown prose paragraph and list item on one source line. Format Markdown tables so their columns and separators align in the source.
 
 ## Changelog
 

Cargo.toml

@@ -41,7 +41,6 @@ cargo_metadata = { version = "0.23.1" }
 clap = { version = "4.6.5" }
 divan = { version = "0.1.21" }
 flume = { version = "0.12.0", default-features = false }
-pollster = { version = "1.0.1" }
 semver = { version = "1.0.28" }
 serde = { version = "1.0.229" }
 tokio = { version = "1.53.1" }

benchmarks/Cargo.toml

@@ -51,7 +51,6 @@ asyncband = { workspace = true, features = [
 ] }
 divan = { workspace = true }
 flume = { workspace = true, features = ["async"] }
-pollster = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "sync"] }
 waitgroup = { workspace = true }
 

benchmarks/asyncband/main.rs

@@ -23,6 +23,10 @@ mod condvar;
 mod event;
 mod latch;
 mod mpmc;
+
+#[allow(dead_code)]
+#[path = "../mpmc/mod.rs"]
+mod mpmc_support;
 mod mpsc;
 mod mutex;
 mod once;

benchmarks/asyncband/mpmc/bounded.rs

@@ -15,35 +15,41 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use asyncband::mpmc;
 use divan::Bencher;
 use divan::counter::ItemsCount;
 
-use super::support::BATCH_MESSAGES;
-use super::support::BOUNDED_CAPACITY;
-use super::support::ConcurrentBatch;
-use super::support::TOPOLOGIES;
-use super::support::Topology;
+use crate::mpmc_support::adapters::Asyncband;
+use crate::mpmc_support::support::BATCH_MESSAGES;
+use crate::mpmc_support::support::BOUNDED_CAPACITY;
+use crate::mpmc_support::support::Bounded;
+use crate::mpmc_support::support::TOPOLOGIES;
+use crate::mpmc_support::support::TaskBatch;
+use crate::mpmc_support::support::ThreadBatch;
+use crate::mpmc_support::support::Topology;
+use crate::mpmc_support::support::runtime;
 
-fn send(sender: &mpmc::BoundedSender<usize>, value: usize) {
-    pollster::block_on(sender.send(value)).expect("benchmark sender disconnected");
-}
-
-fn recv(receiver: &mpmc::BoundedReceiver<usize>) -> usize {
-    pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected")
+#[divan::bench(
+    args = TOPOLOGIES,
+    sample_count = 20,
+    sample_size = 1,
+    counter = ItemsCount::new(BATCH_MESSAGES),
+)]
+fn blocking_threads(bencher: Bencher, topology: Topology) {
+    bencher
+        .with_inputs(|| ThreadBatch::new_bounded::<Asyncband>(BOUNDED_CAPACITY, topology))
+        .bench_local_refs(|batch| batch.run());
 }
 
 #[divan::bench(
+    consts = [0, 4],
     args = TOPOLOGIES,
     sample_count = 20,
     sample_size = 1,
     counter = ItemsCount::new(BATCH_MESSAGES),
 )]
-fn concurrent(bencher: Bencher, topology: Topology) {
+fn tokio_tasks<const WORKERS: usize>(bencher: Bencher, topology: Topology) {
+    let runtime = runtime(WORKERS);
     bencher
-        .with_inputs(|| {
-            let (sender, receiver) = mpmc::bounded(BOUNDED_CAPACITY);
-            ConcurrentBatch::new(sender, receiver, topology, send, recv)
-        })
-        .bench_local_refs(|batch| batch.run());
+        .with_inputs(|| TaskBatch::new::<Bounded<Asyncband>>(&runtime, topology))
+        .bench_local_refs(|batch| runtime.block_on(batch.run()));
 }

benchmarks/asyncband/mpmc/mod.rs

@@ -16,5 +16,4 @@
 // under the License.
 
 mod bounded;
-mod support;
 mod unbounded;

benchmarks/asyncband/mpmc/unbounded.rs

@@ -15,34 +15,40 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use asyncband::mpmc;
 use divan::Bencher;
 use divan::counter::ItemsCount;
 
-use super::support::BATCH_MESSAGES;
-use super::support::ConcurrentBatch;
-use super::support::TOPOLOGIES;
-use super::support::Topology;
+use crate::mpmc_support::adapters::Asyncband;
+use crate::mpmc_support::support::BATCH_MESSAGES;
+use crate::mpmc_support::support::TOPOLOGIES;
+use crate::mpmc_support::support::TaskBatch;
+use crate::mpmc_support::support::ThreadBatch;
+use crate::mpmc_support::support::Topology;
+use crate::mpmc_support::support::Unbounded;
+use crate::mpmc_support::support::runtime;
 
-fn send(sender: &mpmc::UnboundedSender<usize>, value: usize) {
-    sender.send(value).expect("benchmark sender disconnected");
-}
-
-fn recv(receiver: &mpmc::UnboundedReceiver<usize>) -> usize {
-    pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected")
+#[divan::bench(
+    args = TOPOLOGIES,
+    sample_count = 20,
+    sample_size = 1,
+    counter = ItemsCount::new(BATCH_MESSAGES),
+)]
+fn blocking_threads(bencher: Bencher, topology: Topology) {
+    bencher
+        .with_inputs(|| ThreadBatch::new_unbounded::<Asyncband>(topology))
+        .bench_local_refs(|batch| batch.run());
 }
 
 #[divan::bench(
+    consts = [0, 4],
     args = TOPOLOGIES,
     sample_count = 20,
     sample_size = 1,
     counter = ItemsCount::new(BATCH_MESSAGES),
 )]
-fn concurrent(bencher: Bencher, topology: Topology) {
+fn tokio_tasks<const WORKERS: usize>(bencher: Bencher, topology: Topology) {
+    let runtime = runtime(WORKERS);
     bencher
-        .with_inputs(|| {
-            let (sender, receiver) = mpmc::unbounded();
-            ConcurrentBatch::new(sender, receiver, topology, send, recv)
-        })
-        .bench_local_refs(|batch| batch.run());
+        .with_inputs(|| TaskBatch::new::<Unbounded<Asyncband>>(&runtime, topology))
+        .bench_local_refs(|batch| runtime.block_on(batch.run()));
 }

benchmarks/ecosystem/main.rs

@@ -17,6 +17,10 @@
 
 mod broadcast;
 mod mpmc;
+
+#[allow(dead_code)]
+#[path = "../mpmc/mod.rs"]
+mod mpmc_support;
 mod mpsc;
 mod waitgroup;
 mod watch;