From a247358ec17e5eddc1b2b0ebc42720314a7859a2 Mon Sep 17 00:00:00 2001 From: fanyang Date: Wed, 24 Jun 2026 21:09:07 +0800 Subject: [PATCH] test(bench): add counter contention benchmark Adds benches/counter_contention.rs comparing the production CounterHandle against reconstructed baselines (pre-optimization single-atomic + CAS + Mutex) under tokio-task contention, plus counter-only variants. Uses the real stats_manager types so the numbers reflect shipped code. Placed between the sharding and fastant commits so each can be benchmarked independently: at this commit prod = sharded counter + Mutex touch. --- Cargo.lock | 133 +++++++- easytier/Cargo.toml | 5 + easytier/benches/counter_contention.rs | 439 +++++++++++++++++++++++++ 3 files changed, 576 insertions(+), 1 deletion(-) create mode 100644 easytier/benches/counter_contention.rs diff --git a/Cargo.lock b/Cargo.lock index 66581096..7701164f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -129,6 +129,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + [[package]] name = "anstream" version = "0.6.15" @@ -1139,6 +1145,12 @@ dependencies = [ "toml 0.9.12+spec-1.1.0", ] +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cc" version = "1.2.10" @@ -1248,6 +1260,33 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + [[package]] name = "cidr" version = "0.3.1" @@ -1593,6 +1632,42 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "is-terminal", + "itertools 0.10.5", + "num-traits", + "once_cell", + "oorandom", + "plotters", + "rayon", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools 0.10.5", +] + [[package]] name = "critical-section" version = "1.2.0" @@ -2266,6 +2341,7 @@ dependencies = [ "clap_complete", "clap_complete_nushell", "console-subscriber", + "criterion", "crossbeam", "ctor 0.8.0", "dashmap", @@ -4506,6 +4582,17 @@ dependencies = [ "once_cell", ] +[[package]] +name = "is-terminal" +version = "0.4.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" +dependencies = [ + "hermit-abi", + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "is-wsl" version = "0.4.0" @@ -5624,7 +5711,7 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "680998035259dcfcafe653688bf2aa6d3e2dc05e98be6ab46afb089dc84f1df8" dependencies = [ - "proc-macro-crate 2.0.0", + "proc-macro-crate 3.5.0", "proc-macro2", "quote", "syn 2.0.117", @@ -5866,6 +5953,12 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + [[package]] name = "opaque-debug" version = "0.3.1" @@ -6496,6 +6589,34 @@ dependencies = [ "time", ] +[[package]] +name = "plotters" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" +dependencies = [ + "num-traits", + "plotters-backend", + "plotters-svg", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "plotters-backend" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" + +[[package]] +name = "plotters-svg" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" +dependencies = [ + "plotters-backend", +] + [[package]] name = "pnet" version = "0.35.0" @@ -9853,6 +9974,16 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "tinyvec" version = "1.8.0" diff --git a/easytier/Cargo.toml b/easytier/Cargo.toml index c3640ef2..488647de 100644 --- a/easytier/Cargo.toml +++ b/easytier/Cargo.toml @@ -342,6 +342,11 @@ futures-util = "0.3.31" maplit = "1.0.2" tempfile = "3.22.0" ctor = "0.8.0" +criterion = { version = "0.5", features = ["html_reports"] } + +[[bench]] +name = "counter_contention" +harness = false [target.'cfg(target_os = "linux")'.dev-dependencies] defguard_wireguard_rs = "0.4.2" diff --git a/easytier/benches/counter_contention.rs b/easytier/benches/counter_contention.rs new file mode 100644 index 00000000..f026cad3 --- /dev/null +++ b/easytier/benches/counter_contention.rs @@ -0,0 +1,439 @@ +//! Compare counter implementations under tokio-task contention. +//! +//! Groups: +//! - `contention_scaling` : N tokio tasks share one counter, total work fixed. +//! Variants: `single_atomic`, `cas_saturating`, `sharded_atomic`, +//! `thread_local_cell`, `unsafe_cell` (unsound, for reference only). +//! - `single_thread_write`: per-`add` cost with no contention (floor cost). +//! - `read_cost` : per-`get()` cost. +//! - `counter_handle` : the REAL production hot path. Measures the actual +//! `stats_manager::CounterHandle::add` (sharded `fetch_add` + lock-free +//! atomic-millis `touch`) against reconstructed baselines: +//! * `prod` - real `CounterHandle` (this code's version) +//! * `baseline_cas_mutex` - pre-optimization design: single +//! `AtomicU64` with `fetch_update` (CAS) + +//! `Mutex` touch +//! * `baseline_fetchadd_mutex`- `fetch_add` + `Mutex` touch +//! (isolates the sharding + lock-free touch) +//! +//! Run: `cargo bench -p easytier --bench counter_contention` + +use std::cell::{Cell, UnsafeCell}; +use std::hint::black_box; +use std::sync::Arc; +use std::sync::LazyLock; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::thread::available_parallelism; +use std::time::Instant; + +use criterion::{ + BenchmarkId, Criterion, Throughput, criterion_group, criterion_main, measurement::WallTime, +}; +use easytier::common::stats_manager::{CounterHandle, MetricName, StatsManager}; +use parking_lot::Mutex; + +const COUNTER_SHARDS: usize = 16; +const TOTAL_WORK: u64 = 8_000_000; +// The handle path calls `Instant::now()` per `add`, so it is far heavier per op +// than the counter-only groups; use a smaller total to keep the bench fast. +const HANDLE_TOTAL_WORK: u64 = 2_000_000; +const TASK_COUNTS: &[usize] = &[1, 2, 4, 8, 16, 32]; + +trait Counter: Send + Sync { + fn add(&self, delta: u64); + fn get(&self) -> u64; +} + +// --------------------------------------------------------------------------- +// 1. SingleAtomic: one atomic, fetch_add. Baseline; contends across cores. +// --------------------------------------------------------------------------- + +struct SingleAtomic(AtomicU64); + +impl Default for SingleAtomic { + fn default() -> Self { + Self(AtomicU64::new(0)) + } +} + +impl Counter for SingleAtomic { + #[inline(always)] + fn add(&self, delta: u64) { + self.0.fetch_add(delta, Ordering::Relaxed); + } + #[inline(always)] + fn get(&self) -> u64 { + self.0.load(Ordering::Relaxed) + } +} + +// --------------------------------------------------------------------------- +// 2. CasSaturating: fetch_update with saturating_add (the original PR `add`). +// A CAS loop that can retry under contention. +// --------------------------------------------------------------------------- + +struct CasSaturating(AtomicU64); + +impl Default for CasSaturating { + fn default() -> Self { + Self(AtomicU64::new(0)) + } +} + +impl Counter for CasSaturating { + #[inline(always)] + fn add(&self, delta: u64) { + let _ = self.0.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + Some(c.saturating_add(delta)) + }); + } + #[inline(always)] + fn get(&self) -> u64 { + self.0.load(Ordering::Relaxed) + } +} + +// --------------------------------------------------------------------------- +// 3. ShardedAtomic: 16 cache-aligned shards + per-thread shard index. +// Mirrors production `stats_manager::UnsafeCounter`. +// --------------------------------------------------------------------------- + +thread_local! { + static SHARD_IDX: Cell = Cell::new({ + static NEXT: AtomicUsize = AtomicUsize::new(0); + NEXT.fetch_add(1, Ordering::Relaxed) % COUNTER_SHARDS + }); +} + +#[repr(align(64))] +struct Shard { + value: AtomicU64, +} + +struct ShardedAtomic { + shards: Box<[Shard]>, +} + +impl Default for ShardedAtomic { + fn default() -> Self { + let mut shards = Vec::with_capacity(COUNTER_SHARDS); + for _ in 0..COUNTER_SHARDS { + shards.push(Shard { + value: AtomicU64::new(0), + }); + } + Self { + shards: shards.into_boxed_slice(), + } + } +} + +impl Counter for ShardedAtomic { + #[inline(always)] + fn add(&self, delta: u64) { + let i = SHARD_IDX.with(|c| c.get()); + self.shards[i].value.fetch_add(delta, Ordering::Relaxed); + } + #[inline(always)] + fn get(&self) -> u64 { + self.shards + .iter() + .map(|s| s.value.load(Ordering::Relaxed)) + .sum() + } +} + +// --------------------------------------------------------------------------- +// 4. ThreadLocalCell: per-thread Cell accumulation. Zero-atomic writes. +// `get()` flushes the caller thread's local into a shared aggregate, so the +// measured read cost reflects a flush-based read. Exact totals would require +// flushing every thread (not modeled here). +// --------------------------------------------------------------------------- + +thread_local! { + static TLS_DELTA: Cell = Cell::new(0); +} + +struct ThreadLocalCell { + shared: AtomicU64, +} + +impl Default for ThreadLocalCell { + fn default() -> Self { + Self { + shared: AtomicU64::new(0), + } + } +} + +impl Counter for ThreadLocalCell { + #[inline(always)] + fn add(&self, delta: u64) { + TLS_DELTA.with(|c| c.set(c.get() + delta)); + } + #[inline(always)] + fn get(&self) -> u64 { + let local = TLS_DELTA.with(|c| c.replace(0)); + self.shared.fetch_add(local, Ordering::Relaxed) + local + } +} + +// --------------------------------------------------------------------------- +// 5. UnsafeCellCounter: a plain u64 mutated through UnsafeCell with manual +// `unsafe impl Send/Sync`. This is UNSOUND under concurrent access (data +// race / UB) and is exactly what the original code did "for speed". It is +// included only to measure the speed ceiling the author was chasing, and to +// show that its `get()` returns wrong totals under contention (lost updates). +// --------------------------------------------------------------------------- + +struct UnsafeCellCounter(UnsafeCell); + +// SAFETY: deliberately unsound; see above. +unsafe impl Send for UnsafeCellCounter {} +unsafe impl Sync for UnsafeCellCounter {} + +impl Default for UnsafeCellCounter { + fn default() -> Self { + Self(UnsafeCell::new(0)) + } +} + +impl Counter for UnsafeCellCounter { + #[inline(always)] + fn add(&self, delta: u64) { + // SAFETY: UNSOUND under concurrent access (data race). + unsafe { + *self.0.get() += delta; + } + } + #[inline(always)] + fn get(&self) -> u64 { + // SAFETY: UNSOUND under concurrent writers (data race). + unsafe { *self.0.get() } + } +} + +// --------------------------------------------------------------------------- +// Production counter handle + reconstructed baselines for the `counter_handle` +// group. These measure the full hot path (`add` = counter update + `touch` +// timestamp), which is what actually runs per packet in `peer_manager`. +// --------------------------------------------------------------------------- + +// The real production `CounterHandle`. `CounterHandle::add` does a sharded +// `fetch_add` then a lock-free atomic-millis `touch`. +impl Counter for CounterHandle { + #[inline(always)] + fn add(&self, delta: u64) { + // Fully-qualified to avoid infinite recursion through the trait method. + CounterHandle::add(self, delta); + } + #[inline(always)] + fn get(&self) -> u64 { + CounterHandle::get(self) + } +} + +// A faithful replica of the pre-optimization design: a SINGLE `AtomicU64` +// (unsharded) plus a `Mutex` timestamp. `use_cas` selects whether the +// counter write is a `fetch_update` saturating CAS (the original PR `add`) or a +// plain `fetch_add`. +struct BaselineHandle { + counter: AtomicU64, + last_updated: Mutex, + use_cas: bool, +} + +impl Default for BaselineHandle { + fn default() -> Self { + Self { + counter: AtomicU64::new(0), + last_updated: Mutex::new(Instant::now()), + use_cas: false, + } + } +} + +impl Counter for BaselineHandle { + #[inline(always)] + fn add(&self, delta: u64) { + if self.use_cas { + let _ = self.counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + Some(c.saturating_add(delta)) + }); + } else { + self.counter.fetch_add(delta, Ordering::Relaxed); + } + *self.last_updated.lock() = Instant::now(); + } + #[inline(always)] + fn get(&self) -> u64 { + self.counter.load(Ordering::Relaxed) + } +} + +// --------------------------------------------------------------------------- +// Harness: a shared multi-thread tokio runtime sized to the host's parallelism. +// --------------------------------------------------------------------------- + +static RUNTIME: LazyLock = LazyLock::new(|| { + let workers = available_parallelism().map(|n| n.get()).unwrap_or(1); + tokio::runtime::Builder::new_multi_thread() + .worker_threads(workers) + .enable_all() + .build() + .expect("failed to build tokio runtime") +}); + +fn bench_contention( + group: &mut criterion::BenchmarkGroup<'_, WallTime>, + name: &str, + n_tasks: usize, + per_task: u64, +) { + let counter: Arc = Arc::new(C::default()); + group.bench_with_input(BenchmarkId::new(name, n_tasks), &n_tasks, |b, &n| { + b.iter(|| { + let counter = counter.clone(); + RUNTIME.block_on(async move { + let mut handles = Vec::with_capacity(n); + for _ in 0..n { + let c = counter.clone(); + handles.push(tokio::spawn(async move { + for _ in 0..per_task { + c.add(black_box(1)); + } + })); + } + for handle in handles { + let _ = handle.await; + } + black_box(counter.get()); + }); + }); + }); +} + +fn contention_scaling(c: &mut Criterion) { + let mut group = c.benchmark_group("contention_scaling"); + group.throughput(Throughput::Elements(TOTAL_WORK)); + for &n in TASK_COUNTS { + let per = TOTAL_WORK / n as u64; + bench_contention::(&mut group, "single_atomic", n, per); + bench_contention::(&mut group, "cas_saturating", n, per); + bench_contention::(&mut group, "sharded_atomic", n, per); + bench_contention::(&mut group, "thread_local_cell", n, per); + bench_contention::(&mut group, "unsafe_cell", n, per); + } + group.finish(); +} + +fn single_thread_write( + group: &mut criterion::BenchmarkGroup<'_, WallTime>, + name: &str, +) { + let counter = C::default(); + group.bench_function(name, |b| { + b.iter(|| { + counter.add(black_box(1)); + }); + }); +} + +fn single_thread_write_group(c: &mut Criterion) { + let mut group = c.benchmark_group("single_thread_write"); + group.throughput(Throughput::Elements(1)); + single_thread_write::(&mut group, "single_atomic"); + single_thread_write::(&mut group, "cas_saturating"); + single_thread_write::(&mut group, "sharded_atomic"); + single_thread_write::(&mut group, "thread_local_cell"); + single_thread_write::(&mut group, "unsafe_cell"); + group.finish(); +} + +fn read_cost( + group: &mut criterion::BenchmarkGroup<'_, WallTime>, + name: &str, +) { + let counter = C::default(); + counter.add(1000); + group.bench_function(name, |b| { + b.iter(|| black_box(counter.get())); + }); +} + +fn read_cost_group(c: &mut Criterion) { + let mut group = c.benchmark_group("read_cost"); + read_cost::(&mut group, "single_atomic"); + read_cost::(&mut group, "cas_saturating"); + read_cost::(&mut group, "sharded_atomic"); + read_cost::(&mut group, "thread_local_cell"); + read_cost::(&mut group, "unsafe_cell"); + group.finish(); +} + +fn bench_handle( + group: &mut criterion::BenchmarkGroup<'_, WallTime>, + name: &str, + n_tasks: usize, + per_task: u64, + counter: Arc, +) { + group.bench_with_input(BenchmarkId::new(name, n_tasks), &n_tasks, |b, &n| { + b.iter(|| { + let counter = counter.clone(); + RUNTIME.block_on(async move { + let mut handles = Vec::with_capacity(n); + for _ in 0..n { + let c = counter.clone(); + handles.push(tokio::spawn(async move { + for _ in 0..per_task { + c.add(black_box(1)); + } + })); + } + for handle in handles { + let _ = handle.await; + } + black_box(counter.get()); + }); + }); + }); +} + +fn counter_handle(c: &mut Criterion) { + let mut group = c.benchmark_group("counter_handle"); + group.throughput(Throughput::Elements(HANDLE_TOTAL_WORK)); + + // StatsManager::new() spawns a background cleanup task, which needs a tokio + // runtime context; bind it to our shared RUNTIME for the lifetime of the + // group. + let _rt_guard = RUNTIME.enter(); + let stats = StatsManager::new(); + + let prod: Arc = Arc::new(stats.get_simple_counter(MetricName::TrafficBytesTx)); + let cas: Arc = Arc::new(BaselineHandle { + use_cas: true, + ..Default::default() + }); + let fam: Arc = Arc::new(BaselineHandle { + use_cas: false, + ..Default::default() + }); + + for &n in TASK_COUNTS { + let per = HANDLE_TOTAL_WORK / n as u64; + bench_handle(&mut group, "prod", n, per, prod.clone()); + bench_handle(&mut group, "baseline_cas_mutex", n, per, cas.clone()); + bench_handle(&mut group, "baseline_fetchadd_mutex", n, per, fam.clone()); + } + group.finish(); +} + +// Keep the default measurement config; pass CLI flags to speed up a run, e.g. +// `-- --measurement-time 2 --sample-size 30 --warm-up-time 500`. +criterion_group! { + name = benches; + config = Criterion::default(); + targets = contention_scaling, single_thread_write_group, read_cost_group, counter_handle +} +criterion_main!(benches);