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.

easytier/src/peers/mod.rs

@@ -59,8 +59,9 @@ type BoxNicPacketFilter = Box<dyn NicPacketFilter + Send + Sync>;
 pub type PacketRecvChan = tokio::sync::mpsc::Sender<ZCPacket>;
 pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver<ZCPacket>;
 pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
-    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> {

easytier/src/peers/peer.rs

@@ -2,6 +2,9 @@ use std::sync::Arc;
 
 use crossbeam::atomic::AtomicCell;
 use dashmap::{DashMap, DashSet};
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::parking_lot::RwLock;
+#[cfg(not(feature = "hotpath"))]
 use parking_lot::RwLock;
 
 use tokio::{select, sync::mpsc};
@@ -56,7 +59,7 @@ impl Peer {
         let shutdown_notifier = Arc::new(tokio::sync::Notify::new());
         let peer_identity_type = Arc::new(AtomicCell::new(None));
         let peer_identity_type_copy = peer_identity_type.clone();
-        let peer_public_key = Arc::new(RwLock::new(None));
+        let peer_public_key = Arc::new(hotpath::rw_lock!(parking_lot::RwLock::new(None)));
         let peer_public_key_copy = peer_public_key.clone();
 
         let conns_copy = conns.clone();
@@ -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));

easytier/src/peers/peer_conn.rs

@@ -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 _;
@@ -381,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(),
@@ -1463,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?)
     }

easytier/src/peers/peer_manager.rs

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

easytier/src/peers/peer_map.rs

