From cdec67ff53896970dfd1722be476677d18a499dc Mon Sep 17 00:00:00 2001
From: fanyang <fanyang89@outlook.com>
Date: Sun, 28 Jun 2026 19:42:42 +0800
Subject: [PATCH] perf(mpsc): add direct sink path bypassing channel for
 PeerConn
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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
---
 easytier/src/peers/peer_conn.rs |  2 +-
 easytier/src/tunnel/mpsc.rs     | 51 ++++++++++++++++++++++++++-------
 2 files changed, 42 insertions(+), 11 deletions(-)

diff --git a/easytier/src/peers/peer_conn.rs b/easytier/src/peers/peer_conn.rs
index 2e640547..8468cc8a 100644
--- a/easytier/src/peers/peer_conn.rs
+++ b/easytier/src/peers/peer_conn.rs
@@ -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());
 
diff --git a/easytier/src/tunnel/mpsc.rs b/easytier/src/tunnel/mpsc.rs
index 7e7d51ef..82debb03 100644
--- a/easytier/src/tunnel/mpsc.rs
+++ b/easytier/src/tunnel/mpsc.rs
@@ -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> {