chore: Update hotpath (#2412 )

feat(easytier): initialize hotpath profiler via hotpath::main
Gate the binary entry point with hotpath::main when the feature is enabled, except when hotpath-alloc is combined with a custom global allocator (jemalloc/mimalloc), which is already rejected at compile time in lib.rs.
2026-09-22 19:28:58 +00:00 · 2026-07-07 19:45:41 +08:00 · 2026-07-02 21:19:28 +08:00 · 2026-07-02 21:19:28 +08:00 · 2026-07-02 21:19:28 +08:00
33 changed files with 668 additions and 792 deletions
@@ -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",
-]
@@ -263,6 +263,12 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"

+[[package]]
+name = "ascii"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16"
+
 [[package]]
 name = "async-broadcast"
 version = "0.7.2"
@@ -1260,6 +1266,12 @@ dependencies = [
 "windows-targets 0.52.6",
 ]

+[[package]]
+name = "chunked_transfer"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
+
 [[package]]
 name = "ciborium"
 version = "0.2.2"
@@ -2163,6 +2175,16 @@ dependencies = [
 "dirs-sys 0.5.0",
 ]

+[[package]]
+name = "dirs-next"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b98cf8ebf19c3d1b223e151f99a4f9f0690dca41414773390fc824184ac833e1"
+dependencies = [
+ "cfg-if",
+ "dirs-sys-next",
+]
+
 [[package]]
 name = "dirs-sys"
 version = "0.3.7"
@@ -2186,6 +2208,17 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

+[[package]]
+name = "dirs-sys-next"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d"
+dependencies = [
+ "libc",
+ "redox_users 0.4.5",
+ "winapi",
+]
+
 [[package]]
 name = "dispatch2"
 version = "0.3.1"
