style: format noop waker mpsc tests

TCP tunnel uses FramedWriter (not RingSink), but start_send is still
sync (writes to BufList in memory). poll_flush does actual TCP write
syscall — noop_waker returns Ok for Pending (data stays in BufList,
flushed on next send when BufList >= 64).

Add TCP benchmark support via HOTPATH_TUNNEL=tcp. Note: TCP/UDP
convergence requires netns in bench environment (connector multi-bind
address behavior doesn't work for localhost without namespaces).

All 210 peers tests pass. Ring tunnel benchmark: 234K -> 508K pps (+117%).

.rust-analyzer.toml

@@ -1,32 +0,0 @@
-# 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,7 +2435,6 @@ dependencies = [
  "tempfile",
  "terminal_size",
  "thiserror 1.0.63",
- "thread_local",
  "thunk-rs",
  "tikv-jemalloc-ctl",
  "tikv-jemalloc-sys",

easytier/Cargo.toml

@@ -32,10 +32,6 @@ path = "src/lib.rs"
 name = "tx_throughput"
 harness = false
 
-[[bench]]
-name = "packet_bytes_extraction"
-harness = false
-
 [dependencies]
 git-version = "0.3.9"
 
@@ -224,7 +220,6 @@ 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"
 
@@ -345,7 +340,7 @@ zip = "4.0.0"
 
 
 [dev-dependencies]
-criterion = { version = "0.5", features = ["html_reports"] }
+criterion = "0.5.1"
 serial_test = "3.0.0"
 rstest = "0.25.0"
 futures-util = "0.3.31"
@@ -353,14 +348,6 @@ 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/README.md

@@ -1,47 +1,4 @@
-# Benchmarks
-
-Criterion benchmarks for EasyTier hot paths.
-
-| Bench                       | What it measures                                                                 |
-| --------------------------- | -------------------------------------------------------------------------------- |
-| `tx_throughput`             | End-to-end TX injection path through `peer_manager::send_msg_by_ip`              |
-| `packet_bytes_extraction`   | `ZCPacket::payload_bytes` / `tunnel_payload_bytes` extraction (advance hot path)  |
-
-## Packet Bytes Extraction
-
-Criterion benchmark for `ZCPacket` bytes extraction — the methods touched by the
-`advance`-based slicing refactor. Measures `payload_bytes` and
-`tunnel_payload_bytes` at two payload sizes (1280, 4096). Setup
-(`ZCPacket::new_with_payload`) runs in the benchmark harness's preparation
-phase and is excluded from the timed region, so the numbers reflect only the
-extraction call.
-
-### Quick start
-
-```bash
-cargo bench --bench packet_bytes_extraction
-```
-
-Smoke run:
-
-```bash
-PACKET_BYTES_MEASUREMENT_SECS=2 \
-PACKET_BYTES_WARMUP_SECS=1 \
-PACKET_BYTES_SAMPLE_SIZE=10 \
-cargo bench --bench packet_bytes_extraction -- --quiet
-```
-
-### Environment variables
-
-| Variable                        | Default | Notes                        |
-| ------------------------------- | ------- | ---------------------------- |
-| `PACKET_BYTES_MEASUREMENT_SECS` | `10`    | Criterion `measurement_time` |
-| `PACKET_BYTES_WARMUP_SECS`      | `3`     | Criterion `warm_up_time`     |
-| `PACKET_BYTES_SAMPLE_SIZE`      | `10`    | Criterion `sample_size` (min 10) |
-
----
-
-## TX Throughput Benchmark
+# TX Throughput Benchmark
 
 Criterion benchmark for EasyTier's TX injection path (`peer_manager::send_msg_by_ip`).
 

easytier/benches/acl_hotpath.rs

@@ -1,137 +0,0 @@
-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/packet_bytes_extraction.rs

@@ -1,65 +0,0 @@
-use std::hint::black_box;
-use std::time::Duration;
-
-use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main};
-
-use easytier::tunnel::packet_def::ZCPacket;
-
-const PAYLOAD_SIZES: &[usize] = &[1280, 4096];
-
-fn env_parse<T: std::str::FromStr>(key: &str, default: T) -> T {
-    std::env::var(key)
-        .ok()
-        .and_then(|v| v.parse().ok())
-        .unwrap_or(default)
-}
-
-fn bench_payload_bytes(c: &mut Criterion) {
-    let mut group = c.benchmark_group("payload_bytes");
-    for &size in PAYLOAD_SIZES {
-        let data = vec![0u8; size];
-        group.throughput(Throughput::Bytes(size as u64));
-        group.bench_with_input(format!("{size}"), &data, |b, data| {
-            b.iter_batched(
-                || ZCPacket::new_with_payload(black_box(data)),
-                |p| black_box(p).payload_bytes(),
-                BatchSize::SmallInput,
-            )
-        });
-    }
-    group.finish();
-}
-
-fn bench_tunnel_payload_bytes(c: &mut Criterion) {
-    let mut group = c.benchmark_group("tunnel_payload_bytes");
-    for &size in PAYLOAD_SIZES {
-        let data = vec![0u8; size];
-        group.throughput(Throughput::Bytes(size as u64));
-        group.bench_with_input(format!("{size}"), &data, |b, data| {
-            b.iter_batched(
-                || ZCPacket::new_with_payload(black_box(data)),
-                |p| black_box(p).tunnel_payload_bytes(),
-                BatchSize::SmallInput,
-            )
-        });
-    }
-    group.finish();
-}
-
-fn criterion_config() -> Criterion {
-    let measurement_secs = env_parse("PACKET_BYTES_MEASUREMENT_SECS", 10u64);
-    let warmup_secs = env_parse("PACKET_BYTES_WARMUP_SECS", 3u64);
-    let sample_size = env_parse("PACKET_BYTES_SAMPLE_SIZE", 10usize).max(10);
-
-    Criterion::default()
-        .measurement_time(Duration::from_secs(measurement_secs))
-        .warm_up_time(Duration::from_secs(warmup_secs))
-        .sample_size(sample_size)
-}
-
-criterion_group! {
-    name = benches;
-    config = criterion_config();
-    targets = bench_payload_bytes, bench_tunnel_payload_bytes
-}
-criterion_main!(benches);

easytier/benches/zc_packet.rs

@@ -1,51 +0,0 @@
-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,8 +3,7 @@ use std::{
     net::{IpAddr, SocketAddr},
     str::FromStr as _,
     sync::Arc,
-    sync::atomic::{AtomicU64, Ordering::Relaxed},
-    time::Duration,
+    time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
 use quanta::Instant;
@@ -77,41 +76,7 @@ pub struct FastLookupRule {
     pub stateful: bool,
     pub rate_limit: u32,
     pub burst_limit: u32,
-    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),
-            }),
-        }
-    }
+    pub rule_stats: Arc<RuleStats>,
 }
 
 // Cache key combining packet info and chain type
