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-20 03:22:05 +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
29 changed files with 350 additions and 2026 deletions
@@ -263,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"
@@ -1266,12 +1260,6 @@ dependencies = [
 "windows-targets 0.52.6",
 ]

-[[package]]
-name = "chunked_transfer"
-version = "1.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
-
 [[package]]
 name = "ciborium"
 version = "0.2.2"
@@ -2175,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"
@@ -2208,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"
@@ -2398,7 +2365,6 @@ dependencies = [
 "hickory-resolver",
 "hickory-server",
 "hmac",
- "hotpath",
 "http",
 "http_req",
 "humansize",
@@ -2741,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"
@@ -4005,61 +3965,6 @@ dependencies = [
 "windows-sys 0.52.0",
 ]

-[[package]]
-name = "hotpath"
-version = "0.21.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ff6b552a6afa29d9e33f8d555bee9093c142dd449501ae128e6494a303f03dc"
-dependencies = [
- "arc-swap",
- "cfg-if",
- "crossbeam-channel",
- "flate2",
- "flume 0.12.0",
- "futures-util",
- "hdrhistogram",
- "hotpath-macros",
- "hotpath-meta",
- "libc",
- "object",
- "parking_lot",
- "pin-project-lite",
- "prettytable-rs",
- "quanta",
- "regex",
- "rustc-demangle",
- "serde",
- "serde_json",
- "tiny_http",
- "tokio",
-]
-
-[[package]]
-name = "hotpath-macros"
-version = "0.21.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4f15322569d3cfadf84c0de7ef72be435b8f4b4839ee4ace78a7eaca48a87ded"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.117",
-]
-
-[[package]]
-name = "hotpath-macros-meta"
-version = "0.21.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b3675e29d16c844ccad12763672b33e51d9a000c346720c4f354f7a3bdc649a8"
-
-[[package]]
-name = "hotpath-meta"
-version = "0.21.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d558d972ddc9483fb4e713af3dd41634edf895c7cb18ed13dc58c56431d42e27"
-dependencies = [
- "hotpath-macros-meta",
-]
-
 [[package]]
 name = "html5ever"
 version = "0.29.1"
@@ -6028,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"
@@ -6941,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"
@@ -8013,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"
@@ -9943,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"
@@ -10109,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"
@@ -58,8 +58,6 @@ chrono = { version = "0.4.37", features = ["serde"] }
 guarden = "0.2"
 quanta = "0.12"

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

 itertools = "0.14.0"
@@ -412,14 +410,11 @@ tracing = ["tokio/tracing", "dep:console-subscriber"]
 magic-dns = ["dep:hickory-client", "dep:hickory-server"]
 faketcp = ["dep:flume"]
 zstd = ["dep:zstd"]
-hotpath = [
-    "dep:hotpath",
-    "hotpath/hotpath",
-    "hotpath/tokio",
-    "hotpath/parking_lot",
-    "hotpath/flume",
-]
-hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
-hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+# 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 = []
@@ -219,7 +219,6 @@ pub struct GlobalCtx {

    running_listeners: Mutex<Vec<url::Url>>,
    advertised_ipv6_public_addr_prefix: Mutex<Option<cidr::Ipv6Cidr>>,
-    tun_device_name: Mutex<Option<String>>,

    flags: ArcSwap<Flags>,

@@ -337,7 +336,6 @@ impl GlobalCtx {

            running_listeners: Mutex::new(Vec::new()),
            advertised_ipv6_public_addr_prefix: Mutex::new(None),
-            tun_device_name: Mutex::new(None),

            flags: ArcSwap::new(Arc::new(flags)),

@@ -372,24 +370,6 @@ impl GlobalCtx {
        }
    }

-    fn set_tun_device_name(&self, name: Option<String>) {
-        *self.tun_device_name.lock().unwrap() = name;
-    }
-
-    pub(crate) fn set_tun_device_ready(&self, name: String) {
-        self.set_tun_device_name(Some(name.clone()));
-        self.issue_event(GlobalCtxEvent::TunDeviceReady(name));
-    }
-
-    pub(crate) fn set_tun_device_error(&self, error: String) {
-        self.set_tun_device_name(None);
-        self.issue_event(GlobalCtxEvent::TunDeviceError(error));
-    }
-
-    pub fn get_tun_device_name(&self) -> Option<String> {
-        self.tun_device_name.lock().unwrap().clone()
-    }
-
    pub fn check_network_in_whitelist(&self, network_name: &str) -> Result<(), anyhow::Error> {
        if self
            .get_flags()
@@ -845,36 +825,6 @@ pub mod tests {
        );
    }

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

 #[async_trait]
@@ -24,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
 }
