style: format noop waker mpsc tests

perf(mpsc): extend noop_waker sync send to TCP tunnels
TCP tunnel uses FramedWriter (not RingSink), but start_send is still sync (writes to BufList in memory). poll_flush does actual TCP write syscall — noop_waker returns Ok for Pending (data stays in BufList, flushed on next send when BufList >= 64). Add TCP benchmark support via HOTPATH_TUNNEL=tcp. Note: TCP/UDP convergence requires netns in bench environment (connector multi-bind address behavior doesn't work for localhost without namespaces). All 210 peers tests pass. Ring tunnel benchmark: 234K -> 508K pps (+117%).
2026-09-21 03:24:00 +00:00 · 2026-06-30 21:26:33 +08:00 · 2026-06-30 21:26:33 +08:00 · 2026-06-30 21:26:33 +08:00 · 2026-06-30 21:26:33 +08:00 · 2026-06-30 21:26:33 +08:00
46 changed files with 916 additions and 331 deletions
@@ -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"
@@ -257,12 +263,6 @@ 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"
@@ -1145,6 +1145,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"
@@ -1255,10 +1261,31 @@ dependencies = [
 ]

 [[package]]
-name = "chunked_transfer"
-version = "1.5.0"
+name = "ciborium"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
+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"
@@ -1605,6 +1632,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"
@@ -2100,16 +2163,6 @@ 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"
@@ -2133,17 +2186,6 @@ 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"
@@ -2299,6 +2341,7 @@ dependencies = [
 "clap_complete",
 "clap_complete_nushell",
 "console-subscriber",
+ "criterion",
 "crossbeam",
 "ctor 0.8.0",
 "dashmap",
@@ -2322,7 +2365,6 @@ dependencies = [
 "hickory-resolver",
 "hickory-server",
 "hmac",
- "hotpath",
 "http",
 "http_req",
 "humansize",
@@ -2665,12 +2707,6 @@ 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"
@@ -3929,61 +3965,6 @@ dependencies = [
 "windows-sys 0.52.0",
 ]

-[[package]]
-name = "hotpath"
-version = "0.18.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc2c28b1fa962e433f800ed1ea0bf53dc028d3745cf2acec6cfd28b65ac96afa"
-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.18.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a585238d8daf746e27df0f24d1bbdcd2410e9febff63f9a0173f90d7e71c50f6"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.117",
-]
-
-[[package]]
-name = "hotpath-macros-meta"
-version = "0.18.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "309f63c2f755dead454dd4b3ea8ab5c947f14f8ea435fbcd37fa820e17290e80"
-
-[[package]]
-name = "hotpath-meta"
-version = "0.18.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68faa91a9e1114dff668cd90560f332da6bbde40dae37ec28ea1c43ca5ce3be3"
-dependencies = [
- "hotpath-macros-meta",
-]
-
 [[package]]
 name = "html5ever"
 version = "0.29.1"
@@ -5952,15 +5933,6 @@ 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"
@@ -5971,6 +5943,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 +6579,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"
@@ -6831,19 +6837,6 @@ 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"
@@ -7903,12 +7896,6 @@ 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"
@@ -9833,17 +9820,6 @@ 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"
@@ -9999,18 +9975,6 @@ 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"
@@ -10021,6 +9985,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"
@@ -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,7 +56,6 @@ 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"

 delegate = "0.13.5"
@@ -337,6 +340,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,17 +407,14 @@ jemalloc-prof = [
    "jemalloc-sys/stats",
 ]
 tracing = ["tokio/tracing", "dep:console-subscriber"]
-hotpath = [
-    "dep:hotpath",
-    "hotpath/hotpath",
-    "hotpath/tokio",
-    "hotpath/parking_lot",
-    "hotpath/flume",
-]
-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"]
+# 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"]
 # For Network Extension on macOS
 macos-ne = []
@@ -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.
@@ -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);
@@ -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::*;
@@ -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;
@@ -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};
@@ -11,7 +11,7 @@ use std::{
    time::Duration,
 };

-use hotpath::instant::Instant;
+use quanta::Instant;

 use crate::{
    common::{
@@ -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::{
@@ -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;

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

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

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

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

@@ -30,16 +24,6 @@ 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
 }
@@ -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},
@@ -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;

@@ -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.
@@ -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;

@@ -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};
@@ -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};
@@ -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,
@@ -1,40 +0,0 @@
-//! 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
-    };
-}
@@ -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
@@ -5,29 +5,10 @@ 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;
-
-// 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 type Instant = quanta::Instant;
-}
-
 // 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;
