chore: Update hotpath (#2412 )

feat(easytier): initialize hotpath profiler via hotpath::main
Gate the binary entry point with hotpath::main when the feature is enabled, except when hotpath-alloc is combined with a custom global allocator (jemalloc/mimalloc), which is already rejected at compile time in lib.rs.
2026-10-30 08:05:42 +00:00 · 2026-07-07 19:45:41 +08:00 · 2026-07-02 21:19:28 +08:00 · 2026-07-02 21:19:28 +08:00 · 2026-07-02 21:19:28 +08:00 · 2026-07-01 12:34:46 +08:00
29 changed files with 2026 additions and 350 deletions
@@ -263,6 +263,12 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"

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

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

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

+[[package]]
+name = "dirs-sys-next"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d"
+dependencies = [
+ "libc",
+ "redox_users 0.4.5",
+ "winapi",
+]
+
 [[package]]
 name = "dispatch2"
 version = "0.3.1"
@@ -2365,6 +2398,7 @@ dependencies = [
 "hickory-resolver",
 "hickory-server",
 "hmac",
+ "hotpath",
 "http",
 "http_req",
 "humansize",
@@ -2707,6 +2741,12 @@ version = "1.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ef6b89e5b37196644d8796de5268852ff179b44e96276cf4290264843743bb7"

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

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

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

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

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

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

+[[package]]
+name = "tiny_http"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "389915df6413a2e74fb181895f933386023c71110878cd0825588928e64cdc82"
+dependencies = [
+ "ascii",
+ "chunked_transfer",
+ "httpdate",
+ "log",
+]
+
 [[package]]
 name = "tinystr"
 version = "0.7.6"
@@ -58,6 +58,8 @@ chrono = { version = "0.4.37", features = ["serde"] }
 guarden = "0.2"
 quanta = "0.12"

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

 itertools = "0.14.0"
@@ -410,11 +412,14 @@ tracing = ["tokio/tracing", "dep:console-subscriber"]
 magic-dns = ["dep:hickory-client", "dep:hickory-server"]
 faketcp = ["dep:flume"]
 zstd = ["dep:zstd"]
-# Deprecated: hotpath profiling has been removed. These feature aliases are
-# retained as no-ops so existing build scripts using `--features hotpath*`
-# continue to work without pulling in any dependencies.
-hotpath = []
-hotpath-cpu = ["hotpath"]
-hotpath-alloc = ["hotpath"]
+hotpath = [
+    "dep:hotpath",
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/parking_lot",
+    "hotpath/flume",
+]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
 # For Network Extension on macOS
 macos-ne = []
@@ -219,6 +219,7 @@ pub struct GlobalCtx {

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

    flags: ArcSwap<Flags>,

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

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

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

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

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

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

 #[async_trait]
@@ -24,6 +24,16 @@ pub static malloc_conf: &[u8] = b"retain:false\0";
 rust_i18n::i18n!("locales", fallback = "en");

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

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

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

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

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

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

-            let provider_reconcile_should_run =
-                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);
+            if should_run_public_ipv6_provider_reconcile(&global_ctx) {
+                ensure_public_ipv6_provider_reconcile_task(
+                    &global_ctx,
+                    &self.public_ipv6_provider_task,
+                )
+                .await;
            }
        }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

 use base64::Engine as _;
@@ -363,15 +366,7 @@ 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 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 mut mpsc_tunnel = MpscTunnel::new(peer_conn_tunnel, Some(Duration::from_secs(7)));

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

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

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

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

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

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

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

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

            peers,

            peer_rpc_mgr,
            peer_rpc_tspt: rpc_tspt,

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

            route_algo_inst,

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

-        run_ip_in_ns(
-            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",
-            ],
-        );
+        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::SERVER_NS,
@@ -606,19 +643,21 @@ impl PublicIpv6Lab {
                "srv0",
            ],
        );
-        run_ip_in_ns(
-            Self::UPSTREAM_NS,
-            &[
-                "-6",
-                "route",
-                "add",
-                Self::PROVIDER_PREFIX,
-                "via",
-                "2001:db8:ffff:1::2",
-                "dev",
-                "upwan0",
-            ],
-        );
+        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_sysctl_in_ns(Self::PROVIDER_NS, "net.ipv6.conf.all.forwarding=1");
        run_sysctl_in_ns(Self::UPSTREAM_NS, "net.ipv6.conf.all.forwarding=1");