@@ -2365,6 +2398,7 @@ dependencies = [
 "hickory-resolver",
 "hickory-server",
 "hmac",
+ "hotpath",
 "http",
 "http_req",
 "humansize",
@@ -2435,7 +2469,6 @@ dependencies = [
 "tempfile",
 "terminal_size",
 "thiserror 1.0.63",
- "thread_local",
 "thunk-rs",
 "tikv-jemalloc-ctl",
 "tikv-jemalloc-sys",
@@ -2708,6 +2741,12 @@ version = "1.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ef6b89e5b37196644d8796de5268852ff179b44e96276cf4290264843743bb7"

+[[package]]
+name = "encode_unicode"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0"
+
 [[package]]
 name = "encoding"
 version = "0.2.33"
@@ -3966,6 +4005,61 @@ dependencies = [
 "windows-sys 0.52.0",
 ]

+[[package]]
+name = "hotpath"
+version = "0.21.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ff6b552a6afa29d9e33f8d555bee9093c142dd449501ae128e6494a303f03dc"
+dependencies = [
+ "arc-swap",
+ "cfg-if",
+ "crossbeam-channel",
+ "flate2",
+ "flume 0.12.0",
+ "futures-util",
+ "hdrhistogram",
+ "hotpath-macros",
+ "hotpath-meta",
+ "libc",
+ "object",
+ "parking_lot",
+ "pin-project-lite",
+ "prettytable-rs",
+ "quanta",
+ "regex",
+ "rustc-demangle",
+ "serde",
+ "serde_json",
+ "tiny_http",
+ "tokio",
+]
+
+[[package]]
+name = "hotpath-macros"
+version = "0.21.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f15322569d3cfadf84c0de7ef72be435b8f4b4839ee4ace78a7eaca48a87ded"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.117",
+]
+
+[[package]]
+name = "hotpath-macros-meta"
+version = "0.21.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b3675e29d16c844ccad12763672b33e51d9a000c346720c4f354f7a3bdc649a8"
+
+[[package]]
+name = "hotpath-meta"
+version = "0.21.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d558d972ddc9483fb4e713af3dd41634edf895c7cb18ed13dc58c56431d42e27"
+dependencies = [
+ "hotpath-macros-meta",
+]
+
 [[package]]
 name = "html5ever"
 version = "0.29.1"
@@ -5934,6 +6028,15 @@ dependencies = [
 "objc2-foundation",
 ]

+[[package]]
+name = "object"
+version = "0.36.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "once_cell"
 version = "1.21.3"
@@ -6838,6 +6941,19 @@ dependencies = [
 "syn 2.0.117",
 ]

+[[package]]
+name = "prettytable-rs"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eea25e07510aa6ab6547308ebe3c036016d162b8da920dbb079e3ba8acf3d95a"
+dependencies = [
+ "encode_unicode",
+ "is-terminal",
+ "lazy_static",
+ "term",
+ "unicode-width 0.1.11",
+]
+
 [[package]]
 name = "primeorder"
 version = "0.13.6"
@@ -7897,6 +8013,12 @@ dependencies = [
 "wasm-bindgen",
 ]

+[[package]]
+name = "rustc-demangle"
+version = "0.1.27"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b50b8869d9fc858ce7266cce0194bd74df58b9d0e3f6df3a9fc8eb470d95c09d"
+
 [[package]]
 name = "rustc-hash"
 version = "2.1.0"
@@ -9821,6 +9943,17 @@ dependencies = [
 "utf-8",
 ]

+[[package]]
+name = "term"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c59df8ac95d96ff9bede18eb7300b0fda5e5d8d90960e76f8e14ae765eedbf1f"
+dependencies = [
+ "dirs-next",
+ "rustversion",
+ "winapi",
+]
+
 [[package]]
 name = "terminal_size"
 version = "0.4.1"
@@ -9976,6 +10109,18 @@ version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "825f6c8a18bc36d56a62f66af7296385b628c9c5543a8663d4c217fc920bfefd"

+[[package]]
+name = "tiny_http"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "389915df6413a2e74fb181895f933386023c71110878cd0825588928e64cdc82"
+dependencies = [
+ "ascii",
+ "chunked_transfer",
+ "httpdate",
+ "log",
+]
+
 [[package]]
 name = "tinystr"
 version = "0.7.6"
@@ -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"

@@ -62,6 +58,8 @@ chrono = { version = "0.4.37", features = ["serde"] }
 guarden = "0.2"
 quanta = "0.12"

+hotpath = { version = "0.21", default-features = false, optional = true }
+
 delegate = "0.13.5"

 itertools = "0.14.0"
@@ -224,7 +222,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 +342,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 +350,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"
@@ -423,11 +412,14 @@ tracing = ["tokio/tracing", "dep:console-subscriber"]
 magic-dns = ["dep:hickory-client", "dep:hickory-server"]
 faketcp = ["dep:flume"]
 zstd = ["dep:zstd"]
-# Deprecated: hotpath profiling has been removed. These feature aliases are
-# retained as no-ops so existing build scripts using `--features hotpath*`
-# continue to work without pulling in any dependencies.
-hotpath = []
-hotpath-cpu = ["hotpath"]
-hotpath-alloc = ["hotpath"]
+hotpath = [
+    "dep:hotpath",
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/parking_lot",
+    "hotpath/flume",
+]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
 # For Network Extension on macOS
 macos-ne = []
@@ -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`).

@@ -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);
@@ -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);
@@ -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);
@@ -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
@@ -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;
@@ -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);
-    }
-}
@@ -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);
    }
@@ -24,6 +24,16 @@ pub static malloc_conf: &[u8] = b"retain:false\0";
 rust_i18n::i18n!("locales", fallback = "en");

 #[tokio::main(flavor = "current_thread")]
+#[cfg_attr(
+    all(
+        feature = "hotpath",
+        not(all(
+            feature = "hotpath-alloc",
+            any(feature = "jemalloc", feature = "mimalloc")
+        ))
+    ),
+    hotpath::main
+)]
 async fn main() -> std::process::ExitCode {
    core::main().await
 }
@@ -0,0 +1,52 @@
+//! No-op stand-in for the `hotpath` macros used by this crate, selected when
+//! the `hotpath` feature is disabled.
+//!
+//! Keeping `hotpath` as an optional dependency means default builds do not pull
+//! the profiler (or any of its transitive dependencies) into the dependency
+//! graph. These macros expand to their input unchanged, mirroring `hotpath`'s
+//! own disabled mode so call sites compile identically with or without the
+//! feature.
+//!
+//! The macros are `#[macro_export]`-ed so that `lib.rs`' `extern crate self as
+//! hotpath` alias exposes them through the same `hotpath::...` paths used when
+//! the feature is enabled.
+
+/// No-op mirroring `hotpath::channel!`: returns the channel expression
+/// unchanged (dropping any optional trailing `label`/`log`/`capacity` args).
+#[doc(hidden)]
+#[macro_export]
+macro_rules! channel {
+    ($expr:expr $(, $($rest:tt)*)?) => {
+        $expr
+    };
+}
+
+/// No-op mirroring `hotpath::mutex!`: returns the expression unchanged.
+#[doc(hidden)]
+#[macro_export]
+macro_rules! mutex {
+    ($expr:expr $(, $($rest:tt)*)?) => {
+        $expr
+    };
+}
+
+/// No-op mirroring `hotpath::rw_lock!`: returns the expression unchanged.
+#[doc(hidden)]
+#[macro_export]
+macro_rules! rw_lock {
+    ($expr:expr $(, $($rest:tt)*)?) => {
+        $expr
+    };
+}
+
+/// Type-level mirror of `hotpath::wrap` for type positions: with the feature
+/// off, `channel!` returns the original endpoints, so the wrapped endpoint
+/// types are the plain channel types.
+pub(crate) mod wrap {
+    pub(crate) mod tokio {
+        pub(crate) mod sync {
+            pub(crate) use ::tokio::sync::mpsc;
+        }
+    }
+    pub(crate) use ::flume;
+}
@@ -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");
@@ -5,6 +5,22 @@ use std::io;
 use clap::Command;
 use clap_complete::{Generator, Shell};

+// When the `hotpath` feature is off, alias the current crate as `hotpath` so
+// call sites keep using `hotpath::...` paths, and provide a local no-op shim
+// for the profiling macros. This keeps `hotpath` an optional dependency: the
+// profiler is absent from the dependency graph entirely in default builds.
+#[cfg(not(feature = "hotpath"))]
+extern crate self as hotpath;
+#[cfg(not(feature = "hotpath"))]
+mod hotpath_off;
+#[cfg(not(feature = "hotpath"))]
+pub(crate) use hotpath_off::wrap;
+
+// `hotpath-alloc` registers a global profiling allocator, which is mutually
+// exclusive with the `jemalloc`/`mimalloc` global allocators.
+#[cfg(all(feature = "hotpath-alloc", any(feature = "jemalloc", feature = "mimalloc")))]
+compile_error!("feature `hotpath-alloc` cannot be enabled together with `jemalloc` or `mimalloc`");
+
 // Re-export `Instant` at the crate root so public APIs that expose it
 // (e.g. `Route::get_peer_info_last_update_time`) reference a deliberate
 // public type rather than leaking an inaccessible one.
@@ -56,11 +56,12 @@ type BoxNicPacketFilter = Box<dyn NicPacketFilter + Send + Sync>;
 // pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
 //     tachyonix::channel(128)
 // }
-pub type PacketRecvChan = tokio::sync::mpsc::Sender<ZCPacket>;
-pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver<ZCPacket>;
+pub type PacketRecvChan = hotpath::wrap::tokio::sync::mpsc::Sender<ZCPacket>;
+pub type PacketRecvChanReceiver = hotpath::wrap::tokio::sync::mpsc::Receiver<ZCPacket>;
 pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
-    tokio::sync::mpsc::channel(128)
+    hotpath::channel!(tokio::sync::mpsc::channel(128))
 }
