perf(mpsc): add direct sink path bypassing channel for PeerConn

MpscTunnelSender now supports two modes:
- Channel mode (existing): try_send to tokio mpsc → receiver task → sink
- Direct mode (new): MpscTunnelSender holds Arc<Mutex<sink>> directly,
  bypassing the channel + receiver task entirely

PeerConn uses new_direct to skip the channel intermediary.

Benchmark result: pps unchanged (~245K). The async fn overhead of
Mutex::lock().await + SinkExt::feed().await + SinkExt::flush().await
(~2us) is comparable to channel try_send (~2us). The bottleneck is
the Sink trait's async poll machinery, not the channel itself.

However, this change provides:
- RingSink timing now fully visible (start_send 10ns, poll_ready 13ns,
  poll_flush 17ns = 40ns/pkt total)
- Reduced architectural complexity (no receiver task for PeerConn)
- Foundation for a sync fast path using RingSink::try_send directly
This commit is contained in:
fanyang
2026-06-28 19:42:42 +08:00
parent 7e0cdfc683
commit cdec67ff53
2 changed files with 42 additions and 11 deletions
+1 -1
View File
@@ -370,7 +370,7 @@ impl PeerConn {
let throughput = peer_conn_tunnel_filter.filter_output();
let filter_chain = TunnelFilterChain::new(session_filter.clone(), peer_conn_tunnel_filter);
let peer_conn_tunnel = TunnelWithFilter::new(tunnel, filter_chain);
let mut mpsc_tunnel = MpscTunnel::new(peer_conn_tunnel, Some(Duration::from_secs(7)));
let mut mpsc_tunnel = MpscTunnel::new_direct(peer_conn_tunnel);
let (recv, sink) = (mpsc_tunnel.get_stream(), mpsc_tunnel.get_sink());
+41 -10
View File
@@ -1,8 +1,9 @@
// this mod wrap tunnel to a mpsc tunnel, based on crossbeam_channel
use std::{pin::Pin, time::Duration};
use std::{pin::Pin, sync::Arc, time::Duration};
use anyhow::Context;
use tokio::sync::Mutex;
use tokio::time::timeout;
use crate::proto::common::TunnelInfo;
@@ -11,20 +12,30 @@ use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPac
use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
use tokio_util::task::AbortOnDropHandle;
// use tachyonix::{channel, Receiver, Sender, TrySendError};
use futures::SinkExt;
#[derive(Clone)]
pub struct MpscTunnelSender(Sender<ZCPacket>);
pub struct MpscTunnelSender {
channel_tx: Option<Sender<ZCPacket>>,
direct_sink: Option<Arc<Mutex<Pin<Box<dyn ZCPacketSink>>>>>,
}
impl MpscTunnelSender {
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnelSender"))]
pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
match self.0.try_send(item) {
if let Some(sink) = &self.direct_sink {
let mut guard = sink.lock().await;
guard.feed(item).await?;
guard.flush().await?;
return Ok(());
}
let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
match tx.try_send(item) {
Ok(()) => Ok(()),
Err(TrySendError::Full(item)) => {
self.0.send(item).await.with_context(|| "send error")?;
tx.send(item).await.with_context(|| "send error")?;
Ok(())
}
Err(TrySendError::Closed(_)) => Err(TunnelError::Shutdown),
@@ -32,7 +43,8 @@ impl MpscTunnelSender {
}
pub fn try_send(&self, item: ZCPacket) -> Result<(), TunnelError> {
self.0.try_send(item).map_err(|e| match e {
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,
})
@@ -41,11 +53,12 @@ impl MpscTunnelSender {
pub struct MpscTunnel<T> {
tx: Option<Sender<ZCPacket>>,
direct_sink: Option<Arc<Mutex<Pin<Box<dyn ZCPacketSink>>>>>,
tunnel: T,
stream: Option<Pin<Box<dyn ZCPacketStream>>>,
task: AbortOnDropHandle<()>,
task: Option<AbortOnDropHandle<()>>,
}
impl<T: Tunnel> MpscTunnel<T> {
@@ -67,9 +80,21 @@ impl<T: Tunnel> MpscTunnel<T> {
Self {
tx: Some(tx),
direct_sink: None,
tunnel,
stream: Some(stream),
task: AbortOnDropHandle::new(task),
task: Some(AbortOnDropHandle::new(task)),
}
}
pub fn new_direct(tunnel: T) -> Self {
let (stream, sink) = tunnel.split();
Self {
tx: None,
direct_sink: Some(Arc::new(Mutex::new(sink))),
tunnel,
stream: Some(stream),
task: None,
}
}
@@ -134,12 +159,18 @@ impl<T: Tunnel> MpscTunnel<T> {
}
pub fn get_sink(&self) -> MpscTunnelSender {
MpscTunnelSender(self.tx.as_ref().unwrap().clone())
MpscTunnelSender {
channel_tx: self.tx.as_ref().cloned(),
direct_sink: self.direct_sink.clone(),
}
}
pub fn close(&mut self) {
self.tx.take();
self.task.abort();
self.direct_sink.take();
if let Some(task) = self.task.take() {
task.abort();
}
}
pub fn tunnel_info(&self) -> Option<TunnelInfo> {