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); }