refactor(benchmarks): measure fixed workloads in milliseconds

Cargo.toml

@@ -39,7 +39,6 @@ async-broadcast = { version = "0.7.2" }
 async-channel = { version = "2.5.0" }
 cargo_metadata = { version = "0.23.1" }
 clap = { version = "4.6.5" }
-divan = { version = "0.1.21" }
 flume = { version = "0.12.0", default-features = false }
 semver = { version = "1.0.28" }
 serde = { version = "1.0.229" }

README.md

@@ -79,6 +79,10 @@ Public paths stay direct—such as `asyncband::mutex`, `asyncband::pool`, and `a
 
 Runnable examples live in the [`examples`](examples) workspace crate. They demonstrate how to choose and compose Asyncband primitives in complete programs.
 
+## Benchmarks
+
+Use `cargo x bench` to run fixed workloads measured in milliseconds. See the [benchmark guide](benchmarks/README.md) for filtering, repeated comparisons, workload sizes, and timing boundaries.
+
 ## API map
 
 | Area                 | API                                                                                            | Feature        | Use                                                                                                             |

benchmarks/Cargo.toml

@@ -49,7 +49,7 @@ asyncband = { workspace = true, features = [
   "waitgroup",
   "watch",
 ] }
-divan = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
 flume = { workspace = true, features = ["async"] }
 tokio = { workspace = true, features = ["rt-multi-thread", "sync"] }
 
@@ -60,7 +60,6 @@ waitgroup = { workspace = true }
 [lib]
 bench = false
 doctest = false
-test = false
 
 [[bench]]
 harness = false

benchmarks/README.md

@@ -0,0 +1,67 @@
+<!--
+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.
+-->
+
+# Benchmarks
+
+These benchmarks follow the approach in [Benchmark In Milliseconds](https://matklad.github.io/2026/10/05/benchmark-milliseconds.html): run a fixed amount of useful work, aim for roughly 300 ms, and repeat it to develop intuition about a change. Work sizes are checked into the source. The runner never adjusts them in response to measured performance.
+
+## Running
+
+Use `cargo x bench` to build and run the release-profile benchmark executables. Select a target and filter cases for a short feedback loop:
+
+```shell
+# Discover the commands and available workloads.
+cargo x bench --help
+cargo x bench --bench primitives -- --help
+cargo x bench -- --list
+
+# Repeat one primitive workload and inspect every elapsed time.
+cargo x bench --bench primitives -- mutex::lock::uncontended_reuse --repeat 10
+
+# Compare all implementations of the same channel workload.
+cargo x bench --bench ecosystem -- mpsc::bounded::try_send_receive --repeat 10
+
+# Compile only, or exercise every case once per worker without timing.
+cargo x bench --no-run
+cargo x bench -- --test
+```
+
+Filters are case-sensitive substrings; multiple filters must all match. Use `--bench primitives` or `--bench ecosystem` with filters, since each executable rejects an empty selection. `--list` shows the resolved round counts without constructing fixtures or starting workers.
+
+Every measured row prints the integer elapsed milliseconds, total rounds, case name, and repetition number. The default is one measurement per case; `--repeat 10` prints ten independent runs, including the first. There is no hidden warm-up run, averaging, outlier removal, or conversion to a per-operation time.
+
+## Work and timing boundaries
+
+`primitives` exercises operation cycles with controlled polling and reference-counted wakers. It covers ready paths, registration, handoff, cancellation, fan-in and fan-out; selected cases also exercise shared state from multiple OS threads. Hand-polled cases include waker bookkeeping but do not model executor scheduling.
+
+`ecosystem` compares equivalent workloads across Asyncband and other libraries. It includes hand-polled paths, message batches on threads or Tokio tasks, bounded backpressure, retained backlogs, payload sizes, and notification fan-out. Existing semantic differences remain explicit: for example, Tokio broadcast does not participate in lossless bounded backpressure cases.
+
+A round is one invocation of the case's operation closure. It may transfer one message, complete a group of waiters, or run a batch of messages. Channel batches use the message counts in their support modules. The `threads=N` variants divide the total rounds among N workers and start them at a barrier; they share the same fixture and do the same total work at every thread count. Other concurrency settings are part of the case name and describe the workers inside each round.
+
+The clock surrounds the entire workload function, from fixture construction to cleanup. Outputs are consumed and dropped each round. Reusable fixtures survive across rounds, while cases that need fresh state construct and dispose of it inside the loop. Channel workloads that prime reusable workers or storage do so once inside this measured interval. Setup, teardown, allocation, key generation, coordination, and correctness checks are not separately subtracted. This makes the result a workload cost; it does not isolate the nanosecond cost of the operation in the function name.
+
+Each repetition recreates the workload's fixture. Repeated rounds must return it to a usable state: drain messages, release permits, finish workers, and keep retained memory bounded. The smoke mode keeps each case's topology and inner message batch intact while reducing its outer loop to one round per worker.
+
+## Sizing and comparing
+
+Choose round counts that put representative cases in the hundreds of milliseconds on a development machine. Treat 300 ms as a sizing guide, not a deadline or an assertion: implementations, contention, hardware, and schedulers differ. A result below 10 ms needs more work before small differences are useful; a result above a second usually needs fewer rounds for convenient iteration.
+
+Use `--rounds N` to try a different fixed count on a filtered workload, or `--scale N` to multiply the checked-in counts. These options do not change message capacities, payloads, or topology. Copy a useful count into the case's registration when changing the workload definition. Keep it identical for all implementations of the same case and parameter set; never calibrate each implementation or revision independently to the same elapsed time.
+
+Compare the same workload definitions, parameters, round counts, toolchain, and machine. Run without competing builds or benchmarks, repeat a narrow selection, and inspect the spread before attributing a difference to the code. Changing a workload or its timing boundary establishes a new baseline; these results are not directly comparable with the former Divan per-operation measurements. The full matrix is useful for coverage, while a filtered set is the normal optimization workflow.

benchmarks/benches/ecosystem/broadcast/mod.rs

@@ -16,3 +16,9 @@
 // under the License.
 
 mod mpmc;
+
+use benchmarks::harness::Runner;
+
+pub fn register(runner: &mut Runner) {
+    mpmc::register(runner);
+}

benchmarks/benches/ecosystem/broadcast/mpmc/bounded.rs

@@ -17,26 +17,25 @@
 
 // Tokio broadcast overwrites at capacity, so it cannot participate in lossless backpressure cases.
 
+use std::hint::black_box;
+
 use benchmarks::channels::runtime;
+use benchmarks::harness::Bencher;
+use benchmarks::harness::Runner;
 use benchmarks::support::bench_context;
-use divan::Bencher;
-use divan::black_box;
-use divan::counter::ItemsCount;
 use tokio::runtime::Builder;
 
 use crate::broadcast::mpmc::adapters::AsyncBroadcast;
 use crate::broadcast::mpmc::adapters::Asyncband;
 use crate::broadcast::mpmc::adapters::BoundedBroadcastMpmc;
 use crate::broadcast::mpmc::support::BACKPRESSURE_SHAPE;
-use crate::broadcast::mpmc::support::BATCH_MESSAGES;
 use crate::broadcast::mpmc::support::BOUNDED_SHAPES;
 use crate::broadcast::mpmc::support::BoundedConcurrent;
 use crate::broadcast::mpmc::support::BoundedShape;
 use crate::broadcast::mpmc::support::BoundedTasks;
 use crate::broadcast::mpmc::support::ROUND_TRIP_CAPACITY;
 
 // Send-then-receive pairing keeps at most one message retained, so these never reach capacity.
-#[divan::bench(types = [Asyncband, AsyncBroadcast], sample_size = 512)]
 fn try_send_receive<C: BoundedBroadcastMpmc>(bencher: Bencher) {
     let (sender, mut receivers) = C::channel(ROUND_TRIP_CAPACITY, 1);
     let mut receiver = receivers.pop().unwrap();
@@ -47,7 +46,6 @@ fn try_send_receive<C: BoundedBroadcastMpmc>(bencher: Bencher) {
     });
 }
 
-#[divan::bench(types = [Asyncband, AsyncBroadcast], sample_size = 512)]
 fn ready_send_receive<C: BoundedBroadcastMpmc>(bencher: Bencher) {
     let mut context = bench_context();
     let (sender, mut receivers) = C::channel(ROUND_TRIP_CAPACITY, 1);
@@ -59,42 +57,91 @@ fn ready_send_receive<C: BoundedBroadcastMpmc>(bencher: Bencher) {
     });
 }
 
-// Keep one fixture per sample so workers from other fixtures are not alive during timing.
-#[divan::bench(
-    types = [Asyncband, AsyncBroadcast],
-    args = BOUNDED_SHAPES,
-    sample_count = 10,
-    sample_size = 1,
-    counter = ItemsCount::new(BATCH_MESSAGES),
-)]
+// Finish and join each fixture before starting the next round.
 fn thread_batch<C: BoundedBroadcastMpmc>(bencher: Bencher, shape: BoundedShape) {
-    bencher
-        .with_inputs(|| BoundedConcurrent::new::<C>(shape))
-        .bench_local_refs(BoundedConcurrent::run);
+    bencher.bench_local(|| {
+        let mut input = BoundedConcurrent::new::<C>(shape);
+        BoundedConcurrent::run(&mut input)
+    });
 }
 
-#[divan::bench(
-    types = [Asyncband, AsyncBroadcast],
-    args = BOUNDED_SHAPES,
-    sample_count = 10,
-    sample_size = 1,
-    counter = ItemsCount::new(BATCH_MESSAGES),
-)]
 fn task_batch<C: BoundedBroadcastMpmc>(bencher: Bencher, shape: BoundedShape) {
     let runtime = Builder::new_multi_thread()
         .worker_threads(4)
         .build()
         .unwrap();
-    bencher
-        .with_inputs(|| BoundedTasks::new::<C>(&runtime, shape))
-        .bench_local_refs(|tasks| tasks.run(&runtime));
+    bencher.bench_local(|| {
+        let mut tasks = BoundedTasks::new::<C>(&runtime, shape);
+        tasks.run(&runtime)
+    });
 }
 
-#[divan::bench(types = [Asyncband, AsyncBroadcast], sample_count = 10, sample_size = 1,
-    counter = ItemsCount::new(BATCH_MESSAGES))]
 fn capacity_one_task_handoff<C: BoundedBroadcastMpmc>(bencher: Bencher) {
     let runtime = runtime(4);
-    bencher
-        .with_inputs(|| BoundedTasks::new::<C>(&runtime, BACKPRESSURE_SHAPE))
-        .bench_local_refs(|tasks| tasks.run(&runtime));
+    bencher.bench_local(|| {
+        let mut tasks = BoundedTasks::new::<C>(&runtime, BACKPRESSURE_SHAPE);
+        tasks.run(&runtime)
+    });
+}
+
+pub fn register(runner: &mut Runner) {
+    register_implementation::<Asyncband>(runner, "Asyncband");
+    register_implementation::<AsyncBroadcast>(runner, "AsyncBroadcast");
+}
+
+fn register_implementation<C>(runner: &mut Runner, implementation: &str)
+where
+    C: BoundedBroadcastMpmc,
+{
+    runner.case(
+        format!(
+            "{}/{implementation}",
+            concat!(module_path!(), "::try_send_receive")
+        ),
+        9_000_000,
+        1,
+        try_send_receive::<C>,
+    );
+
+    runner.case(
+        format!(
+            "{}/{implementation}",
+            concat!(module_path!(), "::ready_send_receive")
+        ),
+        10_000_000,
+        1,
+        ready_send_receive::<C>,
+    );
+
+    for &arg in BOUNDED_SHAPES {
+        runner.case(
+            format!(
+                "{}/{implementation}/{arg}",
+                concat!(module_path!(), "::thread_batch")
+            ),
+            128,
+            1,
+            |bencher| thread_batch::<C>(bencher, arg),
+        );
+    }
+    for &arg in BOUNDED_SHAPES {
+        runner.case(
+            format!(
+                "{}/{implementation}/{arg}",
+                concat!(module_path!(), "::task_batch")
+            ),
+            320,
+            1,
+            |bencher| task_batch::<C>(bencher, arg),
+        );
+    }
+    runner.case(
+        format!(
+            "{}/{implementation}",
+            concat!(module_path!(), "::capacity_one_task_handoff")
+        ),
+        80,
+        1,
+        capacity_one_task_handoff::<C>,
+    );
 }