diff --git a/Cargo.toml b/Cargo.toml index c223be1..163636b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,7 @@ crate-type = ["cdylib", "rlib"] default = ["getrandom"] getrandom = ["dep:getrandom"] no_random = [] # compile without access to a source of randomness +bench = [] # expose internals for benchmarking python = ["dep:pyo3"] [dependencies] @@ -25,3 +26,8 @@ criterion = { version = "0.5", features = ["html_reports"] } [[bench]] name = "runtime_benchmarks" harness = false + +[[bench]] +name = "channel_benchmarks" +harness = false +required-features = ["bench"] diff --git a/benches/channel_benchmarks.rs b/benches/channel_benchmarks.rs new file mode 100644 index 0000000..f8029ce --- /dev/null +++ b/benches/channel_benchmarks.rs @@ -0,0 +1,722 @@ +use criterion::{ + criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion, Throughput, +}; +use std::sync::{Arc, Barrier}; +use std::thread; +use swactor::channel::{HybridChannel, Receiver, Sender}; + +// --------------------------------------------------------------------------- +// Payload types +// --------------------------------------------------------------------------- + +#[derive(Clone, Debug)] +struct Tiny(u64); + +#[derive(Clone, Debug)] +struct Medium([u8; 256]); + +impl Default for Medium { + fn default() -> Self { + Medium([0xAB; 256]) + } +} + +#[derive(Clone, Debug)] +struct Large(Box<[u8; 4096]>); + +impl Default for Large { + fn default() -> Self { + Large(Box::new([0xCD; 4096])) + } +} + +// =========================================================================== +// 1. SINGLE-THREADED — raw HybridChannel push/pop +// =========================================================================== + +fn st_push_pop(c: &mut Criterion) { + let mut group = c.benchmark_group("st_push_pop"); + + // 1a — Single push + pop (hot path: ring only) + for cap in [64, 256, 1024] { + group.bench_with_input( + BenchmarkId::new("single_roundtrip", cap), + &cap, + |b, &cap| { + let ch = HybridChannel::new(cap); + b.iter(|| { + ch.push(Tiny(42)).unwrap(); + ch.pop().unwrap(); + }); + }, + ); + } + + // 1b — Burst: push N then pop N (all in ring) + for n in [64, 256, 1024] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::new("burst_ring_only", n), &n, |b, &n| { + let ch = HybridChannel::new(n); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + } + + // 1c — Burst: push N into capacity N/2 (forces overflow) + for n in [128, 512, 2048] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input( + BenchmarkId::new("burst_with_overflow", n), + &n, + |b, &n| { + let ch = HybridChannel::new(n / 2); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }, + ); + } + + // 1d — Burst: 100% overflow (capacity=1) + for n in [256, 1024] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input( + BenchmarkId::new("burst_all_overflow", n), + &n, + |b, &n| { + let ch = HybridChannel::new(1); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }, + ); + } + + group.finish(); +} + +// =========================================================================== +// 2. EMPTY-POP COST +// =========================================================================== + +fn empty_pop(c: &mut Criterion) { + let mut group = c.benchmark_group("empty_pop"); + + group.bench_function("empty_channel", |b| { + let ch: HybridChannel = HybridChannel::new(64); + b.iter(|| { + assert!(ch.pop().is_none()); + }); + }); + + // Pop from a channel that previously had items (ring drained, overflow drained) + group.bench_function("after_drain", |b| { + let ch = HybridChannel::new(64); + for i in 0..100u64 { + ch.push(Tiny(i)).unwrap(); + } + while ch.pop().is_some() {} + b.iter(|| { + assert!(ch.pop().is_none()); + }); + }); + + group.finish(); +} + +// =========================================================================== +// 3. PAYLOAD SIZE COMPARISON +// =========================================================================== + +fn payload_sizes(c: &mut Criterion) { + let mut group = c.benchmark_group("payload_size"); + let n = 512usize; + group.throughput(Throughput::Elements(n as u64)); + + // Tiny (8 bytes) + group.bench_function("tiny_8B", |b| { + let ch = HybridChannel::new(n); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + // Medium (256 bytes) + group.bench_function("medium_256B", |b| { + let ch = HybridChannel::new(n); + b.iter(|| { + for _ in 0..n { + ch.push(Medium::default()).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + // Large (4096 bytes, boxed) + group.bench_function("large_4KB_boxed", |b| { + let ch = HybridChannel::new(n); + b.iter(|| { + for _ in 0..n { + ch.push(Large::default()).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + // Box — what the runtime actually uses + group.bench_function("boxed_any", |b| { + let ch: HybridChannel> = HybridChannel::new(n); + b.iter(|| { + for i in 0..n as u64 { + let _ = ch.push(Box::new(Tiny(i))); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + group.finish(); +} + +// =========================================================================== +// 4. SENDER / RECEIVER API +// =========================================================================== + +fn sender_receiver_api(c: &mut Criterion) { + let mut group = c.benchmark_group("sender_receiver"); + + // 4a — Single send + recv + group.bench_function("single_roundtrip", |b| { + let rx: Receiver = Receiver::new(64); + let tx = rx.new_sender(); + b.iter(|| { + tx.try_send(Tiny(42)).unwrap(); + rx.try_recv().unwrap(); + }); + }); + + // 4b — Burst through sender/receiver + for n in [256, 1024, 4096] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::new("burst", n), &n, |b, &n| { + let rx: Receiver = Receiver::new(n); + let tx = rx.new_sender(); + b.iter(|| { + for i in 0..n as u64 { + tx.try_send(Tiny(i)).unwrap(); + } + for _ in 0..n { + rx.try_recv().unwrap(); + } + }); + }); + } + + // 4c — Multiple senders (clone cost + contention-free since ST) + group.bench_function("clone_sender", |b| { + let rx: Receiver = Receiver::new(64); + let tx = rx.new_sender(); + b.iter(|| { + let _tx2 = rx.new_sender(); + tx.try_send(Tiny(1)).unwrap(); + rx.try_recv().unwrap(); + }); + }); + + group.finish(); +} + +// =========================================================================== +// 5. SPSC — one producer thread, one consumer thread +// =========================================================================== + +fn spsc(c: &mut Criterion) { + let mut group = c.benchmark_group("spsc"); + + for n in [1_000usize, 10_000, 100_000] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::new("throughput", n), &n, |b, &n| { + b.iter_batched( + || { + let rx: Receiver = Receiver::new(1024); + let tx = rx.new_sender(); + (tx, rx) + }, + |(tx, rx)| { + let barrier = Arc::new(Barrier::new(2)); + + let b1 = barrier.clone(); + let producer = thread::spawn(move || { + b1.wait(); + for i in 0..n as u64 { + tx.try_send(Tiny(i)).unwrap(); + } + }); + + let b2 = barrier.clone(); + let consumer = thread::spawn(move || { + b2.wait(); + let mut received = 0u64; + while received < n as u64 { + if rx.try_recv().is_some() { + received += 1; + } else { + std::hint::spin_loop(); + } + } + received + }); + + producer.join().unwrap(); + let count = consumer.join().unwrap(); + assert_eq!(count, n as u64); + }, + BatchSize::PerIteration, + ); + }); + } + + // SPSC with varying ring capacities (fixed message count) + let n = 10_000usize; + group.throughput(Throughput::Elements(n as u64)); + for cap in [16, 64, 256, 1024, 8192] { + group.bench_with_input( + BenchmarkId::new("cap_sweep", cap), + &cap, + |b, &cap| { + b.iter_batched( + || { + let rx: Receiver = Receiver::new(cap); + let tx = rx.new_sender(); + (tx, rx) + }, + |(tx, rx)| { + let barrier = Arc::new(Barrier::new(2)); + + let b1 = barrier.clone(); + let producer = thread::spawn(move || { + b1.wait(); + for i in 0..n as u64 { + tx.try_send(Tiny(i)).unwrap(); + } + }); + + let b2 = barrier.clone(); + let consumer = thread::spawn(move || { + b2.wait(); + let mut received = 0u64; + while received < n as u64 { + if rx.try_recv().is_some() { + received += 1; + } else { + std::hint::spin_loop(); + } + } + }); + + producer.join().unwrap(); + consumer.join().unwrap(); + }, + BatchSize::PerIteration, + ); + }, + ); + } + + group.finish(); +} + +// =========================================================================== +// 6. MPSC — multiple producer threads, one consumer thread +// =========================================================================== + +fn mpsc(c: &mut Criterion) { + let mut group = c.benchmark_group("mpsc"); + + // Vary number of producers + let msgs_per_producer = 5_000usize; + for num_producers in [2, 4, 8] { + let total = num_producers * msgs_per_producer; + group.throughput(Throughput::Elements(total as u64)); + group.bench_with_input( + BenchmarkId::new("producers", num_producers), + &num_producers, + |b, &num_producers| { + b.iter_batched( + || { + let rx: Receiver = Receiver::new(1024); + let senders: Vec> = + (0..num_producers).map(|_| rx.new_sender()).collect(); + (senders, rx) + }, + |(senders, rx)| { + let barrier = Arc::new(Barrier::new(num_producers + 1)); + + let handles: Vec<_> = senders + .into_iter() + .map(|tx| { + let b = barrier.clone(); + thread::spawn(move || { + b.wait(); + for i in 0..msgs_per_producer as u64 { + tx.try_send(Tiny(i)).unwrap(); + } + }) + }) + .collect(); + + let total = num_producers * msgs_per_producer; + let b = barrier.clone(); + let consumer = thread::spawn(move || { + b.wait(); + let mut received = 0usize; + while received < total { + if rx.try_recv().is_some() { + received += 1; + } else { + std::hint::spin_loop(); + } + } + received + }); + + for h in handles { + h.join().unwrap(); + } + assert_eq!(consumer.join().unwrap(), total); + }, + BatchSize::PerIteration, + ); + }, + ); + } + + // Heavy contention: many producers, small ring + for cap in [16, 256] { + let num_producers = 8; + let total = num_producers * msgs_per_producer; + group.throughput(Throughput::Elements(total as u64)); + group.bench_with_input( + BenchmarkId::new("contention_cap", cap), + &cap, + |b, &cap| { + b.iter_batched( + || { + let rx: Receiver = Receiver::new(cap); + let senders: Vec> = + (0..num_producers).map(|_| rx.new_sender()).collect(); + (senders, rx) + }, + |(senders, rx)| { + let barrier = Arc::new(Barrier::new(num_producers + 1)); + + let handles: Vec<_> = senders + .into_iter() + .map(|tx| { + let b = barrier.clone(); + thread::spawn(move || { + b.wait(); + for i in 0..msgs_per_producer as u64 { + tx.try_send(Tiny(i)).unwrap(); + } + }) + }) + .collect(); + + let total = num_producers * msgs_per_producer; + let b = barrier.clone(); + let consumer = thread::spawn(move || { + b.wait(); + let mut received = 0usize; + while received < total { + if rx.try_recv().is_some() { + received += 1; + } else { + std::hint::spin_loop(); + } + } + received + }); + + for h in handles { + h.join().unwrap(); + } + assert_eq!(consumer.join().unwrap(), total); + }, + BatchSize::PerIteration, + ); + }, + ); + } + + group.finish(); +} + +// =========================================================================== +// 7. OVERFLOW RATIO — measure how overflow % affects throughput +// =========================================================================== + +fn overflow_ratio(c: &mut Criterion) { + let mut group = c.benchmark_group("overflow_ratio"); + let n = 4096usize; + group.throughput(Throughput::Elements(n as u64)); + + // 0% overflow + group.bench_function("0pct", |b| { + let ch = HybridChannel::new(n); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + // 25% overflow + group.bench_function("25pct", |b| { + let ch = HybridChannel::new(n * 3 / 4); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + // 50% overflow + group.bench_function("50pct", |b| { + let ch = HybridChannel::new(n / 2); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + // 75% overflow + group.bench_function("75pct", |b| { + let ch = HybridChannel::new(n / 4); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + // 100% overflow + group.bench_function("100pct", |b| { + let ch = HybridChannel::new(1); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..n { + ch.pop().unwrap(); + } + }); + }); + + group.finish(); +} + +// =========================================================================== +// 8. INTERLEAVED — alternating push/pop (simulates steady-state runtime) +// =========================================================================== + +fn interleaved(c: &mut Criterion) { + let mut group = c.benchmark_group("interleaved"); + + // Push 1, pop 1 — the most common runtime pattern + for cap in [64, 256, 1024] { + let n = 4096usize; + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input( + BenchmarkId::new("push1_pop1", cap), + &cap, + |b, &cap| { + let ch = HybridChannel::new(cap); + b.iter(|| { + for i in 0..n as u64 { + ch.push(Tiny(i)).unwrap(); + ch.pop().unwrap(); + } + }); + }, + ); + } + + // Push K, pop K — batched steady-state + for k in [4, 16, 64] { + let rounds = 256usize; + let n = rounds * k; + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input( + BenchmarkId::new("batch_k", k), + &k, + |b, &k| { + let ch = HybridChannel::new(256); + b.iter(|| { + for _ in 0..rounds { + for i in 0..k as u64 { + ch.push(Tiny(i)).unwrap(); + } + for _ in 0..k { + ch.pop().unwrap(); + } + } + }); + }, + ); + } + + group.finish(); +} + +// =========================================================================== +// 9. CHANNEL CREATION COST +// =========================================================================== + +fn creation(c: &mut Criterion) { + let mut group = c.benchmark_group("creation"); + + for cap in [16, 64, 256, 1024, 4096] { + group.bench_with_input( + BenchmarkId::new("hybrid_channel", cap), + &cap, + |b, &cap| { + b.iter(|| HybridChannel::::new(cap)); + }, + ); + } + + for cap in [16, 64, 256, 1024, 4096] { + group.bench_with_input( + BenchmarkId::new("receiver_and_sender", cap), + &cap, + |b, &cap| { + b.iter(|| { + let rx: Receiver = Receiver::new(cap); + let _tx = rx.new_sender(); + }); + }, + ); + } + + group.finish(); +} + +// =========================================================================== +// 10. SPSC PINGPONG — latency measurement (two channels, bounce messages) +// =========================================================================== + +fn spsc_pingpong(c: &mut Criterion) { + let mut group = c.benchmark_group("spsc_pingpong"); + + for cap in [64, 1024] { + let rounds = 10_000usize; + group.throughput(Throughput::Elements(rounds as u64)); + group.bench_with_input( + BenchmarkId::new("roundtrips", cap), + &cap, + |b, &cap| { + b.iter_batched( + || { + // Channel A: thread1 → thread2 + let rx_a: Receiver = Receiver::new(cap); + let tx_a = rx_a.new_sender(); + // Channel B: thread2 → thread1 + let rx_b: Receiver = Receiver::new(cap); + let tx_b = rx_b.new_sender(); + (tx_a, rx_a, tx_b, rx_b) + }, + |(tx_a, rx_a, tx_b, rx_b)| { + let barrier = Arc::new(Barrier::new(2)); + + let b1 = barrier.clone(); + let t1 = thread::spawn(move || { + b1.wait(); + for i in 0..rounds as u64 { + tx_a.try_send(Tiny(i)).unwrap(); + while rx_b.try_recv().is_none() { + std::hint::spin_loop(); + } + } + }); + + let b2 = barrier.clone(); + let t2 = thread::spawn(move || { + b2.wait(); + for _ in 0..rounds { + while rx_a.try_recv().is_none() { + std::hint::spin_loop(); + } + tx_b.try_send(Tiny(0)).unwrap(); + } + }); + + t1.join().unwrap(); + t2.join().unwrap(); + }, + BatchSize::PerIteration, + ); + }, + ); + } + + group.finish(); +} + +// =========================================================================== +// Register all groups +// =========================================================================== + +criterion_group!( + benches, + st_push_pop, + empty_pop, + payload_sizes, + sender_receiver_api, + spsc, + mpsc, + overflow_ratio, + interleaved, + creation, + spsc_pingpong, +); +criterion_main!(benches); diff --git a/src/channel.rs b/src/channel.rs index c487ec4..d1ae3b3 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -41,7 +41,7 @@ impl HybridChannel { } -pub(crate) struct Receiver { +pub struct Receiver { queue: Arc>, } @@ -62,7 +62,7 @@ impl Receiver { } } -pub(crate) struct Sender { +pub struct Sender { queue: Arc>, } diff --git a/src/lib.rs b/src/lib.rs index 715af33..259602e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,6 +1,9 @@ pub mod actor; pub mod worker; +#[cfg(feature = "bench")] +pub mod channel; +#[cfg(not(feature = "bench"))] pub(crate) mod channel; pub(crate) mod error; pub use error::Error;