fix: eliminate unsafe code in packet construction and stats counters

Replace all UnsafeCell-based counters with safe alternatives:

- New ShardedCounter: per-thread TLS accumulation (thread_local crate)
  with periodic publish to static AtomicU64. Zero atomic RMW on hot path.
- ACL RuleStatsTracker: remove unsafe Arc<RuleStats> raw pointer
  mutation, use two ShardedCounter fields.
- ZCPacket: replace set_len on uninitialized BytesMut with
  write_bytes + copy_nonoverlapping before set_len.
- StatsManager UnsafeCounter/MetricData: remove UnsafeCell and
  unsafe impl Send/Sync, wrap ShardedCounter. last_updated uses
  AtomicU64 epoch millis.
- Throughput: replace UnsafeCell with AtomicU64.
- secure_datagram: fix grace-window test timestamp.

.rust-analyzer.toml

@@ -0,0 +1,32 @@
+# rust-analyzer config
+# Skip the `easytier-gui` (Tauri) crate because its build scripts require
+# system libraries webkit2gtk-4.1 / javascriptcoregtk-4.1 that are not
+# installed on this host. Excluding it keeps rust-analyzer healthy for the
+# rest of the workspace.
+
+# Override the command used to run build scripts / collect build data.
+[cargo.buildScripts]
+overrideCommand = [
+    "cargo",
+    "check",
+    "--quiet",
+    "--workspace",
+    "--exclude",
+    "easytier-gui",
+    "--message-format=json",
+    "--all-targets",
+    "--keep-going",
+]
+
+# Override the command used for diagnostics-on-save.
+[check]
+overrideCommand = [
+    "cargo",
+    "check",
+    "--workspace",
+    "--exclude",
+    "easytier-gui",
+    "--message-format=json",
+    "--all-targets",
+    "--keep-going",
+]

Cargo.lock

@@ -2435,6 +2435,7 @@ dependencies = [
  "tempfile",
  "terminal_size",
  "thiserror 1.0.63",
+ "thread_local",
  "thunk-rs",
  "tikv-jemalloc-ctl",
  "tikv-jemalloc-sys",

easytier/Cargo.toml

@@ -224,6 +224,7 @@ smoltcp = { git = "https://github.com/smoltcp-rs/smoltcp.git", rev = "0a926767a6
     "async",
 ] }
 parking_lot = { version = "0.12.0" }
+thread_local = "1.1"
 
 wildmatch = "2.3.4"
 
@@ -344,7 +345,7 @@ zip = "4.0.0"
 
 
 [dev-dependencies]
-criterion = "0.5.1"
+criterion = { version = "0.5", features = ["html_reports"] }
 serial_test = "3.0.0"
 rstest = "0.25.0"
 futures-util = "0.3.31"
@@ -352,6 +353,14 @@ maplit = "1.0.2"
 tempfile = "3.22.0"
 ctor = "0.8.0"
 
+[[bench]]
+name = "acl_hotpath"
+harness = false
+
+[[bench]]
+name = "zc_packet"
+harness = false
+
 [target.'cfg(target_os = "linux")'.dev-dependencies]
 defguard_wireguard_rs = "0.4.2"
 tokio-socks = "0.5.2"

easytier/benches/acl_hotpath.rs

@@ -0,0 +1,137 @@
+use std::sync::Arc;
+
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use easytier::common::acl_processor::{AclProcessor, PacketInfo};
+use easytier::proto::acl::*;
+use std::net::{IpAddr, Ipv4Addr};
+
+fn make_acl_config() -> Acl {
+    let mut acl_config = Acl::default();
+    let mut acl_v1 = AclV1::default();
+
+    let mut chain = Chain {
+        name: "bench_inbound".to_string(),
+        chain_type: ChainType::Inbound as i32,
+        enabled: true,
+        ..Default::default()
+    };
+
+    chain.rules.push(Rule {
+        name: "allow_all".to_string(),
+        priority: 100,
+        enabled: true,
+        action: Action::Allow as i32,
+        protocol: Protocol::Any as i32,
+        ..Default::default()
+    });
+
+    acl_v1.chains.push(chain);
+    acl_config.acl_v1 = Some(acl_v1);
+    acl_config
+}
+
+fn make_packet_info() -> PacketInfo {
+    PacketInfo {
+        src_ip: IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)),
+        dst_ip: IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)),
+        src_port: Some(12345),
+        dst_port: Some(80),
+        protocol: Protocol::Tcp,
+        packet_size: 1024,
+        src_groups: Arc::new(vec![]),
+        dst_groups: Arc::new(vec![]),
+    }
+}
+
+fn bench_cache_hit_single(c: &mut Criterion) {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let processor = rt.block_on(async { AclProcessor::new(make_acl_config()) });
+    let packet_info = make_packet_info();
+
+    // Prime the cache
+    let _ = processor.process_packet(&packet_info, ChainType::Inbound);
+
+    c.bench_function("acl_cache_hit_1t", |b| {
+        b.iter(|| {
+            std::hint::black_box(processor.process_packet(&packet_info, ChainType::Inbound));
+        });
+    });
+}
+
+fn bench_cache_hit_multi(c: &mut Criterion) {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let mut group = c.benchmark_group("acl_cache_hit_multi");
+    for threads in [2, 4, 8] {
+        let processor = Arc::new(rt.block_on(async { AclProcessor::new(make_acl_config()) }));
+        let packet_info = Arc::new(make_packet_info());
+
+        // Prime the cache
+        let _ = processor.process_packet(&packet_info, ChainType::Inbound);
+
+        group.bench_with_input(
+            BenchmarkId::from_parameter(threads),
+            &threads,
+            |b, &threads| {
+                b.iter_custom(|iters| {
+                    use std::sync::Barrier;
+                    use std::thread;
+
+                    let barrier = Arc::new(Barrier::new(threads + 1));
+                    let per_thread = (iters / threads as u64) as usize;
+                    let mut handles = Vec::with_capacity(threads);
+
+                    for _ in 0..threads {
+                        let processor = Arc::clone(&processor);
+                        let packet_info = Arc::clone(&packet_info);
+                        let barrier = Arc::clone(&barrier);
+                        handles.push(thread::spawn(move || {
+                            barrier.wait();
+                            for _ in 0..per_thread {
+                                std::hint::black_box(
+                                    processor.process_packet(&packet_info, ChainType::Inbound),
+                                );
+                            }
+                        }));
+                    }
+
+                    let start = std::time::Instant::now();
+                    barrier.wait();
+                    for handle in handles {
+                        handle.join().unwrap();
+                    }
+                    start.elapsed()
+                });
+            },
+        );
+    }
+    group.finish();
+}
+
+fn bench_unique_rule_match(c: &mut Criterion) {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let processor = rt.block_on(async { AclProcessor::new(make_acl_config()) });
+
+    c.bench_function("acl_unique_rule_match_1t", |b| {
+        let mut i = 0usize;
+        b.iter(|| {
+            let mut packet_info = make_packet_info();
+            packet_info.src_port = Some((1024 + (i % 60_000)) as u16);
+            packet_info.src_ip = IpAddr::V4(Ipv4Addr::new(
+                10,
+                ((i >> 16) & 0xff) as u8,
+                ((i >> 8) & 0xff) as u8,
+                (i & 0xff) as u8,
+            ));
+            std::hint::black_box(processor.process_packet(&packet_info, ChainType::Inbound));
+            i = i.wrapping_add(1);
+        });
+    });
+}
+
+criterion_group!(
+    benches,
+    bench_cache_hit_single,
+    bench_cache_hit_multi,
+    bench_unique_rule_match
+);
+criterion_main!(benches);

