mirror of
https://github.com/EasyTier/EasyTier.git
synced 2026-09-02 09:09:17 +00:00
perf(mpsc): use try_lock + merged poll_fn for direct sink path
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.
This commit is contained in:
@@ -1,6 +1,6 @@
|
|||||||
// this mod wrap tunnel to a mpsc tunnel, based on crossbeam_channel
|
// 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 anyhow::Context;
|
||||||
use tokio::sync::Mutex;
|
use tokio::sync::Mutex;
|
||||||
@@ -24,10 +24,27 @@ 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 guard = sink.lock().await;
|
let mut item = Some(item);
|
||||||
guard.feed(item).await?;
|
loop {
|
||||||
guard.flush().await?;
|
if let Ok(mut guard) = sink.try_lock() {
|
||||||
return Ok(());
|
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)?;
|
let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
|
||||||
|
|||||||
Reference in New Issue
Block a user