@@ -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,
@@ -59,7 +59,7 @@ type BoxNicPacketFilter = Box<dyn NicPacketFilter + Send + Sync>;
 pub type PacketRecvChan = tokio::sync::mpsc::Sender<ZCPacket>;
 pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver<ZCPacket>;
 pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
-    hotpath::channel!(tokio::sync::mpsc::channel(128))
+    tokio::sync::mpsc::channel(128)
 }
 pub async fn recv_packet_from_chan(
    packet_recv_chan_receiver: &mut PacketRecvChanReceiver,
@@ -207,7 +207,6 @@ 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,9 +11,6 @@ use std::{
    },
 };

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

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

        let (recv, sink) = (mpsc_tunnel.get_stream(), mpsc_tunnel.get_sink());

@@ -384,12 +389,12 @@ impl PeerConn {
            session_filter,
            noise_handshake_result: None,

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

            tasks: JoinSet::new(),
@@ -1466,7 +1471,6 @@ 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?)
    }
@@ -6,7 +6,7 @@ use std::{
    time::Duration,
 };

-use hotpath::instant::Instant;
+use quanta::Instant;
 use rand::{Rng, thread_rng};
 use tokio::{
    sync::broadcast,
@@ -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,
@@ -11,9 +11,6 @@ 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},
@@ -279,8 +276,8 @@ impl PeerManager {
        let rpc_tspt = Arc::new(RpcTransport {
            my_peer_id,
            peers: Arc::downgrade(&peers),
-            foreign_peers: hotpath::mutex!(tokio::sync::Mutex::new(None)),
-            packet_recv: hotpath::mutex!(tokio::sync::Mutex::new(peer_rpc_tspt_recv)),
+            foreign_peers: Mutex::new(None),
+            packet_recv: Mutex::new(peer_rpc_tspt_recv),
            peer_rpc_tspt_sender,
            encryptor: encryptor.clone(),
            is_secure_mode_enabled,
@@ -412,21 +409,17 @@ impl PeerManager {
            global_ctx,
            nic_channel,

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

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

            peers,

            peer_rpc_mgr,
            peer_rpc_tspt: rpc_tspt,

-            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()
-            ))),
+            peer_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
+            nic_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),

            route_algo_inst,

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

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

            reserved_my_peer_id_map: DashMap::new(),
            recent_have_traffic: Arc::new(DashMap::new()),
