test(benchmarks): separate broadcast contract and contention costs

benchmarks/README.md

@@ -40,7 +40,7 @@ The `ecosystem` target compares bounded and unbounded MPSC, MPMC and SPMC channe
 
 Each table reports a median and the full minimum-to-maximum range in integer milliseconds. The ratio is the implementation's median elapsed time divided by the reference's median, before rounding: `0.80x` means 20% less elapsed time for the same work. The default is five runs per implementation. The runner interleaves implementations and rotates which one runs first. It keeps the first measurement and every outlier; `--raw` includes the individual times in measurement order.
 
-The clock starts before constructing the fixture and stops after cleanup, including joining workers. Each repetition builds fresh state. Some scenarios reuse channels or workers across rounds: the MPSC task cases also run one priming batch inside the timed interval. A task round transfers 16,384 messages for MPSC, MPMC and SPMC, or publishes 4,096 values to every participating broadcast subscriber. A round in `try-roundtrip` is one send and one receive. Scheduled MPSC comparisons use the same Tokio runtime for all adapters; single-threaded ready operations do not measure task scheduling. Lossless bounded broadcast excludes Tokio's overwrite-on-full channel; the no-backpressure comparison gives bounded peers room for the entire workload so no values are lost.
+The clock starts before constructing the fixture and stops after cleanup, including joining workers. Each repetition builds fresh state. Some scenarios reuse channels or workers across rounds: the MPSC task cases also run one priming batch inside the timed interval. A task round transfers 16,384 messages for MPSC, MPMC and SPMC, or publishes 4,096 values to every participating broadcast subscriber. A round in `try-roundtrip` is one send and one receive. Scheduled MPSC comparisons use the same Tokio runtime for all adapters; single-threaded ready operations do not measure task scheduling. Lossless bounded broadcast excludes Tokio's overwrite-on-full channel; the no-backpressure comparison gives bounded peers room for the entire workload so no values are lost. `TokioArc` publishes a fresh `Arc<usize>` per message so Tokio pays for the allocation behind Asyncband's shared `Arc<T>` receive, and `sequential-drain-4r` reads four subscriptions on one thread to separate per-read work from contention between receivers.
 
 MPSC task scenarios include both one and four producers, small and large buffers, and current-thread controls. Use the reported range to decide whether an apparent difference is reproducible. To compare *different* scenarios, set the same round count where their work units are equivalent; independently sized rows are not a throughput ranking.
 

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

@@ -16,13 +16,16 @@
 // under the License.
 
 use std::future::Future;
+use std::sync::Arc;
 use std::task::Context;
 
 use asyncband::blocking::FutureExt;
 use benchmarks::support::poll_ready;
 
 pub struct Asyncband;
 pub struct Tokio;
+/// Tokio publishing the per-message `Arc` that Asyncband's shared receive requires.
+pub struct TokioArc;
 pub struct AsyncBroadcast;
 
 pub trait BroadcastMpmc: Send + Sync + 'static {
@@ -99,6 +102,37 @@ impl BroadcastMpmc for Tokio {
     }
 }
 
+impl BroadcastMpmc for TokioArc {
+    type Receiver = tokio::sync::broadcast::Receiver<Arc<usize>>;
+    type Sender = tokio::sync::broadcast::Sender<Arc<usize>>;
+
+    fn channel(capacity: usize, receiver_count: usize) -> (Self::Sender, Vec<Self::Receiver>) {
+        let (sender, receiver) = tokio::sync::broadcast::channel(capacity);
+        let mut receivers = Vec::with_capacity(receiver_count);
+        receivers.push(receiver);
+        for _ in 1..receiver_count {
+            receivers.push(sender.subscribe());
+        }
+        (sender, receivers)
+    }
+
+    fn send(sender: &Self::Sender, value: usize) {
+        sender.send(Arc::new(value)).unwrap();
+    }
+
+    fn try_recv(receiver: &mut Self::Receiver) -> Option<usize> {
+        match receiver.try_recv() {
+            Ok(value) => Some(*value),
+            Err(tokio::sync::broadcast::error::TryRecvError::Empty) => None,
+            Err(error) => panic!("unexpected Tokio receive error: {error}"),
+        }
+    }
+
+    fn recv_ready(receiver: &mut Self::Receiver, context: &mut Context<'_>) -> usize {
+        *poll_ready(receiver.recv(), context).unwrap()
+    }
+}
+
 impl BroadcastMpmc for AsyncBroadcast {
     type Receiver = async_broadcast::Receiver<usize>;
     type Sender = async_broadcast::Sender<usize>;

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

@@ -28,9 +28,12 @@ use crate::broadcast::mpmc::adapters::AsyncBroadcast;
 use crate::broadcast::mpmc::adapters::Asyncband;
 use crate::broadcast::mpmc::adapters::BroadcastMpmc;
 use crate::broadcast::mpmc::adapters::Tokio;
+use crate::broadcast::mpmc::adapters::TokioArc;
+use crate::broadcast::mpmc::support::BATCH_MESSAGES;
 use crate::broadcast::mpmc::support::ConcurrentSend;
 use crate::broadcast::mpmc::support::Fanout;
 use crate::broadcast::mpmc::support::ROUND_TRIP_CAPACITY;
+use crate::broadcast::mpmc::support::recv;
 
 fn try_send_receive<C: BroadcastMpmc>(bencher: Bencher) {
     let (sender, mut receivers) = C::channel(ROUND_TRIP_CAPACITY, 1);
@@ -67,34 +70,66 @@ fn publish_then_parallel_drain<C: BroadcastMpmc>(bencher: Bencher, receiver_coun
     });
 }
 
