From d11e57c908096e49f7f48aba6a2bd70ed4ee9039 Mon Sep 17 00:00:00 2001 From: tison Date: Fri, 11 Sep 2026 14:38:17 +0800 Subject: [PATCH 1/2] test: benchmark MPMC on Tokio and replace Pollster --- Cargo.lock | 22 -- Cargo.toml | 1 - benchmarks/Cargo.toml | 1 - benchmarks/README.md | 31 ++ benchmarks/asyncband/main.rs | 4 + benchmarks/asyncband/mpmc/bounded.rs | 42 +-- benchmarks/asyncband/mpmc/mod.rs | 1 - benchmarks/asyncband/mpmc/support.rs | 139 --------- benchmarks/asyncband/mpmc/unbounded.rs | 40 +-- benchmarks/ecosystem/main.rs | 4 + benchmarks/ecosystem/mpmc/bounded.rs | 40 ++- benchmarks/ecosystem/mpmc/mod.rs | 2 - benchmarks/ecosystem/mpmc/support.rs | 152 ---------- benchmarks/ecosystem/mpmc/unbounded.rs | 38 ++- benchmarks/ecosystem/mpsc/adapters.rs | 26 +- benchmarks/ecosystem/mpsc/reservation.rs | 3 +- benchmarks/{ecosystem => }/mpmc/adapters.rs | 119 +++++--- benchmarks/mpmc/mod.rs | 19 ++ benchmarks/mpmc/support.rs | 286 ++++++++++++++++++ benchmarks/tests/mpmc_batch.rs | 79 +++-- tests-integration/Cargo.toml | 1 - .../tests/broadcast_mpmc_unbounded_test.rs | 3 +- tests-integration/tests/completion_test.rs | 27 +- tests-integration/tests/oneshot_test/main.rs | 7 +- tests-integration/tests/phaser_test.rs | 3 +- tests-integration/tests/shutdown_test.rs | 21 +- tests-integration/tests/watch_test.rs | 39 +-- 27 files changed, 637 insertions(+), 513 deletions(-) create mode 100644 benchmarks/README.md delete mode 100644 benchmarks/asyncband/mpmc/support.rs delete mode 100644 benchmarks/ecosystem/mpmc/support.rs rename benchmarks/{ecosystem => }/mpmc/adapters.rs (61%) create mode 100644 benchmarks/mpmc/mod.rs create mode 100644 benchmarks/mpmc/support.rs diff --git a/Cargo.lock b/Cargo.lock index c6baa36c..f4d62de3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -105,7 +105,6 @@ dependencies = [ "asyncband", "divan", "flume", - "pollster", "tokio", "waitgroup", ] @@ -679,26 +678,6 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" -[[package]] -name = "pollster" -version = "1.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bc6355899e1c9462875b6757c79f3caa011a1fdae12bbb1a2e72dd1f234f8336" -dependencies = [ - "pollster-macro", -] - -[[package]] -name = "pollster-macro" -version = "1.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "79af046f3496a2db85e4c2fb04f2e9ff77e6b82f29bd43012b398290ce2c811c" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "potential_utf" version = "0.1.5" @@ -975,7 +954,6 @@ name = "tests-integration" version = "0.0.0" dependencies = [ "asyncband", - "pollster", "tokio", "tokio-test", ] diff --git a/Cargo.toml b/Cargo.toml index 05e963ce..8689fca6 100644 --- a/Cargo.toml +++ b/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" } diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 9010ea9e..bcab8f24 100644 --- a/benchmarks/Cargo.toml +++ b/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 } diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 00000000..b236e4c5 --- /dev/null +++ b/benchmarks/README.md @@ -0,0 +1,31 @@ +# MPMC benchmark workloads + +The MPMC benchmarks compare competing queues with the same delivery semantics: asyncband, async-channel, and Flume. Both the asyncband-only and ecosystem suites use the fixtures in `mpmc/` so their execution models stay aligned. + +## Execution models + +| Benchmark | What it measures | +| --- | --- | +| `tokio_tasks<..., 0>` | Producers and consumers spawned on a Tokio current-thread runtime. | +| `tokio_tasks<..., 4>` | Producers and consumers spawned on a Tokio runtime with four worker threads. | +| `blocking_threads` | Dedicated OS threads using asynchronous channel operations through `asyncband::blocking::FutureExt`; unbounded sends use synchronous publication. | + +Use `tokio_tasks` to compare channels used by async tasks. The caller's `Runtime::block_on` only starts the batch and collects task completion; it does not send or receive measured messages. The runtime is reused across samples, while each sample gets a fresh channel and fresh tasks. Consumers compete freely until the last producer drops its sender and the queue drains, with received counts and checksums checked before the batch completes. + +The blocking workload represents synchronous callers bridging into asynchronous channel APIs, such as a pipeline built from dedicated worker threads. It includes the bridge's polling and thread-parking costs and should not be presented as Tokio task performance or as an isolated channel-operation cost. All libraries use the same blocking driver. This workload retains its equal per-consumer message quotas; compare libraries within a workload rather than attributing every difference between workloads solely to their executors. + +Both workloads send 16,384 `usize` values per batch across 1P/1C, 1P/8C, 8P/1C, and 8P/8C topologies. Bounded queues have capacity 64. Channel and worker creation are outside the measured interval; publication, consumption, and start/completion coordination are included. The Tokio workload also includes draining disconnection and joining the tasks. No application work or artificial per-message yield is inserted: on a current-thread runtime, a producer whose sends are always ready can finish its burst before a consumer runs. + +Results describe batch completion time and aggregate message throughput. They do not measure per-message latency, memory retention, arbitrary payload sizes, or a sustained stream reusing one channel. Changing the blocking driver from Pollster to asyncband also changes that composed workload; results from the two driver versions are not a channel-only before/after comparison. + +## Running the comparisons + +Use `cargo x` for the repository workflow and compile both suites before selecting a workload: + +```sh +cargo x bench --no-run +cargo bench --package benchmarks --all-features --bench ecosystem -- '^ecosystem::mpmc::.*tokio_tasks' --sample-count 20 +cargo bench --package benchmarks --all-features --bench ecosystem -- '^ecosystem::mpmc::.*blocking_threads' --sample-count 20 +``` + +The asyncband-only target is `--bench benchmarks` with the `^benchmarks::mpmc::` prefix. Record the commit, toolchain, hardware, execution model, runtime worker count, and sampling settings when reporting results. diff --git a/benchmarks/asyncband/main.rs b/benchmarks/asyncband/main.rs index 549e5e55..a389c0ae 100644 --- a/benchmarks/asyncband/main.rs +++ b/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; diff --git a/benchmarks/asyncband/mpmc/bounded.rs b/benchmarks/asyncband/mpmc/bounded.rs index ba2863be..d4af76fb 100644 --- a/benchmarks/asyncband/mpmc/bounded.rs +++ b/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, value: usize) { - pollster::block_on(sender.send(value)).expect("benchmark sender disconnected"); -} - -fn recv(receiver: &mpmc::BoundedReceiver) -> 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::(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(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::>(&runtime, topology)) + .bench_local_refs(|batch| runtime.block_on(batch.run())); } diff --git a/benchmarks/asyncband/mpmc/mod.rs b/benchmarks/asyncband/mpmc/mod.rs index 551a5993..e0ac8347 100644 --- a/benchmarks/asyncband/mpmc/mod.rs +++ b/benchmarks/asyncband/mpmc/mod.rs @@ -16,5 +16,4 @@ // under the License. mod bounded; -mod support; mod unbounded; diff --git a/benchmarks/asyncband/mpmc/support.rs b/benchmarks/asyncband/mpmc/support.rs deleted file mode 100644 index 554fa1fe..00000000 --- a/benchmarks/asyncband/mpmc/support.rs +++ /dev/null @@ -1,139 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -use std::sync::Arc; -use std::sync::Barrier; -use std::thread; -use std::thread::JoinHandle; - -use divan::black_box; - -pub const BATCH_MESSAGES: usize = 16_384; -pub const BOUNDED_CAPACITY: usize = 64; - -#[derive(Clone, Copy, Debug)] -pub struct Topology { - pub producers: usize, - pub consumers: usize, -} - -pub const TOPOLOGIES: &[Topology] = &[ - Topology { - producers: 1, - consumers: 1, - }, - Topology { - producers: 1, - consumers: 8, - }, - Topology { - producers: 8, - consumers: 1, - }, - Topology { - producers: 8, - consumers: 8, - }, -]; - -pub struct ConcurrentBatch { - start: Arc, - done: Arc, - workers: Vec>, -} - -impl ConcurrentBatch { - pub fn new( - sender: S, - receiver: R, - topology: Topology, - send: fn(&S, usize), - recv: fn(&R) -> usize, - ) -> Self - where - S: Clone + Send + 'static, - R: Clone + Send + 'static, - { - assert_eq!(BATCH_MESSAGES % topology.producers, 0); - assert_eq!(BATCH_MESSAGES % topology.consumers, 0); - - let participants = topology.producers + topology.consumers; - let start = Arc::new(Barrier::new(participants + 1)); - let done = Arc::new(Barrier::new(participants + 1)); - let messages_per_producer = BATCH_MESSAGES / topology.producers; - let messages_per_consumer = BATCH_MESSAGES / topology.consumers; - let mut workers = Vec::with_capacity(participants); - - for producer in 0..topology.producers { - let sender = sender.clone(); - let start = start.clone(); - let done = done.clone(); - workers.push(thread::spawn(move || { - start.wait(); - let first = producer * messages_per_producer; - for offset in 0..messages_per_producer { - send(&sender, black_box(first + offset)); - } - // Close the channel when production ends so pending receivers can finish - // draining it before all workers rendezvous at the completion barrier. - drop(sender); - done.wait(); - })); - } - - for _ in 0..topology.consumers { - let receiver = receiver.clone(); - let start = start.clone(); - let done = done.clone(); - workers.push(thread::spawn(move || { - start.wait(); - let mut checksum = 0usize; - for _ in 0..messages_per_consumer { - checksum = checksum.wrapping_add(recv(&receiver)); - } - black_box(checksum); - done.wait(); - })); - } - - drop(sender); - drop(receiver); - - Self { - start, - done, - workers, - } - } - - pub fn run(&self) { - self.start.wait(); - self.done.wait(); - } -} - -impl Drop for ConcurrentBatch { - fn drop(&mut self) { - let panicking = thread::panicking(); - for worker in self.workers.drain(..) { - let result = worker.join(); - if !panicking { - result.expect("benchmark worker panicked"); - } - } - } -} diff --git a/benchmarks/asyncband/mpmc/unbounded.rs b/benchmarks/asyncband/mpmc/unbounded.rs index 940cc803..965a12ed 100644 --- a/benchmarks/asyncband/mpmc/unbounded.rs +++ b/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, value: usize) { - sender.send(value).expect("benchmark sender disconnected"); -} - -fn recv(receiver: &mpmc::UnboundedReceiver) -> 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::(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(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::>(&runtime, topology)) + .bench_local_refs(|batch| runtime.block_on(batch.run())); } diff --git a/benchmarks/ecosystem/main.rs b/benchmarks/ecosystem/main.rs index f2faee4e..6cc187c2 100644 --- a/benchmarks/ecosystem/main.rs +++ b/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; diff --git a/benchmarks/ecosystem/mpmc/bounded.rs b/benchmarks/ecosystem/mpmc/bounded.rs index 9daa6067..c908b416 100644 --- a/benchmarks/ecosystem/mpmc/bounded.rs +++ b/benchmarks/ecosystem/mpmc/bounded.rs @@ -18,15 +18,18 @@ use divan::Bencher; use divan::counter::ItemsCount; -use super::adapters::AsyncChannel; -use super::adapters::Asyncband; -use super::adapters::BoundedMpmc; -use super::adapters::Flume; -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::AsyncChannel; +use crate::mpmc_support::adapters::Asyncband; +use crate::mpmc_support::adapters::BoundedMpmc; +use crate::mpmc_support::adapters::Flume; +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; #[divan::bench( types = [Asyncband, AsyncChannel, Flume], @@ -35,8 +38,23 @@ use super::support::Topology; sample_size = 1, counter = ItemsCount::new(BATCH_MESSAGES), )] -fn concurrent(bencher: Bencher, topology: Topology) { +fn blocking_threads(bencher: Bencher, topology: Topology) { bencher - .with_inputs(|| ConcurrentBatch::new_bounded::(BOUNDED_CAPACITY, topology)) + .with_inputs(|| ThreadBatch::new_bounded::(BOUNDED_CAPACITY, topology)) .bench_local_refs(|batch| batch.run()); } + +#[divan::bench( + types = [Asyncband, AsyncChannel, Flume], + consts = [0, 4], + args = TOPOLOGIES, + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(BATCH_MESSAGES), +)] +fn tokio_tasks(bencher: Bencher, topology: Topology) { + let runtime = runtime(WORKERS); + bencher + .with_inputs(|| TaskBatch::new::>(&runtime, topology)) + .bench_local_refs(|batch| runtime.block_on(batch.run())); +} diff --git a/benchmarks/ecosystem/mpmc/mod.rs b/benchmarks/ecosystem/mpmc/mod.rs index dd09282b..e0ac8347 100644 --- a/benchmarks/ecosystem/mpmc/mod.rs +++ b/benchmarks/ecosystem/mpmc/mod.rs @@ -15,7 +15,5 @@ // specific language governing permissions and limitations // under the License. -mod adapters; mod bounded; -mod support; mod unbounded; diff --git a/benchmarks/ecosystem/mpmc/support.rs b/benchmarks/ecosystem/mpmc/support.rs deleted file mode 100644 index 48404a70..00000000 --- a/benchmarks/ecosystem/mpmc/support.rs +++ /dev/null @@ -1,152 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -use std::sync::Arc; -use std::sync::Barrier; -use std::thread; -use std::thread::JoinHandle; - -use divan::black_box; - -use super::adapters::BoundedMpmc; -use super::adapters::UnboundedMpmc; - -pub const BATCH_MESSAGES: usize = 16_384; -pub const BOUNDED_CAPACITY: usize = 64; - -#[derive(Clone, Copy, Debug)] -pub struct Topology { - pub producers: usize, - pub consumers: usize, -} - -pub const TOPOLOGIES: &[Topology] = &[ - Topology { - producers: 1, - consumers: 1, - }, - Topology { - producers: 1, - consumers: 8, - }, - Topology { - producers: 8, - consumers: 1, - }, - Topology { - producers: 8, - consumers: 8, - }, -]; - -pub struct ConcurrentBatch { - start: Arc, - done: Arc, - workers: Vec>, -} - -impl ConcurrentBatch { - pub fn new_bounded(capacity: usize, topology: Topology) -> Self { - let (sender, receiver) = C::channel(capacity); - Self::new(sender, receiver, topology, C::send, C::recv) - } - - pub fn new_unbounded(topology: Topology) -> Self { - let (sender, receiver) = C::channel(); - Self::new(sender, receiver, topology, C::send, C::recv) - } - - fn new( - sender: S, - receiver: R, - topology: Topology, - send: fn(&S, usize), - recv: fn(&R) -> usize, - ) -> Self - where - S: Clone + Send + 'static, - R: Clone + Send + 'static, - { - assert_eq!(BATCH_MESSAGES % topology.producers, 0); - assert_eq!(BATCH_MESSAGES % topology.consumers, 0); - - let participants = topology.producers + topology.consumers; - let start = Arc::new(Barrier::new(participants + 1)); - let done = Arc::new(Barrier::new(participants + 1)); - let messages_per_producer = BATCH_MESSAGES / topology.producers; - let messages_per_consumer = BATCH_MESSAGES / topology.consumers; - let mut workers = Vec::with_capacity(participants); - - for producer in 0..topology.producers { - let sender = sender.clone(); - let start = start.clone(); - let done = done.clone(); - workers.push(thread::spawn(move || { - start.wait(); - let first = producer * messages_per_producer; - for offset in 0..messages_per_producer { - send(&sender, black_box(first + offset)); - } - // Close the channel when production ends so pending receivers can finish - // draining it before all workers rendezvous at the completion barrier. - drop(sender); - done.wait(); - })); - } - - for _ in 0..topology.consumers { - let receiver = receiver.clone(); - let start = start.clone(); - let done = done.clone(); - workers.push(thread::spawn(move || { - start.wait(); - let mut checksum = 0usize; - for _ in 0..messages_per_consumer { - checksum = checksum.wrapping_add(recv(&receiver)); - } - black_box(checksum); - done.wait(); - })); - } - - drop(sender); - drop(receiver); - - Self { - start, - done, - workers, - } - } - - pub fn run(&self) { - self.start.wait(); - self.done.wait(); - } -} - -impl Drop for ConcurrentBatch { - fn drop(&mut self) { - let panicking = thread::panicking(); - for worker in self.workers.drain(..) { - let result = worker.join(); - if !panicking { - result.expect("benchmark worker panicked"); - } - } - } -} diff --git a/benchmarks/ecosystem/mpmc/unbounded.rs b/benchmarks/ecosystem/mpmc/unbounded.rs index 4dfc7e11..6013ff55 100644 --- a/benchmarks/ecosystem/mpmc/unbounded.rs +++ b/benchmarks/ecosystem/mpmc/unbounded.rs @@ -18,14 +18,17 @@ use divan::Bencher; use divan::counter::ItemsCount; -use super::adapters::AsyncChannel; -use super::adapters::Asyncband; -use super::adapters::Flume; -use super::adapters::UnboundedMpmc; -use super::support::BATCH_MESSAGES; -use super::support::ConcurrentBatch; -use super::support::TOPOLOGIES; -use super::support::Topology; +use crate::mpmc_support::adapters::AsyncChannel; +use crate::mpmc_support::adapters::Asyncband; +use crate::mpmc_support::adapters::Flume; +use crate::mpmc_support::adapters::UnboundedMpmc; +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; #[divan::bench( types = [Asyncband, AsyncChannel, Flume], @@ -34,8 +37,23 @@ use super::support::Topology; sample_size = 1, counter = ItemsCount::new(BATCH_MESSAGES), )] -fn concurrent(bencher: Bencher, topology: Topology) { +fn blocking_threads(bencher: Bencher, topology: Topology) { bencher - .with_inputs(|| ConcurrentBatch::new_unbounded::(topology)) + .with_inputs(|| ThreadBatch::new_unbounded::(topology)) .bench_local_refs(|batch| batch.run()); } + +#[divan::bench( + types = [Asyncband, AsyncChannel, Flume], + consts = [0, 4], + args = TOPOLOGIES, + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(BATCH_MESSAGES), +)] +fn tokio_tasks(bencher: Bencher, topology: Topology) { + let runtime = runtime(WORKERS); + bencher + .with_inputs(|| TaskBatch::new::>(&runtime, topology)) + .bench_local_refs(|batch| runtime.block_on(batch.run())); +} diff --git a/benchmarks/ecosystem/mpsc/adapters.rs b/benchmarks/ecosystem/mpsc/adapters.rs index f9fe81b6..5fe91df6 100644 --- a/benchmarks/ecosystem/mpsc/adapters.rs +++ b/benchmarks/ecosystem/mpsc/adapters.rs @@ -19,6 +19,8 @@ use std::fmt::Debug; use std::future::Future; use std::task::Context; +use asyncband::blocking::FutureExt; + use crate::support::poll_ready; pub struct Asyncband; @@ -86,11 +88,11 @@ impl BoundedMpsc for Asyncband { } fn send_blocking(sender: &Self::Sender, value: T) { - pollster::block_on(sender.send(value)).unwrap(); + FutureExt::block_on(sender.send(value)).unwrap(); } fn recv_blocking(receiver: &mut Self::Receiver) -> T { - pollster::block_on(receiver.recv()).unwrap() + FutureExt::block_on(receiver.recv()).unwrap() } } @@ -127,11 +129,11 @@ impl BoundedMpsc for Tokio { } fn send_blocking(sender: &Self::Sender, value: T) { - pollster::block_on(sender.send(value)).unwrap(); + FutureExt::block_on(sender.send(value)).unwrap(); } fn recv_blocking(receiver: &mut Self::Receiver) -> T { - pollster::block_on(receiver.recv()).unwrap() + FutureExt::block_on(receiver.recv()).unwrap() } } @@ -168,11 +170,11 @@ impl BoundedMpsc for AsyncChannel { } fn send_blocking(sender: &Self::Sender, value: T) { - pollster::block_on(sender.send(value)).unwrap(); + FutureExt::block_on(sender.send(value)).unwrap(); } fn recv_blocking(receiver: &mut Self::Receiver) -> T { - pollster::block_on(receiver.recv()).unwrap() + FutureExt::block_on(receiver.recv()).unwrap() } } @@ -209,11 +211,11 @@ impl BoundedMpsc for Flume { } fn send_blocking(sender: &Self::Sender, value: T) { - pollster::block_on(sender.send_async(value)).unwrap(); + FutureExt::block_on(sender.send_async(value)).unwrap(); } fn recv_blocking(receiver: &mut Self::Receiver) -> T { - pollster::block_on(receiver.recv_async()).unwrap() + FutureExt::block_on(receiver.recv_async()).unwrap() } } @@ -242,7 +244,7 @@ impl UnboundedMpsc for Asyncband { } fn recv_blocking(receiver: &mut Self::Receiver) -> T { - pollster::block_on(receiver.recv()).unwrap() + FutureExt::block_on(receiver.recv()).unwrap() } } @@ -271,7 +273,7 @@ impl UnboundedMpsc for Tokio { } fn recv_blocking(receiver: &mut Self::Receiver) -> T { - pollster::block_on(receiver.recv()).unwrap() + FutureExt::block_on(receiver.recv()).unwrap() } } @@ -300,7 +302,7 @@ impl UnboundedMpsc for AsyncChannel { } fn recv_blocking(receiver: &mut Self::Receiver) -> T { - pollster::block_on(receiver.recv()).unwrap() + FutureExt::block_on(receiver.recv()).unwrap() } } @@ -329,6 +331,6 @@ impl UnboundedMpsc for Flume { } fn recv_blocking(receiver: &mut Self::Receiver) -> T { - pollster::block_on(receiver.recv_async()).unwrap() + FutureExt::block_on(receiver.recv_async()).unwrap() } } diff --git a/benchmarks/ecosystem/mpsc/reservation.rs b/benchmarks/ecosystem/mpsc/reservation.rs index 32defce4..5900eb1d 100644 --- a/benchmarks/ecosystem/mpsc/reservation.rs +++ b/benchmarks/ecosystem/mpsc/reservation.rs @@ -18,6 +18,7 @@ use std::future::Future; use std::marker::PhantomData; +use asyncband::blocking::FutureExt; use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; @@ -92,7 +93,7 @@ impl ConcurrentMpsc for Reserved usize { C::recv_blocking(receiver) diff --git a/benchmarks/ecosystem/mpmc/adapters.rs b/benchmarks/mpmc/adapters.rs similarity index 61% rename from benchmarks/ecosystem/mpmc/adapters.rs rename to benchmarks/mpmc/adapters.rs index 84f6c12a..209869dd 100644 --- a/benchmarks/ecosystem/mpmc/adapters.rs +++ b/benchmarks/mpmc/adapters.rs @@ -15,26 +15,46 @@ // specific language governing permissions and limitations // under the License. +use std::future::Future; + +use asyncband::blocking::FutureExt; + pub struct Asyncband; pub struct AsyncChannel; pub struct Flume; pub trait BoundedMpmc: Send + Sync + 'static { - type Sender: Clone + Send + 'static; - type Receiver: Clone + Send + 'static; + type Sender: Clone + Send + Sync + 'static; + type Receiver: Clone + Send + Sync + 'static; fn channel(capacity: usize) -> (Self::Sender, Self::Receiver); - fn send(sender: &Self::Sender, value: usize); - fn recv(receiver: &Self::Receiver) -> usize; + fn send_async(sender: &Self::Sender, value: usize) -> impl Future + Send; + fn recv_async(receiver: &Self::Receiver) -> impl Future> + Send; + + fn send(sender: &Self::Sender, value: usize) { + Self::send_async(sender, value).block_on(); + } + + fn recv(receiver: &Self::Receiver) -> usize { + Self::recv_async(receiver) + .block_on() + .expect("benchmark receiver disconnected") + } } pub trait UnboundedMpmc: Send + Sync + 'static { - type Sender: Clone + Send + 'static; - type Receiver: Clone + Send + 'static; + type Sender: Clone + Send + Sync + 'static; + type Receiver: Clone + Send + Sync + 'static; fn channel() -> (Self::Sender, Self::Receiver); fn send(sender: &Self::Sender, value: usize); - fn recv(receiver: &Self::Receiver) -> usize; + fn recv_async(receiver: &Self::Receiver) -> impl Future> + Send; + + fn recv(receiver: &Self::Receiver) -> usize { + Self::recv_async(receiver) + .block_on() + .expect("benchmark receiver disconnected") + } } impl BoundedMpmc for Asyncband { @@ -45,82 +65,91 @@ impl BoundedMpmc for Asyncband { asyncband::mpmc::bounded(capacity) } - fn send(sender: &Self::Sender, value: usize) { - pollster::block_on(sender.send(value)).expect("benchmark sender disconnected"); + async fn send_async(sender: &Self::Sender, value: usize) { + sender + .send(value) + .await + .expect("benchmark sender disconnected"); } - fn recv(receiver: &Self::Receiver) -> usize { - pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") + async fn recv_async(receiver: &Self::Receiver) -> Option { + receiver.recv().await.ok() } } -impl BoundedMpmc for AsyncChannel { - type Receiver = async_channel::Receiver; - type Sender = async_channel::Sender; +impl UnboundedMpmc for Asyncband { + type Receiver = asyncband::mpmc::UnboundedReceiver; + type Sender = asyncband::mpmc::UnboundedSender; - fn channel(capacity: usize) -> (Self::Sender, Self::Receiver) { - async_channel::bounded(capacity) + fn channel() -> (Self::Sender, Self::Receiver) { + asyncband::mpmc::unbounded() } fn send(sender: &Self::Sender, value: usize) { - pollster::block_on(sender.send(value)).expect("benchmark sender disconnected"); + sender.send(value).expect("benchmark sender disconnected"); } - fn recv(receiver: &Self::Receiver) -> usize { - pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") + async fn recv_async(receiver: &Self::Receiver) -> Option { + receiver.recv().await.ok() } } -impl BoundedMpmc for Flume { - type Receiver = flume::Receiver; - type Sender = flume::Sender; +impl BoundedMpmc for AsyncChannel { + type Receiver = async_channel::Receiver; + type Sender = async_channel::Sender; fn channel(capacity: usize) -> (Self::Sender, Self::Receiver) { - flume::bounded(capacity) + async_channel::bounded(capacity) } - fn send(sender: &Self::Sender, value: usize) { - pollster::block_on(sender.send_async(value)).expect("benchmark sender disconnected"); + async fn send_async(sender: &Self::Sender, value: usize) { + sender + .send(value) + .await + .expect("benchmark sender disconnected"); } - fn recv(receiver: &Self::Receiver) -> usize { - pollster::block_on(receiver.recv_async()).expect("benchmark receiver disconnected") + async fn recv_async(receiver: &Self::Receiver) -> Option { + receiver.recv().await.ok() } } -impl UnboundedMpmc for Asyncband { - type Receiver = asyncband::mpmc::UnboundedReceiver; - type Sender = asyncband::mpmc::UnboundedSender; +impl UnboundedMpmc for AsyncChannel { + type Receiver = async_channel::Receiver; + type Sender = async_channel::Sender; fn channel() -> (Self::Sender, Self::Receiver) { - asyncband::mpmc::unbounded() + async_channel::unbounded() } fn send(sender: &Self::Sender, value: usize) { - sender.send(value).expect("benchmark sender disconnected"); + sender + .try_send(value) + .expect("benchmark sender disconnected"); } - fn recv(receiver: &Self::Receiver) -> usize { - pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") + async fn recv_async(receiver: &Self::Receiver) -> Option { + receiver.recv().await.ok() } } -impl UnboundedMpmc for AsyncChannel { - type Receiver = async_channel::Receiver; - type Sender = async_channel::Sender; +impl BoundedMpmc for Flume { + type Receiver = flume::Receiver; + type Sender = flume::Sender; - fn channel() -> (Self::Sender, Self::Receiver) { - async_channel::unbounded() + fn channel(capacity: usize) -> (Self::Sender, Self::Receiver) { + flume::bounded(capacity) } - fn send(sender: &Self::Sender, value: usize) { + async fn send_async(sender: &Self::Sender, value: usize) { sender - .try_send(value) + .send_async(value) + .await .expect("benchmark sender disconnected"); } - fn recv(receiver: &Self::Receiver) -> usize { - pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") + async fn recv_async(receiver: &Self::Receiver) -> Option { + receiver.recv_async().await.ok() } } @@ -136,7 +165,7 @@ impl UnboundedMpmc for Flume { sender.send(value).expect("benchmark sender disconnected"); } - fn recv(receiver: &Self::Receiver) -> usize { - pollster::block_on(receiver.recv_async()).expect("benchmark receiver disconnected") + async fn recv_async(receiver: &Self::Receiver) -> Option { + receiver.recv_async().await.ok() } } diff --git a/benchmarks/mpmc/mod.rs b/benchmarks/mpmc/mod.rs new file mode 100644 index 00000000..440df0ac --- /dev/null +++ b/benchmarks/mpmc/mod.rs @@ -0,0 +1,19 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +pub mod adapters; +pub mod support; diff --git a/benchmarks/mpmc/support.rs b/benchmarks/mpmc/support.rs new file mode 100644 index 00000000..0eb3d714 --- /dev/null +++ b/benchmarks/mpmc/support.rs @@ -0,0 +1,286 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::future::Future; +use std::marker::PhantomData; +use std::sync::Arc; +use std::sync::Barrier; +use std::thread; +use std::thread::JoinHandle; + +use divan::black_box; +use tokio::runtime::Runtime; +use tokio::task::JoinSet; + +use super::adapters::BoundedMpmc; +use super::adapters::UnboundedMpmc; + +pub const BATCH_MESSAGES: usize = 16_384; +pub const BOUNDED_CAPACITY: usize = 64; + +#[derive(Clone, Copy, Debug)] +pub struct Topology { + pub producers: usize, + pub consumers: usize, +} + +pub const TOPOLOGIES: &[Topology] = &[ + Topology { + producers: 1, + consumers: 1, + }, + Topology { + producers: 1, + consumers: 8, + }, + Topology { + producers: 8, + consumers: 1, + }, + Topology { + producers: 8, + consumers: 8, + }, +]; + +pub trait ConcurrentMpmc: Send + Sync + 'static { + type Sender: Clone + Send + Sync + 'static; + type Receiver: Clone + Send + Sync + 'static; + + fn channel() -> (Self::Sender, Self::Receiver); + fn send(sender: &Self::Sender, value: usize) -> impl Future + Send; + fn recv(receiver: &Self::Receiver) -> impl Future> + Send; +} + +pub struct Bounded(PhantomData); + +impl ConcurrentMpmc for Bounded { + type Sender = C::Sender; + type Receiver = C::Receiver; + + fn channel() -> (Self::Sender, Self::Receiver) { + C::channel(CAPACITY) + } + + async fn send(sender: &Self::Sender, value: usize) { + C::send_async(sender, value).await; + } + + async fn recv(receiver: &Self::Receiver) -> Option { + C::recv_async(receiver).await + } +} + +pub struct Unbounded(PhantomData); + +impl ConcurrentMpmc for Unbounded { + type Sender = C::Sender; + type Receiver = C::Receiver; + + fn channel() -> (Self::Sender, Self::Receiver) { + C::channel() + } + + async fn send(sender: &Self::Sender, value: usize) { + C::send(sender, value); + } + + async fn recv(receiver: &Self::Receiver) -> Option { + C::recv_async(receiver).await + } +} + +pub fn runtime(worker_threads: usize) -> Runtime { + if worker_threads == 0 { + tokio::runtime::Builder::new_current_thread() + .build() + .unwrap() + } else { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(worker_threads) + .build() + .unwrap() + } +} + +// The caller only coordinates the batch. All measured sends and receives run in spawned tasks, +// including on the current-thread runtime; no data is received by Runtime::block_on itself. +pub struct TaskBatch { + start: Arc, + workers: JoinSet<(usize, usize)>, +} + +impl TaskBatch { + pub fn new(runtime: &Runtime, topology: Topology) -> Self { + assert_eq!(BATCH_MESSAGES % topology.producers, 0); + let (sender, receiver) = C::channel(); + let start = Arc::new(tokio::sync::Barrier::new( + topology.producers + topology.consumers + 1, + )); + let messages_per_producer = BATCH_MESSAGES / topology.producers; + let mut workers = JoinSet::new(); + for producer in 0..topology.producers { + let sender = sender.clone(); + let start = start.clone(); + workers.spawn_on( + async move { + start.wait().await; + let first = producer * messages_per_producer; + for value in first..first + messages_per_producer { + C::send(&sender, black_box(value)).await; + } + // Completion drops this sender so receivers can observe the end of input. + (0, 0) + }, + runtime.handle(), + ); + } + for _ in 0..topology.consumers { + let receiver = receiver.clone(); + let start = start.clone(); + workers.spawn_on( + async move { + start.wait().await; + let mut count = 0; + let mut checksum = 0usize; + // Competing consumers drain freely, rather than stopping at equal quotas. + while let Some(value) = C::recv(&receiver).await { + count += 1; + checksum = checksum.wrapping_add(value); + } + (count, checksum) + }, + runtime.handle(), + ); + } + drop(sender); + drop(receiver); + Self { start, workers } + } + + pub async fn run(&mut self) -> (usize, usize) { + self.start.wait().await; + let mut count = 0; + let mut checksum = 0usize; + while let Some(result) = self.workers.join_next().await { + let (received, sum) = result.expect("benchmark task panicked"); + count += received; + checksum = checksum.wrapping_add(sum); + } + assert_eq!(count, BATCH_MESSAGES); + assert_eq!(checksum, BATCH_MESSAGES * (BATCH_MESSAGES - 1) / 2); + black_box((count, checksum)) + } +} + +pub struct ThreadBatch { + start: Arc, + done: Arc, + workers: Vec>, +} + +impl ThreadBatch { + pub fn new_bounded(capacity: usize, topology: Topology) -> Self { + let (sender, receiver) = C::channel(capacity); + Self::new(sender, receiver, topology, C::send, C::recv) + } + + pub fn new_unbounded(topology: Topology) -> Self { + let (sender, receiver) = C::channel(); + Self::new(sender, receiver, topology, C::send, C::recv) + } + + fn new( + sender: S, + receiver: R, + topology: Topology, + send: fn(&S, usize), + recv: fn(&R) -> usize, + ) -> Self + where + S: Clone + Send + 'static, + R: Clone + Send + 'static, + { + assert_eq!(BATCH_MESSAGES % topology.producers, 0); + assert_eq!(BATCH_MESSAGES % topology.consumers, 0); + + let participants = topology.producers + topology.consumers; + let start = Arc::new(Barrier::new(participants + 1)); + let done = Arc::new(Barrier::new(participants + 1)); + let messages_per_producer = BATCH_MESSAGES / topology.producers; + let messages_per_consumer = BATCH_MESSAGES / topology.consumers; + let mut workers = Vec::with_capacity(participants); + + for producer in 0..topology.producers { + let sender = sender.clone(); + let start = start.clone(); + let done = done.clone(); + workers.push(thread::spawn(move || { + start.wait(); + let first = producer * messages_per_producer; + for offset in 0..messages_per_producer { + send(&sender, black_box(first + offset)); + } + // Close the channel when production ends so pending receivers can finish + // draining it before all workers rendezvous at the completion barrier. + drop(sender); + done.wait(); + })); + } + + for _ in 0..topology.consumers { + let receiver = receiver.clone(); + let start = start.clone(); + let done = done.clone(); + workers.push(thread::spawn(move || { + start.wait(); + let mut checksum = 0usize; + for _ in 0..messages_per_consumer { + checksum = checksum.wrapping_add(recv(&receiver)); + } + black_box(checksum); + done.wait(); + })); + } + + drop(sender); + drop(receiver); + + Self { + start, + done, + workers, + } + } + + pub fn run(&self) { + self.start.wait(); + self.done.wait(); + } +} + +impl Drop for ThreadBatch { + fn drop(&mut self) { + let panicking = thread::panicking(); + for worker in self.workers.drain(..) { + let result = worker.join(); + if !panicking { + result.expect("benchmark worker panicked"); + } + } + } +} diff --git a/benchmarks/tests/mpmc_batch.rs b/benchmarks/tests/mpmc_batch.rs index 2d855998..d4362790 100644 --- a/benchmarks/tests/mpmc_batch.rs +++ b/benchmarks/tests/mpmc_batch.rs @@ -20,14 +20,11 @@ use std::thread; use std::time::Duration; #[allow(dead_code)] -#[path = "../ecosystem/mpmc/adapters.rs"] -mod adapters; -#[allow(dead_code)] -#[path = "../asyncband/mpmc/support.rs"] -mod asyncband_support; -#[allow(dead_code)] -#[path = "../ecosystem/mpmc/support.rs"] -mod ecosystem_support; +#[path = "../mpmc/mod.rs"] +mod mpmc_support; + +use mpmc_support::adapters; +use mpmc_support::support; // Deliberately drain only after the last producer drops its sender. This makes // retaining senders across the completion barrier deadlock deterministically. @@ -45,6 +42,13 @@ impl adapters::UnboundedMpmc for DrainAfterClose { sender.send(value).unwrap(); } + async fn recv_async(receiver: &Self::Receiver) -> Option { + while !receiver.is_disconnected() { + tokio::task::yield_now().await; + } + receiver.try_recv().ok() + } + fn recv(receiver: &Self::Receiver) -> usize { while !receiver.is_disconnected() { thread::sleep(Duration::from_millis(1)); @@ -66,46 +70,55 @@ fn assert_completes(run: impl FnOnce() + Send + 'static) { } #[test] -fn ecosystem_batch_closes_before_waiting_for_consumers() { +fn thread_batch_closes_before_waiting_for_consumers() { assert_completes(|| { - for &topology in ecosystem_support::TOPOLOGIES { - let batch = - ecosystem_support::ConcurrentBatch::new_unbounded::(topology); + for &topology in support::TOPOLOGIES { + let batch = support::ThreadBatch::new_unbounded::(topology); batch.run(); } }); } #[test] -fn asyncband_batch_closes_before_waiting_for_consumers() { - use adapters::UnboundedMpmc; - +fn flume_batches_complete_with_competing_consumers() { assert_completes(|| { - for &topology in asyncband_support::TOPOLOGIES { - let (sender, receiver) = DrainAfterClose::channel(); - let batch = asyncband_support::ConcurrentBatch::new( - sender, - receiver, - topology, - DrainAfterClose::send, - DrainAfterClose::recv, - ); + for _ in 0..100 { + let batch = support::ThreadBatch::new_unbounded::(support::Topology { + producers: 1, + consumers: 8, + }); batch.run(); } }); } #[test] -fn flume_batches_complete_with_competing_consumers() { +fn tokio_batches_drain_all_messages_before_completion() { + fn check(runtime: &tokio::runtime::Runtime) { + for topology in support::TOPOLOGIES + .iter() + .copied() + .chain([support::Topology { + producers: 1, + consumers: 3, + }]) + { + // Three consumers cannot receive equal quotas from a 16,384-message batch. + let mut batch = support::TaskBatch::new::(runtime, topology); + runtime.block_on(batch.run()); + } + } + assert_completes(|| { - for _ in 0..100 { - let batch = ecosystem_support::ConcurrentBatch::new_unbounded::( - ecosystem_support::Topology { - producers: 1, - consumers: 8, - }, - ); - batch.run(); + for workers in [0, 4] { + let runtime = support::runtime(workers); + check::>(&runtime); + check::>(&runtime); + check::>(&runtime); + check::>(&runtime); + check::>(&runtime); + check::>(&runtime); + check::>(&runtime); } }); } diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index ab53ef3d..44942255 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -51,7 +51,6 @@ asyncband = { workspace = true, features = [ "waitgroup", "watch", ] } -pollster = { workspace = true, features = ["macro"] } tokio-test = { workspace = true } [lib] diff --git a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs b/tests-integration/tests/broadcast_mpmc_unbounded_test.rs index ad200695..afd1d779 100644 --- a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs +++ b/tests-integration/tests/broadcast_mpmc_unbounded_test.rs @@ -23,6 +23,7 @@ use std::task::Poll; use std::task::Waker; use std::thread; +use asyncband::blocking::FutureExt; use asyncband::broadcast::mpmc::*; use tests_integration::WakeCounter; use tests_integration::assert_completes_without_deadlock; @@ -454,7 +455,7 @@ fn concurrent_senders_deliver_every_message_to_every_receiver() { .map(|mut receiver| { thread::spawn(move || { let mut seen = vec![]; - while let Ok(value) = pollster::block_on(receiver.recv()) { + while let Ok(value) = FutureExt::block_on(receiver.recv()) { seen.push(value); } seen diff --git a/tests-integration/tests/completion_test.rs b/tests-integration/tests/completion_test.rs index 106d36de..8200e23b 100644 --- a/tests-integration/tests/completion_test.rs +++ b/tests-integration/tests/completion_test.rs @@ -22,6 +22,7 @@ use std::task::Poll; use std::task::Waker; use std::thread; +use asyncband::blocking::FutureExt; use asyncband::completion; use tests_integration::PanicWake; use tests_integration::WakeCounter; @@ -41,16 +42,16 @@ fn all_observers_borrow_the_same_non_clone_value() { assert!(completer.complete(NotClone(String::from("ready"))).is_ok()); - let first_value = pollster::block_on(first.wait()).unwrap(); - let second_value = pollster::block_on(second.wait()).unwrap(); - let repeated = pollster::block_on(first.wait()).unwrap(); + let first_value = FutureExt::block_on(first.wait()).unwrap(); + let second_value = FutureExt::block_on(second.wait()).unwrap(); + let repeated = FutureExt::block_on(first.wait()).unwrap(); assert_eq!(first_value.0.as_str(), "ready"); assert!(std::ptr::eq(first_value, second_value)); assert!(std::ptr::eq(first_value, repeated)); let late = first.clone(); drop(first); - let late_value = pollster::block_on(late.wait()).unwrap(); + let late_value = FutureExt::block_on(late.wait()).unwrap(); assert!(std::ptr::eq(second_value, late_value)); } @@ -60,7 +61,7 @@ fn completer_transfers_a_send_only_value_between_threads() { let worker = thread::spawn(move || completer.complete(Cell::new(7))); - assert_eq!(pollster::block_on(completion.wait()).unwrap().get(), 7); + assert_eq!(FutureExt::block_on(completion.wait()).unwrap().get(), 7); worker.join().unwrap().unwrap(); } @@ -113,7 +114,7 @@ fn abandonment_wakes_registered_waits_and_is_visible_to_late_observers() { )); let late = first.clone(); - assert!(pollster::block_on(late.wait()).is_err()); + assert!(FutureExt::block_on(late.wait()).is_err()); } #[test] @@ -121,13 +122,13 @@ fn payload_errors_remain_distinct_from_abandonment() { let (completer, completion) = completion::new::>(); completer.complete(Err("domain error")).unwrap(); assert_eq!( - pollster::block_on(completion.wait()), + FutureExt::block_on(completion.wait()), Ok(&Err("domain error")) ); let (completer, completion) = completion::new::>(); drop(completer); - assert!(pollster::block_on(completion.wait()).is_err()); + assert!(FutureExt::block_on(completion.wait()).is_err()); } #[test] @@ -169,7 +170,7 @@ fn cancelling_after_wake_does_not_consume_the_shared_result() { assert_eq!(tracker.count(), 1); drop(wait); - assert_eq!(pollster::block_on(second.wait()), Ok(&9)); + assert_eq!(FutureExt::block_on(second.wait()), Ok(&9)); } #[test] @@ -186,7 +187,7 @@ fn cancelling_after_abandonment_does_not_retain_the_waker() { assert_eq!(Arc::strong_count(&tracker), baseline); drop(wait); assert_eq!(Arc::strong_count(&tracker), baseline); - assert!(pollster::block_on(completion.wait()).is_err()); + assert!(FutureExt::block_on(completion.wait()).is_err()); } #[test] @@ -217,7 +218,7 @@ fn wake_callbacks_run_outside_the_completion_lock() { let (completer, completion) = completion::new(); let callback_completion = completion.clone(); let waker = waker_on_wake(move || { - assert_eq!(pollster::block_on(callback_completion.wait()), Ok(&13)); + assert_eq!(FutureExt::block_on(callback_completion.wait()), Ok(&13)); }); let mut wait = Box::pin(completion.wait()); @@ -229,7 +230,7 @@ fn wake_callbacks_run_outside_the_completion_lock() { let (completer, completion) = completion::new::(); let callback_completion = completion.clone(); let waker = waker_on_wake(move || { - assert!(pollster::block_on(callback_completion.wait()).is_err()); + assert!(FutureExt::block_on(callback_completion.wait()).is_err()); }); let mut wait = Box::pin(completion.wait()); assert!(poll_with(wait.as_mut(), &waker).is_pending()); @@ -270,7 +271,7 @@ fn cancelled_wakers_are_dropped_outside_the_completion_lock() { assert!(poll_with(wait.as_mut(), &waker).is_pending()); drop(waker); drop(wait); - assert!(pollster::block_on(completion.wait()).is_err()); + assert!(FutureExt::block_on(completion.wait()).is_err()); }); } diff --git a/tests-integration/tests/oneshot_test/main.rs b/tests-integration/tests/oneshot_test/main.rs index 98e60bb1..bd3eba01 100644 --- a/tests-integration/tests/oneshot_test/main.rs +++ b/tests-integration/tests/oneshot_test/main.rs @@ -30,6 +30,7 @@ use std::sync::atomic::Ordering; use std::task::Context; use std::task::Poll; +use asyncband::blocking::FutureExt; use asyncband::oneshot; use asyncband::oneshot::TryRecvError; @@ -46,7 +47,7 @@ fn send_before_await() { assert!(!receiver.has_message()); assert!(sender.send(19i128).is_ok()); assert!(receiver.has_message()); - assert_eq!(pollster::block_on(receiver), Ok(19i128)); + assert_eq!(FutureExt::block_on(receiver), Ok(19i128)); } #[test] @@ -56,7 +57,7 @@ fn await_with_dropped_sender() { drop(sender); assert!(receiver.is_disconnected()); assert_eq!( - pollster::block_on(receiver), + FutureExt::block_on(receiver), Err(oneshot::RecvError::Disconnected) ); } @@ -72,7 +73,7 @@ fn try_recv_success_then_disconnected() { assert!(rx.is_disconnected()); assert!(!rx.has_message()); assert_eq!( - pollster::block_on(rx.into_future()), + FutureExt::block_on(rx.into_future()), Err(oneshot::RecvError::Disconnected) ); } diff --git a/tests-integration/tests/phaser_test.rs b/tests-integration/tests/phaser_test.rs index 9730c859..09c5bf20 100644 --- a/tests-integration/tests/phaser_test.rs +++ b/tests-integration/tests/phaser_test.rs @@ -25,6 +25,7 @@ use std::task::Poll; use std::task::Wake; use std::task::Waker; +use asyncband::blocking::FutureExt; use asyncband::phaser::Phaser; fn poll_once(future: std::pin::Pin<&mut F>) -> Poll { @@ -748,7 +749,7 @@ fn arrivals_publish_each_workers_writes_across_threads() { for (id, mut participant) in participants.enumerate() { let values = &values; scope.spawn(move || { - pollster::block_on(async { + FutureExt::block_on(async { for round in 1..=16 { values[id].store(round, Ordering::Relaxed); participant.wait().await.unwrap(); diff --git a/tests-integration/tests/shutdown_test.rs b/tests-integration/tests/shutdown_test.rs index 1aaadfd8..db36080b 100644 --- a/tests-integration/tests/shutdown_test.rs +++ b/tests-integration/tests/shutdown_test.rs @@ -18,6 +18,7 @@ use std::future::pending; use std::pin::pin; +use asyncband::blocking::FutureExt; use asyncband::shutdown::*; use tests_integration::poll_once; use tests_integration::test_runtime; @@ -26,8 +27,8 @@ use tests_integration::test_runtime; fn test_single_pair() { let (shutdown, guard) = new(); let handle = test_runtime().spawn(async move { guard.shutdown_requested().await }); - pollster::block_on(shutdown); - pollster::block_on(handle).unwrap(); + FutureExt::block_on(shutdown); + FutureExt::block_on(handle).unwrap(); } #[test] @@ -38,7 +39,7 @@ fn test_multiple_tasks() { test_runtime().spawn(async move { guard.shutdown_requested().await }); } drop(guard); - pollster::block_on(shutdown); + FutureExt::block_on(shutdown); } #[test] @@ -51,8 +52,8 @@ fn test_multiple_control_handles() { drop(guard); let shutdown_clone = shutdown.clone(); shutdown.request_shutdown(); - pollster::block_on(shutdown); - pollster::block_on(shutdown_clone); + FutureExt::block_on(shutdown); + FutureExt::block_on(shutdown_clone); } #[test] @@ -82,7 +83,7 @@ fn test_shutdown_requested_owned_does_not_capture_self() { _ = run_state(&mut state) => (), } }); - pollster::block_on(shutdown); + FutureExt::block_on(shutdown); } #[test] @@ -90,7 +91,7 @@ fn test_watch_does_not_block_completion() { let (shutdown, guard) = new(); let watch = guard.into_watch(); - pollster::block_on(shutdown); + FutureExt::block_on(shutdown); assert!(watch.is_shutdown_requested()); } @@ -131,7 +132,7 @@ fn test_dropping_polled_shutdown_keeps_request_sticky() { #[test] fn test_disabled_select_branch_does_not_request_shutdown() { - pollster::block_on(async { + FutureExt::block_on(async { let (shutdown, guard) = new(); let watch = guard.watch(); @@ -151,6 +152,6 @@ fn test_watch_observes_shutdown_request() { let handle = test_runtime().spawn(async move { watch.shutdown_requested().await }); drop(guard); - pollster::block_on(shutdown); - pollster::block_on(handle).unwrap(); + FutureExt::block_on(shutdown); + FutureExt::block_on(handle).unwrap(); } diff --git a/tests-integration/tests/watch_test.rs b/tests-integration/tests/watch_test.rs index 72d8ddbc..994eda02 100644 --- a/tests-integration/tests/watch_test.rs +++ b/tests-integration/tests/watch_test.rs @@ -21,6 +21,7 @@ use std::sync::atomic::Ordering; use std::task::Poll; use std::task::Waker; +use asyncband::blocking::FutureExt; use asyncband::watch; use tests_integration::PanicWake; use tests_integration::WakeCounter; @@ -69,7 +70,7 @@ fn initial_value_is_observed_and_updates_coalesce() { tx.send(2).unwrap(); assert_eq!(rx.has_changed(), Ok(true)); - assert_eq!(pollster::block_on(rx.recv()).unwrap(), 2); + assert_eq!(FutureExt::block_on(rx.recv()).unwrap(), 2); assert_eq!(rx.has_changed(), Ok(false)); } @@ -80,7 +81,7 @@ fn equal_values_still_create_a_new_version() { tx.send(1).unwrap(); assert_eq!(rx.has_changed(), Ok(true)); - assert_eq!(pollster::block_on(rx.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(rx.recv()).unwrap(), 1); } #[test] @@ -90,7 +91,7 @@ fn get_does_not_consume_but_recv_does() { assert_eq!(rx.get(), 1); assert_eq!(rx.has_changed(), Ok(true)); - assert_eq!(pollster::block_on(rx.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(rx.recv()).unwrap(), 1); assert_eq!(rx.has_changed(), Ok(false)); } @@ -109,12 +110,12 @@ fn panicking_clone_leaves_the_update_unseen() { panic_next.store(true, Ordering::Relaxed); let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - pollster::block_on(rx.recv()) + FutureExt::block_on(rx.recv()) })); assert!(result.is_err()); assert_eq!(rx.has_changed(), Ok(true)); - assert_eq!(pollster::block_on(rx.recv()).unwrap().value, 1); + assert_eq!(FutureExt::block_on(rx.recv()).unwrap().value, 1); } #[test] @@ -123,13 +124,13 @@ fn cloned_receivers_inherit_then_advance_independently() { tx.send(1).unwrap(); let mut second = first.clone(); - assert_eq!(pollster::block_on(first.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(first.recv()).unwrap(), 1); assert_eq!(first.has_changed(), Ok(false)); assert_eq!(second.has_changed(), Ok(true)); - assert_eq!(pollster::block_on(second.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(second.recv()).unwrap(), 1); tx.send(2).unwrap(); - assert_eq!(pollster::block_on(first.recv()).unwrap(), 2); + assert_eq!(FutureExt::block_on(first.recv()).unwrap(), 2); assert_eq!(second.has_changed(), Ok(true)); } @@ -143,7 +144,7 @@ fn subscriptions_start_at_the_current_version() { assert_eq!(subscribed.has_changed(), Ok(false)); tx.send(2).unwrap(); - assert_eq!(pollster::block_on(subscribed.recv()).unwrap(), 2); + assert_eq!(FutureExt::block_on(subscribed.recv()).unwrap(), 2); } #[test] @@ -157,12 +158,12 @@ fn final_unseen_value_is_reported_before_disconnection() { assert_eq!(first.has_changed(), Ok(true)); assert_eq!(first.get(), 1); assert_eq!(first.has_changed(), Ok(true)); - assert_eq!(pollster::block_on(first.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(first.recv()).unwrap(), 1); assert_eq!(first.has_changed(), Err(watch::RecvError::Disconnected)); - assert_eq!(pollster::block_on(second.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(second.recv()).unwrap(), 1); assert_eq!( - pollster::block_on(second.recv()), + FutureExt::block_on(second.recv()), Err(watch::RecvError::Disconnected) ); } @@ -181,7 +182,7 @@ fn sending_without_receivers_returns_the_value_and_preserves_current() { assert_eq!(replacement.has_changed(), Ok(false)); tx.send(String::from("accepted")).unwrap(); - assert_eq!(pollster::block_on(replacement.recv()).unwrap(), "accepted"); + assert_eq!(FutureExt::block_on(replacement.recv()).unwrap(), "accepted"); } #[test] @@ -189,7 +190,7 @@ fn send_replace_returns_previous_and_publishes_without_receivers() { let (tx, mut rx) = watch::channel(String::from("initial")); assert_eq!(tx.send_replace(String::from("first")), "initial"); - assert_eq!(pollster::block_on(rx.recv()).unwrap(), "first"); + assert_eq!(FutureExt::block_on(rx.recv()).unwrap(), "first"); drop(rx); assert_eq!(tx.send_replace(String::from("retained")), "first"); @@ -199,7 +200,7 @@ fn send_replace_returns_previous_and_publishes_without_receivers() { assert_eq!(subscribed.has_changed(), Ok(false)); assert_eq!(tx.send_replace(String::from("next")), "retained"); - assert_eq!(pollster::block_on(subscribed.recv()).unwrap(), "next"); + assert_eq!(FutureExt::block_on(subscribed.recv()).unwrap(), "next"); } #[test] @@ -217,7 +218,7 @@ fn cancelling_changed_releases_its_waker_without_consuming() { tx.send(1).unwrap(); assert_eq!(tracker.count(), 0); - assert_eq!(pollster::block_on(rx.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(rx.recv()).unwrap(), 1); } #[test] @@ -232,7 +233,7 @@ fn cancelling_after_wake_still_leaves_the_update_unseen() { assert_eq!(tracker.count(), 1); drop(changed); - assert_eq!(pollster::block_on(rx.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(rx.recv()).unwrap(), 1); } #[test] @@ -247,7 +248,7 @@ fn cancelling_recv_after_wake_still_leaves_the_update_unseen() { assert_eq!(tracker.count(), 1); drop(recv); - assert_eq!(pollster::block_on(rx.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(rx.recv()).unwrap(), 1); } #[test] @@ -334,7 +335,7 @@ fn dropping_a_stale_changed_future_keeps_a_new_waiter_registered() { tx.send(1).unwrap(); assert_eq!(first_tracker.count(), 1); - assert_eq!(pollster::block_on(second.recv()).unwrap(), 1); + assert_eq!(FutureExt::block_on(second.recv()).unwrap(), 1); let second_tracker = Arc::new(WakeCounter::default()); let second_waker = Waker::from(second_tracker.clone()); let mut second_changed = Box::pin(second.changed()); From 6e6a492308e128e2879ed2ad54995d8e04e2eb56 Mon Sep 17 00:00:00 2001 From: tison Date: Fri, 11 Sep 2026 14:48:55 +0800 Subject: [PATCH 2/2] docs: clarify Markdown formatting and remove benchmark README --- AGENTS.md | 2 +- benchmarks/README.md | 31 ------------------------------- 2 files changed, 1 insertion(+), 32 deletions(-) delete mode 100644 benchmarks/README.md diff --git a/AGENTS.md b/AGENTS.md index 9ee884c1..c2e722d7 100644 --- a/AGENTS.md +++ b/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 diff --git a/benchmarks/README.md b/benchmarks/README.md deleted file mode 100644 index b236e4c5..00000000 --- a/benchmarks/README.md +++ /dev/null @@ -1,31 +0,0 @@ -# MPMC benchmark workloads - -The MPMC benchmarks compare competing queues with the same delivery semantics: asyncband, async-channel, and Flume. Both the asyncband-only and ecosystem suites use the fixtures in `mpmc/` so their execution models stay aligned. - -## Execution models - -| Benchmark | What it measures | -| --- | --- | -| `tokio_tasks<..., 0>` | Producers and consumers spawned on a Tokio current-thread runtime. | -| `tokio_tasks<..., 4>` | Producers and consumers spawned on a Tokio runtime with four worker threads. | -| `blocking_threads` | Dedicated OS threads using asynchronous channel operations through `asyncband::blocking::FutureExt`; unbounded sends use synchronous publication. | - -Use `tokio_tasks` to compare channels used by async tasks. The caller's `Runtime::block_on` only starts the batch and collects task completion; it does not send or receive measured messages. The runtime is reused across samples, while each sample gets a fresh channel and fresh tasks. Consumers compete freely until the last producer drops its sender and the queue drains, with received counts and checksums checked before the batch completes. - -The blocking workload represents synchronous callers bridging into asynchronous channel APIs, such as a pipeline built from dedicated worker threads. It includes the bridge's polling and thread-parking costs and should not be presented as Tokio task performance or as an isolated channel-operation cost. All libraries use the same blocking driver. This workload retains its equal per-consumer message quotas; compare libraries within a workload rather than attributing every difference between workloads solely to their executors. - -Both workloads send 16,384 `usize` values per batch across 1P/1C, 1P/8C, 8P/1C, and 8P/8C topologies. Bounded queues have capacity 64. Channel and worker creation are outside the measured interval; publication, consumption, and start/completion coordination are included. The Tokio workload also includes draining disconnection and joining the tasks. No application work or artificial per-message yield is inserted: on a current-thread runtime, a producer whose sends are always ready can finish its burst before a consumer runs. - -Results describe batch completion time and aggregate message throughput. They do not measure per-message latency, memory retention, arbitrary payload sizes, or a sustained stream reusing one channel. Changing the blocking driver from Pollster to asyncband also changes that composed workload; results from the two driver versions are not a channel-only before/after comparison. - -## Running the comparisons - -Use `cargo x` for the repository workflow and compile both suites before selecting a workload: - -```sh -cargo x bench --no-run -cargo bench --package benchmarks --all-features --bench ecosystem -- '^ecosystem::mpmc::.*tokio_tasks' --sample-count 20 -cargo bench --package benchmarks --all-features --bench ecosystem -- '^ecosystem::mpmc::.*blocking_threads' --sample-count 20 -``` - -The asyncband-only target is `--bench benchmarks` with the `^benchmarks::mpmc::` prefix. Record the commit, toolchain, hardware, execution model, runtime worker count, and sampling settings when reporting results.