238 lines
8.1 KiB
Rust
238 lines
8.1 KiB
Rust
//! Comparison benchmarks against alternative implementations
|
|
//!
|
|
//! Compares announcement's oneshot broadcast pattern against:
|
|
//! - tokio::sync::oneshot (N separate channels)
|
|
//! - tokio::sync::broadcast (multi-shot channel used for oneshot)
|
|
//! - async-broadcast (runtime-agnostic broadcast)
|
|
//! - oneshot_broadcast crate (direct competitor)
|
|
//!
|
|
//! Each comparison uses the oneshot pattern: send once, N receivers.
|
|
|
|
#![allow(deprecated)]
|
|
#![allow(unused_must_use)]
|
|
|
|
use announcement::Announcement;
|
|
use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
|
|
use std::hint::black_box;
|
|
use std::sync::Arc;
|
|
|
|
/// Compare: announcement vs tokio::sync::oneshot (N separate channels)
|
|
///
|
|
/// Pattern: One announcement to N listeners vs N separate oneshot channels
|
|
fn bench_vs_tokio_oneshot(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("vs_alternatives/tokio_oneshot");
|
|
|
|
for n in [5, 10, 50, 100].iter() {
|
|
group.throughput(Throughput::Elements(*n as u64));
|
|
|
|
// announcement - O(1) channel, O(N) arc clones
|
|
group.bench_with_input(BenchmarkId::new("announcement", n), n, |b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (tx, announcement) = Announcement::new();
|
|
|
|
let mut handles = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
let listener = announcement.listener();
|
|
handles.push(tokio::spawn(async move { listener.listen().await }));
|
|
}
|
|
|
|
tx.announce(Arc::new(42)).unwrap();
|
|
|
|
for h in handles {
|
|
black_box(h.await.unwrap());
|
|
}
|
|
});
|
|
});
|
|
|
|
// N oneshots - O(N) channels, O(N) sends
|
|
group.bench_with_input(BenchmarkId::new("tokio_oneshot", n), n, |b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let mut senders = Vec::with_capacity(n);
|
|
let mut handles = Vec::with_capacity(n);
|
|
|
|
for _ in 0..n {
|
|
let (tx, rx) = tokio::sync::oneshot::channel();
|
|
senders.push(tx);
|
|
handles.push(tokio::spawn(async move { rx.await.unwrap() }));
|
|
}
|
|
|
|
let data = Arc::new(42);
|
|
for tx in senders {
|
|
tx.send(Arc::clone(&data)).unwrap();
|
|
}
|
|
|
|
for h in handles {
|
|
black_box(h.await.unwrap());
|
|
}
|
|
});
|
|
});
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
/// Compare: announcement vs tokio::sync::broadcast (for oneshot pattern)
|
|
///
|
|
/// tokio::broadcast is designed for multi-shot but can be used for oneshot
|
|
fn bench_vs_tokio_broadcast(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("vs_alternatives/tokio_broadcast");
|
|
|
|
for n in [1, 10, 100, 500].iter() {
|
|
group.throughput(Throughput::Elements(*n as u64));
|
|
|
|
// announcement - optimized for oneshot
|
|
group.bench_with_input(BenchmarkId::new("announcement", n), n, |b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (announcer, announcement) = Announcement::new();
|
|
|
|
let mut handles = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
let listener = announcement.listener();
|
|
handles.push(tokio::spawn(async move { listener.listen().await }));
|
|
}
|
|
|
|
announcer.announce(Arc::new(42)).unwrap();
|
|
|
|
for h in handles {
|
|
black_box(h.await.unwrap());
|
|
}
|
|
});
|
|
});
|
|
|
|
// tokio::broadcast - designed for multi-shot
|
|
group.bench_with_input(BenchmarkId::new("tokio_broadcast", n), n, |b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (tx, _) = tokio::sync::broadcast::channel::<Arc<i32>>(1);
|
|
|
|
let mut handles = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
let mut rx = tx.subscribe();
|
|
handles.push(tokio::spawn(async move { rx.recv().await.unwrap() }));
|
|
}
|
|
|
|
tx.send(Arc::new(42)).unwrap();
|
|
drop(tx);
|
|
|
|
for h in handles {
|
|
black_box(h.await.unwrap());
|
|
}
|
|
});
|
|
});
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
/// Compare: announcement vs async-broadcast (runtime-agnostic)
|
|
///
|
|
/// NOTE: async-broadcast benchmark is commented out due to a bug in the crate
|
|
/// where broadcast() fails with SendError even with receivers waiting.
|
|
/// See tests/async_broadcast_race_repro.rs for minimal reproduction.
|
|
fn bench_vs_async_broadcast(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("vs_alternatives/async_broadcast");
|
|
|
|
for n in [1, 10, 100, 500].iter() {
|
|
group.throughput(Throughput::Elements(*n as u64));
|
|
|
|
// announcement
|
|
group.bench_with_input(BenchmarkId::new("announcement", n), n, |b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (announcer, announcement) = Announcement::new();
|
|
|
|
let mut handles = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
let listener = announcement.listener();
|
|
handles.push(tokio::spawn(async move { listener.listen().await }));
|
|
}
|
|
|
|
announcer.announce(Arc::new(42)).unwrap();
|
|
|
|
for h in handles {
|
|
black_box(h.await.unwrap());
|
|
}
|
|
});
|
|
});
|
|
|
|
// async-broadcast - COMMENTED OUT due to crate bug
|
|
// Bug: broadcast() randomly fails with SendError even when receivers exist
|
|
// Reproduction: tests/async_broadcast_race_repro.rs
|
|
/*
|
|
group.bench_with_input(BenchmarkId::new("async_broadcast", n), n, |b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (tx, _) = async_broadcast::broadcast::<Arc<i32>>(1);
|
|
|
|
let mut handles = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
let mut rx = tx.new_receiver();
|
|
handles.push(tokio::spawn(async move { rx.recv().await.unwrap() }));
|
|
}
|
|
|
|
tx.broadcast(Arc::new(42)).await.unwrap();
|
|
|
|
for h in handles {
|
|
black_box(h.await.unwrap());
|
|
}
|
|
});
|
|
});
|
|
*/
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
/// Show where tokio::broadcast excels: multiple announcements
|
|
///
|
|
/// This is NOT what announcement is designed for - this shows the tradeoffs
|
|
fn bench_multishot_pattern(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("vs_alternatives/multishot_pattern");
|
|
|
|
// announcement - must create new channel each time
|
|
group.bench_function("announcement_recreate", |b| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
for i in 0..10 {
|
|
let (announcer, announcement) = Announcement::new();
|
|
let listener = announcement.listener();
|
|
|
|
let handle = tokio::spawn(async move { listener.listen().await });
|
|
announcer.announce(Arc::new(i)).unwrap();
|
|
black_box(handle.await.unwrap());
|
|
}
|
|
});
|
|
});
|
|
|
|
// tokio::broadcast - designed for this
|
|
group.bench_function("tokio_broadcast_multishot", |b| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (tx, _) = tokio::sync::broadcast::channel::<Arc<i32>>(10);
|
|
let mut rx = tx.subscribe();
|
|
|
|
let handle = tokio::spawn(async move {
|
|
for _ in 0..10 {
|
|
black_box(rx.recv().await.unwrap());
|
|
}
|
|
});
|
|
|
|
for i in 0..10 {
|
|
tx.send(Arc::new(i)).unwrap();
|
|
}
|
|
|
|
handle.await.unwrap();
|
|
});
|
|
});
|
|
|
|
group.finish();
|
|
}
|
|
|
|
criterion_group!(
|
|
benches,
|
|
bench_vs_tokio_oneshot,
|
|
bench_vs_tokio_broadcast,
|
|
bench_vs_async_broadcast,
|
|
bench_multishot_pattern,
|
|
);
|
|
criterion_main!(benches);
|