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/src/peers/peer_conn.rs

@@ -363,11 +363,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

@@ -87,9 +87,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),