From 42b6d326a811571c9f8ba1864f40e80ec0c75b0c Mon Sep 17 00:00:00 2001 From: fanyang Date: Mon, 29 Jun 2026 00:35:50 +0800 Subject: [PATCH] perf(recv): try_recv fast path + inline in start_peer_recv MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add try_recv fast path to recv_packet_from_chan: try non-blocking recv first, fall back to recv().await only when channel is empty. Additionally inline the try_recv into start_peer_recv's loop body, eliminating the async fn wrapper overhead for the common case (channel has data). Also add hotpath measure to DefaultCompressor::decompress for receive path visibility. Benchmark (TCP, with hotpath): Before: ~449K pps After: ~448K pps (noise — receiver is not the bottleneck) Key finding: receive side is NOT the bottleneck in one-directional bench. Sender rate (~448K pps with hotpath, ~984K without) limits throughput. Receive optimization matters for bidirectional scenarios. 210 peers tests pass. 6 netns tests fail (require root, unchanged). --- easytier/src/common/compressor.rs | 1 + easytier/src/peers/mod.rs | 14 ++++++++++---- easytier/src/peers/peer_manager.rs | 12 +++++++++++- 3 files changed, 22 insertions(+), 5 deletions(-) diff --git a/easytier/src/common/compressor.rs b/easytier/src/common/compressor.rs index d78972ee..bd46e6f2 100644 --- a/easytier/src/common/compressor.rs +++ b/easytier/src/common/compressor.rs @@ -129,6 +129,7 @@ impl Compressor for DefaultCompressor { Ok(()) } + #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "DefaultCompressor"))] async fn decompress(&self, zc_packet: &mut ZCPacket) -> Result<(), Error> { let pm_header = zc_packet.peer_manager_header().unwrap(); if !pm_header.is_compressed() { diff --git a/easytier/src/peers/mod.rs b/easytier/src/peers/mod.rs index debf7e9d..57c2e7d2 100644 --- a/easytier/src/peers/mod.rs +++ b/easytier/src/peers/mod.rs @@ -61,13 +61,19 @@ pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver; pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) { hotpath::channel!(tokio::sync::mpsc::channel(128)) } +#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PacketRecvChan"))] pub async fn recv_packet_from_chan( packet_recv_chan_receiver: &mut PacketRecvChanReceiver, ) -> Result { - packet_recv_chan_receiver - .recv() - .await - .ok_or(anyhow::anyhow!("recv_packet_from_chan failed")) + use tokio::sync::mpsc::error::TryRecvError; + match packet_recv_chan_receiver.try_recv() { + Ok(pkt) => Ok(pkt), + Err(TryRecvError::Empty) => packet_recv_chan_receiver + .recv() + .await + .ok_or(anyhow::anyhow!("recv_packet_from_chan failed")), + Err(TryRecvError::Disconnected) => Err(anyhow::anyhow!("recv_packet_from_chan failed")), + } } pub const PUBLIC_SERVER_HOSTNAME_PREFIX: &str = "PublicServer_"; diff --git a/easytier/src/peers/peer_manager.rs b/easytier/src/peers/peer_manager.rs index 6bbbfa64..c1d495ba 100644 --- a/easytier/src/peers/peer_manager.rs +++ b/easytier/src/peers/peer_manager.rs @@ -1015,7 +1015,17 @@ impl PeerManager { self.tasks.lock().await.spawn(async move { tracing::trace!("start_peer_recv"); - while let Ok(ret) = recv_packet_from_chan(&mut recv).await { + loop { + let ret = match recv.try_recv() { + Ok(pkt) => pkt, + Err(tokio::sync::mpsc::error::TryRecvError::Empty) => { + match recv.recv().await { + Some(pkt) => pkt, + None => break, + } + } + Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break, + }; let disable_relay_data = global_ctx.flags_arc().disable_relay_data; let Err(mut ret) = Self::try_handle_foreign_network_packet( ret,