easytier/benches/zc_packet.rs

@@ -0,0 +1,51 @@
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use easytier::tunnel::packet_def::{ZCPacket, ZCPacketType};
+
+fn bench_new_with_payload(c: &mut Criterion) {
+    let mut group = c.benchmark_group("zc_new_with_payload");
+    for size in [64usize, 1500] {
+        let payload = vec![0xabu8; size];
+        group.bench_with_input(BenchmarkId::from_parameter(size), &payload, |b, payload| {
+            b.iter(|| {
+                std::hint::black_box(ZCPacket::new_with_payload(std::hint::black_box(payload)));
+            });
+        });
+    }
+    group.finish();
+}
+
+fn bench_new_for_foreign_network(c: &mut Criterion) {
+    let payload = vec![0xabu8; 64];
+    let foreign_packet = ZCPacket::new_with_payload(&payload);
+    let network_name = "bench-network".to_string();
+
+    c.bench_function("zc_new_for_foreign_network_64b", |b| {
+        b.iter(|| {
+            std::hint::black_box(ZCPacket::new_for_foreign_network(
+                std::hint::black_box(&network_name),
+                42,
+                std::hint::black_box(&foreign_packet),
+            ));
+        });
+    });
+}
+
+fn bench_convert_type(c: &mut Criterion) {
+    let payload = vec![0xabu8; 64];
+    let packet = ZCPacket::new_with_payload(&payload);
+
+    c.bench_function("zc_convert_type_tcp_64b", |b| {
+        b.iter(|| {
+            let p = std::hint::black_box(packet.clone());
+            std::hint::black_box(p.convert_type(ZCPacketType::TCP));
+        });
+    });
+}
+
+criterion_group!(
+    benches,
+    bench_new_with_payload,
+    bench_new_for_foreign_network,
+    bench_convert_type
+);
+criterion_main!(benches);

easytier/src/common/acl_processor.rs

@@ -3,7 +3,8 @@ use std::{
     net::{IpAddr, SocketAddr},
     str::FromStr as _,
     sync::Arc,
-    time::{Duration, SystemTime, UNIX_EPOCH},
+    sync::atomic::{AtomicU64, Ordering::Relaxed},
+    time::Duration,
 };
 
 use quanta::Instant;
@@ -76,7 +77,41 @@ pub struct FastLookupRule {
     pub stateful: bool,
     pub rate_limit: u32,
     pub burst_limit: u32,
-    pub rule_stats: Arc<RuleStats>,
+    pub rule_stats: Arc<RuleStatsTracker>,
+}
+
+#[derive(Debug)]
+pub struct RuleStatsTracker {
+    rule: Option<Rule>,
+    packets: AtomicU64,
+    bytes: AtomicU64,
+}
+
+impl RuleStatsTracker {
+    fn new(rule: Option<Rule>) -> Self {
+        Self {
+            rule,
+            packets: AtomicU64::new(0),
+            bytes: AtomicU64::new(0),
+        }
+    }
+
+    #[inline]
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RuleStatsTracker"))]
+    fn increment(&self, packet_size: usize) {
+        self.packets.fetch_add(1, Relaxed);
+        self.bytes.fetch_add(packet_size as u64, Relaxed);
+    }
+
+    fn snapshot(&self) -> RuleStats {
+        RuleStats {
+            rule: self.rule.clone(),
+            stat: Some(StatItem {
+                packet_count: self.packets.load(Relaxed),
+                byte_count: self.bytes.load(Relaxed),
+            }),
+        }
+    }
 }
 
 // Cache key combining packet info and chain type
@@ -108,17 +143,17 @@ impl AclCacheKey {
 }
 
 // Cache entry with timestamp for LRU cleanup
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub(crate) struct AclCacheEntry {
     pub action: Action,
     pub matched_rule: RuleId,
-    pub last_access: Instant,
+    pub last_access: AtomicU64,
     // New fields to track rule characteristics for proper cache behavior
     pub conn_track_key: Option<String>,
     pub rate_limit_keys: Vec<RateLimitKey>,
     pub chain_type: ChainType,
     pub acl_result: Option<AclResult>,
-    pub rule_stats_vec: Vec<Arc<RuleStats>>,
+    pub rule_stats_vec: Vec<Arc<RuleStatsTracker>>,
 }
 
 // Packet info extracted for ACL processing