+// Without worker threads, the cost of each read is measured apart from contention between readers.
+fn publish_then_sequential_drain<C: BroadcastMpmc>(bencher: Bencher, receiver_count: usize) {
+    let (sender, mut receivers) = C::channel(BATCH_MESSAGES, receiver_count);
+
+    bencher.bench_local(|| {
+        for value in 0..BATCH_MESSAGES {
+            C::send(&sender, black_box(value));
+        }
+
+        let mut checksum = 0usize;
+        for receiver in &mut receivers {
+            for _ in 0..BATCH_MESSAGES {
+                checksum = checksum.wrapping_add(recv::<C>(receiver));
+            }
+        }
+        black_box(checksum)
+    });
+}
+
 pub fn register(runner: &mut Runner) {
     runner.compare(
         "broadcast/no-backpressure/try-roundtrip",
         9_100_000,
-        implementations!(
-            [Asyncband, Tokio, AsyncBroadcast],
-            |C, b| try_send_receive::<C>(b)
-        ),
+        implementations!([Asyncband, Tokio, TokioArc, AsyncBroadcast], |C, b| {
+            try_send_receive::<C>(b)
+        }),
     );
     runner.compare(
         "broadcast/no-backpressure/ready-roundtrip",
         8_000_000,
-        implementations!([Asyncband, Tokio, AsyncBroadcast], |C, b| {
+        implementations!([Asyncband, Tokio, TokioArc, AsyncBroadcast], |C, b| {
             ready_send_receive::<C>(b)
         }),
     );
     runner.compare(
         "broadcast/no-backpressure/publish-4p-drain-1r",
         810,
-        implementations!([Asyncband, Tokio, AsyncBroadcast], |C, b| {
+        implementations!([Asyncband, Tokio, TokioArc, AsyncBroadcast], |C, b| {
             publish_batch_then_drain::<C>(b, 4)
         }),
     );
+    runner.compare(
+        "broadcast/no-backpressure/drain-1r",
+        2_100,
+        implementations!([Asyncband, Tokio, TokioArc, AsyncBroadcast], |C, b| {
+            publish_then_parallel_drain::<C>(b, 1)
+        }),
+    );
     runner.compare(
         "broadcast/no-backpressure/drain-4r",
         500,
-        implementations!([Asyncband, Tokio, AsyncBroadcast], |C, b| {
+        implementations!([Asyncband, Tokio, TokioArc, AsyncBroadcast], |C, b| {
             publish_then_parallel_drain::<C>(b, 4)
         }),
     );
+    runner.compare(
+        "broadcast/no-backpressure/sequential-drain-4r",
+        1_600,
+        implementations!([Asyncband, Tokio, TokioArc, AsyncBroadcast], |C, b| {
+            publish_then_sequential_drain::<C>(b, 4)
+        }),
+    );
 }

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

@@ -44,7 +44,7 @@ pub const BACKPRESSURE_SHAPE: BoundedShape = BoundedShape {
     receivers: 4,
 };
 
-fn recv<C: BroadcastMpmc>(receiver: &mut C::Receiver) -> usize {
+pub fn recv<C: BroadcastMpmc>(receiver: &mut C::Receiver) -> usize {
     C::try_recv(receiver).expect("the published benchmark batch must be ready")
 }