From c94d1067141a5b913cd61c73dfdcac43fa656215 Mon Sep 17 00:00:00 2001 From: fanyang Date: Wed, 24 Jun 2026 23:34:30 +0800 Subject: [PATCH] perf(stats): lock-free fastant timestamp and fetch_add counter Switch the counter add() from a fetch_update CAS loop to fetch_add, and replace MetricData's Mutex timestamp with a lock-free AtomicU64 storing millis since a lazily-initialized base. Back now_millis() with fastant (TSC on x86_64 Linux, std fallback elsewhere) so touch() is cheap enough to call per packet. GC and its test compare in the millis domain. This is the change that actually delivers the performance: the Mutex timestamp was the serialization bottleneck, and removing it (plus the TSC clock) is what makes the handle path fast. No sharding. --- Cargo.lock | 17 ++++++++ easytier/Cargo.toml | 1 + easytier/src/common/stats_manager.rs | 60 ++++++++++++++++++---------- 3 files changed, 56 insertions(+), 22 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0ca21153..66581096 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2276,6 +2276,7 @@ dependencies = [ "derive_builder", "derive_more 2.1.1", "encoding", + "fastant", "flume 0.12.0", "forwarded-header-value", "futures", @@ -2848,6 +2849,16 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "fastant" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e825441bfb2d831c47c97d05821552db8832479f44c571b97fededbf0099c07" +dependencies = [ + "small_ctor", + "web-time", +] + [[package]] name = "fastbloom" version = "0.9.0" @@ -8678,6 +8689,12 @@ dependencies = [ "autocfg", ] +[[package]] +name = "small_ctor" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88414a5ca1f85d82cc34471e975f0f74f6aa54c40f062efa42c0080e7f763f81" + [[package]] name = "smallvec" version = "1.13.2" diff --git a/easytier/Cargo.toml b/easytier/Cargo.toml index 19e560ce..c3640ef2 100644 --- a/easytier/Cargo.toml +++ b/easytier/Cargo.toml @@ -215,6 +215,7 @@ smoltcp = { git = "https://github.com/smoltcp-rs/smoltcp.git", rev = "0a926767a6 "async", ] } parking_lot = { version = "0.12.0" } +fastant = "0.1" wildmatch = "2.3.4" diff --git a/easytier/src/common/stats_manager.rs b/easytier/src/common/stats_manager.rs index 0db8edbe..7a280dcf 100644 --- a/easytier/src/common/stats_manager.rs +++ b/easytier/src/common/stats_manager.rs @@ -1,10 +1,10 @@ use dashmap::DashMap; -use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use std::fmt; use std::sync::Arc; +use std::sync::OnceLock; use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::{Duration, Instant}; +use std::time::Duration; use tokio::time::interval; use tokio_util::task::AbortOnDropHandle; @@ -402,11 +402,7 @@ impl UnsafeCounter { /// Increment the counter by the given amount pub fn add(&self, delta: u64) { - let _ = self - .value - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { - Some(current.saturating_add(delta)) - }); + self.value.fetch_add(delta, Ordering::Relaxed); } /// Increment the counter by 1 @@ -430,36 +426,52 @@ impl UnsafeCounter { } } +/// Epoch used to convert a monotonic clock reading into a storable `u64` +/// millisecond count for `MetricData::last_updated`. Lazily initialized on first +/// use. Backed by `fastant`, which uses the TSC on x86_64 Linux (and falls back +/// to `std::time::Instant` elsewhere), making `now_millis()` cheap enough to +/// call per packet. +fn time_base() -> fastant::Instant { + static BASE: OnceLock = OnceLock::new(); + *BASE.get_or_init(fastant::Instant::now) +} + +fn now_millis() -> u64 { + fastant::Instant::now() + .saturating_duration_since(time_base()) + .as_millis() as u64 +} + /// MetricData contains both the counter and last update timestamp #[derive(Debug)] struct MetricData { counter: UnsafeCounter, - last_updated: Mutex, + last_updated: AtomicU64, } impl MetricData { fn new() -> Self { Self { counter: UnsafeCounter::new(), - last_updated: Mutex::new(Instant::now()), + last_updated: AtomicU64::new(now_millis()), } } fn new_with_value(initial: u64) -> Self { Self { counter: UnsafeCounter::new_with_value(initial), - last_updated: Mutex::new(Instant::now()), + last_updated: AtomicU64::new(now_millis()), } } - /// Update the last_updated timestamp + /// Update the last_updated timestamp. Lock-free. fn touch(&self) { - *self.last_updated.lock() = Instant::now(); + self.last_updated.store(now_millis(), Ordering::Relaxed); } - /// Get the last updated timestamp - fn get_last_updated(&self) -> Instant { - *self.last_updated.lock() + /// Last update time as milliseconds since `time_base()`. + fn last_updated_millis(&self) -> u64 { + self.last_updated.load(Ordering::Relaxed) } } @@ -565,9 +577,9 @@ impl StatsManager { loop { interval.tick().await; - let Some(cutoff_time) = Instant::now().checked_sub(Duration::from_secs(180)) else { - continue; - }; + // Drop metrics untouched for 180s and with no live handles. + // Compare in the millis-since-base domain; no Instant alloc. + let cutoff_millis = now_millis().saturating_sub(180_000); let Some(counters) = counters_clone.upgrade() else { break; @@ -575,7 +587,7 @@ impl StatsManager { counters.retain(|_, metric_data: &mut Arc| { Arc::strong_count(metric_data) > 1 - || metric_data.get_last_updated() > cutoff_time + || metric_data.last_updated_millis() > cutoff_millis }); counters.shrink_to_fit(); } @@ -896,11 +908,14 @@ mod tests { let counter = stats.get_simple_counter(MetricName::TrafficBytesForwarded); counter.set(1); - let cutoff_time = Instant::now().checked_add(Duration::from_secs(1)).unwrap(); + // Cutoff 1s in the future, so nothing is stale by timestamp; only live + // handles keep a metric. + let cutoff_millis = now_millis() + 1_000; stats .counters .retain(|_, metric_data: &mut Arc| { - Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > cutoff_time + Arc::strong_count(metric_data) > 1 + || metric_data.last_updated_millis() > cutoff_millis }); assert_eq!(stats.metric_count(), 1); @@ -910,7 +925,8 @@ mod tests { stats .counters .retain(|_, metric_data: &mut Arc| { - Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > cutoff_time + Arc::strong_count(metric_data) > 1 + || metric_data.last_updated_millis() > cutoff_millis }); assert_eq!(stats.metric_count(), 0); }