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

@@ -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();

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)))

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),