From 66abb792ce8249519b24b05bb54dd9965d1d5aed Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 5 Oct 2026 20:48:06 +0700 Subject: [PATCH 1/3] feat(client): adopt an asynchronous Tokio runtime for the client frontend --- crates/irtt-app/Cargo.toml | 4 +- crates/irtt-app/src/cmd/client/mod.rs | 1 + crates/irtt-app/src/cmd/client/run.rs | 464 ++++++------------- crates/irtt-app/src/cmd/client/worker.rs | 133 ++++++ crates/irtt-app/src/entry.rs | 19 +- crates/irtt-app/src/shared/client/session.rs | 8 +- crates/irtt-app/src/signal.rs | 12 + crates/irtt-app/tests/cli.rs | 45 +- 8 files changed, 355 insertions(+), 331 deletions(-) create mode 100644 crates/irtt-app/src/cmd/client/worker.rs diff --git a/crates/irtt-app/Cargo.toml b/crates/irtt-app/Cargo.toml index 64bd9c7..10552d2 100644 --- a/crates/irtt-app/Cargo.toml +++ b/crates/irtt-app/Cargo.toml @@ -42,7 +42,7 @@ doc = false [features] default = ["client", "server", "tui"] -client = ["dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats"] +client = ["dep:tokio", "dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats"] server = ["dep:irtt-server", "dep:tokio"] tui =["dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats", "dep:ratatui", "dep:crossterm"] @@ -55,7 +55,7 @@ irtt-proto = { workspace = true, optional = true } irtt-server = { workspace = true, optional = true } irtt-stats = { workspace = true, optional = true } ratatui = { version = "0.30.2", optional = true } -tokio = { version = "1.53.2", default-features = false, features = ["rt", "time"], optional = true } +tokio = { version = "1.53.2", default-features = false, features = ["rt", "time", "sync", "macros"], optional = true } [dev-dependencies] irtt-server.workspace = true diff --git a/crates/irtt-app/src/cmd/client/mod.rs b/crates/irtt-app/src/cmd/client/mod.rs index 44bbb87..8de33e2 100644 --- a/crates/irtt-app/src/cmd/client/mod.rs +++ b/crates/irtt-app/src/cmd/client/mod.rs @@ -2,6 +2,7 @@ pub mod args; pub mod output; pub mod run; pub mod summary; +mod worker; pub use args::{ClientArgs, HeaderMode, OutputFormat}; pub use run::run_stream; diff --git a/crates/irtt-app/src/cmd/client/run.rs b/crates/irtt-app/src/cmd/client/run.rs index f6ac03c..c9583c1 100644 --- a/crates/irtt-app/src/cmd/client/run.rs +++ b/crates/irtt-app/src/cmd/client/run.rs @@ -1,16 +1,15 @@ use std::{ collections::{BTreeMap, BTreeSet, HashSet, VecDeque}, io::{self, BufRead, Write}, - sync::{atomic::AtomicBool, Arc, Mutex}, + sync::Arc, thread, - time::Duration, }; use irtt_client::{ managed::{ - BlockingManagedClient, ManagedClientHandle, ManagedCommandApplyError, ManagedEndReason, - ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, ManagedStatus, - ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance, + ManagedClient, ManagedCommandApplyError, ManagedEndReason, ManagedEvent, + ManagedEventSubscription, ManagedEventTryRecvError, ManagedStatus, ManagedTargetConfig, + ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance, }, ClientEvent, }; @@ -18,13 +17,14 @@ use irtt_client::{ use super::{ args::ClientArgs, output::{EventRenderStats, OutputConfig}, + worker::ManagedWorker, }; use crate::shared::client::{ - expected_probe_count, is_shutdown_requested, parse_stdin_target_set, + expected_probe_count, parse_stdin_target_set, session::{ - drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close, - request_managed_stop_once, should_print_final_summary, ManagedDrainState, + peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once, + should_print_final_summary, }, ManagedRunSetup, STDIN_MAX_DESIRED_TARGETS, STDIN_OUTCOME_HISTORY_LIMIT, }; @@ -40,12 +40,11 @@ const FINITE_STATS_MEMORY_WARNING_BYTES: u64 = 128 * MIB; const FINITE_STATS_MEMORY_STRONG_WARNING_BYTES: u64 = 512 * MIB; const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB; -const MANAGED_EVENT_WAIT_SLICE: Duration = Duration::from_millis(20); const MAX_STDIN_RECORD_BYTES: usize = 64 * 1024; -pub fn run_stream( +pub async fn run_stream( args: ClientArgs, - shutdown_requested: &AtomicBool, + mut shutdown: tokio::sync::watch::Receiver, ) -> Result<(), Box> { if args.list_columns { print!("{}", OutputConfig::list_columns()); @@ -63,7 +62,7 @@ pub fn run_stream( ) .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; if setup.stdin_controlled { - return run_stdin_stream(setup, output_config, shutdown_requested); + return run_stdin_stream(setup, output_config, shutdown).await; } let continuous = args.is_continuous(); let target_count = setup.target_count(); @@ -71,14 +70,13 @@ pub fn run_stream( if let Some(warning) = finite_stats_memory_warning(&args, target_count) { eprintln!("{warning}"); } - if is_shutdown_requested(shutdown_requested) { + if *shutdown.borrow() { return Ok(()); } - let (owner, mut events) = BlockingManagedClient::start_with_subscription( - setup.managed_config(), - setup.managed_targets(), - )?; - let handle = owner.handle(); + let (task, handle) = ManagedClient::task(setup.managed_config(), setup.managed_targets())?; + let mut events = handle.subscribe()?; + let mut status = handle.subscribe_status(); + let owner = ManagedWorker::start(task, handle.clone())?; let mut stdout = io::LineWriter::new(io::stdout().lock()); let mut stream_output = StreamOutput { config: output_config, @@ -102,9 +100,9 @@ pub fn run_stream( let mut dropped_events = 0_u64; let mut interrupted = false; let mut stop_requested = false; - let mut subscription_closed = false; loop { - if is_shutdown_requested(shutdown_requested) { + let snapshot = status.borrow_and_update().clone(); + if *shutdown.borrow() { interrupted = true; if request_managed_stop_once(&mut stop_requested) { drop(handle.stop()); @@ -113,38 +111,33 @@ pub fn run_stream( if request_managed_stop_for_peer_close( continuous, interrupted, - handle.status().peer_closed_target_outcomes, + snapshot.peer_closed_target_outcomes, &mut stop_requested, ) { drop(handle.stop()); } - let drain_state = drain_events( - &mut events, - &mut stream_output, - &mut stats, - &mut terminal_targets, - &mut dropped_events, - )?; - match drain_state { - ManagedDrainState::Empty => thread::sleep(MANAGED_EVENT_WAIT_SLICE), - ManagedDrainState::Closed => subscription_closed = true, - ManagedDrainState::BudgetExhausted => {} - } - if handle.status().final_outcome.is_some() || subscription_closed { + if snapshot.final_outcome.is_some() { break; } + tokio::select! { + _ = shutdown.changed(), if !interrupted => {} + changed = status.changed() => { if changed.is_err() { break; } } + event = events.recv() => match event { + Ok(event) => process_event(event, &mut stream_output, &mut stats, &mut terminal_targets)?, + Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { + dropped_events = dropped_events.saturating_add(count); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } } if interrupted { eprintln!("interrupted, closing managed run..."); } - let outcome = owner.join()?; - drain_final_events( - &mut events, - &mut stream_output, - &mut stats, - &mut terminal_targets, - &mut dropped_events, - )?; + let outcome = owner.join().await?; + drain_final_events(&mut events, &mut dropped_events, |event| { + process_event(event, &mut stream_output, &mut stats, &mut terminal_targets) + })?; if let Some(warning) = dropped_event_warning(dropped_events) { eprintln!("{warning}"); } @@ -159,7 +152,7 @@ pub fn run_stream( report_target_failure(target); } } - interrupted |= is_shutdown_requested(shutdown_requested); + interrupted |= *shutdown.borrow(); let terminal_error = match &outcome.end_reason { ManagedEndReason::DriverFailed(failure) => { Some(format!("managed driver failed: {failure}")) @@ -201,98 +194,9 @@ pub fn run_stream( } #[derive(Clone)] -struct StdinTargetSet { - revision: u64, - targets: Vec, -} - -#[derive(Default)] -struct StdinMailboxState { - next_revision: u64, - latest: Option, - eof: bool, - fatal: Option, -} - -#[derive(Default)] -struct StdinMailbox { - state: Mutex, -} - -impl StdinMailbox { - fn publish(&self, targets: Vec) { - let mut state = self - .state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - state.next_revision = state.next_revision.saturating_add(1); - state.latest = Some(StdinTargetSet { - revision: state.next_revision, - targets, - }); - } - - fn finish(&self) { - self.state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .eof = true; - } - - fn fail(&self, error: String) { - let mut state = self - .state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - state.fatal = Some(error); - state.latest = None; - } - - fn latest(&self) -> Option { - self.state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .latest - .clone() - } - - fn clear_if_current(&self, revision: u64) { - let mut state = self - .state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if state - .latest - .as_ref() - .is_some_and(|set| set.revision == revision) - { - state.latest = None; - } - } - - fn is_current(&self, revision: u64) -> bool { - self.state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .latest - .as_ref() - .is_some_and(|set| set.revision == revision) - } - - fn fatal(&self) -> Option { - self.state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .fatal - .clone() - } - - fn is_finished(&self) -> bool { - self.state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .eof - } +enum StdinUpdate { + Targets(Vec), + Stop(StdinStop), } fn read_stdin_record( @@ -371,7 +275,10 @@ fn read_stdin_record( } } -fn read_stdin_target_sets(reader: &mut R, mailbox: &StdinMailbox) { +fn read_stdin_target_sets( + reader: &mut R, + updates: &tokio::sync::watch::Sender, +) { let mut line = 0_u64; let mut record = Vec::with_capacity(MAX_STDIN_RECORD_BYTES + 1); loop { @@ -383,101 +290,43 @@ fn read_stdin_target_sets(reader: &mut R, mailbox: &StdinMailbox) { } match parse_stdin_target_set(&record, STDIN_MAX_DESIRED_TARGETS) { Ok(targets) => { - mailbox.publish(targets.into_iter().map(|target| target.managed).collect()) + updates.send_replace(StdinUpdate::Targets( + targets.into_iter().map(|target| target.managed).collect(), + )); } Err(error) => { - mailbox.fail(format!("invalid --targets-stdin line {line}: {error}")); + updates.send_replace(StdinUpdate::Stop(StdinStop::Fatal(format!( + "invalid --targets-stdin line {line}: {error}" + )))); return; } } } Ok(None) => { - mailbox.finish(); + updates.send_replace(StdinUpdate::Stop(StdinStop::Eof)); return; } Err(_) => { - mailbox.fail(format!("failed to read --targets-stdin line {}", line + 1)); + updates.send_replace(StdinUpdate::Stop(StdinStop::Fatal(format!( + "failed to read --targets-stdin line {}", + line + 1 + )))); return; } } } } -fn spawn_stdin_target_reader(mailbox: Arc) -> io::Result<()> { +fn spawn_stdin_target_reader(updates: tokio::sync::watch::Sender) -> io::Result<()> { thread::Builder::new() .name("irtt-targets-stdin".to_owned()) .spawn(move || { let stdin = io::stdin(); - read_stdin_target_sets(&mut stdin.lock(), &mailbox); + read_stdin_target_sets(&mut stdin.lock(), &updates); }) .map(|_| ()) } -#[derive(Clone)] -struct RetryAfterStatus { - revision: u64, - status: Arc, -} - -fn retry_waits_for_status_change( - retry_after: &mut Option, - target_set: &StdinTargetSet, - status: &Arc, -) -> bool { - let Some(retry) = retry_after else { - return false; - }; - if retry.revision != target_set.revision { - *retry_after = None; - return false; - } - Arc::ptr_eq(&retry.status, status) -} - -fn retry_after_capacity_rejection( - mailbox: &StdinMailbox, - target_set: &StdinTargetSet, - status: Arc, -) -> Option { - mailbox - .is_current(target_set.revision) - .then_some(RetryAfterStatus { - revision: target_set.revision, - status, - }) -} - -fn apply_latest_stdin_target_set( - handle: &ManagedClientHandle, - mailbox: &StdinMailbox, - retry_after: &mut Option, -) -> Result<(), String> { - let Some(target_set) = mailbox.latest() else { - *retry_after = None; - return Ok(()); - }; - let status = handle.status(); - if retry_waits_for_status_change(retry_after, &target_set, &status) { - return Ok(()); - } - - let receipt = handle - .update_targets(target_set.targets.clone()) - .map_err(|_| "failed to submit --targets-stdin update".to_owned())?; - match receipt.blocking_wait() { - Ok(_) => { - mailbox.clear_if_current(target_set.revision); - *retry_after = None; - Ok(()) - } - Err(ManagedCommandApplyError::LiveGenerationLimitExceeded { .. }) => { - *retry_after = retry_after_capacity_rejection(mailbox, &target_set, status); - Ok(()) - } - Err(_) => Err("--targets-stdin update was rejected".to_owned()), - } -} - struct BoundedTargetSet { limit: usize, order: VecDeque, @@ -542,37 +391,6 @@ fn process_stdin_event( Ok(()) } -fn drain_stdin_events( - events: &mut ManagedEventSubscription, - stream_output: &mut StreamOutput<'_, W>, - stats: &mut BTreeMap, - terminal_targets: &mut BoundedTargetSet, - dropped_events: &mut u64, -) -> io::Result { - drain_managed_events(events, dropped_events, |event| { - process_stdin_event(event, stream_output, stats, terminal_targets) - }) -} - -fn drain_final_stdin_events( - events: &mut ManagedEventSubscription, - stream_output: &mut StreamOutput<'_, W>, - stats: &mut BTreeMap, - terminal_targets: &mut BoundedTargetSet, - dropped_events: &mut u64, -) -> io::Result<()> { - loop { - match events.try_recv() { - Ok(event) => process_stdin_event(event, stream_output, stats, terminal_targets)?, - Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break, - Err(ManagedEventTryRecvError::Lagged(count)) => { - *dropped_events = dropped_events.saturating_add(count); - } - } - } - Ok(()) -} - fn reconcile_stdin_stats( stats: &mut BTreeMap, status: &ManagedStatus, @@ -613,36 +431,20 @@ enum StdinStop { Fatal(String), } -fn stdin_stop_request( - shutdown_requested: &AtomicBool, - mailbox: &StdinMailbox, -) -> Option { - if is_shutdown_requested(shutdown_requested) { - Some(StdinStop::Interrupted) - } else if let Some(error) = mailbox.fatal() { - Some(StdinStop::Fatal(error)) - } else if mailbox.is_finished() { - Some(StdinStop::Eof) - } else { - None - } -} - -fn run_stdin_stream( +async fn run_stdin_stream( setup: ManagedRunSetup, output_config: OutputConfig, - shutdown_requested: &AtomicBool, + mut shutdown: tokio::sync::watch::Receiver, ) -> Result<(), Box> { - if is_shutdown_requested(shutdown_requested) { + if *shutdown.borrow() { return Ok(()); } - let (owner, mut events) = BlockingManagedClient::start_with_subscription( - setup.managed_config(), - setup.managed_targets(), - )?; - let handle = owner.handle(); - let mailbox = Arc::new(StdinMailbox::default()); - spawn_stdin_target_reader(Arc::clone(&mailbox))?; + let (task, handle) = ManagedClient::task(setup.managed_config(), setup.managed_targets())?; + let mut events = handle.subscribe()?; + let mut status = handle.subscribe_status(); + let owner = ManagedWorker::start(task, handle.clone())?; + let (updates, mut stdin) = tokio::sync::watch::channel(StdinUpdate::Targets(Vec::new())); + spawn_stdin_target_reader(updates)?; let mut stdout = io::LineWriter::new(io::stdout().lock()); let mut stream_output = StreamOutput { @@ -654,59 +456,103 @@ fn run_stdin_stream( }; let mut stats = BTreeMap::new(); let mut terminal_targets = BoundedTargetSet::new(STDIN_OUTCOME_HISTORY_LIMIT); + let mut stdin_changed = false; + let mut desired = None; + let mut pending = None; + let mut submitted = Vec::new(); let mut retry_after = None; let mut dropped_events = 0_u64; let mut stop_requested = false; let mut stop = None; let mut summary_targets = None; - let mut subscription_closed = false; - loop { - let status = handle.status(); + let snapshot = status.borrow_and_update().clone(); if stop.is_none() { - if let Some(requested_stop) = stdin_stop_request(shutdown_requested, &mailbox) { - summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status)); - stop = Some(requested_stop); - if request_managed_stop_once(&mut stop_requested) { - drop(handle.stop()); + if *shutdown.borrow() { + stop = Some(StdinStop::Interrupted); + } else if stdin_changed || stdin.has_changed().unwrap_or(true) { + stdin_changed = false; + match stdin.borrow_and_update().clone() { + StdinUpdate::Targets(targets) => { + desired = Some(targets); + retry_after = None; + } + StdinUpdate::Stop(reason) => stop = Some(reason), } - } else if let Err(error) = - apply_latest_stdin_target_set(&handle, &mailbox, &mut retry_after) + } + if stop.is_none() + && pending.is_none() + && !retry_after + .as_ref() + .is_some_and(|previous| Arc::ptr_eq(previous, &snapshot)) { - summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status)); - stop = Some(StdinStop::Fatal(error)); + if let Some(targets) = desired.take() { + submitted = targets; + match handle.update_targets(submitted.clone()) { + Ok(receipt) => { + pending = Some(receipt); + retry_after = Some(snapshot.clone()); + } + Err(_) => { + stop = Some(StdinStop::Fatal( + "failed to submit --targets-stdin update".to_owned(), + )) + } + } + } + } + if stop.is_some() { + summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &snapshot)); if request_managed_stop_once(&mut stop_requested) { drop(handle.stop()); } } } - - let drain_state = drain_stdin_events( - &mut events, - &mut stream_output, - &mut stats, - &mut terminal_targets, - &mut dropped_events, - )?; - reconcile_stdin_stats(&mut stats, &handle.status(), summary_targets.as_ref()); - match drain_state { - ManagedDrainState::Empty => thread::sleep(MANAGED_EVENT_WAIT_SLICE), - ManagedDrainState::Closed => subscription_closed = true, - ManagedDrainState::BudgetExhausted => {} - } - if handle.status().final_outcome.is_some() || subscription_closed { + reconcile_stdin_stats(&mut stats, &snapshot, summary_targets.as_ref()); + if snapshot.final_outcome.is_some() { break; } + tokio::select! { + _ = shutdown.changed(), if stop.is_none() => {} + _ = stdin.changed(), if stop.is_none() => { stdin_changed = true; } + changed = status.changed() => { if changed.is_err() { break; } } + result = async { pending.as_mut().expect("guarded pending receipt").await }, if pending.is_some() && stop.is_none() => { + pending = None; + match result { + Ok(_) => retry_after = None, + Err(ManagedCommandApplyError::LiveGenerationLimitExceeded { .. }) => { + // A newer desired set supersedes this rejected set. Otherwise, + // retry only after the driver's status advances from submission. + if desired.is_none() { + desired = Some(std::mem::take(&mut submitted)); + } else { + retry_after = None; + } + } + Err(_) => { + stop = Some(StdinStop::Fatal("--targets-stdin update was rejected".to_owned())); + summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status.borrow())); + if request_managed_stop_once(&mut stop_requested) { drop(handle.stop()); } + } + } + } + event = events.recv() => match event { + Ok(event) => { + process_stdin_event(event, &mut stream_output, &mut stats, &mut terminal_targets)?; + reconcile_stdin_stats(&mut stats, &status.borrow(), summary_targets.as_ref()); + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { + dropped_events = dropped_events.saturating_add(count); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } } - let outcome = owner.join()?; - drain_final_stdin_events( - &mut events, - &mut stream_output, - &mut stats, - &mut terminal_targets, - &mut dropped_events, - )?; + let outcome = owner.join().await?; + drain_final_events(&mut events, &mut dropped_events, |event| { + process_stdin_event(event, &mut stream_output, &mut stats, &mut terminal_targets) + })?; if let Some(warning) = dropped_event_warning(dropped_events) { eprintln!("{warning}"); } @@ -782,28 +628,14 @@ fn process_event( Ok(()) } -fn drain_events( +fn drain_final_events( events: &mut ManagedEventSubscription, - stream_output: &mut StreamOutput<'_, W>, - stats: &mut BTreeMap, - terminal_targets: &mut HashSet, dropped_events: &mut u64, -) -> io::Result { - drain_managed_events(events, dropped_events, |event| { - process_event(event, stream_output, stats, terminal_targets) - }) -} - -fn drain_final_events( - events: &mut ManagedEventSubscription, - stream_output: &mut StreamOutput<'_, W>, - stats: &mut BTreeMap, - terminal_targets: &mut HashSet, - dropped_events: &mut u64, -) -> io::Result<()> { + mut process: impl FnMut(ManagedEvent) -> Result<(), E>, +) -> Result<(), E> { loop { match events.try_recv() { - Ok(event) => process_event(event, stream_output, stats, terminal_targets)?, + Ok(event) => process(event)?, Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break, Err(ManagedEventTryRecvError::Lagged(count)) => { *dropped_events = dropped_events.saturating_add(count); diff --git a/crates/irtt-app/src/cmd/client/worker.rs b/crates/irtt-app/src/cmd/client/worker.rs new file mode 100644 index 0000000..d9d646c --- /dev/null +++ b/crates/irtt-app/src/cmd/client/worker.rs @@ -0,0 +1,133 @@ +use std::{io, thread, time::Duration}; + +use irtt_client::managed::{ManagedClientHandle, ManagedClientTask, ManagedOutcome}; +use tokio::sync::oneshot; + +/// Keeps measurement scheduling separate from frontend output and statistics. +/// Drop also stops and joins on frontend errors or cancellation. +pub(super) struct ManagedWorker { + handle: ManagedClientHandle, + thread: Option>, + completion: oneshot::Receiver>, +} + +impl ManagedWorker { + pub(super) fn start(task: ManagedClientTask, handle: ManagedClientHandle) -> io::Result { + let (completed, completion) = oneshot::channel(); + let thread = thread::Builder::new() + .name("irtt-managed".to_owned()) + .spawn(move || { + // Construct and destroy the runtime on its owning thread, including + // startup failures; never drop a runtime inside the frontend executor. + let outcome = (|| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + runtime.block_on(task) + })); + runtime.shutdown_timeout(Duration::from_millis(100)); + match outcome { + Ok(outcome) => Ok(outcome), + Err(panic) => std::panic::resume_unwind(panic), + } + })(); + let _ = completed.send(outcome); + })?; + Ok(Self { + handle, + thread: Some(thread), + completion, + }) + } + + pub(super) async fn join(mut self) -> Result> { + let outcome = (&mut self.completion).await; + self.join_thread()?; + Ok(outcome + .map_err(|_| io::Error::other("managed worker completed without an outcome"))??) + } + + fn join_thread(&mut self) -> io::Result<()> { + self.thread + .take() + .expect("managed worker is joined once") + .join() + .map_err(|_| io::Error::other("managed worker thread panicked")) + } +} + +impl Drop for ManagedWorker { + fn drop(&mut self) { + if self.thread.is_some() { + drop(self.handle.stop()); + let _ = self.join_thread(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use irtt_client::managed::{ + ManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason, + }; + + // The frontend executor must be able to stop a worker while awaiting its + // completion; a blocking join here would deadlock this current-thread runtime. + #[test] + fn completion_leaves_frontend_free_to_stop_and_preserves_outcome() { + let (task, handle) = ManagedClient::task( + ManagedClientConfig { + completion: ManagedCompletionPolicy::ExplicitStop, + ..Default::default() + }, + vec![], + ) + .unwrap(); + let worker = ManagedWorker::start(task, handle.clone()).unwrap(); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let (outcome, ()) = tokio::time::timeout(Duration::from_secs(2), async { + tokio::join!(worker.join(), handle.stop()) + }) + .await + .expect("worker completion must allow frontend stop"); + let outcome = outcome.unwrap(); + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + assert_eq!(handle.status().final_outcome.as_deref(), Some(&outcome)); + }); + } + + // Cancelling the frontend's completion wait must stop and join the worker, + // rather than detaching an ongoing measurement task. + #[test] + fn cancelling_completion_stops_and_joins_worker() { + use std::{ + future::Future, + task::{Context, Waker}, + }; + let (task, handle) = ManagedClient::task( + ManagedClientConfig { + completion: ManagedCompletionPolicy::ExplicitStop, + ..Default::default() + }, + vec![], + ) + .unwrap(); + let worker = ManagedWorker::start(task, handle.clone()).unwrap(); + let mut completion = Box::pin(worker.join()); + assert!(completion + .as_mut() + .poll(&mut Context::from_waker(Waker::noop())) + .is_pending()); + drop(completion); + assert_eq!( + handle.status().final_outcome.as_ref().unwrap().end_reason, + ManagedEndReason::StopRequested + ); + } +} diff --git a/crates/irtt-app/src/entry.rs b/crates/irtt-app/src/entry.rs index 10fac74..6280c95 100644 --- a/crates/irtt-app/src/entry.rs +++ b/crates/irtt-app/src/entry.rs @@ -1,9 +1,9 @@ -#[cfg(any(feature = "client", feature = "server", feature = "tui"))] +#[cfg(any(feature = "server", feature = "tui"))] use std::sync::{atomic::AtomicBool, Arc}; use std::{env, ffi::OsString, process::ExitCode}; use crate::applet::{dispatch_from_argv, AppletDispatch, RequestedApplet}; -#[cfg(any(feature = "client", feature = "server", feature = "tui"))] +#[cfg(any(feature = "server", feature = "tui"))] use crate::signal::install_signal_handler; /// Entry point for the `irtt-rs` multicall dispatcher binary. @@ -75,9 +75,9 @@ fn run_dispatcher_from_env() -> Result<(), Box> { } } -/// Installs the shutdown signal handler shared by every applet and hands the +/// Installs the shutdown signal handler for synchronous applets and hands the /// resulting flag to `f`. -#[cfg(any(feature = "client", feature = "server", feature = "tui"))] +#[cfg(any(feature = "server", feature = "tui"))] fn with_shutdown_flag( f: impl FnOnce(&AtomicBool) -> Result<(), Box>, ) -> Result<(), Box> { @@ -91,10 +91,13 @@ fn with_shutdown_flag( fn run_client_applet(argv: Vec) -> Result<(), Box> { use clap::Parser; - with_shutdown_flag(|shutdown_requested| { - let args = crate::cmd::client::ClientArgs::parse_from(argv); - crate::cmd::client::run_stream(args, shutdown_requested) - }) + let shutdown = crate::signal::install_async_signal_handler() + .map_err(|err| format!("failed to install signal handler: {err}"))?; + let args = crate::cmd::client::ClientArgs::parse_from(argv); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()? + .block_on(crate::cmd::client::run_stream(args, shutdown)) } #[cfg(not(feature = "client"))] diff --git a/crates/irtt-app/src/shared/client/session.rs b/crates/irtt-app/src/shared/client/session.rs index 26cde72..5871771 100644 --- a/crates/irtt-app/src/shared/client/session.rs +++ b/crates/irtt-app/src/shared/client/session.rs @@ -1,12 +1,12 @@ use std::sync::atomic::{AtomicBool, Ordering}; -#[cfg(any(feature = "client", feature = "tui"))] +#[cfg(feature = "tui")] use irtt_client::managed::{ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError}; -#[cfg(any(feature = "client", feature = "tui"))] +#[cfg(feature = "tui")] pub(crate) const MANAGED_EVENT_WORK_BUDGET: usize = 128; -#[cfg(any(feature = "client", feature = "tui"))] +#[cfg(feature = "tui")] #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum ManagedDrainState { Empty, @@ -16,7 +16,7 @@ pub(crate) enum ManagedDrainState { /// Drain a bounded amount of lossy presentation work before returning to the /// frontend control loop. -#[cfg(any(feature = "client", feature = "tui"))] +#[cfg(feature = "tui")] pub(crate) fn drain_managed_events( events: &mut ManagedEventSubscription, dropped_events: &mut u64, diff --git a/crates/irtt-app/src/signal.rs b/crates/irtt-app/src/signal.rs index 9bc6249..8c4e0a2 100644 --- a/crates/irtt-app/src/signal.rs +++ b/crates/irtt-app/src/signal.rs @@ -1,10 +1,22 @@ +#[cfg(any(feature = "server", feature = "tui"))] use std::sync::{ atomic::{AtomicBool, Ordering}, Arc, }; +#[cfg(any(feature = "server", feature = "tui"))] pub fn install_signal_handler(shutdown_requested: Arc) -> Result<(), ctrlc::Error> { ctrlc::set_handler(move || { shutdown_requested.store(true, Ordering::Relaxed); }) } + +/// Preserve ctrlc's SIGINT/SIGTERM policy while waking the async client. +#[cfg(feature = "client")] +pub fn install_async_signal_handler() -> Result, ctrlc::Error> { + let (shutdown, receiver) = tokio::sync::watch::channel(false); + ctrlc::set_handler(move || { + shutdown.send_replace(true); + })?; + Ok(receiver) +} diff --git a/crates/irtt-app/tests/cli.rs b/crates/irtt-app/tests/cli.rs index a66f492..fb453f7 100644 --- a/crates/irtt-app/tests/cli.rs +++ b/crates/irtt-app/tests/cli.rs @@ -442,7 +442,7 @@ fn targets_stdin_supersedes_the_active_set_and_stops_on_eof() { let mut stdin = client.0.stdin.take().unwrap(); let mut rows = Vec::new(); for target in ["first", "second"] { - writeln!(stdin, "{target}={}", server.addr).unwrap(); + write!(stdin, "\r\n{target}={}\r\n", server.addr).unwrap(); stdin.flush().unwrap(); let deadline = Instant::now() + Duration::from_secs(5); loop { @@ -485,3 +485,46 @@ fn targets_stdin_supersedes_the_active_set_and_stops_on_eof() { .any(|row| row.contains("\"target\":\"second\"") && row.contains("\"event\":\"session_closed\""))); } + +// Record bounds and decoding errors must terminate the real stdin frontend, +// even when no managed events are available to wake it. +#[cfg(feature = "client")] +#[test] +fn targets_stdin_invalid_records_are_fatal_and_empty_eof_is_normal() { + use std::{ + io::{Read, Write}, + process::Stdio, + }; + for (record, success, diagnostic) in [ + (Vec::new(), true, ""), + (b"\r\n\n".to_vec(), true, ""), + (b"\n=127.0.0.1:9\n".to_vec(), false, "line 2"), + (vec![0xff, b'\n'], false, "line 1"), + (vec![b'x'; 64 * 1024 + 2], false, "line 1"), + ] { + let mut client = ClientProcess( + Command::new(env!("CARGO_BIN_EXE_irtt-client")) + .env_clear() + .args(["--duration", "0", "--targets-stdin"]) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ); + let mut stdin = client.0.stdin.take().unwrap(); + stdin.write_all(&record).unwrap(); + drop(stdin); + let status = client.wait(); + let mut stderr = String::new(); + client + .0 + .stderr + .take() + .unwrap() + .read_to_string(&mut stderr) + .unwrap(); + assert_eq!(status.success(), success, "{stderr}"); + assert!(stderr.contains(diagnostic), "{stderr}"); + } +} From 5067ceeb1c22b516952b56dd7834202c0d8ed928 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 5 Oct 2026 21:49:36 +0700 Subject: [PATCH 2/3] feat(tui): adopt an asynchronous Tokio runtime for the TUI frontend Refactor TUI event handling and rendering to improve responsiveness and use a unified managed client worker. --- Cargo.lock | 3 + crates/irtt-app/Cargo.toml | 8 +- crates/irtt-app/src/cmd/client/mod.rs | 1 - crates/irtt-app/src/cmd/client/run.rs | 28 +- crates/irtt-app/src/cmd/tui/run.rs | 336 ++++++++---------- crates/irtt-app/src/cmd/tui/ui.rs | 6 - crates/irtt-app/src/entry.rs | 17 +- crates/irtt-app/src/shared/client/mod.rs | 2 +- crates/irtt-app/src/shared/client/session.rs | 32 +- .../src/{cmd => shared}/client/worker.rs | 6 +- crates/irtt-app/src/signal.rs | 8 +- crates/irtt-app/tests/tui.rs | 327 +++++++++++++++++ 12 files changed, 508 insertions(+), 266 deletions(-) rename crates/irtt-app/src/{cmd => shared}/client/worker.rs (96%) create mode 100644 crates/irtt-app/tests/tui.rs diff --git a/Cargo.lock b/Cargo.lock index 063921f..b2f98bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -317,6 +317,7 @@ dependencies = [ "crossterm_winapi", "derive_more", "document-features", + "futures-core", "mio", "parking_lot", "rustix", @@ -753,10 +754,12 @@ dependencies = [ "clap", "crossterm", "ctrlc", + "futures-core", "irtt-client", "irtt-proto", "irtt-server", "irtt-stats", + "nix 0.31.3", "ratatui", "tokio", ] diff --git a/crates/irtt-app/Cargo.toml b/crates/irtt-app/Cargo.toml index 10552d2..1af6ef6 100644 --- a/crates/irtt-app/Cargo.toml +++ b/crates/irtt-app/Cargo.toml @@ -44,11 +44,12 @@ doc = false default = ["client", "server", "tui"] client = ["dep:tokio", "dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats"] server = ["dep:irtt-server", "dep:tokio"] -tui =["dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats", "dep:ratatui", "dep:crossterm"] +tui =["dep:tokio", "dep:futures-core", "dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats", "dep:ratatui", "dep:crossterm"] [dependencies] clap = { version = "4.6.7", features = ["derive", "env"] } -crossterm = { version = "0.29.0", optional = true } +crossterm = { version = "0.29.0", features = ["event-stream"], optional = true } +futures-core = { version = "0.3.34", default-features = false, optional = true } ctrlc = { version = "3.5.2", features = ["termination"] } irtt-client = { workspace = true, optional = true } irtt-proto = { workspace = true, optional = true } @@ -61,6 +62,9 @@ tokio = { version = "1.53.2", default-features = false, features = ["rt", "time" irtt-server.workspace = true tokio = { version = "1.53.2", default-features = false, features = ["rt", "sync"] } +[target.'cfg(unix)'.dev-dependencies] +nix = { version = "0.31.3", features = ["term", "poll", "signal", "process"] } + [profile.release] opt-level = "z" lto = true diff --git a/crates/irtt-app/src/cmd/client/mod.rs b/crates/irtt-app/src/cmd/client/mod.rs index 8de33e2..44bbb87 100644 --- a/crates/irtt-app/src/cmd/client/mod.rs +++ b/crates/irtt-app/src/cmd/client/mod.rs @@ -2,7 +2,6 @@ pub mod args; pub mod output; pub mod run; pub mod summary; -mod worker; pub use args::{ClientArgs, HeaderMode, OutputFormat}; pub use run::run_stream; diff --git a/crates/irtt-app/src/cmd/client/run.rs b/crates/irtt-app/src/cmd/client/run.rs index c9583c1..2f24e35 100644 --- a/crates/irtt-app/src/cmd/client/run.rs +++ b/crates/irtt-app/src/cmd/client/run.rs @@ -7,9 +7,8 @@ use std::{ use irtt_client::{ managed::{ - ManagedClient, ManagedCommandApplyError, ManagedEndReason, ManagedEvent, - ManagedEventSubscription, ManagedEventTryRecvError, ManagedStatus, ManagedTargetConfig, - ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance, + ManagedClient, ManagedCommandApplyError, ManagedEndReason, ManagedEvent, ManagedStatus, + ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance, }, ClientEvent, }; @@ -17,15 +16,15 @@ use irtt_client::{ use super::{ args::ClientArgs, output::{EventRenderStats, OutputConfig}, - worker::ManagedWorker, }; use crate::shared::client::{ expected_probe_count, parse_stdin_target_set, session::{ - peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once, - should_print_final_summary, + drain_final_events, peer_close_run_error, request_managed_stop_for_peer_close, + request_managed_stop_once, should_print_final_summary, }, + worker::ManagedWorker, ManagedRunSetup, STDIN_MAX_DESIRED_TARGETS, STDIN_OUTCOME_HISTORY_LIMIT, }; @@ -628,23 +627,6 @@ fn process_event( Ok(()) } -fn drain_final_events( - events: &mut ManagedEventSubscription, - dropped_events: &mut u64, - mut process: impl FnMut(ManagedEvent) -> Result<(), E>, -) -> Result<(), E> { - loop { - match events.try_recv() { - Ok(event) => process(event)?, - Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break, - Err(ManagedEventTryRecvError::Lagged(count)) => { - *dropped_events = dropped_events.saturating_add(count); - } - } - } - Ok(()) -} - fn report_target_failure(target: &ManagedTargetOutcome) { for message in target_failure_messages(target) { eprintln!("{message}"); diff --git a/crates/irtt-app/src/cmd/tui/run.rs b/crates/irtt-app/src/cmd/tui/run.rs index d7dd6b7..01d7384 100644 --- a/crates/irtt-app/src/cmd/tui/run.rs +++ b/crates/irtt-app/src/cmd/tui/run.rs @@ -1,133 +1,168 @@ -use std::{ - collections::HashSet, - io, - sync::atomic::{AtomicBool, Ordering}, - thread, - time::{Duration, Instant}, -}; +use std::{collections::HashSet, future::poll_fn, io, time::Duration}; -use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal}; +use super::ui::{TuiConfig, TuiState, TuiStatus, TuiTerminal}; use crate::{ cmd::tui::args::TuiArgs, shared::client::{ - is_shutdown_requested, session::{ - drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close, - request_managed_stop_once, ManagedDrainState, + drain_final_events, peer_close_run_error, request_managed_stop_for_peer_close, + request_managed_stop_once, }, + worker::ManagedWorker, }, }; use crossterm::{ - event::{self, Event, KeyCode, KeyEventKind, KeyModifiers}, + event::{Event, EventStream, KeyCode, KeyEventKind, KeyModifiers}, terminal, }; -use irtt_client::managed::{ - BlockingManagedClient, ManagedEndReason, ManagedEvent, ManagedEventSubscription, - ManagedEventTryRecvError, TargetInstance, +use futures_core::Stream; +use irtt_client::{ + managed::{ManagedClient, ManagedEndReason, ManagedEvent, TargetInstance}, + ClientEvent, }; use ratatui::layout::Rect; +use tokio::{ + sync::{broadcast::error::RecvError, watch}, + time::{sleep, Instant}, +}; const RENDER_INTERVAL: Duration = Duration::from_millis(250); -const TUI_WAIT_SLICE: Duration = Duration::from_millis(20); -const INPUT_EVENT_WORK_BUDGET: usize = 128; -pub fn run_tui( +pub async fn run_tui( args: TuiArgs, - shutdown_requested: &AtomicBool, + mut shutdown: watch::Receiver, ) -> Result<(), Box> { let setup = args .prepare() .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; let continuous = args.is_continuous(); + if *shutdown.borrow() { + return Ok(()); + } let mut terminal = TuiTerminal::enter()?; let mut state = TuiState::with_target_labels( TuiConfig::from_args(&args, &setup), setup.targets.iter().map(|target| target.label.clone()), ); - let mut next_render = Instant::now(); - if is_shutdown_requested(shutdown_requested) { - return Ok(()); - } state.set_status(TuiStatus::Opening); - render_if_due(&mut terminal, &state, &mut next_render, true)?; - let (owner, mut events) = match BlockingManagedClient::start_with_subscription( - setup.managed_config(), - setup.managed_targets(), - ) { + terminal.draw(&state)?; + let (task, handle) = match ManagedClient::task(setup.managed_config(), setup.managed_targets()) + { Ok(value) => value, Err(error) => { state.set_error(error.to_string()); - render_if_due(&mut terminal, &state, &mut next_render, true)?; + terminal.draw(&state)?; return Err(Box::new(error)); } }; - let handle = owner.handle(); + // Both subscriptions exist before the measurement task can be polled. + let mut events = handle.subscribe()?; + let mut status = handle.subscribe_status(); + let worker = match ManagedWorker::start(task, handle.clone()) { + Ok(worker) => worker, + Err(error) => { + state.set_run_error(error.to_string()); + terminal.draw(&state)?; + return Err(error.into()); + } + }; + let completion = worker.join(); + tokio::pin!(completion); + // Input and rendering stay on the frontend executor. EventStream's drop + // wakes its reader; no terminal task can outlive the terminal guard. + let mut input = EventStream::new(); + let render = sleep(RENDER_INTERVAL); + tokio::pin!(render); let mut interrupted = false; let mut stop_requested = false; let mut terminal_targets = HashSet::new(); - let mut dropped_events = 0; - let mut subscription_closed = false; - loop { - if is_shutdown_requested(shutdown_requested) { - interrupted = true; - if request_managed_stop_once(&mut stop_requested) { - drop(handle.stop()); + let mut dropped_events = 0_u64; + let mut events_closed = false; + let mut status_closed = false; + let outcome = loop { + let mut force_render = false; + // Default select fairness lets input, status and rendering progress + // even when measurement events are continuously ready. + tokio::select! { + outcome = &mut completion => break outcome, + _ = shutdown.changed(), if !interrupted => { + interrupted = true; + state.set_status(TuiStatus::Interrupted); + force_render = true; } - } - if handle_input(&mut state, shutdown_requested)? { - render_if_due(&mut terminal, &state, &mut next_render, true)?; - } - if state.quit_requested { - interrupted = true; - if request_managed_stop_once(&mut stop_requested) { - drop(handle.stop()); + changed = status.changed(), if !status_closed => { + status_closed = changed.is_err(); + let snapshot = status.borrow_and_update().clone(); + if request_managed_stop_for_peer_close( + continuous, interrupted, snapshot.peer_closed_target_outcomes, + &mut stop_requested, + ) { + drop(handle.stop()); + } + force_render = true; } + event = events.recv(), if !events_closed => match event { + Ok(event) => force_render = process_tui_event( + event, &mut state, &mut terminal_targets, + ), + Err(RecvError::Lagged(count)) => { + dropped_events = dropped_events.saturating_add(count); + state.mark_dropped_managed_events(dropped_events); + force_render = true; + } + Err(RecvError::Closed) => events_closed = true, + }, + event = poll_fn(|cx| std::pin::Pin::new(&mut input).poll_next(cx)) => { + let event = event.ok_or_else(|| io::Error::new( + io::ErrorKind::UnexpectedEof, "terminal input closed", + ))??; + match event { + Event::Resize(_, _) => force_render = true, + Event::Key(key) if key.kind != KeyEventKind::Release => { + match key.code { + KeyCode::Char('q') | KeyCode::Char('c') + if key.code == KeyCode::Char('q') + || key.modifiers.contains(KeyModifiers::CONTROL) => { + interrupted = true; + state.set_status(TuiStatus::Interrupted); + force_render = true; + } + _ => { + let (width, height) = terminal::size()?; + force_render = state.handle_key( + key.code, Rect::new(0, 0, width, height), + ); + } + } + } + _ => {} + } + } + _ = &mut render, if !state.paused => force_render = true, } - if request_managed_stop_for_peer_close( - continuous, - interrupted, - handle.status().peer_closed_target_outcomes, - &mut stop_requested, - ) { + if interrupted && request_managed_stop_once(&mut stop_requested) { drop(handle.stop()); } - let previous_dropped = dropped_events; - let drain_state = drain_tui_events( - &mut events, - &mut state, - &mut terminal_targets, - &mut dropped_events, - ); - if dropped_events != previous_dropped { - // Data loss changes the meaning of displayed statistics, even while - // ordinary measurement refreshes are paused. - render_if_due(&mut terminal, &state, &mut next_render, true)?; + if force_render { + terminal.draw(&state)?; + // Schedule from the completed draw, never catch up missed frames. + render.as_mut().reset(Instant::now() + RENDER_INTERVAL); } - match drain_state { - ManagedDrainState::Empty => { - thread::sleep(managed_tui_wait_duration(&next_render, state.paused)); - } - ManagedDrainState::Closed => subscription_closed = true, - ManagedDrainState::BudgetExhausted => {} - } - if handle.status().final_outcome.is_some() || subscription_closed { - break; - } - render_if_due(&mut terminal, &state, &mut next_render, false)?; - } - if interrupted { - state.set_status(TuiStatus::Interrupted); - render_if_due(&mut terminal, &state, &mut next_render, true)?; - } + }; state.set_status(TuiStatus::Closing); - render_if_due(&mut terminal, &state, &mut next_render, true)?; - let outcome = owner.join()?; - drain_final_tui_events( - &mut events, - &mut state, - &mut terminal_targets, - &mut dropped_events, - ); + terminal.draw(&state)?; + let outcome = match outcome { + Ok(outcome) => outcome, + Err(error) => { + state.set_run_error(error.to_string()); + terminal.draw(&state)?; + return Err(error); + } + }; + drain_final_events(&mut events, &mut dropped_events, |event| { + process_tui_event(event, &mut state, &mut terminal_targets); + Ok::<(), std::convert::Infallible>(()) + })?; state.mark_dropped_managed_events(dropped_events); if outcome.discarded_target_outcomes != 0 { state.set_run_error(format!( @@ -140,7 +175,7 @@ pub fn run_tui( state.process_target_outcome(target); } } - interrupted |= is_shutdown_requested(shutdown_requested); + interrupted |= *shutdown.borrow(); let error = match &outcome.end_reason { ManagedEndReason::DriverFailed(failure) => { Some(format!("managed driver failed: {failure}")) @@ -160,11 +195,11 @@ pub fn run_tui( }; if let Some(error) = error { state.set_run_error(error.clone()); - render_if_due(&mut terminal, &state, &mut next_render, true)?; + terminal.draw(&state)?; return Err(error.into()); } state.set_status(TuiStatus::Complete); - render_if_due(&mut terminal, &state, &mut next_render, true)?; + terminal.draw(&state)?; Ok(()) } @@ -172,117 +207,36 @@ fn process_tui_event( event: ManagedEvent, state: &mut TuiState, terminal_targets: &mut HashSet, -) { +) -> bool { match event { - ManagedEvent::Client { target, event } => state.process_target_event(&target, &event), + ManagedEvent::Client { target, event } => { + let force_render = matches!( + event, + ClientEvent::SessionStarted(_) + | ClientEvent::NoTestCompleted(_) + | ClientEvent::SessionClosed { .. } + | ClientEvent::Warning { .. } + ); + state.process_target_event(&target, &event); + force_render + } ManagedEvent::TargetFinished { outcome } => { terminal_targets.insert(outcome.target.clone()); state.process_target_outcome(&outcome); + true } - _ => {} + _ => false, } } -fn drain_tui_events( - events: &mut ManagedEventSubscription, - state: &mut TuiState, - terminal_targets: &mut HashSet, - dropped_events: &mut u64, -) -> ManagedDrainState { - let previous_dropped = *dropped_events; - let drained = drain_managed_events(events, dropped_events, |event| { - process_tui_event(event, state, terminal_targets); - Ok::<(), std::convert::Infallible>(()) - }) - .expect("processing TUI managed events is infallible"); - if *dropped_events != previous_dropped { - state.mark_dropped_managed_events(*dropped_events); - } - drained -} - -fn drain_final_tui_events( - events: &mut ManagedEventSubscription, - state: &mut TuiState, - terminal_targets: &mut HashSet, - dropped_events: &mut u64, -) { - loop { - match events.try_recv() { - Ok(event) => process_tui_event(event, state, terminal_targets), - Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break, - Err(ManagedEventTryRecvError::Lagged(count)) => { - *dropped_events = dropped_events.saturating_add(count); - } - } - } -} -fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Result { - let mut force_render = false; - for _ in 0..INPUT_EVENT_WORK_BUDGET { - if !event::poll(Duration::ZERO)? { - break; - } - let key = match event::read()? { - Event::Key(key) => key, - Event::Resize(_, _) => { - force_render = true; - continue; - } - _ => continue, - }; - if key.kind == KeyEventKind::Release { - continue; - } - match key.code { - KeyCode::Char('q') | KeyCode::Char('c') - if key.code == KeyCode::Char('q') - || key.modifiers.contains(KeyModifiers::CONTROL) => - { - state.quit_requested = true; - shutdown_requested.store(true, Ordering::Relaxed); - force_render = true; - break; - } - _ => { - let (width, height) = terminal::size()?; - force_render |= state.handle_key(key.code, Rect::new(0, 0, width, height)); - } - } - } - Ok(force_render) -} -fn render_if_due( - terminal: &mut TuiTerminal, - state: &TuiState, - next_render: &mut Instant, - force: bool, -) -> io::Result<()> { - let now = Instant::now(); - if should_render(now, *next_render, state.paused, force) { - terminal.draw(state)?; - *next_render = now + RENDER_INTERVAL; - } - Ok(()) -} -fn managed_tui_wait_duration(next_render: &Instant, paused: bool) -> Duration { - let render_wait = if paused { - TUI_WAIT_SLICE - } else { - next_render.saturating_duration_since(Instant::now()) - }; - render_wait.min(TUI_WAIT_SLICE) -} #[cfg(test)] mod tests { use super::*; use ratatui::{backend::TestBackend, Terminal}; - #[test] - fn subscription_lag_keeps_incomplete_statistics_visible_while_paused() { + #[tokio::test(flavor = "current_thread")] + async fn subscription_lag_keeps_incomplete_statistics_visible_while_paused() { for paused in [false, true] { - // The production subscription is a broadcast receiver. Overflow the - // real channel rather than synthesizing a lag result or UI state. let (sender, mut events) = tokio::sync::broadcast::channel(1); sender.send(ManagedEvent::Started).unwrap(); sender.send(ManagedEvent::Started).unwrap(); @@ -290,14 +244,12 @@ mod tests { if paused { state.toggle_pause(); } - let mut dropped = 0; - let mut terminal_targets = HashSet::new(); - drain_tui_events(&mut events, &mut state, &mut terminal_targets, &mut dropped); - assert_eq!(dropped, 1); + let Err(RecvError::Lagged(dropped)) = events.recv().await else { + panic!("subscription must report real broadcast overflow"); + }; + state.mark_dropped_managed_events(dropped); let mut terminal = Terminal::new(TestBackend::new(200, 40)).unwrap(); for _ in 0..2 { - // Empty drains must not clear the warning on later redraws. - drain_tui_events(&mut events, &mut state, &mut terminal_targets, &mut dropped); terminal .draw(|frame| super::super::ui::draw_dashboard(frame, &state)) .unwrap(); diff --git a/crates/irtt-app/src/cmd/tui/ui.rs b/crates/irtt-app/src/cmd/tui/ui.rs index 088225d..4003a2d 100644 --- a/crates/irtt-app/src/cmd/tui/ui.rs +++ b/crates/irtt-app/src/cmd/tui/ui.rs @@ -106,7 +106,6 @@ pub(super) struct TuiState { details_open: bool, details_scroll: (u16, u16), pub(super) paused: bool, - pub(super) quit_requested: bool, } impl TuiState { @@ -143,7 +142,6 @@ impl TuiState { details_open: false, details_scroll: (0, 0), paused: false, - quit_requested: false, } } @@ -939,10 +937,6 @@ pub(super) fn draw_dashboard(frame: &mut Frame<'_>, state: &TuiState) { frame.render_widget(status_line(state), rows[3]); } -pub(super) fn should_render(now: Instant, next_render: Instant, paused: bool, force: bool) -> bool { - force || (!paused && now >= next_render) -} - fn dashboard_layout(area: Rect, state: &TuiState) -> [Rect; 4] { let footer = 3 + u16::from(state.last_warning.is_some()); let target_rows = state.targets.len().min(6) as u16; diff --git a/crates/irtt-app/src/entry.rs b/crates/irtt-app/src/entry.rs index 6280c95..6ec1f8a 100644 --- a/crates/irtt-app/src/entry.rs +++ b/crates/irtt-app/src/entry.rs @@ -1,9 +1,9 @@ -#[cfg(any(feature = "server", feature = "tui"))] +#[cfg(feature = "server")] use std::sync::{atomic::AtomicBool, Arc}; use std::{env, ffi::OsString, process::ExitCode}; use crate::applet::{dispatch_from_argv, AppletDispatch, RequestedApplet}; -#[cfg(any(feature = "server", feature = "tui"))] +#[cfg(feature = "server")] use crate::signal::install_signal_handler; /// Entry point for the `irtt-rs` multicall dispatcher binary. @@ -77,7 +77,7 @@ fn run_dispatcher_from_env() -> Result<(), Box> { /// Installs the shutdown signal handler for synchronous applets and hands the /// resulting flag to `f`. -#[cfg(any(feature = "server", feature = "tui"))] +#[cfg(feature = "server")] fn with_shutdown_flag( f: impl FnOnce(&AtomicBool) -> Result<(), Box>, ) -> Result<(), Box> { @@ -124,10 +124,13 @@ fn run_server_applet(_argv: Vec) -> Result<(), Box) -> Result<(), Box> { use clap::Parser; - with_shutdown_flag(|shutdown_requested| { - let args = crate::cmd::tui::TuiArgs::parse_from(argv); - crate::cmd::tui::run_tui(args, shutdown_requested) - }) + let shutdown = crate::signal::install_async_signal_handler() + .map_err(|err| format!("failed to install signal handler: {err}"))?; + let args = crate::cmd::tui::TuiArgs::parse_from(argv); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()? + .block_on(crate::cmd::tui::run_tui(args, shutdown)) } #[cfg(not(feature = "tui"))] diff --git a/crates/irtt-app/src/shared/client/mod.rs b/crates/irtt-app/src/shared/client/mod.rs index e72a847..be0b07f 100644 --- a/crates/irtt-app/src/shared/client/mod.rs +++ b/crates/irtt-app/src/shared/client/mod.rs @@ -3,6 +3,7 @@ pub mod config; pub mod prepare; pub mod session; pub mod targets; +pub(crate) mod worker; pub use args::{ parse_dscp, parse_duration, parse_length, parse_server_fill, parse_test_duration, parse_ttl, @@ -13,7 +14,6 @@ pub use prepare::{ prepare_managed_run, ManagedRunSetup, TargetSelection, MANAGED_EVENT_CAPACITY, STDIN_MAX_DESIRED_TARGETS, STDIN_MAX_LIVE_TARGET_GENERATIONS, STDIN_OUTCOME_HISTORY_LIMIT, }; -pub use session::is_shutdown_requested; pub use targets::{ parse_stdin_target_set, parse_target, prepare_managed_targets, target_specs, GroupPacingArg, PreparedTarget, TargetArg, TargetSpec, diff --git a/crates/irtt-app/src/shared/client/session.rs b/crates/irtt-app/src/shared/client/session.rs index 5871771..8aa9da7 100644 --- a/crates/irtt-app/src/shared/client/session.rs +++ b/crates/irtt-app/src/shared/client/session.rs @@ -1,42 +1,20 @@ -use std::sync::atomic::{AtomicBool, Ordering}; - -#[cfg(feature = "tui")] use irtt_client::managed::{ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError}; -#[cfg(feature = "tui")] -pub(crate) const MANAGED_EVENT_WORK_BUDGET: usize = 128; - -#[cfg(feature = "tui")] -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub(crate) enum ManagedDrainState { - Empty, - Closed, - BudgetExhausted, -} - -/// Drain a bounded amount of lossy presentation work before returning to the -/// frontend control loop. -#[cfg(feature = "tui")] -pub(crate) fn drain_managed_events( +pub(crate) fn drain_final_events( events: &mut ManagedEventSubscription, dropped_events: &mut u64, mut process: impl FnMut(ManagedEvent) -> Result<(), E>, -) -> Result { - for _ in 0..MANAGED_EVENT_WORK_BUDGET { +) -> Result<(), E> { + loop { match events.try_recv() { Ok(event) => process(event)?, + Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break, Err(ManagedEventTryRecvError::Lagged(count)) => { *dropped_events = dropped_events.saturating_add(count); } - Err(ManagedEventTryRecvError::Empty) => return Ok(ManagedDrainState::Empty), - Err(ManagedEventTryRecvError::Closed) => return Ok(ManagedDrainState::Closed), } } - Ok(ManagedDrainState::BudgetExhausted) -} - -pub fn is_shutdown_requested(shutdown_requested: &AtomicBool) -> bool { - shutdown_requested.load(Ordering::Relaxed) + Ok(()) } pub fn should_print_final_summary(continuous: bool, interrupted: bool) -> bool { diff --git a/crates/irtt-app/src/cmd/client/worker.rs b/crates/irtt-app/src/shared/client/worker.rs similarity index 96% rename from crates/irtt-app/src/cmd/client/worker.rs rename to crates/irtt-app/src/shared/client/worker.rs index d9d646c..828d7a2 100644 --- a/crates/irtt-app/src/cmd/client/worker.rs +++ b/crates/irtt-app/src/shared/client/worker.rs @@ -5,14 +5,14 @@ use tokio::sync::oneshot; /// Keeps measurement scheduling separate from frontend output and statistics. /// Drop also stops and joins on frontend errors or cancellation. -pub(super) struct ManagedWorker { +pub(crate) struct ManagedWorker { handle: ManagedClientHandle, thread: Option>, completion: oneshot::Receiver>, } impl ManagedWorker { - pub(super) fn start(task: ManagedClientTask, handle: ManagedClientHandle) -> io::Result { + pub(crate) fn start(task: ManagedClientTask, handle: ManagedClientHandle) -> io::Result { let (completed, completion) = oneshot::channel(); let thread = thread::Builder::new() .name("irtt-managed".to_owned()) @@ -41,7 +41,7 @@ impl ManagedWorker { }) } - pub(super) async fn join(mut self) -> Result> { + pub(crate) async fn join(mut self) -> Result> { let outcome = (&mut self.completion).await; self.join_thread()?; Ok(outcome diff --git a/crates/irtt-app/src/signal.rs b/crates/irtt-app/src/signal.rs index 8c4e0a2..ce0563b 100644 --- a/crates/irtt-app/src/signal.rs +++ b/crates/irtt-app/src/signal.rs @@ -1,18 +1,18 @@ -#[cfg(any(feature = "server", feature = "tui"))] +#[cfg(feature = "server")] use std::sync::{ atomic::{AtomicBool, Ordering}, Arc, }; -#[cfg(any(feature = "server", feature = "tui"))] +#[cfg(feature = "server")] pub fn install_signal_handler(shutdown_requested: Arc) -> Result<(), ctrlc::Error> { ctrlc::set_handler(move || { shutdown_requested.store(true, Ordering::Relaxed); }) } -/// Preserve ctrlc's SIGINT/SIGTERM policy while waking the async client. -#[cfg(feature = "client")] +/// Preserve ctrlc's SIGINT/SIGTERM policy while waking the async frontends. +#[cfg(any(feature = "client", feature = "tui"))] pub fn install_async_signal_handler() -> Result, ctrlc::Error> { let (shutdown, receiver) = tokio::sync::watch::channel(false); ctrlc::set_handler(move || { diff --git a/crates/irtt-app/tests/tui.rs b/crates/irtt-app/tests/tui.rs new file mode 100644 index 0000000..4a32c6c --- /dev/null +++ b/crates/irtt-app/tests/tui.rs @@ -0,0 +1,327 @@ +#![cfg(all(unix, feature = "tui"))] + +#[path = "support/in_tree_server.rs"] +mod in_tree_server; + +use irtt_server::ServerConfig; +use nix::{ + poll::{poll, PollFd, PollFlags}, + pty::{openpty, Winsize}, + sys::{ + signal::{kill, Signal}, + termios::{tcgetattr, Termios}, + }, + unistd::Pid, +}; +use std::{ + fs::File, + io::{Read, Write}, + os::{ + fd::{AsFd, AsRawFd}, + unix::process::CommandExt, + }, + process::{Child, Command, ExitStatus, Stdio}, + time::{Duration, Instant}, +}; + +// Real terminal ownership: keep both ends open to inspect restoration after exit, +// drain output while waiting, and always reap the child on assertion failures. +struct TuiProcess { + child: Child, + master: File, + slave: File, + original: Termios, + output: Vec, + cursor_answered: bool, +} +impl TuiProcess { + fn start(mut command: Command) -> Self { + let pty = openpty( + Some(&Winsize { + ws_row: 40, + ws_col: 160, + ws_xpixel: 0, + ws_ypixel: 0, + }), + None, + ) + .unwrap(); + let slave = File::from(pty.slave); + let original = tcgetattr(&slave).unwrap(); + command + .env("TERM", "xterm-256color") + .stdin(Stdio::from(slave.try_clone().unwrap())) + .stdout(Stdio::from(slave.try_clone().unwrap())) + .stderr(Stdio::from(slave.try_clone().unwrap())); + // Only async-signal-safe operations run between fork and exec. + unsafe { + command.pre_exec(|| { + if nix::libc::setsid() == -1 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + }); + } + Self { + child: command.spawn().unwrap(), + master: File::from(pty.master), + slave, + original, + output: vec![], + cursor_answered: false, + } + } + fn pump(&mut self) { + let mut fd = [PollFd::new(self.master.as_fd(), PollFlags::POLLIN)]; + poll(&mut fd, 20_u16).unwrap(); + if fd[0].revents().unwrap().contains(PollFlags::POLLIN) { + let mut buffer = [0; 65536]; + if let Ok(count) = self.master.read(&mut buffer) { + self.output.extend_from_slice(&buffer[..count]); + if !self.cursor_answered && self.output.windows(4).any(|bytes| bytes == b"\x1b[6n") + { + self.master.write_all(b"\x1b[1;1R").unwrap(); + self.cursor_answered = true; + } + } + } + } + fn await_text(&mut self, text: &str) { + let deadline = Instant::now() + Duration::from_secs(5); + while !String::from_utf8_lossy(&self.output).contains(text) { + self.pump(); + assert!( + Instant::now() < deadline, + "missing {text:?}: {}", + String::from_utf8_lossy(&self.output) + ); + assert!( + self.child.try_wait().unwrap().is_none(), + "TUI exited before {text:?}: {}", + String::from_utf8_lossy(&self.output) + ); + } + } + fn keys(&mut self, keys: &[u8]) { + self.master.write_all(keys).unwrap(); + } + fn signal(&self, signal: Signal) { + kill(Pid::from_raw(self.child.id() as i32), signal).unwrap(); + } + fn finish(&mut self, success: bool) -> ExitStatus { + let deadline = Instant::now() + Duration::from_secs(5); + let status = loop { + self.pump(); + if let Some(status) = self.child.try_wait().unwrap() { + break status; + } + assert!( + Instant::now() < deadline, + "TUI did not finish: {}", + String::from_utf8_lossy(&self.output[self.output.len().saturating_sub(3000)..]) + ); + }; + self.pump(); + assert_eq!( + status.success(), + success, + "{}", + String::from_utf8_lossy(&self.output) + ); + let restored = tcgetattr(&self.master).unwrap(); + assert_eq!(restored, self.original, "raw mode was not restored"); + let output = String::from_utf8_lossy(&self.output); + assert!( + output.contains("\x1b[?1049l"), + "alternate screen was not left: {output}" + ); + assert!( + output.contains("\x1b[?25h"), + "cursor was not restored: {output}" + ); + status + } +} +impl Drop for TuiProcess { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} +fn tui(target: &str, duration: &str) -> Command { + let mut cmd = Command::new(env!("CARGO_BIN_EXE_irtt-tui")); + cmd.args(["--duration", duration, "--interval", "20ms", target]); + cmd +} + +#[test] +fn terminal_controls_and_exit_paths_restore_the_terminal() { + let server = in_tree_server::InTreeServer::start(ServerConfig::default()); + let target = server.addr.to_string(); + let mut command = tui(&format!("first={target}"), "0"); + command.arg(format!("second={target}")); + let mut process = TuiProcess::start(command); + process.await_text("active"); + process.keys(b"p"); + process.await_text("paused"); + let size = Winsize { + ws_row: 12, + ws_col: 50, + ws_xpixel: 0, + ws_ypixel: 0, + }; + assert_eq!( + unsafe { nix::libc::ioctl(process.slave.as_raw_fd(), nix::libc::TIOCSWINSZ, &size) }, + 0 + ); + process.signal(Signal::SIGWINCH); + process.await_text("small"); + let size = Winsize { + ws_row: 40, + ws_col: 160, + ws_xpixel: 0, + ws_ypixel: 0, + }; + assert_eq!( + unsafe { nix::libc::ioctl(process.slave.as_raw_fd(), nix::libc::TIOCSWINSZ, &size) }, + 0 + ); + process.signal(Signal::SIGWINCH); + process.keys(b"\td"); + process.await_text("arrows"); + // Each key waits for a real draw, avoiding a timing-dependent escape batch. + for key in [ + b"\x1b[B".as_slice(), + b"\x1b[C", + b"\x1b[6~", + b"\x1b[F", + b"\x1b[H", + b"d", + b"m", + b"+", + b"-", + b"0", + b"\x1b[D", + b"\x1b[C", + b"\x1b[5~", + b"\x1b[6~", + b"\x1b[H", + b"\x1b[F", + b"r", + b"p", + ] { + let frames = process + .output + .windows(6) + .filter(|bytes| *bytes == b"\x1b[?25l") + .count(); + process.keys(key); + let deadline = Instant::now() + Duration::from_secs(5); + while process + .output + .windows(6) + .filter(|bytes| *bytes == b"\x1b[?25l") + .count() + == frames + { + process.pump(); + assert!(Instant::now() < deadline, "input did not redraw: {key:?}"); + } + } + process.keys(b"q"); + process.finish(true); + + for action in [None, Some(b"\x03".as_slice()), Some(b"signal".as_slice())] { + let mut process = + TuiProcess::start(tui(&target, if action.is_none() { "100ms" } else { "0" })); + process.await_text("active"); + match action { + Some(b"signal") => process.signal(Signal::SIGTERM), + Some(keys) => process.keys(keys), + None => {} + } + process.finish(true); + } + // An adverse early-close proxy forwards the real Rust server's OPEN and + // echo bytes. Only the documented CLOSE flag is added to the first echo. + // This exercises the application's continuous peer-close exit policy. + std::thread::scope(|scope| { + use irtt_proto::{decode_request, DecodedRequestKind, FLAG_CLOSE}; + use std::net::UdpSocket; + let proxy = UdpSocket::bind("127.0.0.1:0").unwrap(); + let proxy_addr = proxy.local_addr().unwrap(); + proxy + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let upstream = UdpSocket::bind("127.0.0.1:0").unwrap(); + upstream.connect(server.addr).unwrap(); + upstream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let peer = scope.spawn(move || { + let mut packet = [0; 65536]; + loop { + let (count, client) = proxy.recv_from(&mut packet).unwrap(); + let echo = matches!( + decode_request(&packet[..count]).unwrap().kind, + DecodedRequestKind::Echo { .. } + ); + upstream.send(&packet[..count]).unwrap(); + let count = upstream.recv(&mut packet).unwrap(); + if echo { + packet[3] |= FLAG_CLOSE; + } + proxy.send_to(&packet[..count], client).unwrap(); + if echo { + break; + } + } + }); + let mut process = TuiProcess::start(tui(&proxy_addr.to_string(), "0")); + process.finish(false); + assert!(String::from_utf8_lossy(&process.output) + .contains("continuous run ended because of peer closure")); + peer.join().unwrap(); + }); + + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args(["--exact", "cancelled_frontend_fixture", "--nocapture"]) + .env("IRTT_TUI_CANCEL_TARGET", &target); + TuiProcess::start(command).finish(true); + + // Opening failure must be reported and restore the terminal too. + let mut process = TuiProcess::start(tui("127.0.0.1:99999", "100ms")); + process.finish(false); + assert!(String::from_utf8_lossy(&process.output) + .contains("no managed target completed successfully")); +} + +// The parent runs this fixture in its own PTY. Dropping the +// actual frontend future must restore that terminal and join measurement. +#[test] +fn cancelled_frontend_fixture() { + let Ok(target) = std::env::var("IRTT_TUI_CANCEL_TARGET") else { + return; + }; + use clap::Parser; + use std::{ + future::{poll_fn, Future}, + task::Poll, + }; + let args = irtt_app::cmd::tui::TuiArgs::parse_from(["irtt-tui", "--duration", "0", &target]); + let (_shutdown, receiver) = tokio::sync::watch::channel(false); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let mut frontend = Box::pin(irtt_app::cmd::tui::run_tui(args, receiver)); + poll_fn(|cx| { + assert!(frontend.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + drop(frontend); + }); +} From b1b0b62848cc3cba0c383f740c9a29d8a13e37c8 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 5 Oct 2026 22:04:07 +0700 Subject: [PATCH 3/3] chore(app): remove unnecessary tests --- crates/irtt-app/src/cmd/tui/run.rs | 38 -- crates/irtt-app/src/cmd/tui/ui.rs | 226 -------- crates/irtt-app/src/shared/client/worker.rs | 66 --- crates/irtt-app/tests/cli.rs | 530 ------------------ .../irtt-app/tests/support/in_tree_server.rs | 74 --- crates/irtt-app/tests/tui.rs | 327 ----------- 6 files changed, 1261 deletions(-) delete mode 100644 crates/irtt-app/tests/cli.rs delete mode 100644 crates/irtt-app/tests/support/in_tree_server.rs delete mode 100644 crates/irtt-app/tests/tui.rs diff --git a/crates/irtt-app/src/cmd/tui/run.rs b/crates/irtt-app/src/cmd/tui/run.rs index 01d7384..c0a89d0 100644 --- a/crates/irtt-app/src/cmd/tui/run.rs +++ b/crates/irtt-app/src/cmd/tui/run.rs @@ -228,41 +228,3 @@ fn process_tui_event( _ => false, } } - -#[cfg(test)] -mod tests { - use super::*; - use ratatui::{backend::TestBackend, Terminal}; - - #[tokio::test(flavor = "current_thread")] - async fn subscription_lag_keeps_incomplete_statistics_visible_while_paused() { - for paused in [false, true] { - let (sender, mut events) = tokio::sync::broadcast::channel(1); - sender.send(ManagedEvent::Started).unwrap(); - sender.send(ManagedEvent::Started).unwrap(); - let mut state = TuiState::default(); - if paused { - state.toggle_pause(); - } - let Err(RecvError::Lagged(dropped)) = events.recv().await else { - panic!("subscription must report real broadcast overflow"); - }; - state.mark_dropped_managed_events(dropped); - let mut terminal = Terminal::new(TestBackend::new(200, 40)).unwrap(); - for _ in 0..2 { - terminal - .draw(|frame| super::super::ui::draw_dashboard(frame, &state)) - .unwrap(); - let text = terminal - .backend() - .buffer() - .content() - .iter() - .map(|cell| cell.symbol()) - .collect::(); - assert!(text.contains("incomplete:dropped=1")); - assert_eq!(text.contains("display paused"), paused); - } - } - } -} diff --git a/crates/irtt-app/src/cmd/tui/ui.rs b/crates/irtt-app/src/cmd/tui/ui.rs index 4003a2d..afd0de1 100644 --- a/crates/irtt-app/src/cmd/tui/ui.rs +++ b/crates/irtt-app/src/cmd/tui/ui.rs @@ -1558,229 +1558,3 @@ fn format_optional_hex(value: Option) -> String { fn duration_ns(value: Duration) -> i128 { i128::try_from(value.as_nanos()).unwrap_or(i128::MAX) } - -#[cfg(test)] -mod tests { - use super::*; - use irtt_client::{ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, ServerTiming}; - use ratatui::backend::TestBackend; - use std::time::SystemTime; - - fn measure(state: &mut TuiState, label: &str, seq: u32, at: Instant, effective_ns: i128) { - let target = TargetInstance { - id: label.into(), - generation: 1, - }; - let received_at = ClientTimestamp { - mono: at, - wall: SystemTime::UNIX_EPOCH + Duration::from_secs(1), - }; - let sent_at = ClientTimestamp { - mono: at - Duration::from_millis(10), - wall: received_at.wall - Duration::from_millis(10), - }; - let remote = "127.0.0.1:2112".parse().unwrap(); - state.process_target_event( - &target, - &ClientEvent::EchoSent { - seq, - remote, - scheduled_at: None, - sent_at, - bytes: 64, - send_call: Duration::ZERO, - timer_error: None, - }, - ); - state.process_target_event( - &target, - &ClientEvent::EchoReply { - seq, - remote, - sent_at, - received_at, - rtt: RttSample { - raw: Duration::from_millis(10), - adjusted: Some(SignedDuration::from_nanos(effective_ns)), - effective: SignedDuration::from_nanos(effective_ns), - }, - server_timing: Some(ServerTiming { - receive_wall_ns: None, - receive_mono_ns: None, - send_wall_ns: None, - send_mono_ns: None, - midpoint_wall_ns: None, - midpoint_mono_ns: None, - processing: Some(Duration::from_nanos((10_000_000 - effective_ns) as u64)), - }), - one_way: Some(OneWayDelaySample { - client_to_server: Some(SignedDuration::from_nanos(-3_000_000)), - server_to_client: Some(SignedDuration::from_nanos(effective_ns + 3_000_000)), - }), - received_stats: None, - bytes: 64, - packet_meta: PacketMeta::default(), - }, - ); - } - - fn screen(state: &TuiState, width: u16, height: u16) -> String { - let mut terminal = Terminal::new(TestBackend::new(width, height)).unwrap(); - terminal.draw(|frame| draw_dashboard(frame, state)).unwrap(); - terminal - .backend() - .buffer() - .content() - .chunks(usize::from(width)) - .map(|row| row.iter().map(|cell| cell.symbol()).collect::()) - .collect::>() - .join("\n") - } - - #[test] - fn target_focus_and_details_remain_usable_across_sizes() { - let mut state = TuiState::with_target_labels( - TuiConfig::default(), - (0..8).map(|i| format!("target{i}")), - ); - let now = Instant::now(); - measure( - &mut state, - "target7", - 0, - now - Duration::from_secs(5), - 8_000_000, - ); - measure(&mut state, "target7", 1, now, 5_000_000); - for (width, height) in [(56, 18), (80, 24), (120, 40)] { - let area = Rect::new(0, 0, width, height); - state.handle_key(KeyCode::Home, area); - state.handle_key(KeyCode::BackTab, area); - let text = screen(&state, width, height); - assert!(text.contains(">target7"), "{text}"); - assert!( - text.contains("jitter σ") && text.contains("loss %"), - "{text}" - ); - assert!( - text.contains(&format_optional_ns_i128(Some(-3_000_000))) - && text.contains(&format_optional_ns_i128(Some(5_000_000))), - "{text}" - ); - state.handle_key(KeyCode::Char('d'), area); - assert!(screen(&state, width, height).contains("target: target7")); - state.handle_key(KeyCode::End, area); - assert!(screen(&state, width, height).contains("target7: reply")); - state.handle_key(KeyCode::Tab, area); - // Selecting another target resets detail scrolling and exposes its own data. - assert!(screen(&state, width, height).contains("target: target0")); - state.handle_key(KeyCode::Esc, area); - } - assert!(screen(&state, 40, 10).contains("terminal too small")); - } - - #[test] - fn history_reset_keeps_latest_summary_and_statistics_while_paused() { - let mut state = TuiState::default(); - let area = Rect::new(0, 0, 100, 40); - state.handle_key(KeyCode::Char('p'), area); - measure(&mut state, "target", 0, Instant::now(), 8_000_000); - state.handle_key(KeyCode::Char('r'), area); - let text = screen(&state, 100, 40); - assert!( - text.contains(&format_optional_ns_i128(Some(8_000_000))) - && text.contains("display paused") - ); - assert!(text.contains("waiting for primary replies")); - state.handle_key(KeyCode::Char('d'), area); - let text = screen(&state, 100, 40); - assert!(text.contains("sent 1 received 1 unique 1")); - assert!(text.contains(&format!( - "effective RTT: {}", - format_optional_ns_i128(Some(8_000_000)) - ))); - state.handle_key(KeyCode::Esc, area); - measure(&mut state, "target", 1, Instant::now(), 5_000_000); - assert!(!screen(&state, 100, 40).contains("waiting for primary replies")); - } - - #[test] - fn graph_controls_preserve_signed_metrics_and_history_range() { - let mut state = - TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]); - let area = Rect::new(0, 0, 100, 40); - let now = Instant::now(); - for label in ["a", "b"] { - measure( - &mut state, - label, - 0, - now - Duration::from_secs(120), - -2_000_000, - ); - measure( - &mut state, - label, - 1, - now - Duration::from_secs(60), - -2_000_000, - ); - measure(&mut state, label, 2, now, -2_000_000); - } - state.handle_key(KeyCode::Home, area); - let viewport = state - .graph_viewport - .range(now, state.newest_graph_sample_time()); - assert_eq!(viewport.start, now - Duration::from_secs(120)); - assert_eq!(viewport.end, now - Duration::from_secs(60)); - for metric in [ - GraphMetric::EffectiveRtt, - GraphMetric::RawRtt, - GraphMetric::AdjustedRtt, - GraphMetric::ClientToServer, - GraphMetric::ServerToClient, - GraphMetric::ServerProcessing, - ] { - assert_eq!(state.graph_metric, metric); - let series = state - .targets - .iter() - .enumerate() - .map(|(i, target)| target_metric_series(target, i, viewport, metric).unwrap()) - .collect::>(); - // Both endpoints belong to the viewport, for both targets. - assert!(series.iter().all(|series| series.data.len() == 2)); - if matches!( - metric, - GraphMetric::EffectiveRtt | GraphMetric::AdjustedRtt | GraphMetric::ClientToServer - ) { - assert!(series - .iter() - .all(|series| series.data.iter().all(|(_, y)| *y < 0.0))); - assert!(chart_y_bounds(&series, metric.axis_kind()).0 < 0.0); - } - state.handle_key(KeyCode::Char('m'), area); - } - state.handle_key(KeyCode::PageDown, area); - assert_eq!(state.graph_viewport.range(now, Some(now)).end, now); - state.handle_key(KeyCode::Left, area); - assert_eq!( - state.graph_viewport.range(now, Some(now)).end, - now - Duration::from_secs(15) - ); - state.handle_key(KeyCode::Right, area); - state.handle_key(KeyCode::PageUp, area); - assert_eq!( - state.graph_viewport.range(now, Some(now)).end, - now - Duration::from_secs(60) - ); - state.handle_key(KeyCode::Char('+'), area); - assert_eq!(state.graph_viewport.window, Duration::from_secs(40)); - state.handle_key(KeyCode::Char('-'), area); - assert_eq!(state.graph_viewport.window, DEFAULT_GRAPH_WINDOW); - state.handle_key(KeyCode::Char('0'), area); - assert!(!state.graph_viewport.range(now, Some(now)).is_live); - state.handle_key(KeyCode::End, area); - assert!(state.graph_viewport.range(now, Some(now)).is_live); - } -} diff --git a/crates/irtt-app/src/shared/client/worker.rs b/crates/irtt-app/src/shared/client/worker.rs index 828d7a2..c3e7379 100644 --- a/crates/irtt-app/src/shared/client/worker.rs +++ b/crates/irtt-app/src/shared/client/worker.rs @@ -65,69 +65,3 @@ impl Drop for ManagedWorker { } } } - -#[cfg(test)] -mod tests { - use super::*; - use irtt_client::managed::{ - ManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason, - }; - - // The frontend executor must be able to stop a worker while awaiting its - // completion; a blocking join here would deadlock this current-thread runtime. - #[test] - fn completion_leaves_frontend_free_to_stop_and_preserves_outcome() { - let (task, handle) = ManagedClient::task( - ManagedClientConfig { - completion: ManagedCompletionPolicy::ExplicitStop, - ..Default::default() - }, - vec![], - ) - .unwrap(); - let worker = ManagedWorker::start(task, handle.clone()).unwrap(); - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap() - .block_on(async { - let (outcome, ()) = tokio::time::timeout(Duration::from_secs(2), async { - tokio::join!(worker.join(), handle.stop()) - }) - .await - .expect("worker completion must allow frontend stop"); - let outcome = outcome.unwrap(); - assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); - assert_eq!(handle.status().final_outcome.as_deref(), Some(&outcome)); - }); - } - - // Cancelling the frontend's completion wait must stop and join the worker, - // rather than detaching an ongoing measurement task. - #[test] - fn cancelling_completion_stops_and_joins_worker() { - use std::{ - future::Future, - task::{Context, Waker}, - }; - let (task, handle) = ManagedClient::task( - ManagedClientConfig { - completion: ManagedCompletionPolicy::ExplicitStop, - ..Default::default() - }, - vec![], - ) - .unwrap(); - let worker = ManagedWorker::start(task, handle.clone()).unwrap(); - let mut completion = Box::pin(worker.join()); - assert!(completion - .as_mut() - .poll(&mut Context::from_waker(Waker::noop())) - .is_pending()); - drop(completion); - assert_eq!( - handle.status().final_outcome.as_ref().unwrap().end_reason, - ManagedEndReason::StopRequested - ); - } -} diff --git a/crates/irtt-app/tests/cli.rs b/crates/irtt-app/tests/cli.rs deleted file mode 100644 index fb453f7..0000000 --- a/crates/irtt-app/tests/cli.rs +++ /dev/null @@ -1,530 +0,0 @@ -#[cfg(feature = "client")] -#[path = "support/in_tree_server.rs"] -mod in_tree_server; - -#[cfg(feature = "client")] -use irtt_server::ServerConfig; -use std::process::Command; - -#[cfg(feature = "client")] -struct ClientProcess(std::process::Child); - -#[cfg(feature = "client")] -impl Drop for ClientProcess { - fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); - } -} - -#[cfg(feature = "client")] -impl ClientProcess { - fn wait(&mut self) -> std::process::ExitStatus { - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); - loop { - if let Some(status) = self.0.try_wait().unwrap() { - return status; - } - assert!( - std::time::Instant::now() < deadline, - "client did not terminate within five seconds" - ); - std::thread::sleep(std::time::Duration::from_millis(10)); - } - } -} - -// Dispatch/help/availability belong to the application and run in every build. -#[test] -fn dispatcher_reports_and_runs_the_enabled_applets() { - let dispatcher = env!("CARGO_BIN_EXE_irtt-rs"); - let help = Command::new(dispatcher).arg("--help").output().unwrap(); - assert!(help.status.success()); - let help = String::from_utf8(help.stdout).unwrap(); - for (applet, enabled) in [ - ("client", cfg!(feature = "client")), - ("server", cfg!(feature = "server")), - ("tui", cfg!(feature = "tui")), - ] { - let row = help - .lines() - .find(|line| line.trim_start().starts_with(applet)) - .unwrap(); - assert_eq!(row.contains("not available"), !enabled); - let output = Command::new(dispatcher) - .args([applet, "--help"]) - .output() - .unwrap(); - assert_eq!(output.status.success(), enabled, "{applet}"); - if !enabled { - assert!(String::from_utf8(output.stderr) - .unwrap() - .contains("not available")); - } - } - let binaries = [ - dispatcher, - #[cfg(feature = "client")] - env!("CARGO_BIN_EXE_irtt-client"), - #[cfg(feature = "server")] - env!("CARGO_BIN_EXE_irtt-server"), - #[cfg(feature = "tui")] - env!("CARGO_BIN_EXE_irtt-tui"), - ]; - for binary in binaries { - assert!(Command::new(binary) - .arg("--help") - .output() - .unwrap() - .status - .success()); - assert!(!Command::new(binary) - .arg("--invalid-option") - .output() - .unwrap() - .status - .success()); - } -} - -// The default presentation must reach a meaningful packet summary for a finite run. -#[cfg(feature = "client")] -#[test] -fn default_client_output_emits_a_packet_summary() { - use std::{io::Read, process::Stdio, thread}; - let server = in_tree_server::InTreeServer::start(ServerConfig::default()); - let mut client = ClientProcess( - Command::new(env!("CARGO_BIN_EXE_irtt-client")) - .env_clear() - .args([ - "--duration", - "100ms", - "--interval", - "20ms", - &server.addr.to_string(), - ]) - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .unwrap(), - ); - let mut stdout = client.0.stdout.take().unwrap(); - let stdout = thread::spawn(move || { - let mut text = String::new(); - stdout.read_to_string(&mut text).unwrap(); - text - }); - let mut stderr = client.0.stderr.take().unwrap(); - let stderr = thread::spawn(move || { - let mut text = String::new(); - stderr.read_to_string(&mut text).unwrap(); - text - }); - let status = client.wait(); - let stdout = stdout.join().unwrap(); - let stderr = stderr.join().unwrap(); - assert!(status.success(), "{stderr}"); - assert!(stdout.contains("irtt-rs summary"), "{stdout}"); - let packets = stdout - .lines() - .find(|line| line.starts_with("packets:")) - .expect("default output must include the packet summary"); - for field in ["sent=", "received="] { - let count = packets - .split_whitespace() - .find_map(|word| word.strip_prefix(field)) - .unwrap() - .parse::() - .unwrap(); - assert!( - count > 0, - "summary must report {field} a positive count: {packets}" - ); - } -} - -#[cfg(feature = "client")] -#[test] -fn target_authentication_inherits_overrides_or_disables_the_global_key() { - use clap::Parser; - use irtt_app::cmd::client::args::ClientArgs; - use irtt_client::{managed::TargetAuth, Authentication, HmacKey}; - let args = ClientArgs::try_parse_from([ - "irtt-client", - "--hmac", - "CONSPICUOUS_GLOBAL_HMAC_KEY", - "inherit=127.0.0.1:2112", - "override=127.0.0.1:2113@hmac=CONSPICUOUS_TARGET_HMAC_KEY", - "disable=127.0.0.1:2114@hmac=", - ]) - .unwrap(); - let setup = args.prepare().unwrap(); - assert_eq!( - setup.managed_config().client.auth, - Authentication::Hmac(HmacKey::new(b"CONSPICUOUS_GLOBAL_HMAC_KEY".as_slice())) - ); - let targets = setup.managed_targets(); - let authentication: Vec<_> = targets.iter().map(|target| target.auth.clone()).collect(); - assert_eq!( - authentication, - [ - TargetAuth::Inherit, - TargetAuth::Override(Authentication::Hmac(HmacKey::new( - b"CONSPICUOUS_TARGET_HMAC_KEY".as_slice() - ))), - TargetAuth::Override(Authentication::Unauthenticated) - ] - ); - for output in [format!("{args:?}"), format!("{setup:?}")] { - assert!(!output.contains("CONSPICUOUS_GLOBAL_HMAC_KEY")); - assert!(!output.contains("CONSPICUOUS_TARGET_HMAC_KEY")); - } -} - -// A real authenticated server detects broken client argument-to-config mapping. -#[cfg(feature = "client")] -#[test] -fn client_arguments_export_a_real_client_run() { - use std::{io::Read, process::Stdio, thread}; - let dispatcher = env!("CARGO_BIN_EXE_irtt-rs"); - for option in ["--dscp", "--ttl"] { - assert!(!Command::new(dispatcher) - .args(["client", option, "256", "127.0.0.1:9"]) - .output() - .unwrap() - .status - .success()); - } - let server = - in_tree_server::InTreeServer::start(ServerConfig::default().with_hmac_key(b"cli-test")); - let mut client = ClientProcess( - Command::new(dispatcher) - .args([ - "client", - "--duration", - "100ms", - "--interval", - "20ms", - "--hmac", - "cli-test", - "--length", - "128", - "--sfill", - "pattern:ab", - "--clock", - "wall", - "--tstamp", - "both", - "--stats", - "both", - "--ttl", - "64", - "--format", - "jsonl", - "--columns", - "target,event,bytes", - &format!("edge={}", server.addr), - ]) - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .unwrap(), - ); - // Drain both pipes while waiting so output cannot block client termination. - let mut stdout = client.0.stdout.take().unwrap(); - let stdout = thread::spawn(move || { - let mut bytes = Vec::new(); - stdout.read_to_end(&mut bytes).unwrap(); - bytes - }); - let mut stderr = client.0.stderr.take().unwrap(); - let stderr = thread::spawn(move || { - let mut bytes = Vec::new(); - stderr.read_to_end(&mut bytes).unwrap(); - bytes - }); - let status = client.wait(); - let stdout = stdout.join().unwrap(); - let stderr = stderr.join().unwrap(); - assert!(status.success(), "{}", String::from_utf8_lossy(&stderr)); - let stdout = String::from_utf8(stdout).unwrap(); - let rows: Vec<_> = stdout.lines().collect(); - assert!(!rows.is_empty()); - assert!(rows.iter().all(|row| row.starts_with('{') - && row.ends_with('}') - && row.contains("\"target\":\"edge\""))); - assert!(rows - .iter() - .any(|row| row.contains("\"event\":\"echo_reply\"") && row.contains("\"bytes\":128"))); - assert!(rows - .iter() - .any(|row| row.contains("\"event\":\"session_closed\""))); -} - -#[cfg(any(feature = "client", feature = "tui"))] -#[test] -fn malformed_target_diagnostics_do_not_expose_authentication_keys() { - let secret = "malformed-target-secret-92817"; - for applet in [ - #[cfg(feature = "client")] - "client", - #[cfg(feature = "tui")] - "tui", - ] { - let output = Command::new(env!("CARGO_BIN_EXE_irtt-rs")) - .args([applet, &format!(r"edge=127.0.0.1:9@hmac={secret}\q")]) - .output() - .unwrap(); - assert!(!output.status.success()); - let stderr = String::from_utf8(output.stderr).unwrap(); - assert!( - !stderr.trim().is_empty(), - "malformed target must report a diagnostic" - ); - assert!(!stderr.contains(secret), "target authentication key leaked"); - assert!(!String::from_utf8_lossy(&output.stdout).contains(secret)); - } -} - -#[cfg(feature = "server")] -#[test] -fn server_arguments_resolve_binds_and_map_policy() { - use clap::{CommandFactory, FromArgMatches, Parser}; - use irtt_app::cmd::server::ServerArgs; - use irtt_server::TimestampAllowance; - use std::time::Duration; - // Isolate zero-argument defaults from the test runner's environment without - // mutating process-global environment in this parallel test process. - let defaults = ServerArgs::from_arg_matches( - &ServerArgs::command() - .mut_args(|arg| arg.env(None::<&str>)) - .try_get_matches_from(["irtt-server"]) - .unwrap(), - ) - .unwrap(); - assert_eq!( - defaults.resolve_binds(), - vec![ - "[::]:2112".parse().unwrap(), - "0.0.0.0:2112".parse().unwrap() - ] - ); - assert_eq!( - defaults.server_config(), - irtt_server::ServerConfig::default() - ); - let args = ServerArgs::try_parse_from([ - "irtt-server", - "--bind", - "127.0.0.1:2113", - "--bind", - "[::1]:2114", - "--hmac", - "server-cli-test", - "--max-sessions", - "3", - "--max-packet-length", - "96", - "--min-interval", - "25ms", - "--burst", - "2", - "--idle-timeout", - "8s", - "--max-duration", - "1m", - "--timestamp-allowance", - "single", - "--no-dscp", - ]) - .unwrap(); - assert_eq!( - args.resolve_binds(), - vec![ - "127.0.0.1:2113".parse().unwrap(), - "[::1]:2114".parse().unwrap() - ] - ); - let config = args.server_config(); - assert_eq!(config.hmac_key(), Some(b"server-cli-test".as_slice())); - assert_eq!(config.max_sessions(), 3); - assert_eq!(config.max_packet_length(), 96); - assert_eq!(config.min_send_interval(), Duration::from_millis(25)); - assert_eq!(config.burst_allowance(), 2); - assert_eq!(config.idle_timeout(), Duration::from_secs(8)); - assert_eq!(config.max_test_duration(), Some(Duration::from_secs(60))); - assert_eq!(config.timestamp_allowance(), TimestampAllowance::Single); - assert!(!config.dscp_allowed()); -} - -#[cfg(feature = "server")] -#[test] -fn server_command_line_binds_replace_the_environment_bind() { - use std::net::UdpSocket; - // Occupied ports give bounded, observable startup failures, without starting - // a daemon or needing a process/signal harness merely to inspect arguments. - let environment = UdpSocket::bind("127.0.0.1:0").unwrap(); - let explicit = UdpSocket::bind("127.0.0.1:0").unwrap(); - let environment_addr = environment.local_addr().unwrap().to_string(); - let explicit_addr = explicit.local_addr().unwrap().to_string(); - for use_explicit in [false, true] { - let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-server")); - command - .env_clear() - .env("IRTT_SERVER_BIND", &environment_addr) - .env("IRTT_SERVER_MAX_SESSIONS", "invalid") - .args(["--max-sessions", "3"]); - if use_explicit { - command.args(["--bind", "127.0.0.1:0", "--bind", &explicit_addr]); - } - let output = command.output().unwrap(); - assert!(!output.status.success()); - let stderr = String::from_utf8(output.stderr).unwrap(); - let expected = if use_explicit { - &explicit_addr - } else { - &environment_addr - }; - assert!( - stderr.contains(expected), - "startup must attempt the selected bind: {stderr}" - ); - assert!( - !stderr.contains("listening on"), - "partial startup must not announce listeners" - ); - } -} - -// Exercise stdin replacement and EOF through the actual application process. -#[cfg(feature = "client")] -#[test] -fn targets_stdin_supersedes_the_active_set_and_stops_on_eof() { - use std::{ - io::{BufRead, BufReader, Read, Write}, - process::Stdio, - sync::mpsc, - thread, - time::{Duration, Instant}, - }; - let server = in_tree_server::InTreeServer::start(ServerConfig::default()); - let mut client = ClientProcess( - Command::new(env!("CARGO_BIN_EXE_irtt-client")) - .env_clear() - .args([ - "--duration", - "0", - "--targets-stdin", - "--interval", - "20ms", - "--format", - "jsonl", - "--columns", - "target,event", - ]) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .unwrap(), - ); - let stdout = client.0.stdout.take().unwrap(); - let (tx, rx) = mpsc::channel(); - let reader = thread::spawn(move || { - for line in BufReader::new(stdout).lines() { - if tx.send(line.unwrap()).is_err() { - break; - } - } - }); - let mut stdin = client.0.stdin.take().unwrap(); - let mut rows = Vec::new(); - for target in ["first", "second"] { - write!(stdin, "\r\n{target}={}\r\n", server.addr).unwrap(); - stdin.flush().unwrap(); - let deadline = Instant::now() + Duration::from_secs(5); - loop { - let row = rx - .recv_timeout(deadline.saturating_duration_since(Instant::now())) - .expect("stdin target set did not produce a reply"); - let replied = row.contains(&format!("\"target\":\"{target}\"")) - && row.contains("\"event\":\"echo_reply\""); - rows.push(row); - if replied { - break; - } - } - } - // Retirement can finish after the newer target's first reply. - let deadline = Instant::now() + Duration::from_secs(5); - while !rows.iter().any(|row| { - row.contains("\"target\":\"first\"") && row.contains("\"event\":\"session_closed\"") - }) { - rows.push( - rx.recv_timeout(deadline.saturating_duration_since(Instant::now())) - .expect("superseded stdin target did not retire"), - ); - } - drop(stdin); - let status = client.wait(); - reader.join().unwrap(); - rows.extend(rx.try_iter()); - let mut stderr = String::new(); - client - .0 - .stderr - .take() - .unwrap() - .read_to_string(&mut stderr) - .unwrap(); - assert!(status.success(), "{stderr}"); - assert!(rows - .iter() - .any(|row| row.contains("\"target\":\"second\"") - && row.contains("\"event\":\"session_closed\""))); -} - -// Record bounds and decoding errors must terminate the real stdin frontend, -// even when no managed events are available to wake it. -#[cfg(feature = "client")] -#[test] -fn targets_stdin_invalid_records_are_fatal_and_empty_eof_is_normal() { - use std::{ - io::{Read, Write}, - process::Stdio, - }; - for (record, success, diagnostic) in [ - (Vec::new(), true, ""), - (b"\r\n\n".to_vec(), true, ""), - (b"\n=127.0.0.1:9\n".to_vec(), false, "line 2"), - (vec![0xff, b'\n'], false, "line 1"), - (vec![b'x'; 64 * 1024 + 2], false, "line 1"), - ] { - let mut client = ClientProcess( - Command::new(env!("CARGO_BIN_EXE_irtt-client")) - .env_clear() - .args(["--duration", "0", "--targets-stdin"]) - .stdin(Stdio::piped()) - .stdout(Stdio::null()) - .stderr(Stdio::piped()) - .spawn() - .unwrap(), - ); - let mut stdin = client.0.stdin.take().unwrap(); - stdin.write_all(&record).unwrap(); - drop(stdin); - let status = client.wait(); - let mut stderr = String::new(); - client - .0 - .stderr - .take() - .unwrap() - .read_to_string(&mut stderr) - .unwrap(); - assert_eq!(status.success(), success, "{stderr}"); - assert!(stderr.contains(diagnostic), "{stderr}"); - } -} diff --git a/crates/irtt-app/tests/support/in_tree_server.rs b/crates/irtt-app/tests/support/in_tree_server.rs deleted file mode 100644 index faf4fcd..0000000 --- a/crates/irtt-app/tests/support/in_tree_server.rs +++ /dev/null @@ -1,74 +0,0 @@ -use std::{ - net::{IpAddr, Ipv4Addr, SocketAddr}, - sync::mpsc, - thread::{self, JoinHandle}, - time::Duration, -}; - -use irtt_server::{Server, ServerConfig}; -use tokio::sync::oneshot; - -const STARTUP_TIMEOUT: Duration = Duration::from_secs(2); - -pub struct InTreeServer { - pub addr: SocketAddr, - shutdown: Option>, - done: Option>>, -} - -impl InTreeServer { - pub fn start(config: ServerConfig) -> Self { - let (startup_tx, startup_rx) = mpsc::sync_channel(1); - let (shutdown_tx, shutdown_rx) = oneshot::channel(); - let done = thread::spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("failed to build in-tree test server runtime"); - runtime.block_on(async move { - let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); - let mut server = Server::bind(bind_addr, config).await?; - startup_tx - .send(server.local_addr()?) - .expect("test stopped waiting for in-tree server startup"); - server - .run(async { - let _ = shutdown_rx.await; - }) - .await - }) - }); - - let addr = match startup_rx.recv_timeout(STARTUP_TIMEOUT) { - Ok(addr) => addr, - Err(error) => { - let _ = shutdown_tx.send(()); - let result = done.join().expect("in-tree test server thread panicked"); - panic!("in-tree test server failed before startup ({error}): {result:?}"); - } - }; - - Self { - addr, - shutdown: Some(shutdown_tx), - done: Some(done), - } - } - - fn stop_inner(&mut self) { - if let Some(shutdown) = self.shutdown.take() { - let _ = shutdown.send(()); - } - if let Some(done) = self.done.take() { - done.join() - .expect("in-tree test server thread panicked") - .expect("in-tree test server stopped with an error"); - } - } -} - -impl Drop for InTreeServer { - fn drop(&mut self) { - self.stop_inner(); - } -} diff --git a/crates/irtt-app/tests/tui.rs b/crates/irtt-app/tests/tui.rs deleted file mode 100644 index 4a32c6c..0000000 --- a/crates/irtt-app/tests/tui.rs +++ /dev/null @@ -1,327 +0,0 @@ -#![cfg(all(unix, feature = "tui"))] - -#[path = "support/in_tree_server.rs"] -mod in_tree_server; - -use irtt_server::ServerConfig; -use nix::{ - poll::{poll, PollFd, PollFlags}, - pty::{openpty, Winsize}, - sys::{ - signal::{kill, Signal}, - termios::{tcgetattr, Termios}, - }, - unistd::Pid, -}; -use std::{ - fs::File, - io::{Read, Write}, - os::{ - fd::{AsFd, AsRawFd}, - unix::process::CommandExt, - }, - process::{Child, Command, ExitStatus, Stdio}, - time::{Duration, Instant}, -}; - -// Real terminal ownership: keep both ends open to inspect restoration after exit, -// drain output while waiting, and always reap the child on assertion failures. -struct TuiProcess { - child: Child, - master: File, - slave: File, - original: Termios, - output: Vec, - cursor_answered: bool, -} -impl TuiProcess { - fn start(mut command: Command) -> Self { - let pty = openpty( - Some(&Winsize { - ws_row: 40, - ws_col: 160, - ws_xpixel: 0, - ws_ypixel: 0, - }), - None, - ) - .unwrap(); - let slave = File::from(pty.slave); - let original = tcgetattr(&slave).unwrap(); - command - .env("TERM", "xterm-256color") - .stdin(Stdio::from(slave.try_clone().unwrap())) - .stdout(Stdio::from(slave.try_clone().unwrap())) - .stderr(Stdio::from(slave.try_clone().unwrap())); - // Only async-signal-safe operations run between fork and exec. - unsafe { - command.pre_exec(|| { - if nix::libc::setsid() == -1 { - return Err(std::io::Error::last_os_error()); - } - Ok(()) - }); - } - Self { - child: command.spawn().unwrap(), - master: File::from(pty.master), - slave, - original, - output: vec![], - cursor_answered: false, - } - } - fn pump(&mut self) { - let mut fd = [PollFd::new(self.master.as_fd(), PollFlags::POLLIN)]; - poll(&mut fd, 20_u16).unwrap(); - if fd[0].revents().unwrap().contains(PollFlags::POLLIN) { - let mut buffer = [0; 65536]; - if let Ok(count) = self.master.read(&mut buffer) { - self.output.extend_from_slice(&buffer[..count]); - if !self.cursor_answered && self.output.windows(4).any(|bytes| bytes == b"\x1b[6n") - { - self.master.write_all(b"\x1b[1;1R").unwrap(); - self.cursor_answered = true; - } - } - } - } - fn await_text(&mut self, text: &str) { - let deadline = Instant::now() + Duration::from_secs(5); - while !String::from_utf8_lossy(&self.output).contains(text) { - self.pump(); - assert!( - Instant::now() < deadline, - "missing {text:?}: {}", - String::from_utf8_lossy(&self.output) - ); - assert!( - self.child.try_wait().unwrap().is_none(), - "TUI exited before {text:?}: {}", - String::from_utf8_lossy(&self.output) - ); - } - } - fn keys(&mut self, keys: &[u8]) { - self.master.write_all(keys).unwrap(); - } - fn signal(&self, signal: Signal) { - kill(Pid::from_raw(self.child.id() as i32), signal).unwrap(); - } - fn finish(&mut self, success: bool) -> ExitStatus { - let deadline = Instant::now() + Duration::from_secs(5); - let status = loop { - self.pump(); - if let Some(status) = self.child.try_wait().unwrap() { - break status; - } - assert!( - Instant::now() < deadline, - "TUI did not finish: {}", - String::from_utf8_lossy(&self.output[self.output.len().saturating_sub(3000)..]) - ); - }; - self.pump(); - assert_eq!( - status.success(), - success, - "{}", - String::from_utf8_lossy(&self.output) - ); - let restored = tcgetattr(&self.master).unwrap(); - assert_eq!(restored, self.original, "raw mode was not restored"); - let output = String::from_utf8_lossy(&self.output); - assert!( - output.contains("\x1b[?1049l"), - "alternate screen was not left: {output}" - ); - assert!( - output.contains("\x1b[?25h"), - "cursor was not restored: {output}" - ); - status - } -} -impl Drop for TuiProcess { - fn drop(&mut self) { - let _ = self.child.kill(); - let _ = self.child.wait(); - } -} -fn tui(target: &str, duration: &str) -> Command { - let mut cmd = Command::new(env!("CARGO_BIN_EXE_irtt-tui")); - cmd.args(["--duration", duration, "--interval", "20ms", target]); - cmd -} - -#[test] -fn terminal_controls_and_exit_paths_restore_the_terminal() { - let server = in_tree_server::InTreeServer::start(ServerConfig::default()); - let target = server.addr.to_string(); - let mut command = tui(&format!("first={target}"), "0"); - command.arg(format!("second={target}")); - let mut process = TuiProcess::start(command); - process.await_text("active"); - process.keys(b"p"); - process.await_text("paused"); - let size = Winsize { - ws_row: 12, - ws_col: 50, - ws_xpixel: 0, - ws_ypixel: 0, - }; - assert_eq!( - unsafe { nix::libc::ioctl(process.slave.as_raw_fd(), nix::libc::TIOCSWINSZ, &size) }, - 0 - ); - process.signal(Signal::SIGWINCH); - process.await_text("small"); - let size = Winsize { - ws_row: 40, - ws_col: 160, - ws_xpixel: 0, - ws_ypixel: 0, - }; - assert_eq!( - unsafe { nix::libc::ioctl(process.slave.as_raw_fd(), nix::libc::TIOCSWINSZ, &size) }, - 0 - ); - process.signal(Signal::SIGWINCH); - process.keys(b"\td"); - process.await_text("arrows"); - // Each key waits for a real draw, avoiding a timing-dependent escape batch. - for key in [ - b"\x1b[B".as_slice(), - b"\x1b[C", - b"\x1b[6~", - b"\x1b[F", - b"\x1b[H", - b"d", - b"m", - b"+", - b"-", - b"0", - b"\x1b[D", - b"\x1b[C", - b"\x1b[5~", - b"\x1b[6~", - b"\x1b[H", - b"\x1b[F", - b"r", - b"p", - ] { - let frames = process - .output - .windows(6) - .filter(|bytes| *bytes == b"\x1b[?25l") - .count(); - process.keys(key); - let deadline = Instant::now() + Duration::from_secs(5); - while process - .output - .windows(6) - .filter(|bytes| *bytes == b"\x1b[?25l") - .count() - == frames - { - process.pump(); - assert!(Instant::now() < deadline, "input did not redraw: {key:?}"); - } - } - process.keys(b"q"); - process.finish(true); - - for action in [None, Some(b"\x03".as_slice()), Some(b"signal".as_slice())] { - let mut process = - TuiProcess::start(tui(&target, if action.is_none() { "100ms" } else { "0" })); - process.await_text("active"); - match action { - Some(b"signal") => process.signal(Signal::SIGTERM), - Some(keys) => process.keys(keys), - None => {} - } - process.finish(true); - } - // An adverse early-close proxy forwards the real Rust server's OPEN and - // echo bytes. Only the documented CLOSE flag is added to the first echo. - // This exercises the application's continuous peer-close exit policy. - std::thread::scope(|scope| { - use irtt_proto::{decode_request, DecodedRequestKind, FLAG_CLOSE}; - use std::net::UdpSocket; - let proxy = UdpSocket::bind("127.0.0.1:0").unwrap(); - let proxy_addr = proxy.local_addr().unwrap(); - proxy - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let upstream = UdpSocket::bind("127.0.0.1:0").unwrap(); - upstream.connect(server.addr).unwrap(); - upstream - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let peer = scope.spawn(move || { - let mut packet = [0; 65536]; - loop { - let (count, client) = proxy.recv_from(&mut packet).unwrap(); - let echo = matches!( - decode_request(&packet[..count]).unwrap().kind, - DecodedRequestKind::Echo { .. } - ); - upstream.send(&packet[..count]).unwrap(); - let count = upstream.recv(&mut packet).unwrap(); - if echo { - packet[3] |= FLAG_CLOSE; - } - proxy.send_to(&packet[..count], client).unwrap(); - if echo { - break; - } - } - }); - let mut process = TuiProcess::start(tui(&proxy_addr.to_string(), "0")); - process.finish(false); - assert!(String::from_utf8_lossy(&process.output) - .contains("continuous run ended because of peer closure")); - peer.join().unwrap(); - }); - - let mut command = Command::new(std::env::current_exe().unwrap()); - command - .args(["--exact", "cancelled_frontend_fixture", "--nocapture"]) - .env("IRTT_TUI_CANCEL_TARGET", &target); - TuiProcess::start(command).finish(true); - - // Opening failure must be reported and restore the terminal too. - let mut process = TuiProcess::start(tui("127.0.0.1:99999", "100ms")); - process.finish(false); - assert!(String::from_utf8_lossy(&process.output) - .contains("no managed target completed successfully")); -} - -// The parent runs this fixture in its own PTY. Dropping the -// actual frontend future must restore that terminal and join measurement. -#[test] -fn cancelled_frontend_fixture() { - let Ok(target) = std::env::var("IRTT_TUI_CANCEL_TARGET") else { - return; - }; - use clap::Parser; - use std::{ - future::{poll_fn, Future}, - task::Poll, - }; - let args = irtt_app::cmd::tui::TuiArgs::parse_from(["irtt-tui", "--duration", "0", &target]); - let (_shutdown, receiver) = tokio::sync::watch::channel(false); - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap() - .block_on(async { - let mut frontend = Box::pin(irtt_app::cmd::tui::run_tui(args, receiver)); - poll_fn(|cx| { - assert!(frontend.as_mut().poll(cx).is_pending()); - Poll::Ready(()) - }) - .await; - drop(frontend); - }); -}