From 340145ae5d1ccce936e4736c392324d648498c8b Mon Sep 17 00:00:00 2001
From: fanyang <fanyang89@outlook.com>
Date: Sun, 28 Jun 2026 20:10:18 +0800
Subject: [PATCH] perf(mpsc): use try_lock + merged poll_fn for direct sink
 path
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Replace 3 await points (lock().await + feed().await + flush().await)
with try_lock() (sync) + single poll_fn (merged poll_ready + start_send
+ poll_flush).

parking_lot::Mutex cannot be used because MutexGuard is !Send (cannot
cross await in multi_thread runtime). tokio::sync::Mutex try_lock()
returns synchronously and MutexGuard is Send.

Benchmark: pps 250K → 251K (+0.4%), MpscTunnelSender::send avg
2.07us → 1.98us (-90ns). Improvement is small because tokio async
machinery overhead (Future state machine + poll) dominates over
RingSink's actual 40ns operation cost.
---
 easytier/src/tunnel/mpsc.rs | 27 ++++++++++++++++++++++-----
 1 file changed, 22 insertions(+), 5 deletions(-)

diff --git a/easytier/src/tunnel/mpsc.rs b/easytier/src/tunnel/mpsc.rs
index 5d140fb8..a61f7c26 100644
--- a/easytier/src/tunnel/mpsc.rs
+++ b/easytier/src/tunnel/mpsc.rs
@@ -1,6 +1,6 @@
 // this mod wrap tunnel to a mpsc tunnel, based on crossbeam_channel
 
-use std::{pin::Pin, sync::Arc, time::Duration};
+use std::{future::poll_fn, pin::Pin, sync::Arc, task::Poll, time::Duration};
 
 use anyhow::Context;
 use tokio::sync::Mutex;
@@ -24,10 +24,27 @@ pub struct MpscTunnelSender {
 impl MpscTunnelSender {
     pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
         if let Some(sink) = &self.direct_sink {
-            let mut guard = sink.lock().await;
-            guard.feed(item).await?;
-            guard.flush().await?;
-            return Ok(());
+            let mut item = Some(item);
+            loop {
+                if let Ok(mut guard) = sink.try_lock() {
+                    let result = poll_fn(|cx| {
+                        match guard.as_mut().poll_ready(cx) {
+                            Poll::Ready(Ok(())) => {
+                                let it = item.take().unwrap();
+                                if let Err(e) = guard.as_mut().start_send(it) {
+                                    return Poll::Ready(Err(e));
+                                }
+                                guard.as_mut().poll_flush(cx)
+                            }
+                            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
+                            Poll::Pending => Poll::Pending,
+                        }
+                    })
+                    .await;
+                    return result;
+                }
+                tokio::task::yield_now().await;
+            }
         }
 
         let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;