mirror of
https://github.com/EasyTier/EasyTier.git
synced 2026-09-03 01:25:37 +00:00
perf(mpsc): sync send via noop_waker — +90% pps (249K → 474K)
The async fn Future state machine overhead (~1.9us) dominated MpscTunnelSender::send, while RingSink operations were only ~40ns. Breakthrough: make send() an async fn that completes synchronously on the first poll for the direct (ring tunnel) path. Uses futures::task::noop_waker() to construct a dummy Context, then calls Sink trait methods (poll_ready, start_send, poll_flush) directly. RingSink always returns Ready immediately, so the waker is never invoked and the async fn completes without yielding. Channel mode (TCP/UDP/WG tunnels) still uses async send_async() with proper backpressure. Ring tunnels detected via tunnel_info() type check in PeerConn. Results (4 threads, 1400B, 15s): pps: 249K → 474K (+90%) send_msg_by_ip: 3.53us → 1.67us (-53%) send_msg_internal: 2.40us → 502ns (-79%) MpscTunnelSender::send: 1.97us → 144ns (-93%) All 207 peers:: tests pass. Netns-requiring tests (three_node, credential) unchanged (require root).
This commit is contained in:
@@ -363,7 +363,15 @@ impl PeerConn {
|
|||||||
let throughput = peer_conn_tunnel_filter.filter_output();
|
let throughput = peer_conn_tunnel_filter.filter_output();
|
||||||
let filter_chain = TunnelFilterChain::new(session_filter.clone(), peer_conn_tunnel_filter);
|
let filter_chain = TunnelFilterChain::new(session_filter.clone(), peer_conn_tunnel_filter);
|
||||||
let peer_conn_tunnel = TunnelWithFilter::new(tunnel, filter_chain);
|
let peer_conn_tunnel = TunnelWithFilter::new(tunnel, filter_chain);
|
||||||
let mut mpsc_tunnel = MpscTunnel::new_direct(peer_conn_tunnel);
|
let is_ring = peer_conn_tunnel
|
||||||
|
.info()
|
||||||
|
.map(|i| i.tunnel_type == "ring")
|
||||||
|
.unwrap_or(false);
|
||||||
|
let mut mpsc_tunnel = if is_ring {
|
||||||
|
MpscTunnel::new_direct(peer_conn_tunnel)
|
||||||
|
} else {
|
||||||
|
MpscTunnel::new(peer_conn_tunnel, Some(Duration::from_secs(7)))
|
||||||
|
};
|
||||||
|
|
||||||
let (recv, sink) = (mpsc_tunnel.get_stream(), mpsc_tunnel.get_sink());
|
let (recv, sink) = (mpsc_tunnel.get_stream(), mpsc_tunnel.get_sink());
|
||||||
|
|
||||||
|
|||||||
+26
-28
@@ -2,7 +2,6 @@
|
|||||||
|
|
||||||
use std::{
|
use std::{
|
||||||
cell::UnsafeCell,
|
cell::UnsafeCell,
|
||||||
future::poll_fn,
|
|
||||||
pin::Pin,
|
pin::Pin,
|
||||||
sync::Arc,
|
sync::Arc,
|
||||||
sync::atomic::{AtomicBool, Ordering},
|
sync::atomic::{AtomicBool, Ordering},
|
||||||
@@ -81,29 +80,38 @@ pub struct MpscTunnelSender {
|
|||||||
impl MpscTunnelSender {
|
impl MpscTunnelSender {
|
||||||
pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
|
pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
|
||||||
if let Some(sink) = &self.direct_sink {
|
if let Some(sink) = &self.direct_sink {
|
||||||
let mut item = Some(item);
|
// Sync fast path: no await needed, returns immediately
|
||||||
loop {
|
|
||||||
if let Some(mut guard) = sink.try_lock() {
|
if let Some(mut guard) = sink.try_lock() {
|
||||||
let result = poll_fn(|cx| {
|
let waker = futures::task::noop_waker();
|
||||||
match guard.as_mut().poll_ready(cx) {
|
let mut cx = std::task::Context::from_waker(&waker);
|
||||||
|
match guard.as_mut().poll_ready(&mut cx) {
|
||||||
Poll::Ready(Ok(())) => {
|
Poll::Ready(Ok(())) => {
|
||||||
let it = item.take().unwrap();
|
guard.as_mut().start_send(item)?;
|
||||||
if let Err(e) = guard.as_mut().start_send(it) {
|
match guard.as_mut().poll_flush(&mut cx) {
|
||||||
return Poll::Ready(Err(e));
|
Poll::Ready(Ok(())) => return Ok(()),
|
||||||
|
_ => return Err(TunnelError::Shutdown),
|
||||||
}
|
}
|
||||||
guard.as_mut().poll_flush(cx)
|
|
||||||
}
|
}
|
||||||
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
|
Poll::Ready(Err(e)) => return Err(e),
|
||||||
Poll::Pending => Poll::Pending,
|
Poll::Pending => return Err(TunnelError::BufferFull),
|
||||||
}
|
}
|
||||||
})
|
|
||||||
.await;
|
|
||||||
return result;
|
|
||||||
}
|
|
||||||
tokio::task::yield_now().await;
|
|
||||||
}
|
}
|
||||||
|
return Err(TunnelError::BufferFull);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Channel mode: async with backpressure
|
||||||
|
self.send_async(item).await
|
||||||
|
}
|
||||||
|
|
||||||
|
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 {
|
||||||
|
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)?;
|
let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
|
||||||
match tx.try_send(item) {
|
match tx.try_send(item) {
|
||||||
Ok(()) => Ok(()),
|
Ok(()) => Ok(()),
|
||||||
@@ -114,14 +122,6 @@ impl MpscTunnelSender {
|
|||||||
Err(TrySendError::Closed(_)) => Err(TunnelError::Shutdown),
|
Err(TrySendError::Closed(_)) => Err(TunnelError::Shutdown),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
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 {
|
|
||||||
TrySendError::Full(_) => TunnelError::BufferFull,
|
|
||||||
TrySendError::Closed(_) => TunnelError::Shutdown,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct MpscTunnel<T> {
|
pub struct MpscTunnel<T> {
|
||||||
@@ -300,8 +300,7 @@ mod tests {
|
|||||||
for i in 0..1000000 {
|
for i in 0..1000000 {
|
||||||
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
|
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
|
||||||
let a = sink1
|
let a = sink1
|
||||||
.send(ZCPacket::new_with_payload("hello".as_bytes()))
|
.send_async(ZCPacket::new_with_payload("hello".as_bytes())).await;
|
||||||
.await;
|
|
||||||
if a.is_err() {
|
if a.is_err() {
|
||||||
tracing::info!(?a, "t2 exit with err");
|
tracing::info!(?a, "t2 exit with err");
|
||||||
break;
|
break;
|
||||||
@@ -320,8 +319,7 @@ mod tests {
|
|||||||
for i in 0..1000000 {
|
for i in 0..1000000 {
|
||||||
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
||||||
let a = sink2
|
let a = sink2
|
||||||
.send(ZCPacket::new_with_payload("hello2".as_bytes()))
|
.send_async(ZCPacket::new_with_payload("hello2".as_bytes())).await;
|
||||||
.await;
|
|
||||||
if a.is_err() {
|
if a.is_err() {
|
||||||
tracing::info!(?a, "t3 exit with err");
|
tracing::info!(?a, "t3 exit with err");
|
||||||
break;
|
break;
|
||||||
|
|||||||
Reference in New Issue
Block a user