@@ -1444,7 +1437,6 @@ 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(
@@ -1530,7 +1522,6 @@ 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>,
@@ -2207,7 +2198,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::{
@@ -132,7 +132,6 @@ 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();
@@ -164,7 +163,6 @@ impl PeerMap {
        Ok(())
    }

-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerMap"))]
    pub async fn get_gateway_peer_id(
        &self,
        dst_peer_id: PeerId,
@@ -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,
@@ -1394,7 +1394,6 @@ 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;
@@ -1402,7 +1401,6 @@ 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| {
@@ -7,7 +7,7 @@ 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::{
@@ -376,7 +376,6 @@ impl PeerSession {
        }
    }

-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerSession"))]
    pub fn encrypt_payload(
        &self,
        sender_peer_id: PeerId,
@@ -390,7 +389,6 @@ 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,
@@ -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};
@@ -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},
@@ -701,10 +701,6 @@ impl SecureDatagramSession {
        false
    }

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

-    #[cfg_attr(
-        feature = "hotpath",
-        hotpath::measure(impl_type = "SecureDatagramSession")
-    )]
    pub fn decrypt_payload(
        &self,
        dir: SecureDatagramDirection,
@@ -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;
@@ -1,6 +1,6 @@
 use prost::{Message as _, length_delimiter_len};

-use hotpath::instant::Instant;
+use quanta::Instant;

 use crate::{
    common::{PeerId, compressor::DefaultCompressor},
@@ -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;

@@ -159,7 +159,7 @@ impl Socket {
        ack: Option<u32>,
        state: State,
    ) -> (Socket, flume::Sender<Bytes>) {
-        let (incoming_tx, incoming_rx) = hotpath::channel!(flume::bounded(MPMC_BUFFER_LEN));
+        let (incoming_tx, incoming_rx) = flume::bounded(MPMC_BUFFER_LEN);

        (
            Socket {
@@ -1,6 +1,13 @@
 // this mod wrap tunnel to a mpsc tunnel, based on crossbeam_channel

-use std::{pin::Pin, time::Duration};
+use std::{
+    cell::UnsafeCell,
+    pin::Pin,
+    sync::Arc,
+    sync::atomic::{AtomicBool, Ordering},
+    task::Poll,
+    time::Duration,
+};

 use anyhow::Context;
 use tokio::time::timeout;
@@ -11,39 +18,128 @@ use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPac

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

 use futures::SinkExt;

+/// A simple spinlock protecting a sink. The guard is Send because it only
+/// contains an atomic flag reference (no lifetime-tied borrow like MutexGuard).
+struct SpinSink {
+    locked: AtomicBool,
+    sink: UnsafeCell<Pin<Box<dyn ZCPacketSink>>>,
+}
+
+// SAFETY: access is serialized by the spinlock.
+unsafe impl Send for SpinSink {}
+unsafe impl Sync for SpinSink {}
+
+struct SpinGuard<'a> {
+    spin: &'a SpinSink,
+}
+
+impl<'a> SpinGuard<'a> {
+    fn as_mut(&mut self) -> Pin<&mut dyn ZCPacketSink> {
+        // SAFETY: we hold the spinlock, so we have exclusive access
+        let sink = unsafe { &mut *self.spin.sink.get() };
+        sink.as_mut()
+    }
+}
+
+impl Drop for SpinGuard<'_> {
+    fn drop(&mut self) {
+        self.spin.locked.store(false, Ordering::Release);
+    }
+}
+
+impl SpinSink {
+    fn new(sink: Pin<Box<dyn ZCPacketSink>>) -> Self {
+        Self {
+            locked: AtomicBool::new(false),
+            sink: UnsafeCell::new(sink),
+        }
+    }
+
+    fn try_lock(&self) -> Option<SpinGuard<'_>> {
+        if self
+            .locked
+            .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
+            .is_ok()
+        {
+            Some(SpinGuard { spin: self })
+        } else {
+            None
+        }
+    }
+}
+
 #[derive(Clone)]
-pub struct MpscTunnelSender(Sender<ZCPacket>);
+pub struct MpscTunnelSender {
+    channel_tx: Option<Sender<ZCPacket>>,
+    direct_sink: Option<Arc<SpinSink>>,
+}

 impl MpscTunnelSender {
    pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        self.0.send(item).await.with_context(|| "send error")?;
-        Ok(())
+        if let Some(sink) = &self.direct_sink {
+            // Sync fast path: no await needed, returns immediately
+            if let Some(mut guard) = sink.try_lock() {
+                let waker = futures::task::noop_waker();
+                let mut cx = std::task::Context::from_waker(&waker);
+                match guard.as_mut().poll_ready(&mut cx) {
+                    Poll::Ready(Ok(())) => {
+                        guard.as_mut().start_send(item)?;
+                        // poll_flush may return Pending when the consumer task hasn't
+                        // drained the ring yet. The data is already in the ring buffer
+                        // and will be consumed — treat Pending as success.
+                        match guard.as_mut().poll_flush(&mut cx) {
+                            Poll::Ready(Err(e)) => return Err(e),
+                            _ => return Ok(()),
+                        }
+                    }
+                    Poll::Ready(Err(e)) => return Err(e),
+                    Poll::Pending => return Err(TunnelError::BufferFull),
+                }
+            }
+            return Err(TunnelError::BufferFull);
+        }
+
+        // Channel mode: async with backpressure
+        self.send_async(item).await
    }

    pub fn try_send(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        self.0.try_send(item).map_err(|e| match e {
+        let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
+        tx.try_send(item).map_err(|e| match e {
            TrySendError::Full(_) => TunnelError::BufferFull,
            TrySendError::Closed(_) => TunnelError::Shutdown,
        })
    }
+
+    pub async fn send_async(&self, item: ZCPacket) -> Result<(), TunnelError> {
+        let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
+        match tx.try_send(item) {
+            Ok(()) => Ok(()),
+            Err(TrySendError::Full(item)) => {
+                tx.send(item).await.with_context(|| "send error")?;
+                Ok(())
+            }
+            Err(TrySendError::Closed(_)) => Err(TunnelError::Shutdown),
+        }
+    }
 }

 pub struct MpscTunnel<T> {
    tx: Option<Sender<ZCPacket>>,
+    direct_sink: Option<Arc<SpinSink>>,

    tunnel: T,
    stream: Option<Pin<Box<dyn ZCPacketStream>>>,

-    task: AbortOnDropHandle<()>,
+    task: Option<AbortOnDropHandle<()>>,
 }

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

        let task = tokio::spawn(async move {
@@ -60,13 +156,24 @@ impl<T: Tunnel> MpscTunnel<T> {

        Self {
            tx: Some(tx),
+            direct_sink: None,
            tunnel,
            stream: Some(stream),
-            task: AbortOnDropHandle::new(task),
+            task: Some(AbortOnDropHandle::new(task)),
+        }
+    }
+
+    pub fn new_direct(tunnel: T) -> Self {
+        let (stream, sink) = tunnel.split();
+        Self {
+            tx: None,
+            direct_sink: Some(Arc::new(SpinSink::new(sink))),
+            tunnel,
+            stream: Some(stream),
+            task: None,
        }
    }

-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
    async fn forward_one_round(
        rx: &mut Receiver<ZCPacket>,
        sink: &mut Pin<Box<dyn ZCPacketSink>>,
@@ -80,7 +187,6 @@ 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>>,
@@ -98,7 +204,6 @@ 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>>,
@@ -127,12 +232,18 @@ impl<T: Tunnel> MpscTunnel<T> {
    }

    pub fn get_sink(&self) -> MpscTunnelSender {
-        MpscTunnelSender(self.tx.as_ref().unwrap().clone())
+        MpscTunnelSender {
+            channel_tx: self.tx.as_ref().cloned(),
+            direct_sink: self.direct_sink.clone(),
+        }
    }

    pub fn close(&mut self) {
        self.tx.take();
-        self.task.abort();
+        self.direct_sink.take();
+        if let Some(task) = self.task.take() {
+            task.abort();
+        }
    }

    pub fn tunnel_info(&self) -> Option<TunnelInfo> {
@@ -192,7 +303,7 @@ mod tests {
            for i in 0..1000000 {
                tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
                let a = sink1
-                    .send(ZCPacket::new_with_payload("hello".as_bytes()))
+                    .send_async(ZCPacket::new_with_payload("hello".as_bytes()))
                    .await;
                if a.is_err() {
                    tracing::info!(?a, "t2 exit with err");
@@ -212,7 +323,7 @@ mod tests {
            for i in 0..1000000 {
                tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                let a = sink2
-                    .send(ZCPacket::new_with_payload("hello2".as_bytes()))
+                    .send_async(ZCPacket::new_with_payload("hello2".as_bytes()))
                    .await;
                if a.is_err() {
                    tracing::info!(?a, "t3 exit with err");
@@ -11,7 +11,7 @@ use async_trait::async_trait;
 use futures::{Sink, SinkExt, Stream, StreamExt};
 use once_cell::sync::Lazy;

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

 use uuid::Uuid;

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

 impl RingTunnelListener {
    pub fn new(key: url::Url) -> Self {
-        let (conn_sender, conn_receiver) =
-            hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
+        let (conn_sender, conn_receiver) = unbounded_channel();
        RingTunnelListener {
            listener_addr: key,
            conn_sender,
@@ -15,7 +15,9 @@ use zerocopy::{AsBytes, FromBytes};

 use tokio::{
    net::UdpSocket,
-    sync::mpsc::{Receiver, Sender, UnboundedReceiver, UnboundedSender},
+    sync::mpsc::{
+        Receiver, Sender, UnboundedReceiver, UnboundedSender, channel, unbounded_channel,
+    },
    task::JoinSet,
 };
 use tokio_util::task::AbortOnDropHandle;
@@ -673,9 +675,8 @@ pub struct UdpTunnelListener {

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

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

        let ring_recv = RingStream::new(ring_for_send_udp.clone());
        let ring_sender = RingSink::new(ring_for_recv_udp.clone());
@@ -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,11 @@ use crossbeam::atomic::AtomicCell;
 use dashmap::DashMap;
 use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
 use rand::RngCore;
-use tokio::{net::UdpSocket, sync::Mutex, task::JoinSet};
+use tokio::{
+    net::UdpSocket,
+    sync::{Mutex, mpsc::unbounded_channel},
+    task::JoinSet,
+};

 const MAX_PACKET: usize = 2048;

@@ -470,7 +474,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) = unbounded_channel();
        WgTunnelListener {
            addr,
            config,
Author	SHA1	Message	Date
fanyang	3de3fabcbc	style: format noop waker mpsc tests	2026-06-30 21:26:33 +08:00
fanyang	cb35c17503	perf(mpsc): extend noop_waker sync send to TCP tunnels TCP tunnel uses FramedWriter (not RingSink), but start_send is still sync (writes to BufList in memory). poll_flush does actual TCP write syscall — noop_waker returns Ok for Pending (data stays in BufList, flushed on next send when BufList >= 64). Add TCP benchmark support via HOTPATH_TUNNEL=tcp. Note: TCP/UDP convergence requires netns in bench environment (connector multi-bind address behavior doesn't work for localhost without namespaces). All 210 peers tests pass. Ring tunnel benchmark: 234K -> 508K pps (+117%).	2026-06-30 21:26:33 +08:00
fanyang	6ee717ec08	perf(mpsc): extend noop_waker sync send to UDP tunnels UDP tunnel uses RingSink internally (same as ring tunnel). Extend direct mode to include UDP. Fix poll_flush Pending to return Ok. Add UDP benchmark support via HOTPATH_TUNNEL=udp env variable. All 208 peers tests pass. Netns tests unchanged (require root).	2026-06-30 21:26:33 +08:00
fanyang	c0757977ee	perf(mpsc): sync send via noop_waker — +90% pps (249K → 474K) The async fn Future state machine overhead (~1.9us) dominated MpscTunnelSender::send, while RingSink operations were only ~40ns. Breakthrough: make send() an async fn that completes synchronously on the first poll for the direct (ring tunnel) path. Uses futures::task::noop_waker() to construct a dummy Context, then calls Sink trait methods (poll_ready, start_send, poll_flush) directly. RingSink always returns Ready immediately, so the waker is never invoked and the async fn completes without yielding. Channel mode (TCP/UDP/WG tunnels) still uses async send_async() with proper backpressure. Ring tunnels detected via tunnel_info() type check in PeerConn. Results (4 threads, 1400B, 15s): pps: 249K → 474K (+90%) send_msg_by_ip: 3.53us → 1.67us (-53%) send_msg_internal: 2.40us → 502ns (-79%) MpscTunnelSender::send: 1.97us → 144ns (-93%) All 207 peers:: tests pass. Netns-requiring tests (three_node, credential) unchanged (require root).	2026-06-30 21:26:33 +08:00
fanyang	28dd0e1152	perf(mpsc): replace Mutex with custom SpinSink (AtomicBool spinlock) tokio::sync::Mutex and std::sync::Mutex both have !Send guards that cannot cross await points in multi_thread runtime. Replace with a custom SpinSink using AtomicBool CAS — the SpinGuard contains only a &SpinSink reference (SpinSink: Sync via unsafe impl), so it is Send. Benchmark: pps unchanged (~249K), MpscTunnelSender::send avg 1.97us. The bottleneck is confirmed to be async fn Future state machine overhead (~1.9us), not the lock mechanism. RingSink operations are only ~40ns (poll_ready 15ns + start_send 10ns + poll_flush 15ns). Further breakthrough requires either: - Sync send API (bypassing async entirely) - Concrete type instead of dyn ZCPacketSink (to call RingSink::try_send directly)	2026-06-30 21:26:33 +08:00
fanyang	340145ae5d	perf(mpsc): use try_lock + merged poll_fn for direct sink path Replace 3 await points (lock().await + feed().await + flush().await) with try_lock() (sync) + single poll_fn (merged poll_ready + start_send + poll_flush). parking_lot::Mutex cannot be used because MutexGuard is !Send (cannot cross await in multi_thread runtime). tokio::sync::Mutex try_lock() returns synchronously and MutexGuard is Send. Benchmark: pps 250K → 251K (+0.4%), MpscTunnelSender::send avg 2.07us → 1.98us (-90ns). Improvement is small because tokio async machinery overhead (Future state machine + poll) dominates over RingSink's actual 40ns operation cost.	2026-06-30 21:26:33 +08:00
fanyang	2d86787a55	perf(mpsc): add direct sink path bypassing channel for PeerConn MpscTunnelSender now supports two modes: - Channel mode (existing): try_send to tokio mpsc → receiver task → sink - Direct mode (new): MpscTunnelSender holds Arc<Mutex<sink>> directly, bypassing the channel + receiver task entirely PeerConn uses new_direct to skip the channel intermediary. Benchmark result: pps unchanged (~245K). The async fn overhead of Mutex::lock().await + SinkExt::feed().await + SinkExt::flush().await (~2us) is comparable to channel try_send (~2us). The bottleneck is the Sink trait's async poll machinery, not the channel itself. However, this change provides: - RingSink timing now fully visible (start_send 10ns, poll_ready 13ns, poll_flush 17ns = 40ns/pkt total) - Reduced architectural complexity (no receiver task for PeerConn) - Foundation for a sync fast path using RingSink::try_send directly	2026-06-30 21:26:33 +08:00
fanyang	9473990ca9	bench: add Criterion TX throughput benchmark	2026-06-30 19:47:39 +08:00
fanyang	13412895c5	refactor(core): remove hotpath profiling (#2394 ) hotpath adds noticeable overhead on hot paths, so remove the optional profiling integration while keeping direct quanta::Instant timing.	2026-06-30 09:50:27 +08:00