@@ -211,7 +246,7 @@ pub struct AclProcessor {
     default_outbound_action: Action,
     default_forward_action: Action,
 
-    default_rule_stats: Arc<RuleStats>,
+    default_rule_stats: Arc<RuleStatsTracker>,
 
     // Connection tracking table - shared across different processor instances if needed
     conn_track: Arc<DashMap<String, ConnTrackEntry>>,
@@ -224,6 +259,13 @@ pub struct AclProcessor {
     cache_max_size: usize,
     cache_cleanup_interval: Duration,
 
+    // Coarse monotonic timestamp updated by the cleanup task, used to avoid
+    // calling Instant::now() on every cache hit.
+    coarse_millis: Arc<AtomicU64>,
+
+    // Hot-path counters that bypass the DashMap stats table
+    cache_hits: AtomicU64,
+
     // Statistics
     stats: Arc<DashMap<AclStatKey, u64>>,
 
@@ -259,18 +301,14 @@ impl AclProcessor {
             default_outbound_action,
             default_forward_action,
 
-            default_rule_stats: Arc::new(RuleStats {
+            default_rule_stats: Arc::new(RuleStatsTracker::new(None)),
-                rule: None,
-                stat: Some(StatItem {
-                    packet_count: 0,
-                    byte_count: 0,
-                }),
-            }),
             conn_track: conn_track.unwrap_or_else(|| Arc::new(DashMap::new())),
             rate_limiters: rate_limiters.unwrap_or_else(|| Arc::new(DashMap::new())),
             rule_cache: Arc::new(DashMap::new()), // Always start with fresh cache
             cache_max_size: 1024,                 // Limit cache to 1k entries
             cache_cleanup_interval: Duration::from_secs(20), // Cleanup every 5 minutes
+            coarse_millis: Arc::new(AtomicU64::new(0)),
+            cache_hits: AtomicU64::new(0),
             stats: stats.unwrap_or_else(|| Arc::new(DashMap::new())),
             tasks,
         };
@@ -374,11 +412,14 @@ impl AclProcessor {
         let rule_cache = self.rule_cache.clone();
         let cache_max_size = self.cache_max_size;
         let cleanup_interval = self.cache_cleanup_interval;
+        let coarse_millis = self.coarse_millis.clone();
 
         self.tasks.spawn(async move {
             let mut interval = tokio::time::interval(cleanup_interval);
             loop {
                 interval.tick().await;
+                let now = crate::common::stats_manager::now_monotonic_millis();
+                coarse_millis.store(now, Relaxed);
                 Self::cleanup_cache(&rule_cache, cache_max_size);
                 rule_cache.shrink_to_fit();
 
@@ -401,10 +442,9 @@ impl AclProcessor {
     /// Clean up cache using LRU strategy
     fn cleanup_cache(cache: &DashMap<AclCacheKey, AclCacheEntry>, max_size: usize) {
         // remove cache not be used in last 15 second
-        let expired_timepoint = Instant::now()
+        let now = crate::common::stats_manager::now_monotonic_millis();
-            .checked_sub(Duration::from_secs(15))
+        let cutoff = now.saturating_sub(15_000);
-            .unwrap_or(Instant::now());
+        cache.retain(|_, entry| entry.last_access.load(Relaxed) > cutoff);
-        cache.retain(|_, entry| entry.last_access > expired_timepoint);
 
         let current_size = cache.len();
         if current_size <= max_size {
@@ -412,9 +452,9 @@ impl AclProcessor {
         }
 
         // Remove oldest entries (LRU cleanup)
-        let mut entries: Vec<(AclCacheKey, Instant)> = cache
+        let mut entries: Vec<(AclCacheKey, u64)> = cache
             .iter()
-            .map(|entry| (entry.key().clone(), entry.value().last_access))
+            .map(|entry| (entry.key().clone(), entry.value().last_access.load(Relaxed)))
             .collect();
 
         // Sort by last_access (oldest first)
@@ -433,6 +473,7 @@ impl AclProcessor {
         );
     }
 
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
     pub(crate) fn process_packet_with_cache_entry(
         &self,
         packet_info: &PacketInfo,
@@ -459,42 +500,41 @@ impl AclProcessor {
         cache_entry.acl_result.clone().unwrap()
     }
 
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
     fn inc_cache_entry_stats(&self, cache_entry: &AclCacheEntry, packet_info: &PacketInfo) {
         for rule_stats in cache_entry.rule_stats_vec.iter() {
-            // Use unsafe code to mutate the contents behind the Arc
+            rule_stats.increment(packet_info.packet_size);
-            let stat_ptr = rule_stats.stat.as_ref().unwrap() as *const StatItem as *mut StatItem;
-            unsafe {
-                (*stat_ptr).packet_count += 1;
-                (*stat_ptr).byte_count += packet_info.packet_size as u64;
-            }
         }
     }
 
     pub fn get_rules_stats(&self) -> Vec<RuleStats> {
         let mut stats: Vec<RuleStats> = Vec::new();
         for rule in self.inbound_rules.iter() {
-            stats.push((*rule.rule_stats).clone());
+            stats.push(rule.rule_stats.snapshot());
         }
         for rule in self.outbound_rules.iter() {
-            stats.push((*rule.rule_stats).clone());
+            stats.push(rule.rule_stats.snapshot());
         }
         for rule in self.forward_rules.iter() {
-            stats.push((*rule.rule_stats).clone());
+            stats.push(rule.rule_stats.snapshot());
         }
         stats
     }
 
     /// Process a packet through ACL rules - Now lock-free!
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
     pub fn process_packet(&self, packet_info: &PacketInfo, chain_type: ChainType) -> AclResult {
         // Check cache first for performance
         let cache_key = AclCacheKey::from_packet_info(packet_info, chain_type);
 
-        // If cache hit and can skip checks, return cached result
+        // If cache hit and can skip checks, return cached result.
-        if let Some(mut cached) = self.rule_cache.get_mut(&cache_key) {
+        // Use get() (read lock) instead of get_mut() (write lock) and update
-            // Update last access time for LRU
+        // last_access via AtomicU64, avoiding expensive Instant::now().
-            cached.last_access = Instant::now();
+        if let Some(cached) = self.rule_cache.get(&cache_key) {
-
+            cached
-            self.increment_stat(AclStatKey::CacheHits);
+                .last_access
+                .store(self.coarse_millis.load(Relaxed), Relaxed);
+            self.cache_hits.fetch_add(1, Relaxed);
             return self.process_packet_with_cache_entry(packet_info, &cached);
         }
 
@@ -516,7 +556,7 @@ impl AclProcessor {
         let mut cache_entry = AclCacheEntry {
             action: Action::Allow,
             matched_rule: RuleId::Default,
-            last_access: Instant::now(),
+            last_access: AtomicU64::new(self.coarse_millis.load(Relaxed)),
             conn_track_key: None,
             rate_limit_keys: vec![],
             chain_type,
@@ -581,8 +621,9 @@ impl AclProcessor {
             // Cache the result with rule info
             self.increment_stat(AclStatKey::RuleMatches);
             self.inc_cache_entry_stats(&cache_entry, packet_info);
-            self.cache_result(&cache_key, cache_entry.clone());
+            let result = cache_entry.acl_result.clone().unwrap();
-            return cache_entry.acl_result.clone().unwrap();
+            self.cache_result(&cache_key, cache_entry);
+            return result;
         }
 
         let default_action = match chain_type {
@@ -618,8 +659,9 @@ impl AclProcessor {
 
         // Cache the default result (no rule info)
         self.inc_cache_entry_stats(&cache_entry, packet_info);
-        self.cache_result(&cache_key, cache_entry.clone());
+        let result = cache_entry.acl_result.clone().unwrap();
-        cache_entry.acl_result.clone().unwrap()
+        self.cache_result(&cache_key, cache_entry);
+        result
     }
 
     /// Get shared state for preserving across hot reloads
@@ -743,13 +785,11 @@ impl AclProcessor {
 
     /// Check connection state for stateful rules
     fn check_connection_state(&self, conn_track_key: &str, packet_info: &PacketInfo) {
+        let now = current_unix_secs();
         self.conn_track
             .entry(conn_track_key.to_string())
             .and_modify(|x| {
-                x.last_seen = SystemTime::now()
+                x.last_seen = now;
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap()
-                    .as_secs();
                 x.packet_count += 1;
                 x.byte_count += packet_info.packet_size as u64;
                 x.state = ConnState::Established as i32;
@@ -763,14 +803,8 @@ impl AclProcessor {
                 ),
                 protocol: packet_info.protocol as i32,
                 state: ConnState::New as i32,
-                created_at: SystemTime::now()
+                created_at: now,
-                    .duration_since(UNIX_EPOCH)
+                last_seen: now,
-                    .unwrap()
-                    .as_secs(),
-                last_seen: SystemTime::now()
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap()
-                    .as_secs(),
                 packet_count: 1,
                 byte_count: packet_info.packet_size as u64,
             });
@@ -862,13 +896,7 @@ impl AclProcessor {
             stateful: rule.stateful,
             rate_limit: rule.rate_limit,
             burst_limit: rule.burst_limit,
-            rule_stats: Arc::new(RuleStats {
+            rule_stats: Arc::new(RuleStatsTracker::new(Some(rule.clone()))),
-                rule: Some(rule.clone()),
-                stat: Some(StatItem {
-                    packet_count: 0,
-                    byte_count: 0,
-                }),
-            }),
         }
     }
 
@@ -894,6 +922,10 @@ impl AclProcessor {
             .collect::<HashMap<_, _>>();
 
         // Add cache statistics using enum keys
+        stats.insert(
+            AclStatKey::CacheHits.as_str(),
+            self.cache_hits.load(Relaxed),
+        );
         stats.insert(AclStatKey::CacheSize.as_str(), self.rule_cache.len() as u64);
         stats.insert(
             AclStatKey::CacheMaxSize.as_str(),
@@ -908,14 +940,11 @@ impl AclProcessor {
         conn_track: Arc<DashMap<String, ConnTrackEntry>>,
         timeout_secs: u64,
     ) {
-        let current_time = SystemTime::now()
+        let current_time = current_unix_secs();
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_secs();
         let keys_to_remove: Vec<String> = conn_track
             .iter()
             .filter_map(|entry| {
-                if current_time - entry.last_seen > timeout_secs {
+                if current_time.saturating_sub(entry.last_seen) > timeout_secs {
                     Some(entry.key().clone())
                 } else {
                     None
@@ -930,11 +959,7 @@ impl AclProcessor {
 
     /// Get cache hit rate
     pub fn get_cache_hit_rate(&self) -> f64 {
-        let cache_hits = self
+        let cache_hits = self.cache_hits.load(Relaxed);
-            .stats
-            .get(&AclStatKey::CacheHits)
-            .map(|v| *v.value())
-            .unwrap_or(0);
         let total_requests = cache_hits
             + self
                 .stats
@@ -950,6 +975,13 @@ impl AclProcessor {
     }
 }
 
+fn current_unix_secs() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs()
+}
+
 // 新增辅助函数
 fn parse_port_start(port_strs: &[String]) -> Option<u16> {
     port_strs

easytier/src/common/mod.rs

@@ -22,6 +22,7 @@ pub mod machine_id;
 pub mod netns;
 pub mod network;
 pub mod os_info;
+pub mod sharded_counter;
 pub mod stats_manager;
 pub mod stun;
 pub mod stun_codec_ext;

easytier/src/common/sharded_counter.rs

@@ -0,0 +1,110 @@
+use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
+
+const PUBLISH_INTERVAL: u64 = 256;
+
+/// Counter optimized for long-lived worker threads.
+///
+/// Pending values below `PUBLISH_INTERVAL` live in `ThreadLocal` shards. The
+/// shards are retained until this counter is dropped, so this is intended for
+/// tokio workers or similarly long-lived threads rather than high-churn threads.
+pub struct ShardedCounter {
+    published: AtomicU64,
+    locals: thread_local::ThreadLocal<AtomicU64>,
+}
+
+impl ShardedCounter {
+    pub fn new() -> Self {
+        Self {
+            published: AtomicU64::new(0),
+            locals: thread_local::ThreadLocal::new(),
+        }
+    }
+
+    #[inline]
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "ShardedCounter"))]
+    pub fn add(&self, delta: u64) {
+        let local = self.locals.get_or(|| AtomicU64::new(0));
+        let v = local.load(Relaxed).saturating_add(delta);
+        local.store(v, Relaxed);
+        if v >= PUBLISH_INTERVAL {
+            let pending = local.swap(0, Relaxed);
+            if pending > 0 {
+                self.published.fetch_add(pending, Relaxed);
+            }
+        }
+    }
+
+    #[inline]
+    pub fn inc(&self) {
+        self.add(1);
+    }
+
+    pub fn get(&self) -> u64 {
+        self.locals
+            .iter()
+            .fold(self.published.load(Relaxed), |total, local| {
+                total.saturating_add(local.load(Relaxed))
+            })
+    }
+
+    pub fn reset(&self) {
+        self.published.store(0, Relaxed);
+        for local in self.locals.iter() {
+            local.store(0, Relaxed);
+        }
+    }
+}
+
+impl Default for ShardedCounter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl std::fmt::Debug for ShardedCounter {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ShardedCounter")
+            .field("value", &self.get())
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::{sync::Arc, thread};
+
+    #[test]
+    fn sharded_counter_get_includes_other_thread_locals() {
+        let counter = Arc::new(ShardedCounter::new());
+        let thread_counter = Arc::clone(&counter);
+
+        thread::spawn(move || {
+            for _ in 0..10 {
+                thread_counter.inc();
+            }
+        })
+        .join()
+        .unwrap();
+
+        assert_eq!(counter.get(), 10);
+    }
+
+    #[test]
+    fn sharded_counter_reset_clears_other_thread_locals() {
+        let counter = Arc::new(ShardedCounter::new());
+        let thread_counter = Arc::clone(&counter);
+
+        thread::spawn(move || {
+            for _ in 0..10 {
+                thread_counter.inc();
+            }
+        })
+        .join()
+        .unwrap();
+
+        counter.reset();
+
+        assert_eq!(counter.get(), 0);
+    }
+}

easytier/src/common/stats_manager.rs

@@ -1,13 +1,16 @@
+use crate::common::sharded_counter::ShardedCounter;
 use dashmap::DashMap;
 use quanta::Instant;
 use serde::{Deserialize, Serialize};
-use std::cell::UnsafeCell;
 use std::fmt;
-use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, LazyLock};
 use std::time::Duration;
 use tokio::time::interval;
 use tokio_util::task::AbortOnDropHandle;
 
+static START_INSTANT: LazyLock<Instant> = LazyLock::new(Instant::now);
+
 /// Predefined metric names for type safety
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum MetricName {
@@ -375,10 +378,10 @@ impl Default for LabelSet {
     }
 }
 
-/// UnsafeCounter provides a high-performance counter using UnsafeCell
+/// High-performance counter backed by sharded thread-local accumulation.
 #[derive(Debug)]
 pub struct UnsafeCounter {
-    value: UnsafeCell<u64>,
+    inner: ShardedCounter,
 }
 
 impl Default for UnsafeCounter {
@@ -390,121 +393,56 @@ impl Default for UnsafeCounter {
 impl UnsafeCounter {
     pub fn new() -> Self {
         Self {
-            value: UnsafeCell::new(0),
+            inner: ShardedCounter::new(),
         }
     }
 
-    pub fn new_with_value(initial: u64) -> Self {
+    pub fn add(&self, delta: u64) {
-        Self {
+        self.inner.add(delta);
-            value: UnsafeCell::new(initial),
-        }
     }
 
-    /// Increment the counter by the given amount
+    pub fn inc(&self) {
-    /// # Safety
+        self.inner.inc();
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn add(&self, delta: u64) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = (*ptr).saturating_add(delta);
-        }
     }
 
-    /// Increment the counter by 1
+    pub fn get(&self) -> u64 {
-    /// # Safety
+        self.inner.get()
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn inc(&self) {
-        unsafe {
-            self.add(1);
-        }
     }
 
-    /// Get the current value of the counter
+    pub fn reset(&self) {
-    /// # Safety
+        self.inner.reset();
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is modifying this counter simultaneously.
-    pub unsafe fn get(&self) -> u64 {
-        let ptr = self.value.get();
-        unsafe { *ptr }
-    }
-
-    /// Reset the counter to zero
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn reset(&self) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = 0;
-        }
-    }
-
-    /// Set the counter to a specific value
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn set(&self, value: u64) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = value;
-        }
     }
 }
 
-// UnsafeCounter is Send + Sync because the safety is guaranteed by the caller
-unsafe impl Send for UnsafeCounter {}
-unsafe impl Sync for UnsafeCounter {}
-
 /// MetricData contains both the counter and last update timestamp
-/// Uses UnsafeCell for lock-free access
 #[derive(Debug)]
 struct MetricData {
     counter: UnsafeCounter,
-    last_updated: UnsafeCell<Instant>,
+    last_updated: AtomicU64,
+}
+
+pub(crate) fn now_monotonic_millis() -> u64 {
+    Instant::now().duration_since(*START_INSTANT).as_millis() as u64
 }
 
 impl MetricData {
     fn new() -> Self {
         Self {
             counter: UnsafeCounter::new(),
-            last_updated: UnsafeCell::new(Instant::now()),
+            last_updated: AtomicU64::new(now_monotonic_millis()),
         }
     }
 
-    fn new_with_value(initial: u64) -> Self {
+    fn touch(&self) {
-        Self {
+        self.last_updated
-            counter: UnsafeCounter::new_with_value(initial),
+            .store(now_monotonic_millis(), Ordering::Relaxed);
-            last_updated: UnsafeCell::new(Instant::now()),
-        }
     }
 
-    /// Update the last_updated timestamp
+    fn get_last_updated(&self) -> u64 {
-    /// # Safety
+        self.last_updated.load(Ordering::Relaxed)
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this timestamp simultaneously.
-    unsafe fn touch(&self) {
-        let ptr = self.last_updated.get();
-        unsafe {
-            *ptr = Instant::now();
-        }
-    }
-
-    /// Get the last updated timestamp
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is modifying this timestamp simultaneously.
-    unsafe fn get_last_updated(&self) -> Instant {
-        let ptr = self.last_updated.get();
-        unsafe { *ptr }
     }
 }
 
-// MetricData is Send + Sync because the safety is guaranteed by the caller
-unsafe impl Send for MetricData {}
-unsafe impl Sync for MetricData {}
-
 /// MetricKey uniquely identifies a metric with its name and labels
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct MetricKey {
@@ -545,41 +483,23 @@ impl CounterHandle {
         }
     }
 
-    /// Increment the counter by the given amount
     pub fn add(&self, delta: u64) {
-        unsafe {
+        self.metric_data.counter.add(delta);
-            self.metric_data.counter.add(delta);
+        self.metric_data.touch();
-            self.metric_data.touch();
-        }
     }
 
-    /// Increment the counter by 1
     pub fn inc(&self) {
-        unsafe {
+        self.metric_data.counter.inc();
-            self.metric_data.counter.inc();
+        self.metric_data.touch();
-            self.metric_data.touch();
-        }
     }
 
-    /// Get the current value of the counter
     pub fn get(&self) -> u64 {
-        unsafe { self.metric_data.counter.get() }
+        self.metric_data.counter.get()
     }
 
-    /// Reset the counter to zero
     pub fn reset(&self) {
-        unsafe {
+        self.metric_data.counter.reset();
-            self.metric_data.counter.reset();
+        self.metric_data.touch();
-            self.metric_data.touch();
-        }
-    }
-
-    /// Set the counter to a specific value
-    pub fn set(&self, value: u64) {
-        unsafe {
-            self.metric_data.counter.set(value);
-            self.metric_data.touch();
-        }
     }
 }
 
@@ -615,9 +535,7 @@ impl StatsManager {
             loop {
                 interval.tick().await;
 
-                let Some(cutoff_time) = Instant::now().checked_sub(Duration::from_secs(180)) else {
+                let cutoff_millis = now_monotonic_millis().saturating_sub(180_000);
-                    continue;
-                };
 
                 let Some(counters) = counters_clone.upgrade() else {
                     break;
@@ -625,7 +543,7 @@ impl StatsManager {
 
                 counters.retain(|_, metric_data: &mut Arc<MetricData>| {
                     Arc::strong_count(metric_data) > 1
-                        || unsafe { metric_data.get_last_updated() > cutoff_time }
+                        || metric_data.get_last_updated() >= cutoff_millis
                 });
                 counters.shrink_to_fit();
             }
@@ -663,7 +581,7 @@ impl StatsManager {
             let key = entry.key();
             let metric_data = entry.value();
 
-            let value = unsafe { metric_data.counter.get() };
+            let value = metric_data.counter.get();
 
             metrics.push(MetricSnapshot {
                 name: key.name,
@@ -696,7 +614,7 @@ impl StatsManager {
         let key = MetricKey::new(name, labels.clone());
 
         if let Some(metric_data) = self.counters.get(&key) {
-            let value = unsafe { metric_data.counter.get() };
+            let value = metric_data.counter.get();
             Some(MetricSnapshot {
                 name,
                 labels: labels.clone(),
@@ -797,17 +715,9 @@ mod tests {
     async fn test_unsafe_counter() {
         let counter = UnsafeCounter::new();
 
-        unsafe {
+        assert_eq!(counter.get(), 0);
-            assert_eq!(counter.get(), 0);
+        counter.add(256);
-            counter.inc();
+        assert_eq!(counter.get(), 256);
-            assert_eq!(counter.get(), 1);
-            counter.add(5);
-            assert_eq!(counter.get(), 6);
-            counter.set(10);
-            assert_eq!(counter.get(), 10);
-            counter.reset();
-            assert_eq!(counter.get(), 0);
-        }
     }
 
     #[tokio::test]
@@ -853,11 +763,11 @@ mod tests {
         let stats = StatsManager::new();
 
         let counter1 = stats.get_simple_counter(MetricName::TrafficBytesTx);
-        counter1.set(100);
+        counter1.add(100);
 
         let labels = LabelSet::new().with_label("status", "success");
         let counter2 = stats.get_counter(MetricName::PeerRpcClientTx, labels);
-        counter2.set(50);
+        counter2.add(50);
 
         let traffic_labels = LabelSet::new()
             .with_label_type(LabelType::NetworkName("default".to_string()))
@@ -865,7 +775,7 @@ mod tests {
                 "87ede5a2-9c3d-492d-9bbe-989b9d07e742".to_string(),
             ));
         let counter3 = stats.get_counter(MetricName::TrafficBytesTxByInstance, traffic_labels);
-        counter3.set(25);
+        counter3.add(25);
 
         let prometheus_output = stats.export_prometheus();
 
@@ -885,7 +795,7 @@ mod tests {
 
         let labels = LabelSet::new().with_label("peer", "test");
         let counter = stats.get_counter(MetricName::PeerRpcClientTx, labels.clone());
-        counter.set(42);
+        counter.add(42);
 
         let metric = stats
             .get_metric(MetricName::PeerRpcClientTx, &labels)
@@ -902,11 +812,11 @@ mod tests {
 
         stats
             .get_simple_counter(MetricName::PeerRpcClientTx)
-            .set(10);
+            .add(10);
-        stats.get_simple_counter(MetricName::PeerRpcErrors).set(2);
+        stats.get_simple_counter(MetricName::PeerRpcErrors).add(2);
         stats
             .get_simple_counter(MetricName::TrafficBytesTx)
-            .set(100);
+            .add(100);
 
         let rpc_metrics = stats.get_metrics_by_prefix("peer_rpc");
         assert_eq!(rpc_metrics.len(), 2);
@@ -921,19 +831,15 @@ mod tests {
 
         // 创建一些计数器
         let counter1 = stats.get_simple_counter(MetricName::PeerRpcClientTx);
-        counter1.set(10);
+        counter1.add(10);
 
         let labels = LabelSet::new().with_label("test", "value");
         let counter2 = stats.get_counter(MetricName::TrafficBytesTx, labels);
-        counter2.set(20);
+        counter2.add(20);
 
         // 验证计数器存在
         assert_eq!(stats.metric_count(), 2);
 
-        // 注意：实际的清理测试需要等待3分钟，这在单元测试中不现实
-        // 这里我们只验证清理机制的基本结构是否正确
-        // 清理逻辑在后台线程中运行，会自动删除超过3分钟未更新的条目
-
         // 验证计数器仍然可以正常工作
         counter1.inc();
         assert_eq!(counter1.get(), 11);
@@ -946,14 +852,14 @@ mod tests {
     async fn test_cleanup_keeps_metrics_with_live_handles() {
         let stats = StatsManager::new();
         let counter = stats.get_simple_counter(MetricName::TrafficBytesForwarded);
-        counter.set(1);
+        counter.add(1);
 
-        let cutoff_time = Instant::now().checked_add(Duration::from_secs(1)).unwrap();
+        // Use a future cutoff so last_updated check always fails
+        let future_cutoff = now_monotonic_millis() + 1000;
         stats
             .counters
             .retain(|_, metric_data: &mut Arc<MetricData>| {
-                Arc::strong_count(metric_data) > 1
+                Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > future_cutoff
-                    || unsafe { metric_data.get_last_updated() > cutoff_time }
             });
 
         assert_eq!(stats.metric_count(), 1);
@@ -963,8 +869,7 @@ mod tests {
         stats
             .counters
             .retain(|_, metric_data: &mut Arc<MetricData>| {
-                Arc::strong_count(metric_data) > 1
+                Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > future_cutoff
-                    || unsafe { metric_data.get_last_updated() > cutoff_time }
             });
         assert_eq!(stats.metric_count(), 0);
     }

easytier/src/peers/secure_datagram.rs

@@ -990,13 +990,9 @@ mod tests {
 
         s.sync_root_key(root_key, 2, 2, true);
         assert!(s.check_replay_for_test(2, 0, SecureDatagramDirection::AToB, now + 2));
+        let expires_at = s.sync_rx_grace_expires_at_ms.load(Ordering::Relaxed);
 
-        assert!(!s.check_replay_for_test(
+        assert!(!s.check_replay_for_test(0, 1, SecureDatagramDirection::AToB, expires_at + 1));
-            0,
-            1,
-            SecureDatagramDirection::AToB,
-            now + SecureDatagramSession::SYNC_RX_GRACE_AFTER_MS + 3
-        ));
     }
 
     #[test]

easytier/src/tunnel/packet_def.rs

@@ -1,10 +1,10 @@
 use bytes::Buf;
 use bytes::Bytes;
 use bytes::BytesMut;
+use zerocopy::byteorder::*;
 use zerocopy::AsBytes;
 use zerocopy::FromBytes;
 use zerocopy::FromZeroes;
-use zerocopy::byteorder::*;
 
 type DefaultEndian = LittleEndian;
 
@@ -486,8 +486,16 @@ impl ZCPacket {
         let payload_off = ret.packet_type.get_packet_offsets().payload_offset;
         let total_len = payload_off + payload.len();
         ret.inner.reserve(total_len);
-        unsafe { ret.inner.set_len(total_len) };
+
-        ret.mut_payload().copy_from_slice(payload);
+        // SAFETY: `reserve` guarantees capacity >= total_len.
+        // We zero the header region and copy payload before advancing length,
+        // so every byte in [0..total_len) is initialized before any read.
+        unsafe {
+            let ptr = ret.inner.as_mut_ptr();
+            std::ptr::write_bytes(ptr, 0, payload_off);
+            std::ptr::copy_nonoverlapping(payload.as_ptr(), ptr.add(payload_off), payload.len());
+            ret.inner.set_len(total_len);
+        }
         ret
     }
 
@@ -495,12 +503,12 @@ impl ZCPacket {
         let mut ret = Self::new_nic_packet();
         ret.inner.reserve(cap);
         let total_len = ret.packet_type.get_packet_offsets().payload_offset - packet_info_len;
-        unsafe { ret.inner.set_len(total_len) };
+        ret.inner.resize(total_len, 0);
         ret
     }
 
     pub fn new_for_foreign_network(
-        network_name: &String,
+        network_name: &str,
         dst_peer_id: u32,
         foreign_zc_packet: &ZCPacket,
     ) -> Self {
@@ -509,26 +517,71 @@ impl ZCPacket {
             foreign_network_hdr.get_header_len() + foreign_zc_packet.tunnel_payload().len();
 
         let mut ret = Self::new_nic_packet();
-        let payload_off = ret.packet_type.get_packet_offsets().payload_offset;
+        let offsets = ret.packet_type.get_packet_offsets();
-        ret.inner.reserve(payload_off + total_payload_len);
+        let payload_off = offsets.payload_offset;
-        unsafe { ret.inner.set_len(payload_off + total_payload_len) };
+        let pm_hdr_off = offsets.peer_manager_header_offset;
+        let total_len = payload_off + total_payload_len;
+        ret.inner.reserve(total_len);
 
         let fixed_hdr_len = std::mem::size_of::<ForeignNetworkPacketHeader>();
-        ret.mut_payload()[..fixed_hdr_len].copy_from_slice(foreign_network_hdr.as_bytes());
-
         let name_offset = foreign_network_hdr.network_name_offset.get() as usize;
         let name_len = foreign_network_hdr.network_name_len.get() as usize;
-        ret.mut_payload()[name_offset..name_offset + name_len]
+        let foreign_payload = foreign_zc_packet.tunnel_payload();
-            .copy_from_slice(network_name.as_bytes());
 
-        ret.mut_payload()[foreign_network_hdr.get_header_len()..]
+        // Construct the PeerManagerHeader on the stack so we can write it
-            .copy_from_slice(foreign_zc_packet.tunnel_payload());
+        // directly into the buffer, avoiding a separate mut_peer_manager_header()
+        // call after set_len.
+        let pm_hdr = PeerManagerHeader {
+            from_peer_id: 0.into(),
+            to_peer_id: 0.into(),
+            packet_type: PacketType::ForeignNetworkPacket as u8,
+            flags: 0,
+            forward_counter: 0,
+            reserved: 0,
+            len: U32::new(total_payload_len as u32),
+        };
 
-        let hdr = ret.mut_peer_manager_header().unwrap();
+        // SAFETY: `reserve` guarantees capacity >= total_len.
-        hdr.from_peer_id = 0.into();
+        // We zero only the tunnel-header reserved space [0..pm_hdr_off], write
-        hdr.to_peer_id = 0.into();
+        // the PeerManagerHeader directly at pm_hdr_off, then copy the foreign
-        hdr.packet_type = PacketType::ForeignNetworkPacket as u8;
+        // network header, network name, and payload. Every byte in [0..total_len)
-        hdr.len.set(total_payload_len as u32);
+        // is initialized before set_len.
+        unsafe {
+            let ptr = ret.inner.as_mut_ptr();
+
+            // Zero the tunnel header reserved space only (not the PM header region)
+            std::ptr::write_bytes(ptr, 0, pm_hdr_off);
+
+            // Write PeerManagerHeader directly
+            std::ptr::copy_nonoverlapping(
+                pm_hdr.as_bytes().as_ptr(),
+                ptr.add(pm_hdr_off),
+                std::mem::size_of::<PeerManagerHeader>(),
+            );
+
+            // Copy foreign network fixed header
+            std::ptr::copy_nonoverlapping(
+                foreign_network_hdr.as_bytes().as_ptr(),
+                ptr.add(payload_off),
+                fixed_hdr_len,
+            );
+
+            // Copy network name
+            std::ptr::copy_nonoverlapping(
+                network_name.as_ptr(),
+                ptr.add(payload_off + name_offset),
+                name_len,
+            );
+
+            // Copy foreign payload
+            std::ptr::copy_nonoverlapping(
+                foreign_payload.as_ptr(),
+                ptr.add(payload_off + foreign_network_hdr.get_header_len()),
+                foreign_payload.len(),
+            );
+
+            ret.inner.set_len(total_len);
+        }
 
         ret
     }
@@ -700,7 +753,7 @@ impl ZCPacket {
                 .get_packet_offsets()
                 .peer_manager_header_offset;
             let mut buf = BytesMut::with_capacity(new_pm_offset + tunnel_payload.len());
-            unsafe { buf.set_len(new_pm_offset) };
+            buf.resize(new_pm_offset, 0);
             buf.extend_from_slice(tunnel_payload);
             return Self::new_from_buf(buf, target_packet_type);
         }

easytier/src/tunnel/stats.rs

@@ -1,7 +1,4 @@
-use std::{
+use std::sync::atomic::{AtomicU32, AtomicU64, Ordering::Relaxed};
-    cell::UnsafeCell,
-    sync::atomic::{AtomicU32, Ordering::Relaxed},
-};
 
 pub struct WindowLatency {
     latency_us_window: Vec<AtomicU32>,
@@ -63,34 +60,30 @@ impl WindowLatency {
 
 #[derive(Debug)]
 pub struct Throughput {
-    tx_bytes: UnsafeCell<u64>,
+    tx_bytes: AtomicU64,
-    rx_bytes: UnsafeCell<u64>,
+    rx_bytes: AtomicU64,
-    tx_packets: UnsafeCell<u64>,
+    tx_packets: AtomicU64,
-    rx_packets: UnsafeCell<u64>,
+    rx_packets: AtomicU64,
 }
 
 impl Clone for Throughput {
     fn clone(&self) -> Self {
         Self {
-            tx_bytes: UnsafeCell::new(unsafe { *self.tx_bytes.get() }),
+            tx_bytes: AtomicU64::new(self.tx_bytes.load(Relaxed)),
-            rx_bytes: UnsafeCell::new(unsafe { *self.rx_bytes.get() }),
+            rx_bytes: AtomicU64::new(self.rx_bytes.load(Relaxed)),
-            tx_packets: UnsafeCell::new(unsafe { *self.tx_packets.get() }),
+            tx_packets: AtomicU64::new(self.tx_packets.load(Relaxed)),
-            rx_packets: UnsafeCell::new(unsafe { *self.rx_packets.get() }),
+            rx_packets: AtomicU64::new(self.rx_packets.load(Relaxed)),
         }
     }
 }
 
-// add sync::Send and sync::Sync traits to Throughput
-unsafe impl Send for Throughput {}
-unsafe impl Sync for Throughput {}
-
 impl Default for Throughput {
     fn default() -> Self {
         Self {
-            tx_bytes: UnsafeCell::new(0),
+            tx_bytes: AtomicU64::new(0),
-            rx_bytes: UnsafeCell::new(0),
+            rx_bytes: AtomicU64::new(0),
-            tx_packets: UnsafeCell::new(0),
+            tx_packets: AtomicU64::new(0),
-            rx_packets: UnsafeCell::new(0),
+            rx_packets: AtomicU64::new(0),
         }
     }
 }
@@ -101,32 +94,28 @@ impl Throughput {
     }
 
     pub fn tx_bytes(&self) -> u64 {
-        unsafe { *self.tx_bytes.get() }
+        self.tx_bytes.load(Relaxed)
     }
 
     pub fn rx_bytes(&self) -> u64 {
-        unsafe { *self.rx_bytes.get() }
+        self.rx_bytes.load(Relaxed)
     }
 
     pub fn tx_packets(&self) -> u64 {
-        unsafe { *self.tx_packets.get() }
+        self.tx_packets.load(Relaxed)
     }
 
     pub fn rx_packets(&self) -> u64 {
-        unsafe { *self.rx_packets.get() }
+        self.rx_packets.load(Relaxed)
     }
 
     pub fn record_tx_bytes(&self, bytes: u64) {
-        unsafe {
+        self.tx_bytes.fetch_add(bytes, Relaxed);
-            *self.tx_bytes.get() += bytes;
+        self.tx_packets.fetch_add(1, Relaxed);
-            *self.tx_packets.get() += 1;
-        }
     }
 
     pub fn record_rx_bytes(&self, bytes: u64) {
-        unsafe {
+        self.rx_bytes.fetch_add(bytes, Relaxed);
-            *self.rx_bytes.get() += bytes;
+        self.rx_packets.fetch_add(1, Relaxed);
-            *self.rx_packets.get() += 1;
-        }
     }
 }