style: format noop waker mpsc tests

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

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

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

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"
@@ -1139,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"
@@ -1248,6 +1260,33 @@ dependencies = [
  "windows-targets 0.52.6",
 ]
 
+[[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"
@@ -1593,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"
@@ -2266,6 +2341,7 @@ dependencies = [
  "clap_complete",
  "clap_complete_nushell",
  "console-subscriber",
+ "criterion",
  "crossbeam",
  "ctor 0.8.0",
  "dashmap",
@@ -4496,6 +4572,17 @@ dependencies = [
  "once_cell",
 ]
 
+[[package]]
+name = "is-terminal"
+version = "0.4.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
+dependencies = [
+ "hermit-abi",
+ "libc",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "is-wsl"
 version = "0.4.0"
@@ -5856,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"
@@ -6486,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"
@@ -9864,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"

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"
 
@@ -336,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"

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/peers/peer_conn.rs

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

easytier/src/tunnel/mpsc.rs

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