@@ -1,52 +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
-    };
-}
-
-/// Type-level mirror of `hotpath::wrap` for type positions: with the feature
-/// off, `channel!` returns the original endpoints, so the wrapped endpoint
-/// types are the plain channel types.
-pub(crate) mod wrap {
-    pub(crate) mod tokio {
-        pub(crate) mod sync {
-            pub(crate) use ::tokio::sync::mpsc;
-        }
-    }
-    pub(crate) use ::flume;
-}
@@ -65,9 +65,9 @@ use crate::vpn_portal::{self, VpnPortal};
 use super::dns_server::{MAGIC_DNS_FAKE_IP, runner::DnsRunner};
 use super::listeners::ListenerManager;
 use super::public_ipv6_provider::{
-    PublicIpv6ProviderReconcileTask, reconcile_public_ipv6_provider_runtime,
-    run_public_ipv6_provider_reconcile_task, should_run_public_ipv6_provider_reconcile,
-    validate_public_ipv6_config, validate_public_ipv6_config_values,
+    reconcile_public_ipv6_provider_runtime, run_public_ipv6_provider_reconcile_task,
+    should_run_public_ipv6_provider_reconcile, validate_public_ipv6_config,
+    validate_public_ipv6_config_values,
 };

 #[cfg(feature = "socks5")]
