Decouple output and move TUI off Tokio

Route CLI output through a writer task, move TUI rendering to a
dedicated thread, and simplify event flow with unbounded channels for
non-blocking hot paths

Author: gleb-chipiga

src/cli.rs

@@ -3,9 +3,43 @@ use crate::model::{RunConfig, TestEvent};
 use anyhow::{Context, Result};
 use clap::Parser;
 use rand::RngCore;
+use std::io::Write;
 use std::time::Duration;
 use tokio::sync::mpsc;
 
+enum OutputLine {
+    Stdout(String),
+    Stderr(String),
+}
+
+fn spawn_output_writer() -> (
+    mpsc::UnboundedSender<OutputLine>,
+    tokio::task::JoinHandle<()>,
+) {
+    let (tx, mut rx) = mpsc::unbounded_channel::<OutputLine>();
+    let handle = tokio::task::spawn_blocking(move || {
+        let stdout = std::io::stdout();
+        let stderr = std::io::stderr();
+        let mut out = std::io::BufWriter::new(stdout.lock());
+        let mut err = std::io::BufWriter::new(stderr.lock());
+
+        while let Some(line) = rx.blocking_recv() {
+            match line {
+                OutputLine::Stdout(msg) => {
+                    let _ = writeln!(out, "{}", msg);
+                }
+                OutputLine::Stderr(msg) => {
+                    let _ = writeln!(err, "{}", msg);
+                }
+            }
+        }
+
+        let _ = out.flush();
+        let _ = err.flush();
+    });
+    (tx, handle)
+}
+
 #[derive(Debug, Parser, Clone)]
 #[command(
     name = "cloudflare-speed-cli",
