From 392a970db1186856992414107838a0768871e82b Mon Sep 17 00:00:00 2001 From: fanyang Date: Sun, 28 Jun 2026 18:55:36 +0800 Subject: [PATCH] bench: add configurable pipeline depth via HOTPATH_PIPELINE env FuturesUnordered-based pipeline to overlap encrypt with mpsc_send. Tested depths 1/4/8/16: max +1.6% at depth=4, within noise. Pipeline has limited value because try_send fast path eliminates await gaps that would allow overlap. Default remains depth=1 (serial). --- easytier/examples/cpu_hotspot_ring.rs | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/easytier/examples/cpu_hotspot_ring.rs b/easytier/examples/cpu_hotspot_ring.rs index 490b6e14..7dc883ac 100644 --- a/easytier/examples/cpu_hotspot_ring.rs +++ b/easytier/examples/cpu_hotspot_ring.rs @@ -85,12 +85,29 @@ async fn main() { let pm = inst_a.get_peer_manager(); let send_pkt = make_data_packet(src, "10.144.144.2", pkt_size); + let pipeline_depth: usize = std::env::var("HOTPATH_PIPELINE") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(1); + + println!( + "cpu_hotspot_ring: pipeline_depth={}", + pipeline_depth + ); + let sender_task = tokio::spawn(async move { + use futures::stream::{FuturesUnordered, StreamExt}; + let mut sent: u64 = 0; let start = Instant::now(); + let mut in_flight = FuturesUnordered::new(); + loop { - let pkt = send_pkt.clone(); - let _ = pm.send_msg_by_ip(pkt, dst, false).await; + while in_flight.len() < pipeline_depth { + let pkt = send_pkt.clone(); + in_flight.push(pm.send_msg_by_ip(pkt, dst, false)); + } + in_flight.next().await; sent += 1; if sent % 10000 == 0 { let elapsed = start.elapsed().as_secs_f64();