@@ -6,6 +6,9 @@ use std::{
 use anyhow::Context;
 use dashmap::{DashMap, DashSet};
 use parking_lot::Mutex;
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::tokio::sync::RwLock;
+#[cfg(not(feature = "hotpath"))]
 use tokio::sync::RwLock;
 
 use crate::{
@@ -45,7 +48,7 @@ impl PeerMap {
             my_peer_id,
             peer_map: DashMap::new(),
             packet_send,
-            routes: RwLock::new(Vec::new()),
+            routes: hotpath::rw_lock!(tokio::sync::RwLock::new(Vec::new())),
             alive_client_urls: Arc::new(Mutex::new(multimap::MultiMap::new())),
         }
     }
@@ -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,

easytier/src/peers/peer_ospf_route.rs

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

easytier/src/peers/peer_session.rs

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

easytier/src/peers/relay_peer_map.rs

@@ -144,6 +144,7 @@ impl RelayPeerMap {
         Ok(())
     }
 
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
     async fn send_via_next_hop(
         &self,
         msg: ZCPacket,
@@ -166,6 +167,7 @@ impl RelayPeerMap {
         }
     }
 
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
     pub async fn send_msg(
         self: &Arc<Self>,
         mut msg: ZCPacket,
@@ -613,6 +615,7 @@ impl RelayPeerMap {
         Ok(())
     }
 
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
     pub async fn decrypt_if_needed(self: &Arc<Self>, packet: &mut ZCPacket) -> Result<bool, Error> {
         if !self.is_secure_mode_enabled() {
             return Ok(false);

easytier/src/peers/secure_datagram.rs

@@ -1,11 +1,16 @@
 use std::{
     sync::{
-        Arc, Mutex, RwLock,
         atomic::{AtomicBool, AtomicU32, Ordering},
+        Arc,
     },
     time::{SystemTime, UNIX_EPOCH},
 };
 
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::std::sync::{Mutex, RwLock};
+#[cfg(not(feature = "hotpath"))]
+use std::sync::{Mutex, RwLock};
+
 use anyhow::anyhow;
 use atomic_shim::AtomicU64;
 use hmac::{Hmac, Mac as _};
@@ -14,7 +19,7 @@ use sha2::Sha256;
 use zerocopy::FromBytes;
 
 use crate::{
-    peers::encrypt::{Encryptor, create_encryptor},
+    peers::encrypt::{create_encryptor, Encryptor},
     tunnel::packet_def::{StandardAeadTail, ZCPacket},
 };
 
@@ -228,15 +233,15 @@ pub struct SecureDatagramSession {
 impl std::fmt::Debug for SecureDatagramSession {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("SecureDatagramSession")
-            .field("root_key", &self.root_key)
+            .field("root_key", &*self.root_key.read().unwrap())
             .field("session_generation", &self.session_generation)
             .field("send_epoch", &self.send_epoch)
             .field("send_seq", &self.send_seq)
             .field("send_epoch_started_ms", &self.send_epoch_started_ms)
             .field("send_packets_since_epoch", &self.send_packets_since_epoch)
-            .field("rx_slots", &self.rx_slots)
-            .field("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);

easytier/src/tunnel/fake_tcp/stack.rs

@@ -48,9 +48,13 @@ use std::collections::{HashMap, HashSet};
 use std::fmt;
 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::sync::{
-    Arc, RwLock,
+    Arc,
     atomic::{AtomicU32, Ordering},
 };
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::std::sync::RwLock;
+#[cfg(not(feature = "hotpath"))]
+use std::sync::RwLock;
 use tokio::sync::broadcast;
 use tokio::time;
 use tokio_util::task::AbortOnDropHandle;
@@ -159,7 +163,7 @@ impl Socket {
         ack: Option<u32>,
         state: State,
     ) -> (Socket, flume::Sender<Bytes>) {
-        let (incoming_tx, incoming_rx) = flume::bounded(MPMC_BUFFER_LEN);
+        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(),
         });
 

easytier/src/tunnel/mpsc.rs

@@ -19,6 +19,7 @@ use futures::SinkExt;
 pub struct MpscTunnelSender(Sender<ZCPacket>);
 
 impl MpscTunnelSender {
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnelSender"))]
     pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
         self.0.send(item).await.with_context(|| "send error")?;
         Ok(())
@@ -43,7 +44,7 @@ pub struct MpscTunnel<T> {
 
 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 {
@@ -66,6 +67,7 @@ impl<T: Tunnel> MpscTunnel<T> {
         }
     }
 
+    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
     async fn forward_one_round(
         rx: &mut Receiver<ZCPacket>,
         sink: &mut Pin<Box<dyn ZCPacketSink>>,
@@ -79,6 +81,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>>,
@@ -96,6 +99,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>>,

easytier/src/tunnel/quic.rs

@@ -12,6 +12,9 @@ use crate::tunnel::{
 use anyhow::Context;
 use derivative::Derivative;
 use derive_more::{Deref, DerefMut};
+#[cfg(feature = "hotpath")]
+use hotpath::wrap::parking_lot::RwLock;
+#[cfg(not(feature = "hotpath"))]
 use parking_lot::RwLock;
 use quinn::{
     ClientConfig, ConnectError, Connection, Endpoint, EndpointConfig, ServerConfig,
@@ -312,18 +315,25 @@ struct RwPoolInner<Item> {
     enabled: bool,
 }
 
-#[derive(Debug)]
 struct RwPool<Item> {
     ephemeral: RwLock<RwPoolInner<Item>>,
     persistent: RwLock<RwPoolInner<Item>>,
     capacity: usize,
 }
 
+impl<Item> std::fmt::Debug for RwPool<Item> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RwPool")
+            .field("capacity", &self.capacity)
+            .finish()
+    }
+}
+
 impl<Item> RwPool<Item> {
     fn new(capacity: usize) -> Self {
         Self {
-            ephemeral: RwLock::new(RwPoolInner::default()),
-            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,
         }
     }

easytier/src/tunnel/ring.rs

@@ -196,7 +196,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,

easytier/src/tunnel/udp.rs

@@ -293,6 +293,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 +328,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 +397,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 +544,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 +651,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 +680,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 +921,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());

easytier/src/tunnel/wireguard.rs

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