@@ -164,10 +198,16 @@ pub fn build_config(args: &Cli) -> RunConfig {
 async fn run_test_engine(args: Cli, silent: bool) -> Result<()> {
     let cfg = build_config(&args);
     let network_info = crate::network::gather_network_info(&args);
+    let (out_tx, out_handle) = if silent {
+        (None, None)
+    } else {
+        let (tx, handle) = spawn_output_writer();
+        (Some(tx), Some(handle))
+    };
     let enriched = if silent {
         // In silent mode, spawn task and consume events
-        let (evt_tx, mut evt_rx) = mpsc::channel::<TestEvent>(2048);
-        let (_, ctrl_rx) = mpsc::channel::<EngineControl>(16);
+        let (evt_tx, mut evt_rx) = mpsc::unbounded_channel::<TestEvent>();
+        let (_, ctrl_rx) = mpsc::unbounded_channel::<EngineControl>();
 
         let engine = TestEngine::new(cfg);
         let handle = tokio::spawn(async move { engine.run(evt_tx, ctrl_rx).await });
@@ -185,8 +225,8 @@ async fn run_test_engine(args: Cli, silent: bool) -> Result<()> {
         crate::network::enrich_result(&result, &network_info)
     } else {
         // In JSON mode, directly await the engine (no need to consume events)
-        let (evt_tx, _) = mpsc::channel::<TestEvent>(1024);
-        let (_, ctrl_rx) = mpsc::channel::<EngineControl>(16);
+        let (evt_tx, _) = mpsc::unbounded_channel::<TestEvent>();
+        let (_, ctrl_rx) = mpsc::unbounded_channel::<EngineControl>();
 
         let engine = TestEngine::new(cfg);
         let result = engine
@@ -200,29 +240,38 @@ async fn run_test_engine(args: Cli, silent: bool) -> Result<()> {
     // Handle exports (errors will propagate)
     handle_exports(&args, &enriched)?;
 
-    if !silent {
+    if let Some(tx) = out_tx.as_ref() {
         // Print JSON output in non-silent mode
-        println!("{}", serde_json::to_string_pretty(&enriched)?);
+        let out = serde_json::to_string_pretty(&enriched)?;
+        let _ = tx.send(OutputLine::Stdout(out));
     }
 
     // Save results if auto_save is enabled
     if args.auto_save {
         if silent {
             crate::storage::save_run(&enriched).context("failed to save run results")?;
-        } else {
+        } else if let Some(tx) = out_tx.as_ref() {
             if let Ok(p) = crate::storage::save_run(&enriched) {
-                eprintln!("Saved: {}", p.display());
+                let _ = tx.send(OutputLine::Stderr(format!("Saved: {}", p.display())));
             }
         }
     }
 
+    if let Some(tx) = out_tx {
+        drop(tx);
+    }
+    if let Some(handle) = out_handle {
+        let _ = handle.await;
+    }
+
     Ok(())
 }
 
 async fn run_text(args: Cli) -> Result<()> {
     let cfg = build_config(&args);
-    let (evt_tx, mut evt_rx) = mpsc::channel::<TestEvent>(2048);
-    let (_, ctrl_rx) = mpsc::channel::<EngineControl>(16);
+    let (out_tx, out_handle) = spawn_output_writer();
+    let (evt_tx, mut evt_rx) = mpsc::unbounded_channel::<TestEvent>();
+    let (_, ctrl_rx) = mpsc::unbounded_channel::<EngineControl>();
 
     let engine = TestEngine::new(cfg);
     let handle = tokio::spawn(async move { engine.run(evt_tx, ctrl_rx).await });
@@ -238,7 +287,7 @@ async fn run_text(args: Cli) -> Result<()> {
     while let Some(ev) = evt_rx.recv().await {
         match ev {
             TestEvent::PhaseStarted { phase } => {
-                eprintln!("== {phase:?} ==");
+                let _ = out_tx.send(OutputLine::Stderr(format!("== {phase:?} ==")));
             }
             TestEvent::ThroughputTick {
                 phase,
@@ -251,7 +300,7 @@ async fn run_text(args: Cli) -> Result<()> {
                 ) {
                     let elapsed = run_start.elapsed().as_secs_f64();
                     let mbps = (bps_instant * 8.0) / 1_000_000.0;
-                    eprintln!("{phase:?}: {:.2} Mbps", mbps);
+                    let _ = out_tx.send(OutputLine::Stderr(format!("{phase:?}: {:.2} Mbps", mbps)));
 
                     // Collect throughput points for metrics
                     match phase {
@@ -275,7 +324,10 @@ async fn run_text(args: Cli) -> Result<()> {
                     if let Some(ms) = rtt_ms {
                         match (phase, during) {
                             (crate::model::Phase::IdleLatency, None) => {
-                                eprintln!("Idle latency: {:.1} ms", ms);
+                                let _ = out_tx.send(OutputLine::Stderr(format!(
+                                    "Idle latency: {:.1} ms",
+                                    ms
+                                )));
                                 idle_latency_samples.push(ms);
                             }
                             (
@@ -292,10 +344,13 @@ async fn run_text(args: Cli) -> Result<()> {
                     }
                 }
             }
-            TestEvent::Info { message } => eprintln!("{message}"),
+            TestEvent::Info { message } => {
+                let _ = out_tx.send(OutputLine::Stderr(message));
+            }
             TestEvent::MetaInfo { .. } => {
                 // Meta info is handled in TUI, ignore in text mode
             }
+            TestEvent::RunCompleted { .. } => {}
         }
     }
 
@@ -312,86 +367,90 @@ async fn run_text(args: Cli) -> Result<()> {
         let colo = extracted.colo.as_deref().unwrap_or("-");
         let asn = extracted.asn.as_deref().unwrap_or("-");
         let org = extracted.as_org.as_deref().unwrap_or("-");
-        println!("IP/Colo/ASN: {ip} / {colo} / {asn} ({org})");
+        let _ = out_tx.send(OutputLine::Stdout(format!(
+            "IP/Colo/ASN: {ip} / {colo} / {asn} ({org})"
+        )));
     }
     if let Some(server) = enriched.server.as_deref() {
-        println!("Server: {server}");
+        let _ = out_tx.send(OutputLine::Stdout(format!("Server: {server}")));
     }
     if let Some(comments) = enriched.comments.as_deref() {
         if !comments.trim().is_empty() {
-            println!("Comments: {}", comments);
+            let _ = out_tx.send(OutputLine::Stdout(format!("Comments: {}", comments)));
         }
     }
 
     // Compute and display throughput metrics (mean, median, p25, p75)
     let dl_values: Vec<f64> = dl_points.iter().map(|(_, y)| *y).collect();
     let (dl_mean, dl_median, dl_p25, dl_p75) = crate::metrics::compute_metrics(&dl_values)
         .context("insufficient download throughput data to compute metrics")?;
-    println!(
+    let _ = out_tx.send(OutputLine::Stdout(format!(
         "Download: avg {:.2} med {:.2} p25 {:.2} p75 {:.2}",
         dl_mean, dl_median, dl_p25, dl_p75
-    );
+    )));
 
     let ul_values: Vec<f64> = ul_points.iter().map(|(_, y)| *y).collect();
     let (ul_mean, ul_median, ul_p25, ul_p75) = crate::metrics::compute_metrics(&ul_values)
         .context("insufficient upload throughput data to compute metrics")?;
-    println!(
+    let _ = out_tx.send(OutputLine::Stdout(format!(
         "Upload:   avg {:.2} med {:.2} p25 {:.2} p75 {:.2}",
         ul_mean, ul_median, ul_p25, ul_p75
-    );
+    )));
 
     // Compute and display latency metrics (mean, median, p25, p75)
     let (idle_mean, idle_median, idle_p25, idle_p75) =
         crate::metrics::compute_metrics(&idle_latency_samples)
             .context("insufficient idle latency data to compute metrics")?;
-    println!(
+    let _ = out_tx.send(OutputLine::Stdout(format!(
         "Idle latency: avg {:.1} med {:.1} p25 {:.1} p75 {:.1} ms (loss {:.1}%, jitter {:.1} ms)",
         idle_mean,
         idle_median,
         idle_p25,
         idle_p75,
         enriched.idle_latency.loss * 100.0,
         enriched.idle_latency.jitter_ms.unwrap_or(f64::NAN)
-    );
+    )));
 
     let (dl_lat_mean, dl_lat_median, dl_lat_p25, dl_lat_p75) =
         crate::metrics::compute_metrics(&loaded_dl_latency_samples)
             .context("insufficient loaded download latency data to compute metrics")?;
-    println!(
+    let _ = out_tx.send(OutputLine::Stdout(format!(
         "Loaded latency (download): avg {:.1} med {:.1} p25 {:.1} p75 {:.1} ms (loss {:.1}%, jitter {:.1} ms)",
         dl_lat_mean,
         dl_lat_median,
         dl_lat_p25,
         dl_lat_p75,
         enriched.loaded_latency_download.loss * 100.0,
         enriched.loaded_latency_download.jitter_ms.unwrap_or(f64::NAN)
-    );
+    )));
 
     let (ul_lat_mean, ul_lat_median, ul_lat_p25, ul_lat_p75) =
         crate::metrics::compute_metrics(&loaded_ul_latency_samples)
             .context("insufficient loaded upload latency data to compute metrics")?;
-    println!(
+    let _ = out_tx.send(OutputLine::Stdout(format!(
         "Loaded latency (upload): avg {:.1} med {:.1} p25 {:.1} p75 {:.1} ms (loss {:.1}%, jitter {:.1} ms)",
         ul_lat_mean,
         ul_lat_median,
         ul_lat_p25,
         ul_lat_p75,
         enriched.loaded_latency_upload.loss * 100.0,
         enriched.loaded_latency_upload.jitter_ms.unwrap_or(f64::NAN)
-    );
+    )));
     if let Some(ref exp) = enriched.experimental_udp {
-        println!(
+        let _ = out_tx.send(OutputLine::Stdout(format!(
             "Experimental UDP-like loss probe: loss {:.1}% med {} ms (target {:?})",
             exp.latency.loss * 100.0,
             exp.latency.median_ms.unwrap_or(f64::NAN),
             exp.target
-        );
+        )));
     }
     if args.auto_save {
         if let Ok(p) = crate::storage::save_run(&enriched) {
-            eprintln!("Saved: {}", p.display());
+            let _ = out_tx.send(OutputLine::Stderr(format!("Saved: {}", p.display())));
         }
     }
+    drop(out_tx);
+    let _ = out_handle.await;
     Ok(())
 }
 

src/engine/cloudflare.rs

@@ -1,4 +1,4 @@
-use crate::model::TurnInfo;
+use crate::model::{TestEvent, TurnInfo};
 use anyhow::{Context, Result};
 use reqwest::Url;
 use serde::Deserialize;
@@ -14,7 +14,10 @@ pub struct CloudflareClient {
 }
 
 impl CloudflareClient {
-    pub fn new(cfg: &RunConfig) -> Result<Self> {
+    pub fn new(
+        cfg: &RunConfig,
+        event_tx: Option<tokio::sync::mpsc::UnboundedSender<TestEvent>>,
+    ) -> Result<Self> {
         let base_url = Url::parse(&cfg.base_url).context("invalid base_url")?;
 
         let mut builder = reqwest::Client::builder()
@@ -28,10 +31,14 @@ impl CloudflareClient {
             match network_bind::get_interface_ip(iface) {
                 Ok(ip) => {
                     builder = builder.local_address(ip);
-                    eprintln!(
-                        "Binding HTTP connections to interface {} (IP: {})",
-                        iface, ip
-                    );
+                    if let Some(tx) = event_tx.as_ref() {
+                        let _ = tx.send(TestEvent::Info {
+                            message: format!(
+                                "Binding HTTP connections to interface {} (IP: {})",
+                                iface, ip
+                            ),
+                        });
+                    }
                 }
                 Err(e) => {
                     return Err(anyhow::anyhow!(
@@ -46,7 +53,11 @@ impl CloudflareClient {
             match source_ip.parse::<std::net::IpAddr>() {
                 Ok(ip) => {
                     builder = builder.local_address(ip);
-                    eprintln!("Binding HTTP connections to source IP: {}", ip);
+                    if let Some(tx) = event_tx.as_ref() {
+                        let _ = tx.send(TestEvent::Info {
+                            message: format!("Binding HTTP connections to source IP: {}", ip),
+                        });
+                    }
                 }
                 Err(e) => {
                     return Err(anyhow::anyhow!(

src/engine/latency.rs

@@ -9,14 +9,15 @@ use std::sync::{
 use std::time::{Duration, Instant};
 use tokio::sync::mpsc;
 
+#[allow(clippy::too_many_arguments)]
 pub async fn run_latency_probes(
     client: &CloudflareClient,
     phase: Phase,
     during: Option<Phase>,
     total_duration: Duration,
     interval_ms: u64,
     timeout_ms: u64,
-    event_tx: &mpsc::Sender<TestEvent>,
+    event_tx: &mpsc::UnboundedSender<TestEvent>,
     paused: Arc<AtomicBool>,
     cancel: Arc<AtomicBool>,
 ) -> Result<LatencySummary> {
@@ -48,31 +49,25 @@ pub async fn run_latency_probes(
                 // Extract meta from first successful response
                 if !meta_sent && phase == Phase::IdleLatency {
                     if let Some(meta) = meta_opt {
-                        event_tx.send(TestEvent::MetaInfo { meta }).await.ok();
+                        let _ = event_tx.send(TestEvent::MetaInfo { meta });
                         meta_sent = true;
                     }
                 }
 
-                event_tx
-                    .send(TestEvent::LatencySample {
-                        phase,
-                        during,
-                        rtt_ms: Some(ms),
-                        ok: true,
-                    })
-                    .await
-                    .ok();
+                let _ = event_tx.send(TestEvent::LatencySample {
+                    phase,
+                    during,
+                    rtt_ms: Some(ms),
+                    ok: true,
+                });
             }
             Err(_) => {
-                event_tx
-                    .send(TestEvent::LatencySample {
-                        phase,
-                        during,
-                        rtt_ms: None,
-                        ok: false,
-                    })
-                    .await
-                    .ok();
+                let _ = event_tx.send(TestEvent::LatencySample {
+                    phase,
+                    during,
+                    rtt_ms: None,
+                    ok: false,
+                });
             }
         }