@@ -194,44 +194,6 @@ impl NicCtxContainer {

 #[cfg(feature = "tun")]
 type ArcNicCtx = Arc<Mutex<Option<NicCtxContainer>>>;
-type ArcPublicIpv6ProviderTaskSlot = Arc<PublicIpv6ProviderTaskSlot>;
-
-struct PublicIpv6ProviderTaskSlot {
-    task: Mutex<Option<PublicIpv6ProviderReconcileTask>>,
-    closing: AtomicBool,
-}
-
-impl PublicIpv6ProviderTaskSlot {
-    fn new() -> Self {
-        Self {
-            task: Mutex::new(None),
-            closing: AtomicBool::new(false),
-        }
-    }
-
-    async fn ensure_started(&self, global_ctx: &ArcGlobalCtx) {
-        let mut task = self.task.lock().await;
-        if self.closing.load(Ordering::Acquire) || task.is_some() {
-            return;
-        }
-        *task = run_public_ipv6_provider_reconcile_task(global_ctx);
-    }
-
-    async fn shutdown(&self) {
-        self.closing.store(true, Ordering::Release);
-        let task = self.task.lock().await.take();
-        if let Some(task) = task {
-            task.shutdown().await;
-        }
-    }
-}
-
-async fn ensure_public_ipv6_provider_reconcile_task(
-    global_ctx: &ArcGlobalCtx,
-    task_slot: &ArcPublicIpv6ProviderTaskSlot,
-) {
-    task_slot.ensure_started(global_ctx).await;
-}

 pub struct InstanceRpcServerHook {
    rpc_portal_whitelist: Vec<IpCidr>,
@@ -292,7 +254,6 @@ pub struct InstanceConfigPatcher {
    socks5_server: Weak<Socks5Server>,
    peer_manager: Weak<PeerManager>,
    conn_manager: Weak<ManualConnectorManager>,
-    public_ipv6_provider_task: ArcPublicIpv6ProviderTaskSlot,
 }

 impl InstanceConfigPatcher {
@@ -363,6 +324,7 @@ impl InstanceConfigPatcher {
        self.patch_mapped_listeners(patch.mapped_listeners).await?;
        self.patch_connector(patch.connectors).await?;

+        let provider_reconcile_was_running = should_run_public_ipv6_provider_reconcile(&global_ctx);
        let mut provider_config_changed = false;
        if let Some(hostname) = patch.hostname {
            global_ctx.set_hostname(hostname.clone());
@@ -400,12 +362,10 @@ impl InstanceConfigPatcher {
        if provider_config_changed {
            reconcile_public_ipv6_provider_runtime(&global_ctx).await;

-            if should_run_public_ipv6_provider_reconcile(&global_ctx) {
-                ensure_public_ipv6_provider_reconcile_task(
-                    &global_ctx,
-                    &self.public_ipv6_provider_task,
-                )
-                .await;
+            let provider_reconcile_should_run =
+                should_run_public_ipv6_provider_reconcile(&global_ctx);
+            if !provider_reconcile_was_running && provider_reconcile_should_run {
+                run_public_ipv6_provider_reconcile_task(&global_ctx);
            }
        }

@@ -687,7 +647,6 @@ pub struct Instance {
    socks5_server: Arc<Socks5Server>,

    proxy_cidrs_monitor: Option<AbortOnDropHandle<()>>,
-    public_ipv6_provider_task: ArcPublicIpv6ProviderTaskSlot,

    global_ctx: ArcGlobalCtx,
 }
@@ -775,7 +734,6 @@ impl Instance {
            socks5_server,

            proxy_cidrs_monitor: None,
-            public_ipv6_provider_task: Arc::new(PublicIpv6ProviderTaskSlot::new()),

            global_ctx,
        }
@@ -1076,11 +1034,7 @@ impl Instance {
            .await?;
        self.listener_manager.lock().await.run().await?;
        self.peer_manager.run().await?;
-        ensure_public_ipv6_provider_reconcile_task(
-            &self.global_ctx,
-            &self.public_ipv6_provider_task,
-        )
-        .await;
+        run_public_ipv6_provider_reconcile_task(&self.global_ctx);

        #[cfg(feature = "tun")]
        {
@@ -1393,7 +1347,6 @@ impl Instance {
            socks5_server: Arc::downgrade(&self.socks5_server),
            peer_manager: Arc::downgrade(&self.peer_manager),
            conn_manager: Arc::downgrade(&self.conn_manager),
-            public_ipv6_provider_task: self.public_ipv6_provider_task.clone(),
        }
    }

@@ -1649,7 +1602,6 @@ impl Instance {
    }

    pub async fn clear_resources(&mut self) {
-        self.public_ipv6_provider_task.shutdown().await;
        self.peer_manager.clear_resources().await;
        #[cfg(feature = "tun")]
        let _ = self.nic_ctx.lock().await.take();
@@ -1835,21 +1787,6 @@ mod tests {
        );
    }

-    #[tokio::test]
-    async fn public_ipv6_provider_task_slot_does_not_restart_after_shutdown() {
-        let global_ctx = get_mock_global_ctx();
-        let slot = std::sync::Arc::new(super::PublicIpv6ProviderTaskSlot::new());
-        global_ctx.config.set_ipv6_public_addr_provider(true);
-        global_ctx
-            .config
-            .set_ipv6_public_addr_prefix(Some("2001:db8::/48".parse().unwrap()));
-
-        slot.shutdown().await;
-        super::ensure_public_ipv6_provider_reconcile_task(&global_ctx, &slot).await;
-
-        assert!(slot.task.lock().await.is_none());
-    }
-
    #[tokio::test]
    async fn validate_public_ipv6_patch_allows_enabling_auto_with_manual_ipv6() {
        let global_ctx = get_mock_global_ctx();
@@ -1361,11 +1361,12 @@ impl NicCtx {
                    }

                    self.global_ctx
-                        .set_tun_device_ready(nic.ifname().to_string());
+                        .issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
                    ret
                }
                Err(err) => {
-                    self.global_ctx.set_tun_device_error(err.to_string());
+                    self.global_ctx
+                        .issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
                    return Err(err);
                }
            }
@@ -1404,11 +1405,12 @@ impl NicCtx {
            match nic.create_dev_for_mobile(tun_fd).await {
                Ok(ret) => {
                    self.global_ctx
-                        .set_tun_device_ready(nic.ifname().to_string());
+                        .issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
                    ret
                }
                Err(err) => {
-                    self.global_ctx.set_tun_device_error(err.to_string());
+                    self.global_ctx
+                        .issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
                    return Err(err);
                }
            }
@@ -173,12 +173,7 @@ impl EasyTierLauncher {
        #[cfg(mobile)]
        Self::run_routine_for_mobile(&instance, &data, &mut tasks).await;

-        if let Err(err) = instance.run().await {
-            tasks.abort_all();
-            drop(tasks);
-            instance.clear_resources().await;
-            return Err(err.into());
-        }
+        instance.run().await?;

        #[cfg(feature = "ffi-dataplane")]
        data.data_plane
@@ -5,22 +5,6 @@ use std::io;
 use clap::Command;
 use clap_complete::{Generator, Shell};

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

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

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

        let conns_copy = conns.clone();
@@ -210,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(
+            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?)
    }
@@ -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()),
@@ -963,7 +956,6 @@ impl PeerManager {
        Self::is_relay_data_packet(hdr.packet_type)
    }

-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
    async fn start_peer_recv(&self) {
        let mut recv = self.packet_recv.lock().await.take().unwrap();
        let my_peer_id = self.my_peer_id;
@@ -1445,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(
@@ -1531,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>,
@@ -1698,7 +1688,6 @@ impl PeerManager {
        (dst_peers, is_exit_node)
    }

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

 use crate::{
@@ -48,7 +45,7 @@ impl PeerMap {
            my_peer_id,
            peer_map: DashMap::new(),
            packet_send,
-            routes: hotpath::rw_lock!(tokio::sync::RwLock::new(Vec::new())),
+            routes: RwLock::new(Vec::new()),
            alive_client_urls: Arc::new(Mutex::new(multimap::MultiMap::new())),
        }
    }
@@ -135,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();
@@ -167,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,
@@ -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| {
@@ -1,11 +1,7 @@
 use std::sync::{
-    Arc,
+    Arc, RwLock,
    atomic::{AtomicBool, Ordering},
 };
-#[cfg(feature = "hotpath")]
-use hotpath::wrap::std::sync::RwLock;
-#[cfg(not(feature = "hotpath"))]
-use std::sync::RwLock;
 use std::time::Duration;

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

-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerSession"))]
    pub fn encrypt_payload(
        &self,
        sender_peer_id: PeerId,
@@ -394,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,
@@ -144,7 +144,6 @@ impl RelayPeerMap {
        Ok(())
    }

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

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

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

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

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

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

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

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

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

        let plaintext = b"pkt2";
        let mut pkt2 = ZCPacket::new_with_payload(plaintext);
@@ -49,7 +49,7 @@ async fn test_route_peer_info_ipv6() {
 #[tokio::test]
 async fn test_peer_manager_ipv6() {
    let global_ctx = get_mock_global_ctx();
-    let (packet_sender, _packet_receiver) = crate::peers::create_packet_recv_chan();
+    let (packet_sender, _packet_receiver) = tokio::sync::mpsc::channel(100);
    let peer_mgr = crate::peers::peer_manager::PeerManager::new(
        RouteAlgoType::Ospf,
        global_ctx.clone(),
@@ -477,12 +477,6 @@ struct PublicIpv6Lab {
    extra_bridges: [&'static str; 2],
 }

-#[derive(Clone, Copy)]
-enum PublicIpv6LabTopology {
-    DelegatedPrefix,
-    OnLinkPrefix,
-}
-
 impl PublicIpv6Lab {
    const PROVIDER_NS: &'static str = "net_a";
    const CLIENT_NS: &'static str = "net_b";
@@ -496,13 +490,11 @@ impl PublicIpv6Lab {
    const PROVIDER_DEFAULT_FROM: &'static str = "2001:db8:100::/64";
    const PROVIDER_WAN_ADDR: &'static str = "2001:db8:ffff:1::2/64";
    const UPSTREAM_WAN_ADDR: &'static str = "2001:db8:ffff:1::1/64";
-    const ON_LINK_PROVIDER_WAN_ADDR: &'static str = "2001:db8:100::2/64";
-    const ON_LINK_UPSTREAM_WAN_ADDR: &'static str = "2001:db8:100::1/64";
    const UPSTREAM_SERVER_ADDR: &'static str = "2001:db8:ffff:2::1/64";
    const SERVER_ADDR: &'static str = "2001:db8:ffff:2::100/64";
    const SERVER_IP: &'static str = "2001:db8:ffff:2::100";

-    fn setup_with_topology(topology: PublicIpv6LabTopology) -> Self {
+    fn setup() -> Self {
        prepare_linux_namespaces();

        del_netns(Self::UPSTREAM_NS);
@@ -552,23 +544,13 @@ impl PublicIpv6Lab {
            Self::SERVER_BRIDGE,
        );

-        let (provider_wan_addr, upstream_wan_addr) = match topology {
-            PublicIpv6LabTopology::DelegatedPrefix => {
-                (Self::PROVIDER_WAN_ADDR, Self::UPSTREAM_WAN_ADDR)
-            }
-            PublicIpv6LabTopology::OnLinkPrefix => (
-                Self::ON_LINK_PROVIDER_WAN_ADDR,
-                Self::ON_LINK_UPSTREAM_WAN_ADDR,
-            ),
-        };
-
        run_ip_in_ns(
            Self::PROVIDER_NS,
-            &["addr", "add", provider_wan_addr, "dev", "pubwan0"],
+            &["addr", "add", Self::PROVIDER_WAN_ADDR, "dev", "pubwan0"],
        );
        run_ip_in_ns(
            Self::UPSTREAM_NS,
-            &["addr", "add", upstream_wan_addr, "dev", "upwan0"],
+            &["addr", "add", Self::UPSTREAM_WAN_ADDR, "dev", "upwan0"],
        );
        run_ip_in_ns(
            Self::UPSTREAM_NS,
@@ -579,56 +561,37 @@ impl PublicIpv6Lab {
            &["addr", "add", Self::SERVER_ADDR, "dev", "srv0"],
        );

-        match topology {
-            PublicIpv6LabTopology::DelegatedPrefix => {
-                run_ip_in_ns(
-                    Self::PROVIDER_NS,
-                    &["link", "add", "pubprefix0", "type", "dummy"],
-                );
-                run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
-                run_ip_in_ns(
-                    Self::PROVIDER_NS,
-                    &[
-                        "-6",
-                        "route",
-                        "add",
-                        Self::PROVIDER_PREFIX,
-                        "dev",
-                        "pubprefix0",
-                    ],
-                );
-                run_ip_in_ns(
-                    Self::PROVIDER_NS,
-                    &[
-                        "-6",
-                        "route",
-                        "add",
-                        "default",
-                        "from",
-                        Self::PROVIDER_DEFAULT_FROM,
-                        "via",
-                        "2001:db8:ffff:1::1",
-                        "dev",
-                        "pubwan0",
-                    ],
-                );
-            }
-            PublicIpv6LabTopology::OnLinkPrefix => {
-                run_ip_in_ns(
-                    Self::PROVIDER_NS,
-                    &[
-                        "-6",
-                        "route",
-                        "add",
-                        "default",
-                        "via",
-                        "2001:db8:100::1",
-                        "dev",
-                        "pubwan0",
-                    ],
-                );
-            }
-        }
+        run_ip_in_ns(
+            Self::PROVIDER_NS,
+            &["link", "add", "pubprefix0", "type", "dummy"],
+        );
+        run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
+        run_ip_in_ns(
+            Self::PROVIDER_NS,
+            &[
+                "-6",
+                "route",
+                "add",
+                Self::PROVIDER_PREFIX,
+                "dev",
+                "pubprefix0",
+            ],
+        );
+        run_ip_in_ns(
+            Self::PROVIDER_NS,
+            &[
+                "-6",
+                "route",
+                "add",
+                "default",
+                "from",
+                Self::PROVIDER_DEFAULT_FROM,
+                "via",
+                "2001:db8:ffff:1::1",
+                "dev",
+                "pubwan0",
+            ],
+        );

        run_ip_in_ns(
            Self::SERVER_NS,
@@ -643,21 +606,19 @@ impl PublicIpv6Lab {
                "srv0",
            ],
        );
-        if matches!(topology, PublicIpv6LabTopology::DelegatedPrefix) {
-            run_ip_in_ns(
-                Self::UPSTREAM_NS,
-                &[
-                    "-6",
-                    "route",
-                    "add",
-                    Self::PROVIDER_PREFIX,
-                    "via",
-                    "2001:db8:ffff:1::2",
-                    "dev",
-                    "upwan0",
-                ],
-            );
-        }
+        run_ip_in_ns(
+            Self::UPSTREAM_NS,
+            &[
+                "-6",
+                "route",
+                "add",
+                Self::PROVIDER_PREFIX,
+                "via",
+                "2001:db8:ffff:1::2",
+                "dev",
+                "upwan0",
+            ],
+        );

        run_sysctl_in_ns(Self::PROVIDER_NS, "net.ipv6.conf.all.forwarding=1");
        run_sysctl_in_ns(Self::UPSTREAM_NS, "net.ipv6.conf.all.forwarding=1");
@@ -711,15 +672,7 @@ fn get_public_ipv6_config(
 async fn init_public_ipv6_two_node(
    client_inst_id: uuid::Uuid,
 ) -> (PublicIpv6Lab, Instance, Instance) {
-    init_public_ipv6_two_node_with_topology(client_inst_id, PublicIpv6LabTopology::DelegatedPrefix)
-        .await
-}
-
-async fn init_public_ipv6_two_node_with_topology(
-    client_inst_id: uuid::Uuid,
-    topology: PublicIpv6LabTopology,
-) -> (PublicIpv6Lab, Instance, Instance) {
-    let lab = PublicIpv6Lab::setup_with_topology(topology);
+    let lab = PublicIpv6Lab::setup();

    let provider_cfg = get_public_ipv6_config(
        "provider_public_ipv6",
@@ -803,13 +756,6 @@ fn addr_exists_in_ns(ns: &str, dev: &str, needle: &str) -> bool {
    run_ip_in_ns_output(ns, &["-6", "addr", "show", "dev", dev]).contains(needle)
 }

-fn ndp_proxy_exists_in_ns(ns: &str, dev: &str, addr: std::net::Ipv6Addr) -> bool {
-    let addr = addr.to_string();
-    run_ip_in_ns_output(ns, &["-6", "neigh", "show", "proxy", "dev", dev])
-        .lines()
-        .any(|line| line.split_whitespace().next() == Some(addr.as_str()))
-}
-
 #[tokio::test]
 #[serial_test::serial]
 pub async fn public_ipv6_auto_addr_end_to_end() {
@@ -932,67 +878,6 @@ pub async fn public_ipv6_auto_addr_end_to_end() {
    drop_insts(vec![provider, client]).await;
 }

-#[tokio::test]
-#[serial_test::serial]
-pub async fn public_ipv6_auto_addr_on_link_ndp_proxy_end_to_end() {
-    let client_id = uuid::Uuid::parse_str("44444444-4444-4444-4444-444444444444").unwrap();
-    let (_lab, provider, client) =
-        init_public_ipv6_two_node_with_topology(client_id, PublicIpv6LabTopology::OnLinkPrefix)
-            .await;
-
-    wait_for_condition(
-        || async {
-            provider
-                .get_global_ctx()
-                .get_advertised_ipv6_public_addr_prefix()
-                == Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap())
-        },
-        Duration::from_secs(10),
-    )
-    .await;
-
-    let leased = wait_for_public_ipv6_addr(&client).await;
-    wait_for_public_ipv6_route(&provider, leased).await;
-
-    wait_for_condition(
-        || async {
-            addr_exists_in_ns(
-                PublicIpv6Lab::CLIENT_NS,
-                PublicIpv6Lab::CLIENT_TUN,
-                &leased.to_string(),
-            ) && route_exists_in_ns(
-                PublicIpv6Lab::PROVIDER_NS,
-                &format!("{} dev {}", leased.address(), PublicIpv6Lab::PROVIDER_TUN),
-            )
-        },
-        Duration::from_secs(10),
-    )
-    .await;
-
-    wait_for_condition(
-        || async {
-            ndp_proxy_exists_in_ns(PublicIpv6Lab::PROVIDER_NS, "pubwan0", leased.address())
-        },
-        Duration::from_secs(20),
-    )
-    .await;
-
-    wait_for_condition(
-        || async {
-            ping6_test(
-                PublicIpv6Lab::SERVER_NS,
-                leased.address().to_string().as_str(),
-                None,
-            )
-            .await
-        },
-        Duration::from_secs(20),
-    )
-    .await;
-
-    drop_insts(vec![provider, client]).await;
-}
-
 #[tokio::test]
 #[serial_test::serial]
 pub async fn public_ipv6_auto_addr_reconnect_reuses_same_address() {
@@ -48,13 +48,9 @@ use std::collections::{HashMap, HashSet};
 use std::fmt;
 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::sync::{
-    Arc,
+    Arc, RwLock,
    atomic::{AtomicU32, Ordering},
 };
-#[cfg(feature = "hotpath")]
-use hotpath::wrap::std::sync::RwLock;
-#[cfg(not(feature = "hotpath"))]
-use std::sync::RwLock;
 use tokio::sync::broadcast;
 use tokio::time;
 use tokio_util::task::AbortOnDropHandle;
@@ -89,7 +85,7 @@ impl AddrTuple {

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

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

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

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

        loop {
            let mut buf = BytesMut::new();
@@ -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;
@@ -9,43 +16,130 @@ use crate::proto::common::TunnelInfo;

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

-use hotpath::wrap::tokio::sync::mpsc::{Receiver, Sender};
-use tokio::sync::mpsc::{channel, error::TrySendError};
+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 {
-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnelSender"))]
    pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        self.0.send(item).await.with_context(|| "send error")?;
-        Ok(())
+        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 {
@@ -62,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>>,
@@ -82,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>>,
@@ -100,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>>,
@@ -129,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> {
@@ -194,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");
@@ -214,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");
@@ -12,9 +12,6 @@ use crate::tunnel::{
 use anyhow::Context;
 use derivative::Derivative;
 use derive_more::{Deref, DerefMut};
-#[cfg(feature = "hotpath")]
-use hotpath::wrap::parking_lot::RwLock;
-#[cfg(not(feature = "hotpath"))]
 use parking_lot::RwLock;
 use quinn::{
    ClientConfig, ConnectError, Connection, Endpoint, EndpointConfig, ServerConfig,
@@ -315,25 +312,18 @@ struct RwPoolInner<Item> {
    enabled: bool,
 }

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

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

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

 use uuid::Uuid;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

 impl WgTunnelListener {
    pub fn new(addr: url::Url, config: WgConfig) -> Self {
-        let (conn_send, conn_recv) = hotpath::channel!(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