From 27f6b90fa1775a805ec587819f288f6ab4d359aa Mon Sep 17 00:00:00 2001 From: fanyang89 Date: Thu, 2 Jul 2026 00:07:16 +0800 Subject: [PATCH] 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. --- Cargo.lock | 146 ++++++++++++++++++++++++++ easytier/src/peers/mod.rs | 3 +- easytier/src/peers/peer.rs | 6 +- easytier/src/peers/peer_conn.rs | 10 +- easytier/src/peers/peer_manager.rs | 25 +++-- easytier/src/peers/peer_map.rs | 7 +- easytier/src/peers/peer_ospf_route.rs | 2 + easytier/src/peers/peer_session.rs | 12 ++- easytier/src/peers/relay_peer_map.rs | 3 + easytier/src/peers/secure_datagram.rs | 41 +++++--- easytier/src/tunnel/fake_tcp/stack.rs | 12 ++- easytier/src/tunnel/mpsc.rs | 6 +- easytier/src/tunnel/quic.rs | 16 ++- easytier/src/tunnel/ring.rs | 2 +- easytier/src/tunnel/udp.rs | 12 ++- easytier/src/tunnel/wireguard.rs | 22 ++-- 16 files changed, 276 insertions(+), 49 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ef1042b4..e326d9e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "467479f30ae7262ad93eff1f6653a58ecfca4b44dd9ff82ef4b469ba1ad00017" +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.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7edb8da2a00454786fbb75ab24c4f72380afeb9b5136776caa7737df9895bfa7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "hotpath-macros-meta" +version = "0.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e804ccbbd13922ff16b752ab72043d15ea38411a03a82917b8ea5b36b6b45655" + +[[package]] +name = "hotpath-meta" +version = "0.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f76b1b9a176e5677878243b0c9052d2e0a62405359bd6990bbd85b59cafa2aa" +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" diff --git a/easytier/src/peers/mod.rs b/easytier/src/peers/mod.rs index c94a65de..b08a76c3 100644 --- a/easytier/src/peers/mod.rs +++ b/easytier/src/peers/mod.rs @@ -59,8 +59,9 @@ type BoxNicPacketFilter = Box; pub type PacketRecvChan = tokio::sync::mpsc::Sender; pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver; 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 { diff --git a/easytier/src/peers/peer.rs b/easytier/src/peers/peer.rs index 63af93f8..7c16ca57 100644 --- a/easytier/src/peers/peer.rs +++ b/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)); diff --git a/easytier/src/peers/peer_conn.rs b/easytier/src/peers/peer_conn.rs index 8b9b6fe6..bfba0fcc 100644 --- a/easytier/src/peers/peer_conn.rs +++ b/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, - )), + ))), 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?) } diff --git a/easytier/src/peers/peer_manager.rs b/easytier/src/peers/peer_manager.rs index 42ff5396..9de6582d 100644 --- a/easytier/src/peers/peer_manager.rs +++ b/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, foreign_network_client: &Arc, @@ -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, diff --git a/easytier/src/peers/peer_map.rs b/easytier/src/peers/peer_map.rs index 46e55aaf..ed72ad6a 100644 --- a/easytier/src/peers/peer_map.rs +++ b/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, diff --git a/easytier/src/peers/peer_ospf_route.rs b/easytier/src/peers/peer_ospf_route.rs index 8afa660a..f9b8d26d 100644 --- a/easytier/src/peers/peer_ospf_route.rs +++ b/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 { 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 { let cur_version = self.next_hop_map_version.get(); self.next_hop_map.get(&dst_peer_id).and_then(|x| { diff --git a/easytier/src/peers/peer_session.rs b/easytier/src/peers/peer_session.rs index 0e3e7f38..c255c220 100644 --- a/easytier/src/peers/peer_session.rs +++ b/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, diff --git a/easytier/src/peers/relay_peer_map.rs b/easytier/src/peers/relay_peer_map.rs index ec5562c3..eda863f6 100644 --- a/easytier/src/peers/relay_peer_map.rs +++ b/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, 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, packet: &mut ZCPacket) -> Result { if !self.is_secure_mode_enabled() { return Ok(false); diff --git a/easytier/src/peers/secure_datagram.rs b/easytier/src/peers/secure_datagram.rs index f4722e44..1b48d352 100644 --- a/easytier/src/peers/secure_datagram.rs +++ b/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); diff --git a/easytier/src/tunnel/fake_tcp/stack.rs b/easytier/src/tunnel/fake_tcp/stack.rs index a7f1b779..b7387704 100644 --- a/easytier/src/tunnel/fake_tcp/stack.rs +++ b/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, state: State, ) -> (Socket, flume::Sender) { - 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(), }); diff --git a/easytier/src/tunnel/mpsc.rs b/easytier/src/tunnel/mpsc.rs index e15231ae..1d95b4c9 100644 --- a/easytier/src/tunnel/mpsc.rs +++ b/easytier/src/tunnel/mpsc.rs @@ -19,6 +19,7 @@ use futures::SinkExt; pub struct MpscTunnelSender(Sender); 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 { impl MpscTunnel { pub fn new(tunnel: T, send_timeout: Option) -> 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 MpscTunnel { } } + #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))] async fn forward_one_round( rx: &mut Receiver, sink: &mut Pin>, @@ -79,6 +81,7 @@ impl MpscTunnel { } } + #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))] async fn forward_one_round_no_timeout( rx: &mut Receiver, sink: &mut Pin>, @@ -96,6 +99,7 @@ impl MpscTunnel { sink.flush().await } + #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))] async fn forward_one_round_with_timeout( rx: &mut Receiver, sink: &mut Pin>, diff --git a/easytier/src/tunnel/quic.rs b/easytier/src/tunnel/quic.rs index da68c66c..b9571eca 100644 --- a/easytier/src/tunnel/quic.rs +++ b/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 { enabled: bool, } -#[derive(Debug)] struct RwPool { ephemeral: RwLock>, persistent: RwLock>, capacity: usize, } +impl std::fmt::Debug for RwPool { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RwPool") + .field("capacity", &self.capacity) + .finish() + } +} + impl RwPool { 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, } } diff --git a/easytier/src/tunnel/ring.rs b/easytier/src/tunnel/ring.rs index 101b67c6..4f0fb324 100644 --- a/easytier/src/tunnel/ring.rs +++ b/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, diff --git a/easytier/src/tunnel/udp.rs b/easytier/src/tunnel/udp.rs index d473afb8..fc78329a 100644 --- a/easytier/src/tunnel/udp.rs +++ b/easytier/src/tunnel/udp.rs @@ -293,6 +293,7 @@ fn get_zcpacket_from_buf(buf: BytesMut, allow_stun: bool) -> Result, @@ -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()); diff --git a/easytier/src/tunnel/wireguard.rs b/easytier/src/tunnel/wireguard.rs index 7efc81f6..3c1da62e 100644 --- a/easytier/src/tunnel/wireguard.rs +++ b/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>>>, + sink: StdMutex>>>, data: Option, tasks: JoinSet<()>, @@ -358,19 +366,19 @@ struct WgPeer { impl WgPeer { fn new(udp: Arc, 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(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,