From 6d019085931c27b2844743aaab3fee2aff5a2406 Mon Sep 17 00:00:00 2001 From: fanyang Date: Sun, 28 Jun 2026 22:00:27 +0800 Subject: [PATCH] perf(mpsc): extend noop_waker sync send to UDP tunnels UDP tunnel uses RingSink internally (same as ring tunnel). Extend direct mode to include UDP. Fix poll_flush Pending to return Ok. Add UDP benchmark support via HOTPATH_TUNNEL=udp env variable. All 208 peers tests pass. Netns tests unchanged (require root). --- easytier/examples/cpu_hotspot_ring.rs | 44 +++++++++++++++++++++------ easytier/src/peers/peer_conn.rs | 6 ++-- easytier/src/tunnel/mpsc.rs | 7 +++-- 3 files changed, 43 insertions(+), 14 deletions(-) diff --git a/easytier/examples/cpu_hotspot_ring.rs b/easytier/examples/cpu_hotspot_ring.rs index 7dc883ac..0b70ea9e 100644 --- a/easytier/examples/cpu_hotspot_ring.rs +++ b/easytier/examples/cpu_hotspot_ring.rs @@ -28,10 +28,11 @@ use std::time::{Duration, Instant}; use bytes::BytesMut; -use easytier::common::config::{ConfigLoader, TomlConfigLoader}; +use easytier::common::config::{ConfigLoader, PeerConfig, TomlConfigLoader}; use easytier::instance::instance::Instance; use easytier::tunnel::packet_def::ZCPacket; use easytier::tunnel::ring::RingTunnelConnector; +use easytier::tunnel::udp::UdpTunnelConnector; #[tokio::main(flavor = "multi_thread", worker_threads = 4)] #[cfg_attr(feature = "hotpath", hotpath::main)] @@ -46,16 +47,41 @@ async fn main() { .and_then(|s| s.parse().ok()) .unwrap_or(1400); - let mut inst_a = Instance::new(no_tun_config("hot-a", "10.144.144.1")); - let mut inst_b = Instance::new(no_tun_config("hot-b", "10.144.144.2")); + let tunnel_type = std::env::var("HOTPATH_TUNNEL") + .ok() + .unwrap_or_else(|| "ring".to_string()); + + let (inst_a_config, inst_b_config) = if tunnel_type == "udp" { + let mut a = no_tun_config("hot-a", "10.144.144.1"); + a.set_listeners(vec!["udp://0.0.0.0:35521".parse().unwrap()]); + + let b = no_tun_config("hot-b", "10.144.144.2"); + (a, b) + } else { + ( + no_tun_config("hot-a", "10.144.144.1"), + no_tun_config("hot-b", "10.144.144.2"), + ) + }; + + let mut inst_a = Instance::new(inst_a_config); + let mut inst_b = Instance::new(inst_b_config); inst_a.run().await.expect("inst_a run"); inst_b.run().await.expect("inst_b run"); - let ring_url = format!("ring://{}", inst_a.id()); - inst_b - .get_conn_manager() - .add_connector(RingTunnelConnector::new(ring_url.parse().unwrap())); + tokio::time::sleep(Duration::from_secs(1)).await; + + if tunnel_type == "ring" { + let ring_url = format!("ring://{}", inst_a.id()); + inst_b + .get_conn_manager() + .add_connector(RingTunnelConnector::new(ring_url.parse().unwrap())); + } else if tunnel_type == "udp" { + inst_b.get_conn_manager().add_connector( + UdpTunnelConnector::new("udp://127.0.0.1:35521".parse().unwrap()), + ); + } let dst: IpAddr = "10.144.144.2".parse().unwrap(); let src = "10.144.144.1"; @@ -78,8 +104,8 @@ async fn main() { } println!( - "cpu_hotspot_ring: flooding {}s, pkt_size={} (converged={})", - duration, pkt_size, converged + "cpu_hotspot_ring: flooding {}s, pkt_size={}, tunnel={} (converged={})", + duration, pkt_size, tunnel_type, converged ); let pm = inst_a.get_peer_manager(); diff --git a/easytier/src/peers/peer_conn.rs b/easytier/src/peers/peer_conn.rs index 30a9a72d..044a4e96 100644 --- a/easytier/src/peers/peer_conn.rs +++ b/easytier/src/peers/peer_conn.rs @@ -370,11 +370,11 @@ 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 is_ring = peer_conn_tunnel + let supports_direct = peer_conn_tunnel .info() - .map(|i| i.tunnel_type == "ring") + .map(|i| matches!(i.tunnel_type.as_str(), "ring" | "udp")) .unwrap_or(false); - let mut mpsc_tunnel = if is_ring { + let mut mpsc_tunnel = if supports_direct { MpscTunnel::new_direct(peer_conn_tunnel) } else { MpscTunnel::new(peer_conn_tunnel, Some(Duration::from_secs(7))) diff --git a/easytier/src/tunnel/mpsc.rs b/easytier/src/tunnel/mpsc.rs index 1796cc16..0da7f96a 100644 --- a/easytier/src/tunnel/mpsc.rs +++ b/easytier/src/tunnel/mpsc.rs @@ -88,9 +88,12 @@ impl MpscTunnelSender { match guard.as_mut().poll_ready(&mut cx) { Poll::Ready(Ok(())) => { guard.as_mut().start_send(item)?; + // poll_flush may return Pending when the consumer task hasn't + // drained the ring yet. The data is already in the ring buffer + // and will be consumed — treat Pending as success. match guard.as_mut().poll_flush(&mut cx) { - Poll::Ready(Ok(())) => return Ok(()), - _ => return Err(TunnelError::Shutdown), + Poll::Ready(Err(e)) => return Err(e), + _ => return Ok(()), } } Poll::Ready(Err(e)) => return Err(e),