@@ -143,17 +108,17 @@ impl AclCacheKey {
 }
 
 // Cache entry with timestamp for LRU cleanup
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub(crate) struct AclCacheEntry {
     pub action: Action,
     pub matched_rule: RuleId,
-    pub last_access: AtomicU64,
+    pub last_access: Instant,
     // 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<RuleStatsTracker>>,
+    pub rule_stats_vec: Vec<Arc<RuleStats>>,
 }
 
 // Packet info extracted for ACL processing
@@ -246,7 +211,7 @@ pub struct AclProcessor {
     default_outbound_action: Action,
     default_forward_action: Action,
 
-    default_rule_stats: Arc<RuleStatsTracker>,
+    default_rule_stats: Arc<RuleStats>,
 
     // Connection tracking table - shared across different processor instances if needed
     conn_track: Arc<DashMap<String, ConnTrackEntry>>,
@@ -259,13 +224,6 @@ 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>>,
 
@@ -301,14 +259,18 @@ impl AclProcessor {
             default_outbound_action,
             default_forward_action,
 
-            default_rule_stats: Arc::new(RuleStatsTracker::new(None)),
+            default_rule_stats: Arc::new(RuleStats {
+                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,
         };
@@ -412,14 +374,11 @@ 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();
 
@@ -442,9 +401,10 @@ 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 now = crate::common::stats_manager::now_monotonic_millis();
-        let cutoff = now.saturating_sub(15_000);
-        cache.retain(|_, entry| entry.last_access.load(Relaxed) > cutoff);
+        let expired_timepoint = Instant::now()
+            .checked_sub(Duration::from_secs(15))
+            .unwrap_or(Instant::now());
+        cache.retain(|_, entry| entry.last_access > expired_timepoint);
 
         let current_size = cache.len();
         if current_size <= max_size {
@@ -452,9 +412,9 @@ impl AclProcessor {
         }
 
         // Remove oldest entries (LRU cleanup)
-        let mut entries: Vec<(AclCacheKey, u64)> = cache
+        let mut entries: Vec<(AclCacheKey, Instant)> = cache
             .iter()
-            .map(|entry| (entry.key().clone(), entry.value().last_access.load(Relaxed)))
+            .map(|entry| (entry.key().clone(), entry.value().last_access))
             .collect();
 
         // Sort by last_access (oldest first)
@@ -473,7 +433,6 @@ impl AclProcessor {
         );
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
     pub(crate) fn process_packet_with_cache_entry(
         &self,
         packet_info: &PacketInfo,
@@ -500,41 +459,42 @@ 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() {
-            rule_stats.increment(packet_info.packet_size);
+            // Use unsafe code to mutate the contents behind the Arc
+            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.snapshot());
+            stats.push((*rule.rule_stats).clone());
         }
         for rule in self.outbound_rules.iter() {
-            stats.push(rule.rule_stats.snapshot());
+            stats.push((*rule.rule_stats).clone());
         }
         for rule in self.forward_rules.iter() {
-            stats.push(rule.rule_stats.snapshot());
+            stats.push((*rule.rule_stats).clone());
         }
         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.
-        // Use get() (read lock) instead of get_mut() (write lock) and update
-        // last_access via AtomicU64, avoiding expensive Instant::now().
-        if let Some(cached) = self.rule_cache.get(&cache_key) {
-            cached
-                .last_access
-                .store(self.coarse_millis.load(Relaxed), Relaxed);
-            self.cache_hits.fetch_add(1, Relaxed);
+        // If cache hit and can skip checks, return cached result
+        if let Some(mut cached) = self.rule_cache.get_mut(&cache_key) {
+            // Update last access time for LRU
+            cached.last_access = Instant::now();
+
+            self.increment_stat(AclStatKey::CacheHits);
             return self.process_packet_with_cache_entry(packet_info, &cached);
         }
 
@@ -556,7 +516,7 @@ impl AclProcessor {
         let mut cache_entry = AclCacheEntry {
             action: Action::Allow,
             matched_rule: RuleId::Default,
-            last_access: AtomicU64::new(self.coarse_millis.load(Relaxed)),
+            last_access: Instant::now(),
             conn_track_key: None,
             rate_limit_keys: vec![],
             chain_type,
@@ -621,9 +581,8 @@ impl AclProcessor {
             // Cache the result with rule info
             self.increment_stat(AclStatKey::RuleMatches);
             self.inc_cache_entry_stats(&cache_entry, packet_info);
-            let result = cache_entry.acl_result.clone().unwrap();
-            self.cache_result(&cache_key, cache_entry);
-            return result;
+            self.cache_result(&cache_key, cache_entry.clone());
+            return cache_entry.acl_result.clone().unwrap();
         }
 
         let default_action = match chain_type {
@@ -659,9 +618,8 @@ impl AclProcessor {
 
         // Cache the default result (no rule info)
         self.inc_cache_entry_stats(&cache_entry, packet_info);
-        let result = cache_entry.acl_result.clone().unwrap();
-        self.cache_result(&cache_key, cache_entry);
-        result
+        self.cache_result(&cache_key, cache_entry.clone());
+        cache_entry.acl_result.clone().unwrap()
     }
 
     /// Get shared state for preserving across hot reloads
@@ -785,11 +743,13 @@ 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 = now;
+                x.last_seen = SystemTime::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;
@@ -803,8 +763,14 @@ impl AclProcessor {
                 ),
                 protocol: packet_info.protocol as i32,
                 state: ConnState::New as i32,
-                created_at: now,
-                last_seen: now,
+                created_at: SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .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,
             });
@@ -896,7 +862,13 @@ impl AclProcessor {
             stateful: rule.stateful,
             rate_limit: rule.rate_limit,
             burst_limit: rule.burst_limit,
-            rule_stats: Arc::new(RuleStatsTracker::new(Some(rule.clone()))),
+            rule_stats: Arc::new(RuleStats {
+                rule: Some(rule.clone()),
+                stat: Some(StatItem {
+                    packet_count: 0,
+                    byte_count: 0,
+                }),
+            }),
         }
     }
 
@@ -922,10 +894,6 @@ 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(),
@@ -940,11 +908,14 @@ impl AclProcessor {
         conn_track: Arc<DashMap<String, ConnTrackEntry>>,
         timeout_secs: u64,
     ) {
-        let current_time = current_unix_secs();
+        let current_time = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs();
         let keys_to_remove: Vec<String> = conn_track
             .iter()
             .filter_map(|entry| {
-                if current_time.saturating_sub(entry.last_seen) > timeout_secs {
+                if current_time - entry.last_seen > timeout_secs {
                     Some(entry.key().clone())
                 } else {
                     None
@@ -959,7 +930,11 @@ impl AclProcessor {
 
     /// Get cache hit rate
     pub fn get_cache_hit_rate(&self) -> f64 {
-        let cache_hits = self.cache_hits.load(Relaxed);
+        let cache_hits = self
+            .stats
+            .get(&AclStatKey::CacheHits)
+            .map(|v| *v.value())
+            .unwrap_or(0);
         let total_requests = cache_hits
             + self
                 .stats
@@ -975,13 +950,6 @@ 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/global_ctx.rs

@@ -219,7 +219,6 @@ pub struct GlobalCtx {
 
     running_listeners: Mutex<Vec<url::Url>>,
     advertised_ipv6_public_addr_prefix: Mutex<Option<cidr::Ipv6Cidr>>,
-    tun_device_name: Mutex<Option<String>>,
 
     flags: ArcSwap<Flags>,
 
@@ -337,7 +336,6 @@ impl GlobalCtx {
 
             running_listeners: Mutex::new(Vec::new()),
             advertised_ipv6_public_addr_prefix: Mutex::new(None),
-            tun_device_name: Mutex::new(None),
 
             flags: ArcSwap::new(Arc::new(flags)),
 
@@ -372,24 +370,6 @@ impl GlobalCtx {
         }
     }
 
-    fn set_tun_device_name(&self, name: Option<String>) {
-        *self.tun_device_name.lock().unwrap() = name;
-    }
-
-    pub(crate) fn set_tun_device_ready(&self, name: String) {
-        self.set_tun_device_name(Some(name.clone()));
-        self.issue_event(GlobalCtxEvent::TunDeviceReady(name));
-    }
-
-    pub(crate) fn set_tun_device_error(&self, error: String) {
-        self.set_tun_device_name(None);
-        self.issue_event(GlobalCtxEvent::TunDeviceError(error));
-    }
-
-    pub fn get_tun_device_name(&self) -> Option<String> {
-        self.tun_device_name.lock().unwrap().clone()
-    }
-
     pub fn check_network_in_whitelist(&self, network_name: &str) -> Result<(), anyhow::Error> {
         if self
             .get_flags()
@@ -845,36 +825,6 @@ pub mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn test_tun_device_name_tracks_explicit_runtime_state() {
-        let config = TomlConfigLoader::default();
-        let global_ctx = GlobalCtx::new(config);
-
-        assert_eq!(global_ctx.get_tun_device_name(), None);
-
-        global_ctx.issue_event(GlobalCtxEvent::TunDeviceReady("ignored".to_string()));
-        assert_eq!(global_ctx.get_tun_device_name(), None);
-
-        let mut subscriber = global_ctx.subscribe();
-
-        global_ctx.set_tun_device_ready("easytier0".to_string());
-        assert_eq!(
-            global_ctx.get_tun_device_name(),
-            Some("easytier0".to_string())
-        );
-        assert_eq!(
-            subscriber.recv().await.unwrap(),
-            GlobalCtxEvent::TunDeviceReady("easytier0".to_string())
-        );
-
-        global_ctx.set_tun_device_error("closed".to_string());
-        assert_eq!(global_ctx.get_tun_device_name(), None);
-        assert_eq!(
-            subscriber.recv().await.unwrap(),
-            GlobalCtxEvent::TunDeviceError("closed".to_string())
-        );
-    }
-
     #[tokio::test]
     async fn trusted_key_source_lookup_is_precise() {
         let config = TomlConfigLoader::default();

easytier/src/common/ifcfg/mod.rs

@@ -177,20 +177,3 @@ pub(crate) fn list_ipv6_route_messages()
 pub(crate) fn get_interface_index(name: &str) -> Result<u32, Error> {
     netlink::NetlinkIfConfiger::get_interface_index(name)
 }
-
-#[cfg(target_os = "linux")]
-pub(crate) fn add_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
-    netlink::NetlinkIfConfiger::add_ipv6_ndp_proxy(name, address)
-}
-
-#[cfg(target_os = "linux")]
-pub(crate) fn remove_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
-    netlink::NetlinkIfConfiger::remove_ipv6_ndp_proxy(name, address)
-}
-
-#[cfg(target_os = "linux")]
-pub(crate) fn list_ipv6_ndp_proxy(
-    name: &str,
-) -> Result<std::collections::BTreeSet<Ipv6Addr>, Error> {
-    netlink::NetlinkIfConfiger::list_ipv6_ndp_proxy(name)
-}

easytier/src/common/ifcfg/netlink.rs

@@ -1,5 +1,4 @@
 use std::{
-    collections::BTreeSet,
     ffi::CString,
     fmt::Debug,
     net::{IpAddr, Ipv4Addr, Ipv6Addr},
@@ -17,10 +16,6 @@ use netlink_packet_core::{
 use netlink_packet_route::{
     AddressFamily, RouteNetlinkMessage,
     address::{AddressAttribute, AddressMessage},
-    neighbour::{
-        NeighbourAddress, NeighbourAttribute, NeighbourFlags, NeighbourHeader, NeighbourMessage,
-        NeighbourState,
-    },
     route::{
         RouteAddress, RouteAttribute, RouteHeader, RouteMessage, RouteProtocol, RouteScope,
         RouteType,
@@ -380,105 +375,6 @@ impl NetlinkIfConfiger {
     pub(crate) fn list_ipv6_route_messages() -> Result<Vec<RouteMessage>, Error> {
         Self::list_route_messages(AddressFamily::Inet6)
     }
-
-    fn ipv6_ndp_proxy_message(name: &str, address: Ipv6Addr) -> Result<NeighbourMessage, Error> {
-        let mut message = NeighbourMessage::default();
-        message.header = NeighbourHeader {
-            family: AddressFamily::Inet6,
-            ifindex: Self::get_interface_index(name)?,
-            state: NeighbourState::Permanent,
-            flags: NeighbourFlags::Proxy,
-            kind: RouteType::Unicast,
-        };
-        message
-            .attributes
-            .push(NeighbourAttribute::Destination(NeighbourAddress::Inet6(
-                address,
-            )));
-        Ok(message)
-    }
-
-    pub(crate) fn add_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
-        send_netlink_req_and_wait_one_resp(
-            RouteNetlinkMessage::NewNeighbour(Self::ipv6_ndp_proxy_message(name, address)?),
-            false,
-        )
-    }
-
-    pub(crate) fn remove_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
-        send_netlink_req_and_wait_one_resp(
-            RouteNetlinkMessage::DelNeighbour(Self::ipv6_ndp_proxy_message(name, address)?),
-            true,
-        )
-    }
-
-    fn list_neighbour_messages(
-        address_family: AddressFamily,
-    ) -> Result<Vec<NeighbourMessage>, Error> {
-        let mut message = NeighbourMessage::default();
-        message.header.family = address_family;
-        message.header.flags = NeighbourFlags::Proxy;
-
-        let s = send_netlink_req(
-            RouteNetlinkMessage::GetNeighbour(message),
-            NLM_F_REQUEST | NLM_F_DUMP,
-        )?;
-
-        let mut ret_vec = vec![];
-        let mut resp = Vec::<u8>::new();
-        loop {
-            if resp.is_empty() {
-                let (new_resp, _) = s.recv_from_full()?;
-                resp = new_resp;
-            }
-
-            let ret = NetlinkMessage::<RouteNetlinkMessage>::deserialize(&resp)
-                .with_context(|| "Failed to deserialize netlink neighbour message")?;
-            resp = resp.split_off(ret.buffer_len());
-
-            tracing::debug!("net link response <<< {:?}", ret);
-
-            match ret.payload {
-                NetlinkPayload::Error(e) => {
-                    if e.code == NonZero::new(0) {
-                        continue;
-                    } else {
-                        return Err(e.to_io().into());
-                    }
-                }
-                NetlinkPayload::InnerMessage(RouteNetlinkMessage::NewNeighbour(m)) => {
-                    ret_vec.push(m);
-                }
-                NetlinkPayload::Done(_) => {
-                    break;
-                }
-                p => {
-                    tracing::error!("Unexpected netlink response: {:?}", p);
-                    return Err(anyhow::anyhow!("Unexpected netlink response").into());
-                }
-            }
-        }
-
-        Ok(ret_vec)
-    }
-
-    pub(crate) fn list_ipv6_ndp_proxy(name: &str) -> Result<BTreeSet<Ipv6Addr>, Error> {
-        let ifindex = Self::get_interface_index(name)?;
-
-        Ok(Self::list_neighbour_messages(AddressFamily::Inet6)?
-            .into_iter()
-            .filter(|message| {
-                message.header.ifindex == ifindex
-                    && message.header.flags.contains(NeighbourFlags::Proxy)
-            })
-            .filter_map(|message| {
-                message.attributes.into_iter().find_map(|attr| match attr {
-                    NeighbourAttribute::Destination(NeighbourAddress::Inet6(addr)) => Some(addr),
-                    _ => None,
-                })
-            })
-            .collect())
-    }
 }
 
 #[async_trait]

easytier/src/common/mod.rs

@@ -22,7 +22,6 @@ 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

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

easytier/src/instance/instance.rs

@@ -65,9 +65,9 @@ use crate::vpn_portal::{self, VpnPortal};
 use super::dns_server::{MAGIC_DNS_FAKE_IP, runner::DnsRunner};
 use super::listeners::ListenerManager;
 use super::public_ipv6_provider::{
-    PublicIpv6ProviderReconcileTask, reconcile_public_ipv6_provider_runtime,
-    run_public_ipv6_provider_reconcile_task, should_run_public_ipv6_provider_reconcile,
-    validate_public_ipv6_config, validate_public_ipv6_config_values,
+    reconcile_public_ipv6_provider_runtime, run_public_ipv6_provider_reconcile_task,
+    should_run_public_ipv6_provider_reconcile, validate_public_ipv6_config,
+    validate_public_ipv6_config_values,
 };
 
 #[cfg(feature = "socks5")]
@@ -194,44 +194,6 @@ impl NicCtxContainer {
 
 #[cfg(feature = "tun")]
 type ArcNicCtx = Arc<Mutex<Option<NicCtxContainer>>>;
-type ArcPublicIpv6ProviderTaskSlot = Arc<PublicIpv6ProviderTaskSlot>;
-
-struct PublicIpv6ProviderTaskSlot {
-    task: Mutex<Option<PublicIpv6ProviderReconcileTask>>,
-    closing: AtomicBool,
-}
-
-impl PublicIpv6ProviderTaskSlot {
-    fn new() -> Self {
-        Self {
-            task: Mutex::new(None),
-            closing: AtomicBool::new(false),
-        }
-    }
-
-    async fn ensure_started(&self, global_ctx: &ArcGlobalCtx) {
-        let mut task = self.task.lock().await;
-        if self.closing.load(Ordering::Acquire) || task.is_some() {
-            return;
-        }
-        *task = run_public_ipv6_provider_reconcile_task(global_ctx);
-    }
-
-    async fn shutdown(&self) {
-        self.closing.store(true, Ordering::Release);
-        let task = self.task.lock().await.take();
-        if let Some(task) = task {
-            task.shutdown().await;
-        }
-    }
-}
-
-async fn ensure_public_ipv6_provider_reconcile_task(
-    global_ctx: &ArcGlobalCtx,
-    task_slot: &ArcPublicIpv6ProviderTaskSlot,
-) {
-    task_slot.ensure_started(global_ctx).await;
-}
 
 pub struct InstanceRpcServerHook {
     rpc_portal_whitelist: Vec<IpCidr>,
@@ -292,7 +254,6 @@ pub struct InstanceConfigPatcher {
     socks5_server: Weak<Socks5Server>,
     peer_manager: Weak<PeerManager>,
     conn_manager: Weak<ManualConnectorManager>,
-    public_ipv6_provider_task: ArcPublicIpv6ProviderTaskSlot,
 }
 
 impl InstanceConfigPatcher {
@@ -363,6 +324,7 @@ impl InstanceConfigPatcher {
         self.patch_mapped_listeners(patch.mapped_listeners).await?;
         self.patch_connector(patch.connectors).await?;
 
+        let provider_reconcile_was_running = should_run_public_ipv6_provider_reconcile(&global_ctx);
         let mut provider_config_changed = false;
         if let Some(hostname) = patch.hostname {
             global_ctx.set_hostname(hostname.clone());
@@ -400,12 +362,10 @@ impl InstanceConfigPatcher {
         if provider_config_changed {
             reconcile_public_ipv6_provider_runtime(&global_ctx).await;
 
-            if should_run_public_ipv6_provider_reconcile(&global_ctx) {
-                ensure_public_ipv6_provider_reconcile_task(
-                    &global_ctx,
-                    &self.public_ipv6_provider_task,
-                )
-                .await;
+            let provider_reconcile_should_run =
+                should_run_public_ipv6_provider_reconcile(&global_ctx);
+            if !provider_reconcile_was_running && provider_reconcile_should_run {
+                run_public_ipv6_provider_reconcile_task(&global_ctx);
             }
         }
 
@@ -687,7 +647,6 @@ pub struct Instance {
     socks5_server: Arc<Socks5Server>,
 
     proxy_cidrs_monitor: Option<AbortOnDropHandle<()>>,
-    public_ipv6_provider_task: ArcPublicIpv6ProviderTaskSlot,
 
     global_ctx: ArcGlobalCtx,
 }
@@ -775,7 +734,6 @@ impl Instance {
             socks5_server,
 
             proxy_cidrs_monitor: None,
-            public_ipv6_provider_task: Arc::new(PublicIpv6ProviderTaskSlot::new()),
 
             global_ctx,
         }
@@ -1076,11 +1034,7 @@ impl Instance {
             .await?;
         self.listener_manager.lock().await.run().await?;
         self.peer_manager.run().await?;
-        ensure_public_ipv6_provider_reconcile_task(
-            &self.global_ctx,
-            &self.public_ipv6_provider_task,
-        )
-        .await;
+        run_public_ipv6_provider_reconcile_task(&self.global_ctx);
 
         #[cfg(feature = "tun")]
         {
@@ -1393,7 +1347,6 @@ impl Instance {
             socks5_server: Arc::downgrade(&self.socks5_server),
             peer_manager: Arc::downgrade(&self.peer_manager),
             conn_manager: Arc::downgrade(&self.conn_manager),
-            public_ipv6_provider_task: self.public_ipv6_provider_task.clone(),
         }
     }
 
@@ -1649,7 +1602,6 @@ impl Instance {
     }
 
     pub async fn clear_resources(&mut self) {
-        self.public_ipv6_provider_task.shutdown().await;
         self.peer_manager.clear_resources().await;
         #[cfg(feature = "tun")]
         let _ = self.nic_ctx.lock().await.take();
@@ -1835,21 +1787,6 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn public_ipv6_provider_task_slot_does_not_restart_after_shutdown() {
-        let global_ctx = get_mock_global_ctx();
-        let slot = std::sync::Arc::new(super::PublicIpv6ProviderTaskSlot::new());
-        global_ctx.config.set_ipv6_public_addr_provider(true);
-        global_ctx
-            .config
-            .set_ipv6_public_addr_prefix(Some("2001:db8::/48".parse().unwrap()));
-
-        slot.shutdown().await;
-        super::ensure_public_ipv6_provider_reconcile_task(&global_ctx, &slot).await;
-
-        assert!(slot.task.lock().await.is_none());
-    }
-
     #[tokio::test]
     async fn validate_public_ipv6_patch_allows_enabling_auto_with_manual_ipv6() {
         let global_ctx = get_mock_global_ctx();

easytier/src/instance/virtual_nic.rs

@@ -24,7 +24,7 @@ use crate::{
 };
 
 use byteorder::WriteBytesExt as _;
-use bytes::{Buf, BufMut, BytesMut};
+use bytes::{BufMut, BytesMut};
 use cidr::{Ipv4Inet, Ipv6Inet};
 use futures::{SinkExt, Stream, StreamExt, lock::BiLock, ready};
 use pin_project_lite::pin_project;
@@ -180,13 +180,12 @@ impl ZCPacketToBytes for TunZCPacketToBytes {
         assert!(payload_offset >= 4);
 
         let ret = if self.has_packet_info {
-            inner.advance(payload_offset - 4);
+            let mut inner = inner.split_off(payload_offset - 4);
             let proto = infer_proto(&inner[4..]);
             self.fill_packet_info(&mut inner[0..4], proto)?;
             inner
         } else {
-            inner.advance(payload_offset);
-            inner
+            inner.split_off(payload_offset)
         };
 
         tracing::debug!(?ret, ?payload_offset, "convert zc packet to tun packet");
@@ -1362,11 +1361,12 @@ impl NicCtx {
                     }
 
                     self.global_ctx
-                        .set_tun_device_ready(nic.ifname().to_string());
+                        .issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
                     ret
                 }
                 Err(err) => {
-                    self.global_ctx.set_tun_device_error(err.to_string());
+                    self.global_ctx
+                        .issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
                     return Err(err);
                 }
             }
@@ -1405,11 +1405,12 @@ impl NicCtx {
             match nic.create_dev_for_mobile(tun_fd).await {
                 Ok(ret) => {
                     self.global_ctx
-                        .set_tun_device_ready(nic.ifname().to_string());
+                        .issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
                     ret
                 }
                 Err(err) => {
-                    self.global_ctx.set_tun_device_error(err.to_string());
+                    self.global_ctx
+                        .issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
                     return Err(err);
                 }
             }

easytier/src/launcher.rs

@@ -173,12 +173,7 @@ impl EasyTierLauncher {
         #[cfg(mobile)]
         Self::run_routine_for_mobile(&instance, &data, &mut tasks).await;
 
-        if let Err(err) = instance.run().await {
-            tasks.abort_all();
-            drop(tasks);
-            instance.clear_resources().await;
-            return Err(err.into());
-        }
+        instance.run().await?;
 
         #[cfg(feature = "ffi-dataplane")]
         data.data_plane

easytier/src/peers/peer_conn.rs

@@ -363,7 +363,15 @@ impl PeerConn {
         let throughput = peer_conn_tunnel_filter.filter_output();
         let filter_chain = TunnelFilterChain::new(session_filter.clone(), peer_conn_tunnel_filter);
         let peer_conn_tunnel = TunnelWithFilter::new(tunnel, filter_chain);
-        let mut mpsc_tunnel = MpscTunnel::new(peer_conn_tunnel, Some(Duration::from_secs(7)));
+        let supports_direct = peer_conn_tunnel
+            .info()
+            .map(|i| matches!(i.tunnel_type.as_str(), "ring" | "udp" | "tcp"))
+            .unwrap_or(false);
+        let mut mpsc_tunnel = if supports_direct {
+            MpscTunnel::new_direct(peer_conn_tunnel)
+        } else {
+            MpscTunnel::new(peer_conn_tunnel, Some(Duration::from_secs(7)))
+        };
 
         let (recv, sink) = (mpsc_tunnel.get_stream(), mpsc_tunnel.get_sink());
 

easytier/src/peers/secure_datagram.rs

@@ -990,9 +990,13 @@ 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(0, 1, SecureDatagramDirection::AToB, expires_at + 1));
+        assert!(!s.check_replay_for_test(
+            0,
+            1,
+            SecureDatagramDirection::AToB,
+            now + SecureDatagramSession::SYNC_RX_GRACE_AFTER_MS + 3
+        ));
     }
 
     #[test]

easytier/src/tests/three_node.rs

@@ -477,12 +477,6 @@ struct PublicIpv6Lab {
     extra_bridges: [&'static str; 2],
 }
 
-#[derive(Clone, Copy)]
-enum PublicIpv6LabTopology {
-    DelegatedPrefix,
-    OnLinkPrefix,
-}
-
 impl PublicIpv6Lab {
     const PROVIDER_NS: &'static str = "net_a";
     const CLIENT_NS: &'static str = "net_b";
@@ -496,13 +490,11 @@ impl PublicIpv6Lab {
     const PROVIDER_DEFAULT_FROM: &'static str = "2001:db8:100::/64";
     const PROVIDER_WAN_ADDR: &'static str = "2001:db8:ffff:1::2/64";
     const UPSTREAM_WAN_ADDR: &'static str = "2001:db8:ffff:1::1/64";
-    const ON_LINK_PROVIDER_WAN_ADDR: &'static str = "2001:db8:100::2/64";
-    const ON_LINK_UPSTREAM_WAN_ADDR: &'static str = "2001:db8:100::1/64";
     const UPSTREAM_SERVER_ADDR: &'static str = "2001:db8:ffff:2::1/64";
     const SERVER_ADDR: &'static str = "2001:db8:ffff:2::100/64";
     const SERVER_IP: &'static str = "2001:db8:ffff:2::100";
 
-    fn setup_with_topology(topology: PublicIpv6LabTopology) -> Self {
+    fn setup() -> Self {
         prepare_linux_namespaces();
 
         del_netns(Self::UPSTREAM_NS);
@@ -552,23 +544,13 @@ impl PublicIpv6Lab {
             Self::SERVER_BRIDGE,
         );
 
-        let (provider_wan_addr, upstream_wan_addr) = match topology {
-            PublicIpv6LabTopology::DelegatedPrefix => {
-                (Self::PROVIDER_WAN_ADDR, Self::UPSTREAM_WAN_ADDR)
-            }
-            PublicIpv6LabTopology::OnLinkPrefix => (
-                Self::ON_LINK_PROVIDER_WAN_ADDR,
-                Self::ON_LINK_UPSTREAM_WAN_ADDR,
-            ),
-        };
-
         run_ip_in_ns(
             Self::PROVIDER_NS,
-            &["addr", "add", provider_wan_addr, "dev", "pubwan0"],
+            &["addr", "add", Self::PROVIDER_WAN_ADDR, "dev", "pubwan0"],
         );
         run_ip_in_ns(
             Self::UPSTREAM_NS,
-            &["addr", "add", upstream_wan_addr, "dev", "upwan0"],
+            &["addr", "add", Self::UPSTREAM_WAN_ADDR, "dev", "upwan0"],
         );
         run_ip_in_ns(
             Self::UPSTREAM_NS,
@@ -579,56 +561,37 @@ impl PublicIpv6Lab {
             &["addr", "add", Self::SERVER_ADDR, "dev", "srv0"],
         );
 
-        match topology {
-            PublicIpv6LabTopology::DelegatedPrefix => {
-                run_ip_in_ns(
-                    Self::PROVIDER_NS,
-                    &["link", "add", "pubprefix0", "type", "dummy"],
-                );
-                run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
-                run_ip_in_ns(
-                    Self::PROVIDER_NS,
-                    &[
-                        "-6",
-                        "route",
-                        "add",
-                        Self::PROVIDER_PREFIX,
-                        "dev",
-                        "pubprefix0",
-                    ],
-                );
-                run_ip_in_ns(
-                    Self::PROVIDER_NS,
-                    &[
-                        "-6",
-                        "route",
-                        "add",
-                        "default",
-                        "from",
-                        Self::PROVIDER_DEFAULT_FROM,
-                        "via",
-                        "2001:db8:ffff:1::1",
-                        "dev",
-                        "pubwan0",
-                    ],
-                );
-            }
-            PublicIpv6LabTopology::OnLinkPrefix => {
-                run_ip_in_ns(
-                    Self::PROVIDER_NS,
-                    &[
-                        "-6",
-                        "route",
-                        "add",
-                        "default",
-                        "via",
-                        "2001:db8:100::1",
-                        "dev",
-                        "pubwan0",
-                    ],
-                );
-            }
-        }
+        run_ip_in_ns(
+            Self::PROVIDER_NS,
+            &["link", "add", "pubprefix0", "type", "dummy"],
+        );
+        run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
+        run_ip_in_ns(
+            Self::PROVIDER_NS,
+            &[
+                "-6",
+                "route",
+                "add",
+                Self::PROVIDER_PREFIX,
+                "dev",
+                "pubprefix0",
+            ],
+        );
+        run_ip_in_ns(
+            Self::PROVIDER_NS,
+            &[
+                "-6",
+                "route",
+                "add",
+                "default",
+                "from",
+                Self::PROVIDER_DEFAULT_FROM,
+                "via",
+                "2001:db8:ffff:1::1",
+                "dev",
+                "pubwan0",
+            ],
+        );
 
         run_ip_in_ns(
             Self::SERVER_NS,
@@ -643,21 +606,19 @@ impl PublicIpv6Lab {
                 "srv0",
             ],
         );
-        if matches!(topology, PublicIpv6LabTopology::DelegatedPrefix) {
-            run_ip_in_ns(
-                Self::UPSTREAM_NS,
-                &[
-                    "-6",
-                    "route",
-                    "add",
-                    Self::PROVIDER_PREFIX,
-                    "via",
-                    "2001:db8:ffff:1::2",
-                    "dev",
-                    "upwan0",
-                ],
-            );
-        }
+        run_ip_in_ns(
+            Self::UPSTREAM_NS,
+            &[
+                "-6",
+                "route",
+                "add",
+                Self::PROVIDER_PREFIX,
+                "via",
+                "2001:db8:ffff:1::2",
+                "dev",
+                "upwan0",
+            ],
+        );
 
         run_sysctl_in_ns(Self::PROVIDER_NS, "net.ipv6.conf.all.forwarding=1");
         run_sysctl_in_ns(Self::UPSTREAM_NS, "net.ipv6.conf.all.forwarding=1");
@@ -711,15 +672,7 @@ fn get_public_ipv6_config(
 async fn init_public_ipv6_two_node(
     client_inst_id: uuid::Uuid,
 ) -> (PublicIpv6Lab, Instance, Instance) {
-    init_public_ipv6_two_node_with_topology(client_inst_id, PublicIpv6LabTopology::DelegatedPrefix)
-        .await
-}
-
-async fn init_public_ipv6_two_node_with_topology(
-    client_inst_id: uuid::Uuid,
-    topology: PublicIpv6LabTopology,
-) -> (PublicIpv6Lab, Instance, Instance) {
-    let lab = PublicIpv6Lab::setup_with_topology(topology);
+    let lab = PublicIpv6Lab::setup();
 
     let provider_cfg = get_public_ipv6_config(
         "provider_public_ipv6",
@@ -803,13 +756,6 @@ fn addr_exists_in_ns(ns: &str, dev: &str, needle: &str) -> bool {
     run_ip_in_ns_output(ns, &["-6", "addr", "show", "dev", dev]).contains(needle)
 }
 
-fn ndp_proxy_exists_in_ns(ns: &str, dev: &str, addr: std::net::Ipv6Addr) -> bool {
-    let addr = addr.to_string();
-    run_ip_in_ns_output(ns, &["-6", "neigh", "show", "proxy", "dev", dev])
-        .lines()
-        .any(|line| line.split_whitespace().next() == Some(addr.as_str()))
-}
-
 #[tokio::test]
 #[serial_test::serial]
 pub async fn public_ipv6_auto_addr_end_to_end() {
@@ -932,67 +878,6 @@ pub async fn public_ipv6_auto_addr_end_to_end() {
     drop_insts(vec![provider, client]).await;
 }
 
-#[tokio::test]
-#[serial_test::serial]
-pub async fn public_ipv6_auto_addr_on_link_ndp_proxy_end_to_end() {
-    let client_id = uuid::Uuid::parse_str("44444444-4444-4444-4444-444444444444").unwrap();
-    let (_lab, provider, client) =
-        init_public_ipv6_two_node_with_topology(client_id, PublicIpv6LabTopology::OnLinkPrefix)
-            .await;
-
-    wait_for_condition(
-        || async {
-            provider
-                .get_global_ctx()
-                .get_advertised_ipv6_public_addr_prefix()
-                == Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap())
-        },
-        Duration::from_secs(10),
-    )
-    .await;
-
-    let leased = wait_for_public_ipv6_addr(&client).await;
-    wait_for_public_ipv6_route(&provider, leased).await;
-
-    wait_for_condition(
-        || async {
-            addr_exists_in_ns(
-                PublicIpv6Lab::CLIENT_NS,
-                PublicIpv6Lab::CLIENT_TUN,
-                &leased.to_string(),
-            ) && route_exists_in_ns(
-                PublicIpv6Lab::PROVIDER_NS,
-                &format!("{} dev {}", leased.address(), PublicIpv6Lab::PROVIDER_TUN),
-            )
-        },
-        Duration::from_secs(10),
-    )
-    .await;
-
-    wait_for_condition(
-        || async {
-            ndp_proxy_exists_in_ns(PublicIpv6Lab::PROVIDER_NS, "pubwan0", leased.address())
-        },
-        Duration::from_secs(20),
-    )
-    .await;
-
-    wait_for_condition(
-        || async {
-            ping6_test(
-                PublicIpv6Lab::SERVER_NS,
-                leased.address().to_string().as_str(),
-                None,
-            )
-            .await
-        },
-        Duration::from_secs(20),
-    )
-    .await;
-
-    drop_insts(vec![provider, client]).await;
-}
-
 #[tokio::test]
 #[serial_test::serial]
 pub async fn public_ipv6_auto_addr_reconnect_reuses_same_address() {

easytier/src/tunnel/mpsc.rs

@@ -1,6 +1,13 @@
 // this mod wrap tunnel to a mpsc tunnel, based on crossbeam_channel
 
-use std::{pin::Pin, time::Duration};
+use std::{
+    cell::UnsafeCell,
+    pin::Pin,
+    sync::Arc,
+    sync::atomic::{AtomicBool, Ordering},
+    task::Poll,
+    time::Duration,
+};
 
 use anyhow::Context;
 use tokio::time::timeout;
@@ -11,34 +18,123 @@ use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPac
 
 use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
 use tokio_util::task::AbortOnDropHandle;
-// use tachyonix::{channel, Receiver, Sender, TrySendError};
 
 use futures::SinkExt;
 
+/// A simple spinlock protecting a sink. The guard is Send because it only
+/// contains an atomic flag reference (no lifetime-tied borrow like MutexGuard).
+struct SpinSink {
+    locked: AtomicBool,
+    sink: UnsafeCell<Pin<Box<dyn ZCPacketSink>>>,
+}
+
+// SAFETY: access is serialized by the spinlock.
+unsafe impl Send for SpinSink {}
+unsafe impl Sync for SpinSink {}
+
+struct SpinGuard<'a> {
+    spin: &'a SpinSink,
+}
+
+impl<'a> SpinGuard<'a> {
+    fn as_mut(&mut self) -> Pin<&mut dyn ZCPacketSink> {
+        // SAFETY: we hold the spinlock, so we have exclusive access
+        let sink = unsafe { &mut *self.spin.sink.get() };
+        sink.as_mut()
+    }
+}
+
+impl Drop for SpinGuard<'_> {
+    fn drop(&mut self) {
+        self.spin.locked.store(false, Ordering::Release);
+    }
+}
+
+impl SpinSink {
+    fn new(sink: Pin<Box<dyn ZCPacketSink>>) -> Self {
+        Self {
+            locked: AtomicBool::new(false),
+            sink: UnsafeCell::new(sink),
+        }
+    }
+
+    fn try_lock(&self) -> Option<SpinGuard<'_>> {
+        if self
+            .locked
+            .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
+            .is_ok()
+        {
+            Some(SpinGuard { spin: self })
+        } else {
+            None
+        }
+    }
+}
+
 #[derive(Clone)]
-pub struct MpscTunnelSender(Sender<ZCPacket>);
+pub struct MpscTunnelSender {
+    channel_tx: Option<Sender<ZCPacket>>,
+    direct_sink: Option<Arc<SpinSink>>,
+}
 
 impl MpscTunnelSender {
     pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        self.0.send(item).await.with_context(|| "send error")?;
-        Ok(())
+        if let Some(sink) = &self.direct_sink {
+            // Sync fast path: no await needed, returns immediately
+            if let Some(mut guard) = sink.try_lock() {
+                let waker = futures::task::noop_waker();
+                let mut cx = std::task::Context::from_waker(&waker);
+                match guard.as_mut().poll_ready(&mut cx) {
+                    Poll::Ready(Ok(())) => {
+                        guard.as_mut().start_send(item)?;
+                        // poll_flush may return Pending when the consumer task hasn't
+                        // drained the ring yet. The data is already in the ring buffer
+                        // and will be consumed — treat Pending as success.
+                        match guard.as_mut().poll_flush(&mut cx) {
+                            Poll::Ready(Err(e)) => return Err(e),
+                            _ => return Ok(()),
+                        }
+                    }
+                    Poll::Ready(Err(e)) => return Err(e),
+                    Poll::Pending => return Err(TunnelError::BufferFull),
+                }
+            }
+            return Err(TunnelError::BufferFull);
+        }
+
+        // Channel mode: async with backpressure
+        self.send_async(item).await
     }
 
     pub fn try_send(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        self.0.try_send(item).map_err(|e| match e {
+        let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
+        tx.try_send(item).map_err(|e| match e {
             TrySendError::Full(_) => TunnelError::BufferFull,
             TrySendError::Closed(_) => TunnelError::Shutdown,
         })
     }
+
+    pub async fn send_async(&self, item: ZCPacket) -> Result<(), TunnelError> {
+        let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
+        match tx.try_send(item) {
+            Ok(()) => Ok(()),
+            Err(TrySendError::Full(item)) => {
+                tx.send(item).await.with_context(|| "send error")?;
+                Ok(())
+            }
+            Err(TrySendError::Closed(_)) => Err(TunnelError::Shutdown),
+        }
+    }
 }
 
 pub struct MpscTunnel<T> {
     tx: Option<Sender<ZCPacket>>,
+    direct_sink: Option<Arc<SpinSink>>,
 
     tunnel: T,
     stream: Option<Pin<Box<dyn ZCPacketStream>>>,
 
-    task: AbortOnDropHandle<()>,
+    task: Option<AbortOnDropHandle<()>>,
 }
 
 impl<T: Tunnel> MpscTunnel<T> {
@@ -60,9 +156,21 @@ impl<T: Tunnel> MpscTunnel<T> {
 
         Self {
             tx: Some(tx),
+            direct_sink: None,
             tunnel,
             stream: Some(stream),
-            task: AbortOnDropHandle::new(task),
+            task: Some(AbortOnDropHandle::new(task)),
+        }
+    }
+
+    pub fn new_direct(tunnel: T) -> Self {
+        let (stream, sink) = tunnel.split();
+        Self {
+            tx: None,
+            direct_sink: Some(Arc::new(SpinSink::new(sink))),
+            tunnel,
+            stream: Some(stream),
+            task: None,
         }
     }
 
@@ -124,12 +232,18 @@ impl<T: Tunnel> MpscTunnel<T> {
     }
 
     pub fn get_sink(&self) -> MpscTunnelSender {
-        MpscTunnelSender(self.tx.as_ref().unwrap().clone())
+        MpscTunnelSender {
+            channel_tx: self.tx.as_ref().cloned(),
+            direct_sink: self.direct_sink.clone(),
+        }
     }
 
     pub fn close(&mut self) {
         self.tx.take();
-        self.task.abort();
+        self.direct_sink.take();
+        if let Some(task) = self.task.take() {
+            task.abort();
+        }
     }
 
     pub fn tunnel_info(&self) -> Option<TunnelInfo> {
@@ -189,7 +303,7 @@ mod tests {
             for i in 0..1000000 {
                 tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
                 let a = sink1
-                    .send(ZCPacket::new_with_payload("hello".as_bytes()))
+                    .send_async(ZCPacket::new_with_payload("hello".as_bytes()))
                     .await;
                 if a.is_err() {
                     tracing::info!(?a, "t2 exit with err");
@@ -209,7 +323,7 @@ mod tests {
             for i in 0..1000000 {
                 tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                 let a = sink2
-                    .send(ZCPacket::new_with_payload("hello2".as_bytes()))
+                    .send_async(ZCPacket::new_with_payload("hello2".as_bytes()))
                     .await;
                 if a.is_err() {
                     tracing::info!(?a, "t3 exit with err");

easytier/src/tunnel/packet_def.rs

@@ -1,10 +1,9 @@
-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,16 +485,8 @@ 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);
-
-        // 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);
-        }
+        unsafe { ret.inner.set_len(total_len) };
+        ret.mut_payload()[..payload.len()].copy_from_slice(payload);
         ret
     }
 
@@ -503,12 +494,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;
-        ret.inner.resize(total_len, 0);
+        unsafe { ret.inner.set_len(total_len) };
         ret
     }
 
     pub fn new_for_foreign_network(
-        network_name: &str,
+        network_name: &String,
         dst_peer_id: u32,
         foreign_zc_packet: &ZCPacket,
     ) -> Self {
@@ -517,71 +508,26 @@ impl ZCPacket {
             foreign_network_hdr.get_header_len() + foreign_zc_packet.tunnel_payload().len();
 
         let mut ret = Self::new_nic_packet();
-        let offsets = ret.packet_type.get_packet_offsets();
-        let payload_off = offsets.payload_offset;
-        let pm_hdr_off = offsets.peer_manager_header_offset;
-        let total_len = payload_off + total_payload_len;
-        ret.inner.reserve(total_len);
+        let payload_off = ret.packet_type.get_packet_offsets().payload_offset;
+        ret.inner.reserve(payload_off + total_payload_len);
+        unsafe { ret.inner.set_len(payload_off + total_payload_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;
-        let foreign_payload = foreign_zc_packet.tunnel_payload();
+        ret.mut_payload()[name_offset..name_offset + name_len]
+            .copy_from_slice(network_name.as_bytes());
 
-        // Construct the PeerManagerHeader on the stack so we can write it
-        // 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),
-        };
+        ret.mut_payload()[foreign_network_hdr.get_header_len()..]
+            .copy_from_slice(foreign_zc_packet.tunnel_payload());
 
-        // SAFETY: `reserve` guarantees capacity >= total_len.
-        // We zero only the tunnel-header reserved space [0..pm_hdr_off], write
-        // the PeerManagerHeader directly at pm_hdr_off, then copy the foreign
-        // network header, network name, and payload. Every byte in [0..total_len)
-        // 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);
-        }
+        let hdr = ret.mut_peer_manager_header().unwrap();
+        hdr.from_peer_id = 0.into();
+        hdr.to_peer_id = 0.into();
+        hdr.packet_type = PacketType::ForeignNetworkPacket as u8;
+        hdr.len.set(total_payload_len as u32);
 
         ret
     }
@@ -641,8 +587,7 @@ impl ZCPacket {
     }
 
     pub fn payload_bytes(mut self) -> BytesMut {
-        self.inner.advance(self.payload_offset());
-        self.inner
+        self.inner.split_off(self.payload_offset())
     }
 
     pub fn peer_manager_header(&self) -> Option<&PeerManagerHeader> {
@@ -707,12 +652,11 @@ impl ZCPacket {
     }
 
     pub fn tunnel_payload_bytes(mut self) -> BytesMut {
-        self.inner.advance(
+        self.inner.split_off(
             self.packet_type
                 .get_packet_offsets()
                 .peer_manager_header_offset,
-        );
-        self.inner
+        )
     }
 
     pub fn convert_type(mut self, target_packet_type: ZCPacketType) -> Self {
@@ -753,13 +697,12 @@ impl ZCPacket {
                 .get_packet_offsets()
                 .peer_manager_header_offset;
             let mut buf = BytesMut::with_capacity(new_pm_offset + tunnel_payload.len());
-            buf.resize(new_pm_offset, 0);
+            unsafe { buf.set_len(new_pm_offset) };
             buf.extend_from_slice(tunnel_payload);
             return Self::new_from_buf(buf, target_packet_type);
         }
 
-        self.inner.advance(new_offset);
-        Self::new_from_buf(self.inner, target_packet_type)
+        Self::new_from_buf(self.inner.split_off(new_offset), target_packet_type)
     }
 
     pub fn into_bytes(self) -> Bytes {
@@ -805,10 +748,8 @@ impl ZCPacket {
         let foreign_hdr_len = hdr.get_header_len();
 
         Self::new_from_buf(
-            {
-                self.inner.advance(foreign_hdr_len + self.payload_offset());
-                self.inner
-            },
+            self.inner
+                .split_off(foreign_hdr_len + self.payload_offset()),
             ZCPacketType::DummyTunnel,
         )
     }

easytier/src/tunnel/stats.rs

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