@@ -672,7 +711,15 @@ fn get_public_ipv6_config(
 async fn init_public_ipv6_two_node(
    client_inst_id: uuid::Uuid,
 ) -> (PublicIpv6Lab, Instance, Instance) {
-    let lab = PublicIpv6Lab::setup();
+    init_public_ipv6_two_node_with_topology(client_inst_id, PublicIpv6LabTopology::DelegatedPrefix)
+        .await
+}
+
+async fn init_public_ipv6_two_node_with_topology(
+    client_inst_id: uuid::Uuid,
+    topology: PublicIpv6LabTopology,
+) -> (PublicIpv6Lab, Instance, Instance) {
+    let lab = PublicIpv6Lab::setup_with_topology(topology);

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

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

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

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

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

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

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

        loop {
            let mut buf = BytesMut::new();
@@ -1,13 +1,6 @@
 // this mod wrap tunnel to a mpsc tunnel, based on crossbeam_channel

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

 use anyhow::Context;
 use tokio::time::timeout;
@@ -16,130 +9,43 @@ use crate::proto::common::TunnelInfo;

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

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

 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 {
-    channel_tx: Option<Sender<ZCPacket>>,
-    direct_sink: Option<Arc<SpinSink>>,
-}
+pub struct MpscTunnelSender(Sender<ZCPacket>);

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

    pub fn try_send(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
-        tx.try_send(item).map_err(|e| match e {
+        self.0.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: Option<AbortOnDropHandle<()>>,
+    task: AbortOnDropHandle<()>,
 }

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

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

        Self {
            tx: Some(tx),
-            direct_sink: None,
            tunnel,
            stream: Some(stream),
-            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,
+            task: AbortOnDropHandle::new(task),
        }
    }

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

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

+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
    async fn forward_one_round_with_timeout(
        rx: &mut Receiver<ZCPacket>,
        sink: &mut Pin<Box<dyn ZCPacketSink>>,
@@ -232,18 +129,12 @@ impl<T: Tunnel> MpscTunnel<T> {
    }

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

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

    pub fn tunnel_info(&self) -> Option<TunnelInfo> {
@@ -303,7 +194,7 @@ mod tests {
            for i in 0..1000000 {
                tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
                let a = sink1
-                    .send_async(ZCPacket::new_with_payload("hello".as_bytes()))
+                    .send(ZCPacket::new_with_payload("hello".as_bytes()))
                    .await;
                if a.is_err() {
                    tracing::info!(?a, "t2 exit with err");
@@ -323,7 +214,7 @@ mod tests {
            for i in 0..1000000 {
                tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                let a = sink2
-                    .send_async(ZCPacket::new_with_payload("hello2".as_bytes()))
+                    .send(ZCPacket::new_with_payload("hello2".as_bytes()))
                    .await;
                if a.is_err() {
                    tracing::info!(?a, "t3 exit with err");
@@ -12,6 +12,9 @@ use crate::tunnel::{
 use anyhow::Context;
 use derivative::Derivative;
 use derive_more::{Deref, DerefMut};
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::parking_lot::RwLock;
+#[cfg(not(feature = "hotpath"))]
 use parking_lot::RwLock;
 use quinn::{
    ClientConfig, ConnectError, Connection, Endpoint, EndpointConfig, ServerConfig,
@@ -312,18 +315,25 @@ struct RwPoolInner<Item> {
    enabled: bool,
 }

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

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

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

 use uuid::Uuid;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

 impl WgTunnelListener {
    pub fn new(addr: url::Url, config: WgConfig) -> Self {
-        let (conn_send, conn_recv) = unbounded_channel();
+        let (conn_send, conn_recv) = hotpath::channel!(unbounded_channel());
        WgTunnelListener {
            addr,
            config,
Author	SHA1	Message	Date
Paweł Urbanek	cb04cefb2a	chore: Update hotpath (#2412 )	2026-07-07 19:45:41 +08:00
fanyang89	20873fc62e	feat(easytier): initialize hotpath profiler via hotpath::main Gate the binary entry point with hotpath::main when the feature is enabled, except when hotpath-alloc is combined with a custom global allocator (jemalloc/mimalloc), which is already rejected at compile time in lib.rs.	2026-07-02 21:19:28 +08:00
fanyang89	27f6b90fa1	feat(easytier): instrument hot-path locks, channels, and functions Wrap the per-packet locks and channels behind hotpath's drop-in wrappers () so lock contention and channel flow become visible when the feature is on, while staying zero-cost in default builds via cfg-gated dual imports and the no-op // macros. Annotate the hottest send/recv, encrypt/decrypt, and forward functions with . Coverage: peer_conn/peer_manager/peer_map/peer/peer_session/secure_datagram locks, mpsc/ring/udp/wireguard/fake_tcp channels, quic connection pool, relay/foreign send paths, and OSPF route lookup (function-level only; its parking_lot upgradable guards have no hotpath wrapper). Debug impls that formatted lock fields are updated to dereference the inner value, and quic's RwPool switches to a manual Debug that skips the locks.	2026-07-02 21:19:28 +08:00
fanyang89	a602125d97	chore(easytier): gate hotpath-rs 0.19 dependency and off-mode shim Re-enable the hotpath profiler as an optional dependency gated behind the feature (default off). In off builds the crate stays out of the dependency graph entirely: plus a local module provide no-op // macros so call sites compile unchanged. Also add a guarding the mutual exclusivity of with the / global allocators.	2026-07-02 21:19:28 +08:00
ririyeye Claude KKRainbow	7756a15cbe	feat: add interface address fallback for IPv6 prefix detection (DHCPv6 IA_NA / SLAAC) (#2334 ) feat: add IPv6 prefix fallback and NDP proxy for SLAAC/IA_NA Add interface address fallback for IPv6 prefix detection when route-based detection fails (no delegated prefix on LAN). Covers DHCPv6 IA_NA / SLAAC where prefix is assigned to WAN without PD. Uses getifaddrs() and Ipv6Inet to handle host bits. Add NDP proxy sync for on-link prefixes. ISP router uses NDP to resolve MACs; EasyTier /128 addresses on tun0 won't answer on physical WAN. Periodic sync (30s) reads /128 routes via netlink and manages proxy neigh entries. Auto-enables proxy_ndp sysctl. Returns WAN ifindex from detection; static NDP_WAN_IFINDEX for sync task. Spawns from provider reconcile when auto-detection enabled. Add tests: fallback default route interface, on-link NDP proxy, interface prefix selection. Fixes #2333 Co-authored-by: ririyeye <200610237@qq.com> Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: KKRainbow <443152178@qq.com>	2026-07-01 12:34:46 +08:00
fanyang	741460e1e4	bench: add Criterion TX throughput benchmark (#2397 )	2026-06-30 22:10:34 +08:00