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); + }); +}