mirror of
https://github.com/EasyTier/EasyTier.git
synced 2026-09-02 17:15:43 +00:00
perf(recv): try_recv fast path + inline in start_peer_recv
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).
This commit is contained in:
@@ -129,6 +129,7 @@ impl Compressor for DefaultCompressor {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "DefaultCompressor"))]
|
||||||
async fn decompress(&self, zc_packet: &mut ZCPacket) -> Result<(), Error> {
|
async fn decompress(&self, zc_packet: &mut ZCPacket) -> Result<(), Error> {
|
||||||
let pm_header = zc_packet.peer_manager_header().unwrap();
|
let pm_header = zc_packet.peer_manager_header().unwrap();
|
||||||
if !pm_header.is_compressed() {
|
if !pm_header.is_compressed() {
|
||||||
|
|||||||
@@ -61,13 +61,19 @@ pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver<ZCPacket>;
|
|||||||
pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
|
pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
|
||||||
hotpath::channel!(tokio::sync::mpsc::channel(128))
|
hotpath::channel!(tokio::sync::mpsc::channel(128))
|
||||||
}
|
}
|
||||||
|
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PacketRecvChan"))]
|
||||||
pub async fn recv_packet_from_chan(
|
pub async fn recv_packet_from_chan(
|
||||||
packet_recv_chan_receiver: &mut PacketRecvChanReceiver,
|
packet_recv_chan_receiver: &mut PacketRecvChanReceiver,
|
||||||
) -> Result<ZCPacket, anyhow::Error> {
|
) -> Result<ZCPacket, anyhow::Error> {
|
||||||
packet_recv_chan_receiver
|
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()
|
.recv()
|
||||||
.await
|
.await
|
||||||
.ok_or(anyhow::anyhow!("recv_packet_from_chan failed"))
|
.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_";
|
pub const PUBLIC_SERVER_HOSTNAME_PREFIX: &str = "PublicServer_";
|
||||||
|
|||||||
@@ -1015,7 +1015,17 @@ impl PeerManager {
|
|||||||
|
|
||||||
self.tasks.lock().await.spawn(async move {
|
self.tasks.lock().await.spawn(async move {
|
||||||
tracing::trace!("start_peer_recv");
|
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 disable_relay_data = global_ctx.flags_arc().disable_relay_data;
|
||||||
let Err(mut ret) = Self::try_handle_foreign_network_packet(
|
let Err(mut ret) = Self::try_handle_foreign_network_packet(
|
||||||
ret,
|
ret,
|
||||||
|
|||||||
Reference in New Issue
Block a user