+#[cfg_attr(feature = "hotpath", hotpath::measure())]
 pub async fn recv_packet_from_chan(
    packet_recv_chan_receiver: &mut PacketRecvChanReceiver,
 ) -> Result<ZCPacket, anyhow::Error> {
@@ -2,6 +2,9 @@ use std::sync::Arc;

 use crossbeam::atomic::AtomicCell;
 use dashmap::{DashMap, DashSet};
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::parking_lot::RwLock;
+#[cfg(not(feature = "hotpath"))]
 use parking_lot::RwLock;

 use tokio::{select, sync::mpsc};
@@ -56,7 +59,7 @@ impl Peer {
        let shutdown_notifier = Arc::new(tokio::sync::Notify::new());
        let peer_identity_type = Arc::new(AtomicCell::new(None));
        let peer_identity_type_copy = peer_identity_type.clone();
-        let peer_public_key = Arc::new(RwLock::new(None));
+        let peer_public_key = Arc::new(hotpath::rw_lock!(parking_lot::RwLock::new(None)));
        let peer_public_key_copy = peer_public_key.clone();

        let conns_copy = conns.clone();
@@ -207,6 +210,7 @@ impl Peer {
            .map(|conn| conn.clone())
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "Peer"))]
    pub async fn send_msg(&self, msg: ZCPacket) -> Result<(), Error> {
        let Some(conn) = self.select_conn().await else {
            return Err(Error::PeerNoConnectionError(self.peer_node_id));
@@ -11,6 +11,9 @@ use std::{
    },
 };

+#[cfg(feature = "hotpath")]
+use hotpath::wrap::tokio::sync::Mutex;
+#[cfg(not(feature = "hotpath"))]
 use tokio::sync::Mutex;

 use base64::Engine as _;
@@ -381,12 +384,12 @@ impl PeerConn {
            session_filter,
            noise_handshake_result: None,

-            tunnel: Arc::new(Mutex::new(
+            tunnel: Arc::new(hotpath::mutex!(tokio::sync::Mutex::new(
                Box::new(guard!([mut mpsc_tunnel] mpsc_tunnel.close()))
                    as Box<dyn Any + Send + 'static>,
-            )),
+            ))),
            sink,
-            recv: Mutex::new(Some(recv)),
+            recv: hotpath::mutex!(tokio::sync::Mutex::new(Some(recv))),
            tunnel_info,

            tasks: JoinSet::new(),
@@ -1463,6 +1466,7 @@ impl PeerConn {
        });
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerConn"))]
    pub async fn send_msg(&self, msg: ZCPacket) -> Result<(), Error> {
        Ok(self.sink.send(msg).await?)
    }
@@ -11,6 +11,9 @@ use std::{
    time::{Duration, SystemTime},
 };

+#[cfg(feature = "hotpath")]
+use hotpath::wrap::tokio::sync::{Mutex, RwLock};
+#[cfg(not(feature = "hotpath"))]
 use tokio::sync::{Mutex, RwLock};
 use tokio::{
    sync::mpsc::{self, UnboundedReceiver, UnboundedSender},
@@ -276,8 +279,8 @@ impl PeerManager {
        let rpc_tspt = Arc::new(RpcTransport {
            my_peer_id,
            peers: Arc::downgrade(&peers),
-            foreign_peers: Mutex::new(None),
-            packet_recv: Mutex::new(peer_rpc_tspt_recv),
+            foreign_peers: hotpath::mutex!(tokio::sync::Mutex::new(None)),
+            packet_recv: hotpath::mutex!(tokio::sync::Mutex::new(peer_rpc_tspt_recv)),
            peer_rpc_tspt_sender,
            encryptor: encryptor.clone(),
            is_secure_mode_enabled,
@@ -409,17 +412,21 @@ impl PeerManager {
            global_ctx,
            nic_channel,

-            tasks: Mutex::new(JoinSet::new()),
+            tasks: hotpath::mutex!(tokio::sync::Mutex::new(JoinSet::new())),

-            packet_recv: Arc::new(Mutex::new(Some(packet_recv))),
+            packet_recv: Arc::new(hotpath::mutex!(tokio::sync::Mutex::new(Some(packet_recv)))),

            peers,

            peer_rpc_mgr,
            peer_rpc_tspt: rpc_tspt,

-            peer_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
-            nic_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
+            peer_packet_process_pipeline: Arc::new(hotpath::rw_lock!(tokio::sync::RwLock::new(
+                Vec::new()
+            ))),
+            nic_packet_process_pipeline: Arc::new(hotpath::rw_lock!(tokio::sync::RwLock::new(
+                Vec::new()
+            ))),

            route_algo_inst,

@@ -430,7 +437,7 @@ impl PeerManager {
            encryptor,
            data_compress_algo,

-            exit_nodes: RwLock::new(exit_nodes),
+            exit_nodes: hotpath::rw_lock!(tokio::sync::RwLock::new(exit_nodes)),

            reserved_my_peer_id_map: DashMap::new(),
            recent_have_traffic: Arc::new(DashMap::new()),
@@ -956,6 +963,7 @@ impl PeerManager {
        Self::is_relay_data_packet(hdr.packet_type)
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
    async fn start_peer_recv(&self) {
        let mut recv = self.packet_recv.lock().await.take().unwrap();
        let my_peer_id = self.my_peer_id;
@@ -1437,6 +1445,7 @@ impl PeerManager {
        self.get_route().get_foreign_network_summary().await
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
    async fn run_nic_packet_process_pipeline(&self, data: &mut ZCPacket) -> bool {
        // Enforce ACL for outbound (NIC-originated) packets. If ACL denies, stop processing.
        if !self.global_ctx.get_acl_filter().process_packet_with_acl(
@@ -1522,6 +1531,7 @@ impl PeerManager {
        result
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
    async fn send_msg_internal(
        peers: &Arc<PeerMap>,
        foreign_network_client: &Arc<ForeignNetworkClient>,
@@ -1688,6 +1698,7 @@ impl PeerManager {
        (dst_peers, is_exit_node)
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
    pub async fn try_compress_and_encrypt(
        compress_algo: CompressorAlgo,
        encryptor: &Arc<dyn Encryptor + 'static>,
@@ -6,6 +6,9 @@ use std::{
 use anyhow::Context;
 use dashmap::{DashMap, DashSet};
 use parking_lot::Mutex;
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::tokio::sync::RwLock;
+#[cfg(not(feature = "hotpath"))]
 use tokio::sync::RwLock;

 use crate::{
@@ -45,7 +48,7 @@ impl PeerMap {
            my_peer_id,
            peer_map: DashMap::new(),
            packet_send,
-            routes: RwLock::new(Vec::new()),
+            routes: hotpath::rw_lock!(tokio::sync::RwLock::new(Vec::new())),
            alive_client_urls: Arc::new(Mutex::new(multimap::MultiMap::new())),
        }
    }
@@ -132,6 +135,7 @@ impl PeerMap {
        peer_id == self.my_peer_id || self.peer_map.contains_key(&peer_id)
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerMap"))]
    pub async fn send_msg_directly(&self, msg: ZCPacket, dst_peer_id: PeerId) -> Result<(), Error> {
        if dst_peer_id == self.my_peer_id {
            let packet_send = self.packet_send.clone();
@@ -163,6 +167,7 @@ impl PeerMap {
        Ok(())
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerMap"))]
    pub async fn get_gateway_peer_id(
        &self,
        dst_peer_id: PeerId,
@@ -1394,6 +1394,7 @@ impl RouteTable {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RouteTable"))]
    fn get_next_hop(&self, dst_peer_id: PeerId) -> Option<NextHopInfo> {
        if self.suppressed_peer_ids.contains_key(&dst_peer_id) {
            return None;
@@ -1401,6 +1402,7 @@ impl RouteTable {
        self.get_topology_next_hop(dst_peer_id)
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RouteTable"))]
    fn get_topology_next_hop(&self, dst_peer_id: PeerId) -> Option<NextHopInfo> {
        let cur_version = self.next_hop_map_version.get();
        self.next_hop_map.get(&dst_peer_id).and_then(|x| {
@@ -1,7 +1,11 @@
 use std::sync::{
-    Arc, RwLock,
+    Arc,
    atomic::{AtomicBool, Ordering},
 };
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::std::sync::RwLock;
+#[cfg(not(feature = "hotpath"))]
+use std::sync::RwLock;
 use std::time::Duration;

 use anyhow::anyhow;
@@ -262,7 +266,7 @@ impl std::fmt::Debug for PeerSession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PeerSession")
            .field("peer_id", &self.peer_id)
-            .field("peer_static_pubkey", &self.peer_static_pubkey)
+            .field("peer_static_pubkey", &*self.peer_static_pubkey.read().unwrap())
            .field("datagram", &self.datagram)
            .finish()
    }
@@ -282,7 +286,7 @@ impl PeerSession {
    ) -> Self {
        Self {
            peer_id,
-            peer_static_pubkey: RwLock::new(peer_static_pubkey),
+            peer_static_pubkey: hotpath::rw_lock!(std::sync::RwLock::new(peer_static_pubkey)),
            datagram: SecureDatagramSession::new(
                root_key,
                session_generation,
@@ -376,6 +380,7 @@ impl PeerSession {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerSession"))]
    pub fn encrypt_payload(
        &self,
        sender_peer_id: PeerId,
@@ -389,6 +394,7 @@ impl PeerSession {
            .encrypt_payload(Self::dir_for_sender(sender_peer_id, receiver_peer_id), pkt)
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerSession"))]
    pub fn decrypt_payload(
        &self,
        sender_peer_id: PeerId,
@@ -144,6 +144,7 @@ impl RelayPeerMap {
        Ok(())
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
    async fn send_via_next_hop(
        &self,
        msg: ZCPacket,
@@ -166,6 +167,7 @@ impl RelayPeerMap {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
    pub async fn send_msg(
        self: &Arc<Self>,
        mut msg: ZCPacket,
@@ -613,6 +615,7 @@ impl RelayPeerMap {
        Ok(())
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
    pub async fn decrypt_if_needed(self: &Arc<Self>, packet: &mut ZCPacket) -> Result<bool, Error> {
        if !self.is_secure_mode_enabled() {
            return Ok(false);
@@ -1,11 +1,16 @@
 use std::{
    sync::{
-        Arc, Mutex, RwLock,
        atomic::{AtomicBool, AtomicU32, Ordering},
+        Arc,
    },
    time::{SystemTime, UNIX_EPOCH},
 };

+#[cfg(feature = "hotpath")]
+use hotpath::wrap::std::sync::{Mutex, RwLock};
+#[cfg(not(feature = "hotpath"))]
+use std::sync::{Mutex, RwLock};
+
 use anyhow::anyhow;
 use atomic_shim::AtomicU64;
 use hmac::{Hmac, Mac as _};
@@ -14,7 +19,7 @@ use sha2::Sha256;
 use zerocopy::FromBytes;

 use crate::{
-    peers::encrypt::{Encryptor, create_encryptor},
+    peers::encrypt::{create_encryptor, Encryptor},
    tunnel::packet_def::{StandardAeadTail, ZCPacket},
 };

@@ -228,15 +233,15 @@ pub struct SecureDatagramSession {
 impl std::fmt::Debug for SecureDatagramSession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SecureDatagramSession")
-            .field("root_key", &self.root_key)
+            .field("root_key", &*self.root_key.read().unwrap())
            .field("session_generation", &self.session_generation)
            .field("send_epoch", &self.send_epoch)
            .field("send_seq", &self.send_seq)
            .field("send_epoch_started_ms", &self.send_epoch_started_ms)
            .field("send_packets_since_epoch", &self.send_packets_since_epoch)
-            .field("rx_slots", &self.rx_slots)
-            .field("key_cache", &self.key_cache)
-            .field("sync_rx_grace", &self.sync_rx_grace)
+            .field("rx_slots", &*self.rx_slots.lock().unwrap())
+            .field("key_cache", &*self.key_cache.lock().unwrap())
+            .field("sync_rx_grace", &*self.sync_rx_grace.lock().unwrap())
            .field(
                "sync_rx_grace_expires_at_ms",
                &self.sync_rx_grace_expires_at_ms,
@@ -272,15 +277,15 @@ impl SecureDatagramSession {
        ];
        let now_ms = now_ms();
        Self {
-            root_key: RwLock::new(root_key),
+            root_key: hotpath::rw_lock!(std::sync::RwLock::new(root_key)),
            session_generation: AtomicU32::new(session_generation),
            send_epoch: AtomicU32::new(initial_epoch),
            send_seq: [AtomicU64::new(0), AtomicU64::new(0)],
            send_epoch_started_ms: AtomicU64::new(now_ms),
            send_packets_since_epoch: AtomicU64::new(0),
-            rx_slots: Mutex::new(rx_slots),
-            key_cache: Mutex::new(key_cache),
-            sync_rx_grace: Mutex::new(SyncRxGrace::default()),
+            rx_slots: hotpath::mutex!(std::sync::Mutex::new(rx_slots)),
+            key_cache: hotpath::mutex!(std::sync::Mutex::new(key_cache)),
+            sync_rx_grace: hotpath::mutex!(std::sync::Mutex::new(SyncRxGrace::default())),
            sync_rx_grace_expires_at_ms: AtomicU64::new(0),
            send_cipher_algorithm,
            recv_cipher_algorithm,
@@ -701,6 +706,10 @@ impl SecureDatagramSession {
        false
    }

+    #[cfg_attr(
+        feature = "hotpath",
+        hotpath::measure(impl_type = "SecureDatagramSession")
+    )]
    pub fn encrypt_payload(
        &self,
        dir: SecureDatagramDirection,
@@ -719,6 +728,10 @@ impl SecureDatagramSession {
        Ok(())
    }

+    #[cfg_attr(
+        feature = "hotpath",
+        hotpath::measure(impl_type = "SecureDatagramSession")
+    )]
    pub fn decrypt_payload(
        &self,
        dir: SecureDatagramDirection,
@@ -884,11 +897,9 @@ mod tests {
        let nonce_offset = payload.len() - StandardAeadTail::NONCE_SIZE;
        payload[nonce_offset..].copy_from_slice(&poisoned_nonce);

-        assert!(
-            receiver
-                .decrypt_payload(SecureDatagramDirection::AToB, &mut forged)
-                .is_err()
-        );
+        assert!(receiver
+            .decrypt_payload(SecureDatagramDirection::AToB, &mut forged)
+            .is_err());

        let plaintext = b"pkt2";
        let mut pkt2 = ZCPacket::new_with_payload(plaintext);
@@ -990,9 +1001,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]
@@ -49,7 +49,7 @@ async fn test_route_peer_info_ipv6() {
 #[tokio::test]
 async fn test_peer_manager_ipv6() {
    let global_ctx = get_mock_global_ctx();
-    let (packet_sender, _packet_receiver) = tokio::sync::mpsc::channel(100);
+    let (packet_sender, _packet_receiver) = crate::peers::create_packet_recv_chan();
    let peer_mgr = crate::peers::peer_manager::PeerManager::new(
        RouteAlgoType::Ospf,
        global_ctx.clone(),
@@ -48,9 +48,13 @@ use std::collections::{HashMap, HashSet};
 use std::fmt;
 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::sync::{
-    Arc, RwLock,
+    Arc,
    atomic::{AtomicU32, Ordering},
 };
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::std::sync::RwLock;
+#[cfg(not(feature = "hotpath"))]
+use std::sync::RwLock;
 use tokio::sync::broadcast;
 use tokio::time;
 use tokio_util::task::AbortOnDropHandle;
@@ -85,7 +89,7 @@ impl AddrTuple {

 #[derive(Default)]
 struct StackState {
-    tuples: HashMap<AddrTuple, flume::Sender<Bytes>>,
+    tuples: HashMap<AddrTuple, hotpath::wrap::flume::Sender<Bytes>>,
    closed: bool,
 }

@@ -129,7 +133,7 @@ pub enum State {
 pub struct Socket {
    shared: Arc<Shared>,
    tun: Arc<dyn Tun>,
-    incoming: flume::Receiver<Bytes>,
+    incoming: hotpath::wrap::flume::Receiver<Bytes>,
    local_addr: SocketAddr,
    remote_addr: SocketAddr,
    local_mac: MacAddr,
@@ -158,8 +162,8 @@ impl Socket {
        remote_mac: Option<MacAddr>,
        ack: Option<u32>,
        state: State,
-    ) -> (Socket, flume::Sender<Bytes>) {
-        let (incoming_tx, incoming_rx) = flume::bounded(MPMC_BUFFER_LEN);
+    ) -> (Socket, hotpath::wrap::flume::Sender<Bytes>) {
+        let (incoming_tx, incoming_rx) = hotpath::channel!(flume::bounded(MPMC_BUFFER_LEN));

        (
            Socket {
@@ -430,9 +434,9 @@ impl Stack {
    ) -> Stack {
        let (tuples_purge_tx, _tuples_purge_rx) = broadcast::channel(16);
        let shared = Arc::new(Shared {
-            state: RwLock::new(StackState::default()),
+            state: hotpath::rw_lock!(std::sync::RwLock::new(StackState::default())),
            tun: tun.clone(),
-            listening: RwLock::new(HashSet::new()),
+            listening: hotpath::rw_lock!(std::sync::RwLock::new(HashSet::new())),
            tuples_purge: tuples_purge_tx.clone(),
        });

@@ -501,7 +505,7 @@ impl Stack {
        shared: Arc<Shared>,
        mut tuples_purge: broadcast::Receiver<AddrTuple>,
    ) {
-        let mut tuples: HashMap<AddrTuple, flume::Sender<Bytes>> = HashMap::new();
+        let mut tuples: HashMap<AddrTuple, hotpath::wrap::flume::Sender<Bytes>> = HashMap::new();

        loop {
            let mut buf = BytesMut::new();
@@ -9,7 +9,8 @@ use crate::proto::common::TunnelInfo;

 use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPacket};

-use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
+use hotpath::wrap::tokio::sync::mpsc::{Receiver, Sender};
+use tokio::sync::mpsc::{channel, error::TrySendError};
 use tokio_util::task::AbortOnDropHandle;
 // use tachyonix::{channel, Receiver, Sender, TrySendError};

@@ -19,6 +20,7 @@ use futures::SinkExt;
 pub struct MpscTunnelSender(Sender<ZCPacket>);

 impl MpscTunnelSender {
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnelSender"))]
    pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
        self.0.send(item).await.with_context(|| "send error")?;
        Ok(())
@@ -43,7 +45,7 @@ pub struct MpscTunnel<T> {

 impl<T: Tunnel> MpscTunnel<T> {
    pub fn new(tunnel: T, send_timeout: Option<Duration>) -> Self {
-        let (tx, mut rx) = channel(32);
+        let (tx, mut rx) = hotpath::channel!(channel(32));
        let (stream, mut sink) = tunnel.split();

        let task = tokio::spawn(async move {
@@ -66,6 +68,7 @@ impl<T: Tunnel> MpscTunnel<T> {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
    async fn forward_one_round(
        rx: &mut Receiver<ZCPacket>,
        sink: &mut Pin<Box<dyn ZCPacketSink>>,
@@ -79,6 +82,7 @@ impl<T: Tunnel> MpscTunnel<T> {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
    async fn forward_one_round_no_timeout(
        rx: &mut Receiver<ZCPacket>,
        sink: &mut Pin<Box<dyn ZCPacketSink>>,
@@ -96,6 +100,7 @@ impl<T: Tunnel> MpscTunnel<T> {
        sink.flush().await
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
    async fn forward_one_round_with_timeout(
        rx: &mut Receiver<ZCPacket>,
        sink: &mut Pin<Box<dyn ZCPacketSink>>,
@@ -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,
        )
    }
@@ -12,6 +12,9 @@ use crate::tunnel::{
 use anyhow::Context;
 use derivative::Derivative;
 use derive_more::{Deref, DerefMut};
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::parking_lot::RwLock;
+#[cfg(not(feature = "hotpath"))]
 use parking_lot::RwLock;
 use quinn::{
    ClientConfig, ConnectError, Connection, Endpoint, EndpointConfig, ServerConfig,
@@ -312,18 +315,25 @@ struct RwPoolInner<Item> {
    enabled: bool,
 }

-#[derive(Debug)]
 struct RwPool<Item> {
    ephemeral: RwLock<RwPoolInner<Item>>,
    persistent: RwLock<RwPoolInner<Item>>,
    capacity: usize,
 }

+impl<Item> std::fmt::Debug for RwPool<Item> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RwPool")
+            .field("capacity", &self.capacity)
+            .finish()
+    }
+}
+
 impl<Item> RwPool<Item> {
    fn new(capacity: usize) -> Self {
        Self {
-            ephemeral: RwLock::new(RwPoolInner::default()),
-            persistent: RwLock::new(RwPoolInner::default()),
+            ephemeral: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
+            persistent: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
            capacity,
        }
    }
@@ -11,7 +11,8 @@ use async_trait::async_trait;
 use futures::{Sink, SinkExt, Stream, StreamExt};
 use once_cell::sync::Lazy;

-use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
+use hotpath::wrap::tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::sync::mpsc::unbounded_channel;

 use uuid::Uuid;

@@ -196,7 +197,7 @@ pub struct RingTunnelListener {

 impl RingTunnelListener {
    pub fn new(key: url::Url) -> Self {
-        let (conn_sender, conn_receiver) = unbounded_channel();
+        let (conn_sender, conn_receiver) = hotpath::channel!(unbounded_channel());
        RingTunnelListener {
            listener_addr: key,
            conn_sender,
@@ -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;
+        }
    }
 }
@@ -13,11 +13,10 @@ use futures::{StreamExt, stream::FuturesUnordered};
 use rand::{Rng, SeedableRng};
 use zerocopy::{AsBytes, FromBytes};

+use hotpath::wrap::tokio::sync::mpsc::{Receiver, Sender, UnboundedReceiver, UnboundedSender};
 use tokio::{
    net::UdpSocket,
-    sync::mpsc::{
-        Receiver, Sender, UnboundedReceiver, UnboundedSender, channel, unbounded_channel,
-    },
+    sync::mpsc::{channel, unbounded_channel},
    task::JoinSet,
 };
 use tokio_util::task::AbortOnDropHandle;
@@ -293,6 +292,7 @@ fn get_zcpacket_from_buf(buf: BytesMut, allow_stun: bool) -> Result<ZCPacket, Tu
 }

 #[instrument]
+#[cfg_attr(feature = "hotpath", hotpath::measure())]
 async fn forward_from_ring_to_udp(
    mut ring_recv: RingStream,
    socket: &Arc<UdpSocket>,
@@ -327,6 +327,7 @@ async fn forward_from_ring_to_udp(
    }
 }

+#[cfg_attr(feature = "hotpath", hotpath::measure())]
 async fn udp_recv_from_socket_forward_task(
    socket: &UdpSocket,
    buf: &mut BytesMut,
@@ -395,6 +396,7 @@ impl UdpConnection {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "UdpTunnel"))]
    pub fn handle_packet_from_remote(&mut self, zc_packet: ZCPacket) -> Result<(), TunnelError> {
        let header = zc_packet.udp_tunnel_header().unwrap();
        let conn_id = header.conn_id.get();
@@ -541,6 +543,7 @@ impl UdpTunnelListenerData {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "UdpTunnelListener"))]
    fn do_forward_one_packet_to_conn(&self, zc_packet: ZCPacket, addr: SocketAddr) {
        let header = zc_packet.udp_tunnel_header().unwrap();
        if header.msg_type == UdpPacketType::Syn as u8 {
@@ -647,6 +650,7 @@ impl UdpTunnelListenerData {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "UdpTunnelListener"))]
    async fn do_forward_task(self) {
        let socket = self.socket.as_ref().unwrap().clone();
        let mut buf = BytesMut::new();
@@ -675,8 +679,8 @@ pub struct UdpTunnelListener {

 impl UdpTunnelListener {
    pub fn new(addr: url::Url) -> Self {
-        let (close_event_send, close_event_recv) = unbounded_channel();
-        let (conn_send, conn_recv) = channel(100);
+        let (close_event_send, close_event_recv) = hotpath::channel!(unbounded_channel());
+        let (conn_send, conn_recv) = hotpath::channel!(channel(100));
        Self {
            addr: addr.clone(),
            socket: None,
@@ -916,7 +920,8 @@ impl UdpTunnelConnector {
            "udp build tunnel for connector"
        );

-        let (close_event_sender, mut close_event_recv) = unbounded_channel();
+        let (close_event_sender, mut close_event_recv) =
+            hotpath::channel!(unbounded_channel());

        let ring_recv = RingStream::new(ring_for_send_udp.clone());
        let ring_sender = RingSink::new(ring_for_recv_udp.clone());
@@ -1179,7 +1184,7 @@ mod tests {
        let dst_addr = "127.0.0.1:1".parse().unwrap();
        let ring_for_send_udp = Arc::new(RingTunnel::new(8));
        let ring_for_recv_udp = Arc::new(RingTunnel::new(8));
-        let (close_event_sender, _close_event_recv) = tokio::sync::mpsc::unbounded_channel();
+        let (close_event_sender, _close_event_recv) = hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
        let mut conn = UdpConnection::new(
            socket,
            7,
@@ -37,9 +37,17 @@ use crossbeam::atomic::AtomicCell;
 use dashmap::DashMap;
 use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
 use rand::RngCore;
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::std::sync::Mutex as StdMutex;
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::tokio::sync::Mutex;
+#[cfg(not(feature = "hotpath"))]
+use std::sync::Mutex as StdMutex;
+#[cfg(not(feature = "hotpath"))]
+use tokio::sync::Mutex;
 use tokio::{
    net::UdpSocket,
-    sync::{Mutex, mpsc::unbounded_channel},
+    sync::mpsc::unbounded_channel,
    task::JoinSet,
 };

@@ -347,7 +355,7 @@ struct WgPeer {
    config: WgConfig,
    endpoint: SocketAddr,

-    sink: std::sync::Mutex<Option<Pin<Box<dyn ZCPacketSink>>>>,
+    sink: StdMutex<Option<Pin<Box<dyn ZCPacketSink>>>>,

    data: Option<WgPeerData>,
    tasks: JoinSet<()>,
@@ -358,19 +366,19 @@ struct WgPeer {
 impl WgPeer {
    fn new(udp: Arc<UdpSocket>, config: WgConfig, endpoint: SocketAddr) -> Self {
        WgPeer {
-            tunn: Some(Mutex::new(Tunn::new(
+            tunn: Some(hotpath::mutex!(tokio::sync::Mutex::new(Tunn::new(
                config.my_secret_key.clone(),
                config.peer_public_key,
                None,
                None,
                rand::thread_rng().next_u32(),
                None,
-            ))),
+            )))),

            udp,
            config,
            endpoint,
-            sink: std::sync::Mutex::new(None),
+            sink: hotpath::mutex!(std::sync::Mutex::new(None)),

            data: None,
            tasks: JoinSet::new(),
@@ -379,6 +387,7 @@ impl WgPeer {
        }
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "WgTunnel"))]
    async fn handle_packet_from_me<S: ZCPacketStream + Unpin>(mut stream: S, data: WgPeerData) {
        while let Some(Ok(packet)) = stream.next().await {
            let ret = data.handle_one_packet_from_me(packet).await;
@@ -390,6 +399,7 @@ impl WgPeer {
            .store(true, std::sync::atomic::Ordering::Relaxed);
    }

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "WgTunnel"))]
    async fn handle_packet_from_peer(&self, packet: &[u8]) {
        self.access_time.store(Instant::now());
        tracing::trace!("Received {} bytes from peer", packet.len());
@@ -455,8 +465,8 @@ impl WgPeer {
    }
 }

-type ConnSender = tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
-type ConnReceiver = tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
+type ConnSender = hotpath::wrap::tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
+type ConnReceiver = hotpath::wrap::tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;

 pub struct WgTunnelListener {
    addr: url::Url,
@@ -474,7 +484,7 @@ pub struct WgTunnelListener {

 impl WgTunnelListener {
    pub fn new(addr: url::Url, config: WgConfig) -> Self {
-        let (conn_send, conn_recv) = unbounded_channel();
+        let (conn_send, conn_recv) = hotpath::channel!(unbounded_channel());
        WgTunnelListener {
            addr,
            config,
Author	SHA1	Message	Date
Paweł Urbanek	cb04cefb2a	chore: Update hotpath (#2412 )	2026-07-07 19:45:41 +08:00
fanyang89	20873fc62e	feat(easytier): initialize hotpath profiler via hotpath::main Gate the binary entry point with hotpath::main when the feature is enabled, except when hotpath-alloc is combined with a custom global allocator (jemalloc/mimalloc), which is already rejected at compile time in lib.rs.	2026-07-02 21:19:28 +08:00
fanyang89	27f6b90fa1	feat(easytier): instrument hot-path locks, channels, and functions Wrap the per-packet locks and channels behind hotpath's drop-in wrappers () so lock contention and channel flow become visible when the feature is on, while staying zero-cost in default builds via cfg-gated dual imports and the no-op // macros. Annotate the hottest send/recv, encrypt/decrypt, and forward functions with . Coverage: peer_conn/peer_manager/peer_map/peer/peer_session/secure_datagram locks, mpsc/ring/udp/wireguard/fake_tcp channels, quic connection pool, relay/foreign send paths, and OSPF route lookup (function-level only; its parking_lot upgradable guards have no hotpath wrapper). Debug impls that formatted lock fields are updated to dereference the inner value, and quic's RwPool switches to a manual Debug that skips the locks.	2026-07-02 21:19:28 +08:00
fanyang89	a602125d97	chore(easytier): gate hotpath-rs 0.19 dependency and off-mode shim Re-enable the hotpath profiler as an optional dependency gated behind the feature (default off). In off builds the crate stays out of the dependency graph entirely: plus a local module provide no-op // macros so call sites compile unchanged. Also add a guarding the mutual exclusivity of with the / global allocators.	2026-07-02 21:19:28 +08:00