chore: Update hotpath (#2412)

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.

Cargo.lock

@@ -129,6 +129,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "anes"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"
+
 [[package]]
 name = "anstream"
 version = "0.6.15"
@@ -1145,6 +1151,12 @@ dependencies = [
  "toml 0.9.12+spec-1.1.0",
 ]
 
+[[package]]
+name = "cast"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
+
 [[package]]
 name = "cc"
 version = "1.2.10"
@@ -1260,6 +1272,33 @@ version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
 
+[[package]]
+name = "ciborium"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e"
+dependencies = [
+ "ciborium-io",
+ "ciborium-ll",
+ "serde",
+]
+
+[[package]]
+name = "ciborium-io"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757"
+
+[[package]]
+name = "ciborium-ll"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9"
+dependencies = [
+ "ciborium-io",
+ "half",
+]
+
 [[package]]
 name = "cidr"
 version = "0.3.1"
@@ -1605,6 +1644,42 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "criterion"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f"
+dependencies = [
+ "anes",
+ "cast",
+ "ciborium",
+ "clap",
+ "criterion-plot",
+ "is-terminal",
+ "itertools 0.10.5",
+ "num-traits",
+ "once_cell",
+ "oorandom",
+ "plotters",
+ "rayon",
+ "regex",
+ "serde",
+ "serde_derive",
+ "serde_json",
+ "tinytemplate",
+ "walkdir",
+]
+
+[[package]]
+name = "criterion-plot"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1"
+dependencies = [
+ "cast",
+ "itertools 0.10.5",
+]
+
 [[package]]
 name = "critical-section"
 version = "1.2.0"
@@ -2299,6 +2374,7 @@ dependencies = [
  "clap_complete",
  "clap_complete_nushell",
  "console-subscriber",
+ "criterion",
  "crossbeam",
  "ctor 0.8.0",
  "dashmap",
@@ -3931,9 +4007,9 @@ dependencies = [
 
 [[package]]
 name = "hotpath"
-version = "0.18.0"
+version = "0.21.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc2c28b1fa962e433f800ed1ea0bf53dc028d3745cf2acec6cfd28b65ac96afa"
+checksum = "1ff6b552a6afa29d9e33f8d555bee9093c142dd449501ae128e6494a303f03dc"
 dependencies = [
  "arc-swap",
  "cfg-if",
@@ -3960,9 +4036,9 @@ dependencies = [
 
 [[package]]
 name = "hotpath-macros"
-version = "0.18.0"
+version = "0.21.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a585238d8daf746e27df0f24d1bbdcd2410e9febff63f9a0173f90d7e71c50f6"
+checksum = "4f15322569d3cfadf84c0de7ef72be435b8f4b4839ee4ace78a7eaca48a87ded"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3971,15 +4047,15 @@ dependencies = [
 
 [[package]]
 name = "hotpath-macros-meta"
-version = "0.18.0"
+version = "0.21.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "309f63c2f755dead454dd4b3ea8ab5c947f14f8ea435fbcd37fa820e17290e80"
+checksum = "b3675e29d16c844ccad12763672b33e51d9a000c346720c4f354f7a3bdc649a8"
 
 [[package]]
 name = "hotpath-meta"
-version = "0.18.0"
+version = "0.21.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68faa91a9e1114dff668cd90560f332da6bbde40dae37ec28ea1c43ca5ce3be3"
+checksum = "d558d972ddc9483fb4e713af3dd41634edf895c7cb18ed13dc58c56431d42e27"
 dependencies = [
  "hotpath-macros-meta",
 ]
@@ -5971,6 +6047,12 @@ dependencies = [
  "portable-atomic",
 ]
 
+[[package]]
+name = "oorandom"
+version = "11.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
+
 [[package]]
 name = "opaque-debug"
 version = "0.3.1"
@@ -6601,6 +6683,34 @@ dependencies = [
  "time",
 ]
 
+[[package]]
+name = "plotters"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747"
+dependencies = [
+ "num-traits",
+ "plotters-backend",
+ "plotters-svg",
+ "wasm-bindgen",
+ "web-sys",
+]
+
+[[package]]
+name = "plotters-backend"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a"
+
+[[package]]
+name = "plotters-svg"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670"
+dependencies = [
+ "plotters-backend",
+]
+
 [[package]]
 name = "pnet"
 version = "0.35.0"
@@ -10021,6 +10131,16 @@ dependencies = [
  "zerovec",
 ]
 
+[[package]]
+name = "tinytemplate"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc"
+dependencies = [
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "tinyvec"
 version = "1.8.0"

easytier/Cargo.toml

@@ -28,6 +28,10 @@ path = "src/easytier-cli.rs"
 name = "easytier"
 path = "src/lib.rs"
 
+[[bench]]
+name = "tx_throughput"
+harness = false
+
 [dependencies]
 git-version = "0.3.9"
 
@@ -52,9 +56,10 @@ toml = "0.8.12"
 chrono = { version = "0.4.37", features = ["serde"] }
 
 guarden = "0.2"
-hotpath = { version = "0.18", default-features = false, optional = true }
 quanta = "0.12"
 
+hotpath = { version = "0.21", default-features = false, optional = true }
+
 delegate = "0.13.5"
 
 itertools = "0.14.0"
@@ -337,6 +342,7 @@ zip = "4.0.0"
 
 
 [dev-dependencies]
+criterion = "0.5.1"
 serial_test = "3.0.0"
 rstest = "0.25.0"
 futures-util = "0.3.31"
@@ -403,6 +409,9 @@ jemalloc-prof = [
     "jemalloc-sys/stats",
 ]
 tracing = ["tokio/tracing", "dep:console-subscriber"]
+magic-dns = ["dep:hickory-client", "dep:hickory-server"]
+faketcp = ["dep:flume"]
+zstd = ["dep:zstd"]
 hotpath = [
     "dep:hotpath",
     "hotpath/hotpath",
@@ -412,8 +421,5 @@ hotpath = [
 ]
 hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
-magic-dns = ["dep:hickory-client", "dep:hickory-server"]
-faketcp = ["dep:flume"]
-zstd = ["dep:zstd"]
 # For Network Extension on macOS
 macos-ne = []

easytier/benches/README.md

@@ -0,0 +1,119 @@
+# TX Throughput Benchmark
+
+Criterion benchmark for EasyTier's TX injection path (`peer_manager::send_msg_by_ip`).
+
+## What it measures
+
+The benchmark sets up two EasyTier instances (`hot-a` / `hot-b`) and drives
+packets from `hot-a` to `hot-b` via `peer_manager.send_msg_by_ip`. This is the
+same entry point `easytier-core` uses for daily forwarded traffic, so the
+numbers reflect the real TX hot path: NIC pipeline → route lookup →
+compress/encrypt → peer connection → tunnel send.
+
+Two variants are reported per tunnel kind:
+
+| Bench                             | What it measures                                                                                                                                                                            |
+| --------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `tx_throughput/<tunnel>`          | Serial baseline. One send in flight at a time. Reports per-packet CPU cost (TX injection latency).                                                                                          |
+| `tx_throughput/<tunnel>-saturate` | Spawns `TX_THROUGHPUT_INFLIGHT` tokio tasks that independently pump `send_msg_by_ip`. Reports the aggregate throughput ceiling the peer manager + tunnel can sustain across worker threads. |
+
+> **Out of scope (by design):** TUN read/write (`no_tun = true`), compression
+> (default `None`), reverse/RX-side measurement, multi-peer fanout. Add
+> separate benchmarks if you need those.
+
+## Quick start
+
+### ring tunnel (no root, fastest)
+
+```bash
+cargo bench --bench tx_throughput
+```
+
+Smoke run (faster iteration):
+
+```bash
+TX_THROUGHPUT_MEASUREMENT_SECS=2 \
+TX_THROUGHPUT_WARMUP_SECS=1 \
+TX_THROUGHPUT_SAMPLE_SIZE=10 \
+cargo bench --bench tx_throughput -- --quiet
+```
+
+### tcp / udp tunnels (requires Docker + root)
+
+The benchmark creates a Docker network and registers each container's netns
+under `/var/run/netns`, which requires root. Run the whole command under
+`sudo`:
+
+```bash
+sudo TX_THROUGHPUT_TUNNEL=tcp \
+     TX_THROUGHPUT_MEASUREMENT_SECS=5 \
+     TX_THROUGHPUT_WARMUP_SECS=2 \
+     TX_THROUGHPUT_INFLIGHT=64 \
+     cargo bench --bench tx_throughput -- --quiet
+
+sudo TX_THROUGHPUT_TUNNEL=udp cargo bench --bench tx_throughput -- --quiet
+```
+
+> If `sudo` cannot find `cargo`, use `sudo -E` or the absolute path
+> (`$(which cargo)`).
+
+## Environment variables
+
+| Variable                         | Default               | Notes                                  |
+| -------------------------------- | --------------------- | -------------------------------------- |
+| `TX_THROUGHPUT_TUNNEL`           | `ring`                | `ring` / `tcp` / `udp`                 |
+| `TX_THROUGHPUT_PKT_SIZE`         | `1400`                | IP total length in bytes               |
+| `TX_THROUGHPUT_WORKER_THREADS`   | `4`                   | tokio worker threads                   |
+| `TX_THROUGHPUT_INFLIGHT`         | `64`                  | saturate-mode concurrency (task count) |
+| `TX_THROUGHPUT_TUNNEL_PORT`      | `35521`               | tcp/udp listen port                    |
+| `TX_THROUGHPUT_MEASUREMENT_SECS` | `10`                  | Criterion `measurement_time`           |
+| `TX_THROUGHPUT_WARMUP_SECS`      | `3`                   | Criterion `warm_up_time`               |
+| `TX_THROUGHPUT_SAMPLE_SIZE`      | `10`                  | Criterion `sample_size` (min 10)       |
+| `TX_THROUGHPUT_DOCKER_IMAGE`     | `busybox:latest`      | tcp/udp only                           |
+| `TX_THROUGHPUT_DOCKER_NET`       | `easytier-bench-<id>` | auto-generated unique name             |
+| `TX_THROUGHPUT_DOCKER_SUBNET`    | `172.31.250.0/24`     |                                        |
+| `TX_THROUGHPUT_DOCKER_IP_A`      | `172.31.250.2`        |                                        |
+| `TX_THROUGHPUT_DOCKER_IP_B`      | `172.31.250.3`        |                                        |
+
+## Parameter sweeps
+
+```bash
+# Packet size
+for sz in 64 256 1400 9000; do
+  TX_THROUGHPUT_PKT_SIZE=$sz cargo bench --bench tx_throughput -- --quick
+done
+
+# Inflight depth (self-check: depth=1 should match serial baseline)
+for d in 1 4 16 64 256; do
+  TX_THROUGHPUT_INFLIGHT=$d cargo bench --bench tx_throughput -- --quick
+done
+
+# Worker threads
+for w in 1 2 4 8; do
+  TX_THROUGHPUT_WORKER_THREADS=$w cargo bench --bench tx_throughput -- --quick
+done
+```
+
+## Interpreting results
+
+- **`<tunnel>`** reports per-packet latency. Lower is better. Throughput
+  column here is "what one in-flight sender sustains".
+- **`<tunnel>-saturate`** reports aggregate throughput across
+  `TX_THROUGHPUT_INFLIGHT` concurrent senders. If this matches the serial
+  baseline, the TX path is bottlenecked on an internal serialization point
+  (lock, single-threaded queue, etc.) rather than CPU or link bandwidth.
+
+### Known finding (ring, single peer)
+
+On the ring tunnel with a single destination peer, saturate does **not** beat
+serial (observed ~277 MiB/s saturate vs ~288 MiB/s serial on a 4-worker
+runtime). This points to a serialization point inside the peer-connection TX
+path. Tunnels with real I/O await points (tcp/udp via Docker) are expected to
+show a saturate > serial gap; verify with the sudo commands above.
+
+## Output artifacts
+
+Criterion writes HTML reports + SVG plots under
+`easytier/target/criterion/`. Open `tx_throughput/<tunnel>/report/index.html`
+or `.../<tunnel>-saturate/report/index.html` in a browser to inspect
+distributions and regressions across runs.

easytier/benches/tx_throughput.rs

@@ -0,0 +1,472 @@
+use std::{
+    net::IpAddr,
+    path::PathBuf,
+    process::{Command, Stdio},
+    str::FromStr,
+    sync::Arc,
+    sync::atomic::{AtomicU64, Ordering},
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+};
+
+use bytes::BytesMut;
+use criterion::{Criterion, Throughput, criterion_group, criterion_main};
+
+use easytier::{
+    common::config::{ConfigLoader, TomlConfigLoader},
+    instance::instance::Instance,
+    tunnel::{
+        packet_def::ZCPacket, ring::RingTunnelConnector, tcp::TcpTunnelConnector,
+        udp::UdpTunnelConnector,
+    },
+};
+
+const VIRTUAL_IP_A: &str = "10.144.144.1";
+const VIRTUAL_IP_B: &str = "10.144.144.2";
+const DEFAULT_DOCKER_SUBNET: &str = "172.31.250.0/24";
+const DEFAULT_DOCKER_IP_A: &str = "172.31.250.2";
+const DEFAULT_DOCKER_IP_B: &str = "172.31.250.3";
+const DEFAULT_TUNNEL_PORT: u16 = 35521;
+
+#[derive(Clone, Copy, Debug)]
+enum TunnelKind {
+    Ring,
+    Tcp,
+    Udp,
+}
+
+impl TunnelKind {
+    fn as_str(self) -> &'static str {
+        match self {
+            TunnelKind::Ring => "ring",
+            TunnelKind::Tcp => "tcp",
+            TunnelKind::Udp => "udp",
+        }
+    }
+}
+
+impl FromStr for TunnelKind {
+    type Err = String;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "ring" => Ok(TunnelKind::Ring),
+            "tcp" => Ok(TunnelKind::Tcp),
+            "udp" => Ok(TunnelKind::Udp),
+            other => Err(format!(
+                "unsupported TX_THROUGHPUT_TUNNEL={other:?}; expected ring, tcp, or udp"
+            )),
+        }
+    }
+}
+
+struct BenchTopology {
+    _docker: Option<DockerNetns>,
+    inst_a: Instance,
+    _inst_b: Instance,
+    dst: IpAddr,
+    packet: ZCPacket,
+}
+
+struct DockerNetns {
+    network: String,
+    container_a: String,
+    container_b: String,
+    netns_a: String,
+    netns_b: String,
+    ip_a: String,
+    netns_a_path: PathBuf,
+    netns_b_path: PathBuf,
+}
+
+impl DockerNetns {
+    fn create() -> Self {
+        let id = unique_id();
+        let image = env_string("TX_THROUGHPUT_DOCKER_IMAGE", "busybox:latest");
+        let network = env_string("TX_THROUGHPUT_DOCKER_NET", &format!("easytier-bench-{id}"));
+        let subnet = env_string("TX_THROUGHPUT_DOCKER_SUBNET", DEFAULT_DOCKER_SUBNET);
+        let ip_a = env_string("TX_THROUGHPUT_DOCKER_IP_A", DEFAULT_DOCKER_IP_A);
+        let ip_b = env_string("TX_THROUGHPUT_DOCKER_IP_B", DEFAULT_DOCKER_IP_B);
+        let container_a = format!("easytier-bench-a-{id}");
+        let container_b = format!("easytier-bench-b-{id}");
+        let netns_a = format!("easytier-bench-a-{id}");
+        let netns_b = format!("easytier-bench-b-{id}");
+
+        docker(&[
+            "network", "create", "--driver", "bridge", "--subnet", &subnet, &network,
+        ]);
+
+        let mut docker_netns = Self {
+            network,
+            container_a,
+            container_b,
+            netns_a,
+            netns_b,
+            ip_a: ip_a.clone(),
+            netns_a_path: PathBuf::new(),
+            netns_b_path: PathBuf::new(),
+        };
+
+        docker_netns.start_container(&docker_netns.container_a, &ip_a, &image);
+        docker_netns.start_container(&docker_netns.container_b, &ip_b, &image);
+
+        let pid_a = docker(&["inspect", "-f", "{{.State.Pid}}", &docker_netns.container_a]);
+        let pid_b = docker(&["inspect", "-f", "{{.State.Pid}}", &docker_netns.container_b]);
+
+        docker_netns.netns_a_path = register_netns(&docker_netns.netns_a, &pid_a);
+        docker_netns.netns_b_path = register_netns(&docker_netns.netns_b, &pid_b);
+        docker_netns
+    }
+
+    fn start_container(&self, name: &str, ip: &str, image: &str) {
+        docker(&[
+            "run",
+            "-d",
+            "--name",
+            name,
+            "--network",
+            &self.network,
+            "--ip",
+            ip,
+            image,
+            "sleep",
+            "3600",
+        ]);
+    }
+}
+
+impl Drop for DockerNetns {
+    fn drop(&mut self) {
+        let _ = std::fs::remove_file(&self.netns_a_path);
+        let _ = std::fs::remove_file(&self.netns_b_path);
+        docker_ignore(&["rm", "-f", &self.container_a, &self.container_b]);
+        docker_ignore(&["network", "rm", &self.network]);
+    }
+}
+
+fn bench_tx_throughput(c: &mut Criterion) {
+    let tunnel = env_string("TX_THROUGHPUT_TUNNEL", "ring")
+        .parse::<TunnelKind>()
+        .unwrap_or_else(|err| panic!("{err}"));
+    let packet_size = env_parse("TX_THROUGHPUT_PKT_SIZE", 1400usize);
+    const MIN_PKT_SIZE: usize = 28; // IPv4 (20) + UDP (8) header
+    assert!(
+        packet_size >= MIN_PKT_SIZE,
+        "TX_THROUGHPUT_PKT_SIZE={packet_size} is smaller than the minimum {MIN_PKT_SIZE} (IPv4+UDP headers)"
+    );
+    let worker_threads = env_parse("TX_THROUGHPUT_WORKER_THREADS", 4usize);
+    let inflight_depth = env_parse("TX_THROUGHPUT_INFLIGHT", 64usize).max(1);
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(worker_threads)
+        .enable_all()
+        .build()
+        .expect("create tokio runtime");
+
+    let topology = runtime.block_on(setup_topology(tunnel, packet_size));
+    let peer_manager = topology.inst_a.get_peer_manager();
+    let packet = topology.packet.clone();
+    let dst = topology.dst;
+
+    eprintln!(
+        "tx_throughput: tunnel={} inflight={} workers={} pkt_size={}",
+        tunnel.as_str(),
+        inflight_depth.max(1),
+        worker_threads,
+        packet_size
+    );
+
+    let mut group = c.benchmark_group("tx_throughput");
+    group.throughput(Throughput::Bytes(packet_size as u64));
+
+    // Serial baseline: one packet in flight at a time.
+    // Measures per-packet CPU cost (TX injection latency).
+    group.bench_function(tunnel.as_str(), |b| {
+        b.iter_custom(|iterations| {
+            let pm = peer_manager.clone();
+            let pkt = packet.clone();
+            runtime.block_on(async move {
+                let start = Instant::now();
+                for _ in 0..iterations {
+                    pm.send_msg_by_ip(pkt.clone(), dst, false)
+                        .await
+                        .expect("send packet by EasyTier IP");
+                }
+                start.elapsed()
+            })
+        });
+    });
+
+    // Saturate: spawn TX_THROUGHPUT_INFLIGHT worker tasks, each independently
+    // pumping send_msg_by_ip. Work is distributed across tokio worker threads,
+    // exposing the peer manager + tunnel's true aggregate throughput ceiling.
+    // With TX_THROUGHPUT_INFLIGHT=1 it degrades to the serial baseline.
+    group.bench_function(format!("{}-saturate", tunnel.as_str()), |b| {
+        b.iter_custom(|iterations| {
+            let pm = peer_manager.clone();
+            let pkt = packet.clone();
+            let concurrency = inflight_depth.min(iterations as usize).max(1);
+            runtime.block_on(async move {
+                let counter = Arc::new(AtomicU64::new(iterations));
+                let start = Instant::now();
+                let mut handles = Vec::with_capacity(concurrency);
+                for _ in 0..concurrency {
+                    let pm = pm.clone();
+                    let pkt = pkt.clone();
+                    let counter = counter.clone();
+                    handles.push(tokio::spawn(async move {
+                        loop {
+                            if counter
+                                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |cur| {
+                                    if cur > 0 { Some(cur - 1) } else { None }
+                                })
+                                .is_err()
+                            {
+                                return;
+                            }
+                            pm.send_msg_by_ip(pkt.clone(), dst, false)
+                                .await
+                                .expect("send packet by EasyTier IP");
+                        }
+                    }));
+                }
+                for h in handles {
+                    h.await.expect("saturate worker task panicked");
+                }
+                start.elapsed()
+            })
+        });
+    });
+
+    group.finish();
+
+    runtime.block_on(async move {
+        drop(topology);
+    });
+}
+
+async fn setup_topology(tunnel: TunnelKind, packet_size: usize) -> BenchTopology {
+    let tunnel_port = env_parse("TX_THROUGHPUT_TUNNEL_PORT", DEFAULT_TUNNEL_PORT);
+    let docker = match tunnel {
+        TunnelKind::Ring => None,
+        TunnelKind::Tcp | TunnelKind::Udp => Some(DockerNetns::create()),
+    };
+
+    let (netns_a, netns_b) = match &docker {
+        Some(docker) => (Some(docker.netns_a.clone()), Some(docker.netns_b.clone())),
+        None => (None, None),
+    };
+    let listeners_a = match tunnel {
+        TunnelKind::Ring => Vec::new(),
+        TunnelKind::Tcp | TunnelKind::Udp => vec![
+            format!("{}://0.0.0.0:{}", tunnel.as_str(), tunnel_port)
+                .parse()
+                .unwrap(),
+        ],
+    };
+
+    let mut inst_a = Instance::new(no_tun_config("hot-a", VIRTUAL_IP_A, netns_a, listeners_a));
+    let mut inst_b = Instance::new(no_tun_config("hot-b", VIRTUAL_IP_B, netns_b, Vec::new()));
+
+    inst_a.run().await.expect("inst_a run");
+    inst_b.run().await.expect("inst_b run");
+
+    match tunnel {
+        TunnelKind::Ring => inst_b
+            .get_conn_manager()
+            .add_connector(RingTunnelConnector::new(
+                format!("ring://{}", inst_a.id()).parse().unwrap(),
+            )),
+        TunnelKind::Tcp => inst_b
+            .get_conn_manager()
+            .add_connector(TcpTunnelConnector::new(
+                format!(
+                    "tcp://{}:{}",
+                    docker.as_ref().expect("tcp benchmark needs Docker").ip_a,
+                    tunnel_port
+                )
+                .parse()
+                .unwrap(),
+            )),
+        TunnelKind::Udp => inst_b
+            .get_conn_manager()
+            .add_connector(UdpTunnelConnector::new(
+                format!(
+                    "udp://{}:{}",
+                    docker.as_ref().expect("udp benchmark needs Docker").ip_a,
+                    tunnel_port
+                )
+                .parse()
+                .unwrap(),
+            )),
+    }
+
+    wait_for_routes(&inst_a, &inst_b).await;
+
+    BenchTopology {
+        _docker: docker,
+        inst_a,
+        _inst_b: inst_b,
+        dst: VIRTUAL_IP_B.parse().unwrap(),
+        packet: make_data_packet(VIRTUAL_IP_A, VIRTUAL_IP_B, packet_size),
+    }
+}
+
+async fn wait_for_routes(inst_a: &Instance, inst_b: &Instance) {
+    tokio::time::timeout(Duration::from_secs(15), async {
+        loop {
+            let routes_a = inst_a.get_peer_manager().list_routes().await;
+            let routes_b = inst_b.get_peer_manager().list_routes().await;
+            if !routes_a.is_empty() && !routes_b.is_empty() {
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+    })
+    .await
+    .expect("EasyTier routes did not converge within 15s");
+}
+
+fn make_data_packet(src: &str, dst: &str, total_size: usize) -> ZCPacket {
+    use std::net::Ipv4Addr;
+
+    let hdr_len = 28;
+    let payload_len = total_size.saturating_sub(hdr_len);
+    let ip_total_len = (hdr_len + payload_len) as u16;
+    let mut buf = BytesMut::with_capacity(total_size);
+
+    buf.extend_from_slice(&[
+        0x45,
+        0x00,
+        (ip_total_len >> 8) as u8,
+        (ip_total_len & 0xff) as u8,
+        0x00,
+        0x00,
+        0x40,
+        0x00,
+        0x40,
+        0x11,
+        0x00,
+        0x00,
+    ]);
+    let src: Ipv4Addr = src.parse().unwrap();
+    buf.extend_from_slice(&src.octets());
+    let dst: Ipv4Addr = dst.parse().unwrap();
+    buf.extend_from_slice(&dst.octets());
+
+    let udp_len = (8 + payload_len) as u16;
+    buf.extend_from_slice(&[
+        0x30,
+        0x39,
+        0xd4,
+        0x31,
+        (udp_len >> 8) as u8,
+        (udp_len & 0xff) as u8,
+        0x00,
+        0x00,
+    ]);
+
+    buf.resize(total_size, 0xaa);
+    ZCPacket::new_with_payload(&buf)
+}
+
+fn no_tun_config(
+    name: &str,
+    ipv4: &str,
+    netns: Option<String>,
+    listeners: Vec<url::Url>,
+) -> TomlConfigLoader {
+    let config = TomlConfigLoader::default();
+    config.set_inst_name(name.to_owned());
+    config.set_netns(netns);
+    config.set_ipv4(Some(ipv4.parse().unwrap()));
+    config.set_listeners(listeners);
+    let mut flags = config.get_flags();
+    flags.no_tun = true;
+    config.set_flags(flags);
+    config
+}
+
+fn register_netns(name: &str, pid: &str) -> PathBuf {
+    #[cfg(target_os = "linux")]
+    {
+        let dir = PathBuf::from("/var/run/netns");
+        std::fs::create_dir_all(&dir).expect("create /var/run/netns");
+        let path = dir.join(name);
+        let _ = std::fs::remove_file(&path);
+        std::os::unix::fs::symlink(format!("/proc/{pid}/ns/net"), &path)
+            .expect("link Docker netns into /var/run/netns");
+        path
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    {
+        let _ = (name, pid);
+        panic!("Docker netns benchmark requires Linux");
+    }
+}
+
+fn docker(args: &[&str]) -> String {
+    let output = Command::new("docker")
+        .args(args)
+        .output()
+        .unwrap_or_else(|err| panic!("failed to run docker {args:?}: {err}"));
+    if !output.status.success() {
+        panic!(
+            "docker {:?} failed with status {:?}: {}",
+            args,
+            output.status.code(),
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+    String::from_utf8_lossy(&output.stdout).trim().to_owned()
+}
+
+fn docker_ignore(args: &[&str]) {
+    let _ = Command::new("docker")
+        .args(args)
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .status();
+}
+
+fn env_string(name: &str, default: &str) -> String {
+    std::env::var(name).unwrap_or_else(|_| default.to_owned())
+}
+
+fn env_parse<T>(name: &str, default: T) -> T
+where
+    T: FromStr,
+    T::Err: std::fmt::Display,
+{
+    match std::env::var(name) {
+        Ok(value) => value
+            .parse()
+            .unwrap_or_else(|err| panic!("invalid {name}={value:?}: {err}")),
+        Err(_) => default,
+    }
+}
+
+fn criterion_config() -> Criterion {
+    let measurement_secs = env_parse("TX_THROUGHPUT_MEASUREMENT_SECS", 10u64);
+    let warmup_secs = env_parse("TX_THROUGHPUT_WARMUP_SECS", 3u64);
+    let sample_size = env_parse("TX_THROUGHPUT_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)
+}
+
+fn unique_id() -> String {
+    let nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("system clock before UNIX epoch")
+        .as_nanos();
+    format!("{}-{nanos}", std::process::id())
+}
+
+criterion_group! {
+    name = benches;
+    config = criterion_config();
+    targets = bench_tx_throughput
+}
+criterion_main!(benches);

easytier/src/common/acl_processor.rs

@@ -6,7 +6,7 @@ use std::{
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
-use hotpath::instant::Instant;
+use quanta::Instant;
 
 use crate::common::{config::ConfigLoader, global_ctx::ArcGlobalCtx, token_bucket::TokenBucket};
 use crate::proto::acl::*;

easytier/src/common/global_ctx.rs

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

easytier/src/common/ifcfg/mod.rs

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

easytier/src/common/ifcfg/netlink.rs

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

easytier/src/common/stats_manager.rs

@@ -1,5 +1,5 @@
 use dashmap::DashMap;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use serde::{Deserialize, Serialize};
 use std::cell::UnsafeCell;
 use std::fmt;

easytier/src/common/stun.rs

@@ -8,7 +8,7 @@ use crate::proto::common::{NatType, StunInfo};
 use anyhow::Context;
 use chrono::Local;
 use crossbeam::atomic::AtomicCell;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use rand::seq::IteratorRandom;
 use socket2::{SockAddr, SockRef};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};

easytier/src/connector/direct.rs

@@ -11,7 +11,7 @@ use std::{
     time::Duration,
 };
 
-use hotpath::instant::Instant;
+use quanta::Instant;
 
 use crate::{
     common::{

easytier/src/connector/manual.rs

@@ -6,7 +6,7 @@ use std::{
 };
 
 use dashmap::DashSet;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use tokio::{sync::mpsc, task::JoinSet, time::timeout};
 
 use crate::{

easytier/src/connector/tcp_hole_punch.rs

@@ -5,7 +5,7 @@ use std::{
 };
 
 use anyhow::{Context, Error};
-use hotpath::instant::Instant;
+use quanta::Instant;
 use rand::Rng as _;
 use tokio::task::JoinSet;
 

easytier/src/connector/udp_hole_punch/both_easy_sym.rs

@@ -5,7 +5,7 @@ use std::{
 };
 
 use anyhow::Context;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use tokio::sync::Mutex;
 use tokio_util::task::AbortOnDropHandle;
 

easytier/src/connector/udp_hole_punch/common.rs

@@ -7,7 +7,7 @@ use std::{
 use crossbeam::atomic::AtomicCell;
 use dashmap::{DashMap, DashSet};
 use guarden::defer;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use rand::seq::SliceRandom as _;
 use tokio::{net::UdpSocket, sync::Mutex, task::JoinSet};
 use tracing::{Instrument, Level, instrument};

easytier/src/connector/udp_hole_punch/cone.rs

@@ -1,7 +1,7 @@
 use std::{sync::Arc, time::Duration};
 
 use anyhow::Context;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use tokio::net::UdpSocket;
 use tokio_util::task::AbortOnDropHandle;
 

easytier/src/connector/udp_hole_punch/mod.rs

@@ -8,8 +8,8 @@ use both_easy_sym::{PunchBothEasySymHoleClient, PunchBothEasySymHoleServer};
 use common::{PunchHoleServerCommon, UdpNatType, UdpPunchClientMethod};
 use cone::{PunchConeHoleClient, PunchConeHoleServer};
 use dashmap::DashMap;
-use hotpath::instant::Instant;
 use once_cell::sync::Lazy;
+use quanta::Instant;
 use sym_to_cone::{PunchSymToConeHoleClient, PunchSymToConeHoleServer};
 use tokio::{sync::Mutex, task::JoinHandle};
 

easytier/src/connector/udp_hole_punch/sym_to_cone.rs

@@ -10,7 +10,7 @@ use std::{
 
 use anyhow::Context;
 use guarden::defer;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use rand::{Rng, seq::SliceRandom};
 use tokio::{net::UdpSocket, sync::RwLock};
 use tokio_util::task::AbortOnDropHandle;

easytier/src/easytier-core.rs

@@ -1,11 +1,5 @@
 use easytier::core;
 
-#[cfg(all(
-    feature = "hotpath-alloc",
-    any(feature = "jemalloc", feature = "mimalloc")
-))]
-compile_error!("feature `hotpath-alloc` cannot be enabled together with `jemalloc` or `mimalloc`");
-
 #[cfg(all(feature = "mimalloc", not(feature = "jemalloc")))]
 use mimalloc::MiMalloc;
 

easytier/src/gateway/icmp_proxy.rs

@@ -7,13 +7,13 @@ use std::{
 };
 
 use anyhow::Context;
-use hotpath::instant::Instant;
 use pnet::packet::{
     Packet,
     icmp::{self, IcmpCode, IcmpTypes, MutableIcmpPacket, echo_reply::MutableEchoReplyPacket},
     ip::IpNextHeaderProtocols,
     ipv4::Ipv4Packet,
 };
+use quanta::Instant;
 use socket2::Socket;
 use tokio::{
     sync::{Mutex, mpsc::UnboundedSender},

easytier/src/gateway/ip_reassembler.rs

@@ -1,8 +1,8 @@
 use dashmap::DashMap;
-use hotpath::instant::Instant;
 use pnet::packet::Packet;
 use pnet::packet::ip::IpNextHeaderProtocol;
 use pnet::packet::ipv4::{self, Ipv4Flags, Ipv4Packet, MutableIpv4Packet};
+use quanta::Instant;
 use std::net::Ipv4Addr;
 use std::time::Duration;
 

easytier/src/gateway/quic_proxy.rs

@@ -1018,7 +1018,7 @@ impl TcpProxyRpc for QuicProxyDstRpcService {
 mod tests {
     use super::*;
     use bytes::Buf;
-    use hotpath::instant::Instant;
+    use quanta::Instant;
 
     /// Helper function: Create a pair of interconnected QuicSockets.
     /// Data sent by socket_a will enter socket_b's rx, and vice versa.

easytier/src/gateway/socks5.rs

@@ -9,9 +9,9 @@ use std::{
 };
 
 use crossbeam::atomic::AtomicCell;
-use hotpath::instant::Instant;
 #[cfg(feature = "kcp")]
 use kcp_sys::{endpoint::KcpEndpoint, stream::KcpStream};
+use quanta::Instant;
 use tokio_util::sync::{CancellationToken, DropGuard};
 use tokio_util::task::AbortOnDropHandle;
 

easytier/src/gateway/socks5/dataplane.rs

@@ -26,7 +26,7 @@ use std::{
 };
 
 use anyhow::Context as _;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use tokio::io::{AsyncRead, AsyncWrite};
 
 use crate::{common::error::Error, gateway::fast_socks5::server::AsyncTcpConnector};

easytier/src/gateway/tcp_proxy.rs

@@ -3,12 +3,12 @@ use cidr::Ipv4Inet;
 use core::panic;
 use crossbeam::atomic::AtomicCell;
 use dashmap::DashMap;
-use hotpath::instant::Instant;
 use pnet::packet::MutablePacket;
 use pnet::packet::Packet;
 use pnet::packet::ip::IpNextHeaderProtocols;
 use pnet::packet::ipv4::{Ipv4Packet, MutableIpv4Packet};
 use pnet::packet::tcp::{MutableTcpPacket, TcpPacket, ipv4_checksum};
+use quanta::Instant;
 use socket2::{SockRef, TcpKeepalive};
 use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::sync::atomic::{AtomicBool, AtomicU16};

easytier/src/gateway/udp_proxy.rs

@@ -8,13 +8,13 @@ use bytes::{BufMut, BytesMut};
 use cidr::Ipv4Inet;
 use crossbeam::atomic::AtomicCell;
 use dashmap::DashMap;
-use hotpath::instant::Instant;
 use pnet::packet::{
     Packet,
     ip::IpNextHeaderProtocols,
     ipv4::Ipv4Packet,
     udp::{self, MutableUdpPacket},
 };
+use quanta::Instant;
 use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
 use tokio::{
     net::UdpSocket,

easytier/src/hotpath_off.rs

@@ -38,3 +38,15 @@ macro_rules! rw_lock {
         $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;
+}

easytier/src/instance/instance.rs

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

easytier/src/instance/proxy_cidrs_monitor.rs

@@ -3,7 +3,7 @@ use std::sync::{Arc, Weak};
 
 use crate::common::global_ctx::{ArcGlobalCtx, GlobalCtxEvent};
 use crate::peers::peer_manager::PeerManager;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use tokio_util::task::AbortOnDropHandle;
 
 /// ProxyCidrsMonitor monitors changes in proxy CIDRs from peer routes

easytier/src/instance/virtual_nic.rs

@@ -1361,12 +1361,11 @@ impl NicCtx {
                     }
 
                     self.global_ctx
-                        .issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
+                        .set_tun_device_ready(nic.ifname().to_string());
                     ret
                 }
                 Err(err) => {
-                    self.global_ctx
+                    self.global_ctx.set_tun_device_error(err.to_string());
-                        .issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
                     return Err(err);
                 }
             }
@@ -1405,12 +1404,11 @@ impl NicCtx {
             match nic.create_dev_for_mobile(tun_fd).await {
                 Ok(ret) => {
                     self.global_ctx
-                        .issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
+                        .set_tun_device_ready(nic.ifname().to_string());
                     ret
                 }
                 Err(err) => {
-                    self.global_ctx
+                    self.global_ctx.set_tun_device_error(err.to_string());
-                        .issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
                     return Err(err);
                 }
             }

easytier/src/launcher.rs

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

easytier/src/lib.rs

@@ -13,21 +13,18 @@ use clap_complete::{Generator, Shell};
 extern crate self as hotpath;
 #[cfg(not(feature = "hotpath"))]
 mod hotpath_off;
-
-// When the `hotpath` feature is off, expose a local `instant` module backed by
-// `quanta::Instant` so call sites can uniformly write `use hotpath::instant::Instant;`
-// regardless of whether the feature is enabled. With the feature on, the real
-// `hotpath` crate provides the same path (also `quanta::Instant` on Linux), so
-// the two modes resolve to the identical type.
 #[cfg(not(feature = "hotpath"))]
-pub mod instant {
+pub(crate) use hotpath_off::wrap;
-    pub type Instant = quanta::Instant;
+
-}
+// `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.
-pub use hotpath::instant::Instant;
+pub use quanta::Instant;
 
 mod arch;
 mod gateway;

easytier/src/peers/acl_filter.rs

@@ -7,11 +7,11 @@ use std::{
 
 use arc_swap::ArcSwap;
 use dashmap::DashMap;
-use hotpath::instant::Instant;
 use pnet::packet::ipv6::Ipv6Packet;
 use pnet::packet::{
     Packet as _, ip::IpNextHeaderProtocols, ipv4::Ipv4Packet, tcp::TcpPacket, udp::UdpPacket,
 };
+use quanta::Instant;
 
 use crate::proto::acl::{AclStats, Protocol};
 use crate::tunnel::packet_def::PacketType;
@@ -404,7 +404,7 @@ mod tests {
         sync::Arc,
     };
 
-    use hotpath::instant::Instant;
+    use quanta::Instant;
 
     use crate::{
         common::acl_processor::PacketInfo,

easytier/src/peers/mod.rs

@@ -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 PacketRecvChan = hotpath::wrap::tokio::sync::mpsc::Sender<ZCPacket>;
-pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver<ZCPacket>;
+pub type PacketRecvChanReceiver = hotpath::wrap::tokio::sync::mpsc::Receiver<ZCPacket>;
 pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
     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> {

easytier/src/peers/peer.rs

@@ -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();

easytier/src/peers/peer_conn.rs

@@ -384,10 +384,10 @@ impl PeerConn {
             session_filter,
             noise_handshake_result: None,
 
-            tunnel: Arc::new(hotpath::mutex!(tokio::sync::Mutex::new(Box::new(
+            tunnel: Arc::new(hotpath::mutex!(tokio::sync::Mutex::new(
-                guard!([mut mpsc_tunnel] mpsc_tunnel.close()),
+                Box::new(guard!([mut mpsc_tunnel] mpsc_tunnel.close()))
-            )
+                    as Box<dyn Any + Send + 'static>,
-                as Box<dyn Any + Send + 'static>))),
+            ))),
             sink,
             recv: hotpath::mutex!(tokio::sync::Mutex::new(Some(recv))),
             tunnel_info,

easytier/src/peers/peer_conn_ping.rs

@@ -6,7 +6,7 @@ use std::{
     time::Duration,
 };
 
-use hotpath::instant::Instant;
+use quanta::Instant;
 use rand::{Rng, thread_rng};
 use tokio::{
     sync::broadcast,

easytier/src/peers/peer_manager.rs

@@ -2,7 +2,7 @@ use anyhow::Context;
 use async_trait::async_trait;
 use cidr::{Ipv4Cidr, Ipv6Cidr};
 use dashmap::DashMap;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use std::collections::BTreeSet;
 use std::{
     fmt::Debug,
@@ -963,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;
@@ -1697,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>,
@@ -2207,7 +2209,7 @@ mod tests {
     use base64::Engine;
     use std::{collections::HashMap, fmt::Debug, sync::Arc, time::Duration};
 
-    use hotpath::instant::Instant;
+    use quanta::Instant;
 
     use crate::{
         common::{

easytier/src/peers/peer_map.rs

@@ -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())),
         }
     }

easytier/src/peers/peer_ospf_route.rs

@@ -13,7 +13,6 @@ use arc_swap::ArcSwap;
 use cidr::{IpCidr, Ipv4Cidr, Ipv6Cidr, Ipv6Inet};
 use crossbeam::atomic::AtomicCell;
 use dashmap::DashMap;
-use hotpath::instant::Instant;
 use ordered_hash_map::OrderedHashMap;
 use parking_lot::{RwLock, lock_api::RwLockUpgradableReadGuard};
 use petgraph::{
@@ -25,6 +24,7 @@ use petgraph::{
 use prefix_trie::PrefixMap;
 use prost::Message;
 use prost_reflect::{DynamicMessage, ReflectMessage};
+use quanta::Instant;
 use tokio::{
     select,
     sync::Mutex,

easytier/src/peers/peer_session.rs

@@ -1,13 +1,17 @@
 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;
 use crossbeam::atomic::AtomicCell;
 use dashmap::DashMap;
-use hotpath::instant::Instant;
+use quanta::Instant;
 
 use super::secure_datagram::{SecureDatagramDirection, SecureDatagramSession};
 use crate::{
@@ -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,

easytier/src/peers/relay_peer_map.rs

@@ -1,8 +1,8 @@
 use std::sync::Arc;
 
 use dashmap::DashMap;
-use hotpath::instant::Instant;
 use prost::Message;
+use quanta::Instant;
 use snow::params::NoiseParams;
 use tokio::sync::{Mutex, OwnedMutexGuard, oneshot};
 use tokio::time::{Duration, timeout};
@@ -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);

easytier/src/peers/route_trait.rs

@@ -1,7 +1,7 @@
 use cidr::Ipv6Inet;
 use cidr::{Ipv4Cidr, Ipv6Cidr};
 use dashmap::DashMap;
-use hotpath::instant::Instant;
+use quanta::Instant;
 use std::{
     collections::BTreeSet,
     net::{Ipv4Addr, Ipv6Addr},

easytier/src/peers/secure_datagram.rs

@@ -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("rx_slots", &*self.rx_slots.lock().unwrap())
-            .field("key_cache", &self.key_cache)
+            .field("key_cache", &*self.key_cache.lock().unwrap())
-            .field("sync_rx_grace", &self.sync_rx_grace)
+            .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),
+            rx_slots: hotpath::mutex!(std::sync::Mutex::new(rx_slots)),
-            key_cache: Mutex::new(key_cache),
+            key_cache: hotpath::mutex!(std::sync::Mutex::new(key_cache)),
-            sync_rx_grace: Mutex::new(SyncRxGrace::default()),
+            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,
@@ -892,11 +897,9 @@ mod tests {
         let nonce_offset = payload.len() - StandardAeadTail::NONCE_SIZE;
         payload[nonce_offset..].copy_from_slice(&poisoned_nonce);
 
-        assert!(
+        assert!(receiver
-            receiver
+            .decrypt_payload(SecureDatagramDirection::AToB, &mut forged)
-                .decrypt_payload(SecureDatagramDirection::AToB, &mut forged)
+            .is_err());
-                .is_err()
-        );
 
         let plaintext = b"pkt2";
         let mut pkt2 = ZCPacket::new_with_payload(plaintext);

easytier/src/proto/rpc_impl/client.rs

@@ -5,8 +5,8 @@ use std::sync::{Arc, Mutex};
 use bytes::Bytes;
 use dashmap::DashMap;
 use guarden::defer;
-use hotpath::instant::Instant;
 use prost::Message;
+use quanta::Instant;
 use tokio::sync::mpsc;
 use tokio::task::JoinSet;
 use tokio::time::timeout;

easytier/src/proto/rpc_impl/packet.rs

@@ -1,6 +1,6 @@
 use prost::{Message as _, length_delimiter_len};
 
-use hotpath::instant::Instant;
+use quanta::Instant;
 
 use crate::{
     common::{PeerId, compressor::DefaultCompressor},

easytier/src/proto/rpc_impl/server.rs

@@ -5,8 +5,8 @@ use std::{
 
 use bytes::Bytes;
 use dashmap::DashMap;
-use hotpath::instant::Instant;
 use prost::Message;
+use quanta::Instant;
 use tokio::{task::JoinSet, time::timeout};
 use tokio_stream::StreamExt;
 

easytier/src/tests/ipv6_test.rs

@@ -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(),

easytier/src/tests/three_node.rs

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

easytier/src/tunnel/fake_tcp/stack.rs

@@ -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,7 +162,7 @@ impl Socket {
         remote_mac: Option<MacAddr>,
         ack: Option<u32>,
         state: State,
-    ) -> (Socket, flume::Sender<Bytes>) {
+    ) -> (Socket, hotpath::wrap::flume::Sender<Bytes>) {
         let (incoming_tx, incoming_rx) = hotpath::channel!(flume::bounded(MPMC_BUFFER_LEN));
 
         (
@@ -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();

easytier/src/tunnel/mpsc.rs

@@ -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(())

easytier/src/tunnel/quic.rs

@@ -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()),
+            ephemeral: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
-            persistent: RwLock::new(RwPoolInner::default()),
+            persistent: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
             capacity,
         }
     }

easytier/src/tunnel/ring.rs

@@ -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};
+use hotpath::wrap::tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::sync::mpsc::unbounded_channel;
 
 use uuid::Uuid;
 
@@ -196,8 +197,7 @@ pub struct RingTunnelListener {
 
 impl RingTunnelListener {
     pub fn new(key: url::Url) -> Self {
-        let (conn_sender, conn_receiver) =
+        let (conn_sender, conn_receiver) = hotpath::channel!(unbounded_channel());
-            hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
         RingTunnelListener {
             listener_addr: key,
             conn_sender,

easytier/src/tunnel/udp.rs

@@ -13,9 +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},
+    sync::mpsc::{channel, unbounded_channel},
     task::JoinSet,
 };
 use tokio_util::task::AbortOnDropHandle;
@@ -291,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>,
@@ -325,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,
@@ -393,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();
@@ -539,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 {
@@ -645,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();
@@ -673,9 +679,8 @@ pub struct UdpTunnelListener {
 
 impl UdpTunnelListener {
     pub fn new(addr: url::Url) -> Self {
-        let (close_event_send, close_event_recv) =
+        let (close_event_send, close_event_recv) = hotpath::channel!(unbounded_channel());
-            hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
+        let (conn_send, conn_recv) = hotpath::channel!(channel(100));
-        let (conn_send, conn_recv) = hotpath::channel!(tokio::sync::mpsc::channel(100));
         Self {
             addr: addr.clone(),
             socket: None,
@@ -916,7 +921,7 @@ impl UdpTunnelConnector {
         );
 
         let (close_event_sender, mut close_event_recv) =
-            hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
+            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,

easytier/src/tunnel/wireguard.rs

@@ -6,7 +6,7 @@ use std::{
     time::Duration,
 };
 
-use hotpath::instant::Instant;
+use quanta::Instant;
 
 use super::{
     FromUrl, IpVersion, Tunnel, TunnelError, TunnelInfo, TunnelListener, TunnelUrl, ZCPacketSink,
@@ -37,7 +37,19 @@ use crossbeam::atomic::AtomicCell;
 use dashmap::DashMap;
 use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
 use rand::RngCore;
-use tokio::{net::UdpSocket, sync::Mutex, task::JoinSet};
+#[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::mpsc::unbounded_channel,
+    task::JoinSet,
+};
 
 const MAX_PACKET: usize = 2048;
 
@@ -343,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<()>,
@@ -354,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(),
@@ -375,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;
@@ -386,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());
@@ -451,8 +465,8 @@ impl WgPeer {
     }
 }
 
-type ConnSender = tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
+type ConnSender = hotpath::wrap::tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
-type ConnReceiver = tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
+type ConnReceiver = hotpath::wrap::tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
 
 pub struct WgTunnelListener {
     addr: url::Url,
@@ -470,7 +484,7 @@ pub struct WgTunnelListener {
 
 impl WgTunnelListener {
     pub fn new(addr: url::Url, config: WgConfig) -> Self {
-        let (conn_send, conn_recv) = hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
+        let (conn_send, conn_recv) = hotpath::channel!(unbounded_channel());
         WgTunnelListener {
             addr,
             config,