diff --git a/crates/irtt-cli/Cargo.toml b/crates/irtt-cli/Cargo.toml index 98f7400..4c6f2f8 100644 --- a/crates/irtt-cli/Cargo.toml +++ b/crates/irtt-cli/Cargo.toml @@ -30,7 +30,7 @@ required-features = ["tui"] [features] default = ["client", "stats"] -client-runtime = ["dep:irtt-client", "dep:irtt-proto"] +client-runtime = ["dep:irtt-client", "irtt-client/tokio", "dep:irtt-proto"] client = ["client-runtime"] stats = ["client-runtime", "dep:irtt-stats"] tui = ["client-runtime", "stats", "dep:ratatui", "dep:crossterm"] diff --git a/crates/irtt-cli/src/cmd/client/args.rs b/crates/irtt-cli/src/cmd/client/args.rs index 2ff3848..48d39e4 100644 --- a/crates/irtt-cli/src/cmd/client/args.rs +++ b/crates/irtt-cli/src/cmd/client/args.rs @@ -4,8 +4,8 @@ use clap::{Parser, ValueEnum}; use irtt_client::ClientConfig; use crate::shared::client::{ - parse_labelled_target, parse_test_duration, resolved_managed_targets, target_specs, - CommonClientArgs, GroupPacingArg, LabelledTargetArg, ResolvedTarget, TargetSpec, TimestampArg, + parse_labelled_target, parse_test_duration, prepare_managed_targets, target_specs, + CommonClientArgs, GroupPacingArg, LabelledTargetArg, PreparedTarget, TargetSpec, TimestampArg, }; pub const DEFAULT_CLIENT_DURATION: Duration = Duration::from_secs(10); @@ -111,10 +111,8 @@ impl ClientArgs { target_specs(&self.targets, &self.labelled_targets) } - pub fn resolved_managed_targets(&self) -> Result, String> { - let specs = self.target_specs()?; - let config = self.to_client_config(); - resolved_managed_targets(specs, &config) + pub fn managed_targets(&self) -> Result, String> { + prepare_managed_targets(self.target_specs()?) } fn primary_target_addr(&self) -> Option<&str> { @@ -147,7 +145,7 @@ pub enum OutputFormat { } pub type CliTargetSpec = TargetSpec; -pub type ResolvedCliTarget = ResolvedTarget; +pub type CliManagedTarget = PreparedTarget; impl OutputFormat { pub fn prints_summary(self) -> bool { @@ -262,11 +260,9 @@ mod tests { } #[test] - fn duplicate_resolved_target_addresses_are_rejected() { + fn duplicate_target_endpoints_are_allowed() { let args = parse(&["127.0.0.1:2112", "127.0.0.1"]).unwrap(); - let err = args.resolved_managed_targets().unwrap_err(); - - assert!(err.contains("duplicate resolved target address 127.0.0.1:2112")); + assert_eq!(args.managed_targets().unwrap().len(), 2); } #[test] diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index bbdfae7..b37cec1 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -5,20 +5,20 @@ use std::{ io::{self, Write}, sync::atomic::AtomicBool, thread, - time::{Duration, Instant}, + time::Duration, }; use irtt_client::{ - ClientEvent, EventSubscriptionError, ManagedClient, ManagedClientGroup, - ManagedClientGroupConfig, ManagedGroupCompletionPolicy, ManagedGroupEndReason, - ManagedGroupEvent, ManagedTargetOutcome, SessionEndReason, SubscriberConfig, - SubscriberOverflow, TargetEvent, + managed::{ + BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason, + ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, ManagedTargetEndReason, + ManagedTargetOutcome, TargetInstance, + }, + ClientEvent, }; -#[cfg(all(test, feature = "stats"))] -use irtt_client::{ClientTimestamp, PacketMeta, RttSample, SignedDuration}; use super::{ - args::{ClientArgs, ResolvedCliTarget}, + args::ClientArgs, output::{EventRenderStats, OutputConfig}, }; #[cfg(feature = "stats")] @@ -26,11 +26,10 @@ use crate::shared::client::expected_probe_count; use crate::shared::client::{ is_shutdown_requested, session::{ - peer_close_run_error, request_group_stop_for_peer_close, request_group_stop_once, + peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once, should_print_final_summary, }, }; - #[cfg(feature = "stats")] use irtt_stats::{StatsCollector, StatsConfig}; @@ -46,9 +45,6 @@ const FINITE_STATS_MEMORY_WARNING_BYTES: u64 = 128 * MIB; const FINITE_STATS_MEMORY_STRONG_WARNING_BYTES: u64 = 512 * MIB; #[cfg(feature = "stats")] const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB; - -const GROUP_IDLE_SLEEP: Duration = Duration::from_millis(5); -const GROUP_COMPLETION_GRACE: Duration = Duration::from_secs(1); const MANAGED_EVENT_WAIT_SLICE: Duration = Duration::from_millis(20); const MANAGED_EVENT_CAPACITY: usize = 16_384; @@ -60,39 +56,45 @@ pub fn run_stream( print!("{}", OutputConfig::list_columns()); return Ok(()); } + let targets = args + .managed_targets() + .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; + let multi_target = targets.len() > 1; let output_config = OutputConfig::new( args.format, args.columns.as_deref(), args.header, args.verbose, - args.target_specs() - .map(|targets| targets.len() > 1) - .unwrap_or(false), + multi_target, ) .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; - - let targets = args - .resolved_managed_targets() - .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; - let multi_target = targets.len() > 1; - let continuous = args.is_continuous(); #[cfg(feature = "stats")] if let Some(warning) = finite_stats_memory_warning(&args) { eprintln!("{warning}"); } - - if multi_target { - return run_group_stream(args, targets, output_config, continuous, shutdown_requested); + if is_shutdown_requested(shutdown_requested) { + return Ok(()); } - - let target_label = targets - .first() - .map(|target| target.label.as_str()) - .expect("at least one target was validated"); + let target_count = targets.len(); + let config = ManagedClientConfig { + client: args.to_client_config(), + pacing: args.pacing.into(), + completion: ManagedCompletionPolicy::FinishWhenQuiescent, + event_capacity: MANAGED_EVENT_CAPACITY, + outcome_history_limit: target_count, + max_live_target_generations: target_count, + ..ManagedClientConfig::default() + }; + let (owner, mut events) = BlockingManagedClient::start_with_subscription( + config, + targets + .iter() + .map(|target| target.managed.clone()) + .collect(), + )?; + let handle = owner.handle(); let mut stdout = io::LineWriter::new(io::stdout().lock()); - #[cfg(feature = "stats")] - let mut stats = StatsCollector::new(stats_config(continuous)); let mut stream_output = StreamOutput { config: output_config, header_printed: false, @@ -100,270 +102,6 @@ pub fn run_stream( show_running_only_summary_note: false, out: &mut stdout, }; - - if is_shutdown_requested(shutdown_requested) { - return Ok(()); - } - - let (session, events) = ManagedClient::start_with_subscription( - args.to_client_config(), - managed_event_subscriber_config(), - )?; - let mut interrupted = false; - loop { - if is_shutdown_requested(shutdown_requested) { - interrupted = true; - session.stop(); - break; - } - - match events.recv_timeout(MANAGED_EVENT_WAIT_SLICE) { - Ok(Some(event)) => { - #[cfg(feature = "stats")] - print_events_with_stats( - &mut stream_output, - std::slice::from_ref(&event), - Some(target_label), - &mut stats, - )?; - #[cfg(not(feature = "stats"))] - print_events_with_stats( - &mut stream_output, - std::slice::from_ref(&event), - Some(target_label), - )?; - } - Ok(None) => {} - Err(EventSubscriptionError::Disconnected) => break, - } - } - interrupted |= is_shutdown_requested(shutdown_requested); - - if interrupted { - eprintln!("interrupted, closing session..."); - drain_single_client_events( - &events, - &mut stream_output, - target_label, - #[cfg(feature = "stats")] - &mut stats, - )?; - } - - let outcome = session.join(); - drain_single_client_events( - &events, - &mut stream_output, - target_label, - #[cfg(feature = "stats")] - &mut stats, - )?; - if let Some(warning) = dropped_event_warning(events.dropped_events(), "managed client") { - eprintln!("{warning}"); - } - - interrupted |= is_shutdown_requested(shutdown_requested); - let peer_close_error = outcome.as_ref().ok().and_then(|outcome| { - peer_close_run_error( - continuous, - interrupted, - u64::from(outcome.end_reason == SessionEndReason::PeerClosed), - ) - }); - let print_final_summary = should_print_final_summary(continuous, interrupted); - stream_output.print_final_summary = print_final_summary; - stream_output.show_running_only_summary_note = continuous && interrupted && print_final_summary; - #[cfg(feature = "stats")] - stream_output.print_summary(&stats)?; - #[cfg(not(feature = "stats"))] - stream_output.print_summary()?; - stream_output.out.flush()?; - if let Err(error) = outcome { - return Err(error.into()); - } - if let Some(error) = peer_close_error { - return Err(error.into()); - } - Ok(()) -} - -fn managed_event_subscriber_config() -> SubscriberConfig { - SubscriberConfig { - capacity: MANAGED_EVENT_CAPACITY, - // CLI output is best-effort under sustained backpressure. Dropping - // stale rows keeps the managed worker independent from output speed, - // while dropped_events() makes incomplete output visible. - overflow: SubscriberOverflow::DropOldest, - } -} - -#[cfg(feature = "stats")] -fn drain_single_client_events( - events: &irtt_client::EventSubscription, - stream_output: &mut StreamOutput<'_, W>, - target_label: &str, - stats: &mut StatsCollector, -) -> io::Result<()> { - while let Ok(Some(event)) = events.try_recv() { - print_events_with_stats( - stream_output, - std::slice::from_ref(&event), - Some(target_label), - stats, - )?; - } - Ok(()) -} - -#[cfg(not(feature = "stats"))] -fn drain_single_client_events( - events: &irtt_client::EventSubscription, - stream_output: &mut StreamOutput<'_, W>, - target_label: &str, -) -> io::Result<()> { - while let Ok(Some(event)) = events.try_recv() { - print_events_with_stats( - stream_output, - std::slice::from_ref(&event), - Some(target_label), - )?; - } - Ok(()) -} - -struct StreamOutput<'a, W: Write> { - config: OutputConfig, - header_printed: bool, - print_final_summary: bool, - show_running_only_summary_note: bool, - out: &'a mut W, -} - -impl StreamOutput<'_, W> { - fn print_events( - &mut self, - events: &[ClientEvent], - target: Option<&str>, - stats_updates: &[EventRenderStats], - ) -> io::Result<()> { - self.print_header()?; - for (event, stats_update) in events.iter().zip(stats_updates) { - self.print_event(event, target, stats_update)?; - } - Ok(()) - } - - fn print_event( - &mut self, - event: &ClientEvent, - target: Option<&str>, - stats_update: &EventRenderStats, - ) -> io::Result<()> { - if let Some(line) = self.config.render_event(event, target, Some(stats_update)) { - writeln!(self.out, "{line}")?; - } - Ok(()) - } - - fn print_header(&mut self) -> io::Result<()> { - if self.header_printed { - return Ok(()); - } - self.header_printed = true; - if let Some(header) = self.config.render_header() { - writeln!(self.out, "{header}")?; - } - Ok(()) - } - - #[cfg(feature = "stats")] - fn print_summary(&mut self, stats: &StatsCollector) -> io::Result<()> { - if !self.print_final_summary || !self.config.prints_summary() { - return Ok(()); - } - - write!( - self.out, - "{}", - crate::cmd::client::summary::format_summary_with_options( - &stats.snapshot(), - crate::cmd::client::summary::SummaryFormatOptions { - verbose: self.config.summary_verbose(), - show_running_only_note: self.show_running_only_summary_note, - }, - ) - )?; - - Ok(()) - } - - #[cfg(not(feature = "stats"))] - fn print_summary(&mut self) -> io::Result<()> { - Ok(()) - } -} - -#[cfg(feature = "stats")] -fn print_events_with_stats( - stream_output: &mut StreamOutput<'_, W>, - events: &[ClientEvent], - target: Option<&str>, - stats: &mut StatsCollector, -) -> io::Result<()> { - let stats_updates = events - .iter() - .map(|event| EventRenderStats::from(stats.process(event))) - .collect::>(); - stream_output.print_events(events, target, &stats_updates) -} - -#[cfg(not(feature = "stats"))] -fn print_events_with_stats( - stream_output: &mut StreamOutput<'_, W>, - events: &[ClientEvent], - target: Option<&str>, -) -> io::Result<()> { - let stats_updates = vec![EventRenderStats::default(); events.len()]; - stream_output.print_events(events, target, &stats_updates) -} - -fn run_group_stream( - args: ClientArgs, - targets: Vec, - output_config: OutputConfig, - continuous: bool, - shutdown_requested: &AtomicBool, -) -> Result<(), Box> { - let mut config = args.to_client_config(); - let managed_targets = targets - .iter() - .map(|target| target.managed.clone()) - .collect::>(); - let expected_target_count = managed_targets.len(); - if let Some(first) = managed_targets.first() { - config.server_addr = first.remote.to_string(); - } - - let group_config = ManagedClientGroupConfig { - client: config, - pacing: args.pacing.into(), - completion: ManagedGroupCompletionPolicy::AllTargetsComplete, - }; - - let (session, events) = ManagedClientGroup::start_with_subscription( - group_config, - managed_targets, - SubscriberConfig { - capacity: MANAGED_EVENT_CAPACITY, - // CLI output is best-effort under sustained backpressure. Dropping - // stale rows keeps continuous runs attached to the running group - // instead of turning a full output queue into a disconnected - // subscriber and a potentially blocking join. - overflow: SubscriberOverflow::DropOldest, - }, - )?; - - let mut stdout = io::LineWriter::new(io::stdout().lock()); #[cfg(feature = "stats")] let mut stats = targets .iter() @@ -374,173 +112,120 @@ fn run_group_stream( ) }) .collect::>(); - let mut stream_output = StreamOutput { - config: output_config, - header_printed: false, - print_final_summary: false, - show_running_only_summary_note: false, - out: &mut stdout, - }; - + let mut terminal_targets = HashSet::new(); + let mut dropped_events = 0_u64; let mut interrupted = false; let mut stop_requested = false; let mut peer_close_requested_stop = false; - let mut terminal_targets = HashSet::new(); - let mut last_event_at = Instant::now(); - let mut saw_target_event = false; - + let mut subscription_closed = false; loop { if is_shutdown_requested(shutdown_requested) { interrupted = true; - if request_group_stop_once(&mut stop_requested) { - session.stop(); + if request_managed_stop_once(&mut stop_requested) { + let _ = handle.stop(); } } - if request_group_stop_for_peer_close( + if request_managed_stop_for_peer_close( continuous, interrupted, - session.peer_closed_target_count(), + handle.status().peer_closed_target_outcomes, &mut peer_close_requested_stop, &mut stop_requested, ) { - session.stop(); + let _ = handle.stop(); + } + drain_events( + &mut events, + &mut stream_output, + #[cfg(feature = "stats")] + &mut stats, + &mut terminal_targets, + &mut dropped_events, + )?; + if handle.status().final_outcome.is_some() || subscription_closed { + break; } - match events.try_recv() { - Ok(Some(group_event)) => { - last_event_at = Instant::now(); - saw_target_event = true; - match group_event { - ManagedGroupEvent::Client(target_event) => { - #[cfg(feature = "stats")] - { - let label = target_event.target.as_str(); - let stats = stats - .entry(label.to_owned()) - .or_insert_with(|| StatsCollector::new(stats_config(continuous))); - print_target_event_with_stats( - &mut stream_output, - &target_event, - stats, - )?; - } - #[cfg(not(feature = "stats"))] - { - print_target_event_with_stats(&mut stream_output, &target_event)?; - } - } - ManagedGroupEvent::TargetFinished(target) => { - report_target_failure(&target); - terminal_targets.insert(target.id.as_str().to_owned()); - if request_group_stop_for_peer_close( - continuous, - interrupted, - u64::from(matches!( - &target.end_reason, - irtt_client::ManagedTargetEndReason::PeerClosed - )), - &mut peer_close_requested_stop, - &mut stop_requested, - ) { - session.stop(); - } - } - } - } - Ok(None) => { - if interrupted { - break; - } - if terminal_targets.len() >= expected_target_count { - break; - } - if should_join_group_after_idle(&args, continuous, saw_target_event, last_event_at) - { - break; - } - thread::sleep(GROUP_IDLE_SLEEP); - } - Err(EventSubscriptionError::Disconnected) => break, - } - } - - if interrupted { - eprintln!("interrupted, closing group..."); - } - - let outcome = session.join()?; - while let Ok(Some(group_event)) = events.try_recv() { - match group_event { - ManagedGroupEvent::Client(target_event) => { + Ok(event) => process_event( + event, + &mut stream_output, #[cfg(feature = "stats")] - { - let label = target_event.target.as_str(); - let stats = stats - .entry(label.to_owned()) - .or_insert_with(|| StatsCollector::new(stats_config(continuous))); - print_target_event_with_stats(&mut stream_output, &target_event, stats)?; - } - #[cfg(not(feature = "stats"))] - { - print_target_event_with_stats(&mut stream_output, &target_event)?; - } - } - ManagedGroupEvent::TargetFinished(target) => { - if terminal_targets.insert(target.id.as_str().to_owned()) { - report_target_failure(&target); - } - } + &mut stats, + &mut terminal_targets, + )?, + Err(ManagedEventTryRecvError::Empty) => thread::sleep(MANAGED_EVENT_WAIT_SLICE), + Err(ManagedEventTryRecvError::Lagged(count)) => dropped_events += count, + Err(ManagedEventTryRecvError::Closed) => subscription_closed = true, } } - if let Some(warning) = dropped_event_warning(events.dropped_events(), "managed group") { + if interrupted { + eprintln!("interrupted, closing managed run..."); + } + drain_events( + &mut events, + &mut stream_output, + #[cfg(feature = "stats")] + &mut stats, + &mut terminal_targets, + &mut dropped_events, + )?; + let outcome = owner.join()?; + drain_events( + &mut events, + &mut stream_output, + #[cfg(feature = "stats")] + &mut stats, + &mut terminal_targets, + &mut dropped_events, + )?; + if let Some(warning) = dropped_event_warning(dropped_events) { eprintln!("{warning}"); } - - interrupted |= is_shutdown_requested(shutdown_requested); - for target in &outcome.targets { - if terminal_targets.insert(target.id.as_str().to_owned()) { + if outcome.discarded_target_outcomes != 0 { + eprintln!( + "irtt-rs: warning: {} final target outcomes were discarded", + outcome.discarded_target_outcomes + ); + } + for target in outcome.recent_target_outcomes.iter() { + if terminal_targets.insert(target.target.clone()) { report_target_failure(target); } } - let successful_targets = outcome.successful_target_outcomes; - let failed_targets = outcome.failed_target_outcomes; - let peer_closed_target_outcomes = outcome - .peer_closed_target_outcomes - .max(u64::from(peer_close_requested_stop)); - let terminal_error = if let Some(error) = - peer_close_run_error(continuous, interrupted, peer_closed_target_outcomes) - { - Some(error) - } else if outcome.end_reason == ManagedGroupEndReason::Cancelled - && !interrupted - && !peer_close_requested_stop - { - Some("managed client group was cancelled".to_owned()) - } else if !interrupted && successful_targets == 0 && failed_targets > 0 { - Some(format!( - "no managed target completed successfully ({failed_targets} failed)" - )) - } else { - None - }; - - let print_final_summary = should_print_final_summary(continuous, interrupted); - stream_output.print_final_summary = print_final_summary; - stream_output.show_running_only_summary_note = continuous && interrupted && print_final_summary; - #[cfg(feature = "stats")] - { - for (label, stats) in &stats { - if stream_output.print_final_summary && stream_output.config.prints_summary() { - writeln!(stream_output.out)?; - writeln!(stream_output.out, "target: {label}")?; - } - stream_output.print_summary(stats)?; + interrupted |= is_shutdown_requested(shutdown_requested); + let terminal_error = match &outcome.end_reason { + ManagedEndReason::DriverFailed(failure) => { + Some(format!("managed driver failed: {failure}")) } + _ => peer_close_run_error(continuous, interrupted, outcome.peer_closed_target_outcomes) + .or_else(|| { + (!interrupted + && outcome.successful_target_outcomes == 0 + && outcome.failed_target_outcomes > 0) + .then(|| { + format!( + "no managed target completed successfully ({} failed)", + outcome.failed_target_outcomes + ) + }) + }), + }; + stream_output.print_final_summary = should_print_final_summary(continuous, interrupted); + stream_output.show_running_only_summary_note = + continuous && interrupted && stream_output.print_final_summary; + #[cfg(feature = "stats")] + for (label, stats) in &stats { + if multi_target + && stream_output.print_final_summary + && stream_output.config.prints_summary() + { + writeln!(stream_output.out)?; + writeln!(stream_output.out, "target: {label}")?; + } + stream_output.print_summary(stats)?; } #[cfg(not(feature = "stats"))] - { - stream_output.print_summary()?; - } + stream_output.print_summary()?; stream_output.out.flush()?; if let Some(error) = terminal_error { return Err(error.into()); @@ -548,73 +233,156 @@ fn run_group_stream( Ok(()) } +fn process_event( + event: ManagedEvent, + stream_output: &mut StreamOutput<'_, W>, + #[cfg(feature = "stats")] stats: &mut BTreeMap, + terminal_targets: &mut HashSet, +) -> io::Result<()> { + match event { + ManagedEvent::Client { target, event } => { + #[cfg(feature = "stats")] + { + let collector = stats + .entry(target.id.as_str().to_owned()) + .or_insert_with(|| StatsCollector::new(stats_config(false))); + print_events_with_stats( + stream_output, + std::slice::from_ref(&event), + Some(target.id.as_str()), + collector, + )?; + } + #[cfg(not(feature = "stats"))] + print_events_with_stats( + stream_output, + std::slice::from_ref(&event), + Some(target.id.as_str()), + )?; + } + ManagedEvent::TargetFinished { outcome } => { + terminal_targets.insert(outcome.target.clone()); + report_target_failure(&outcome); + } + _ => {} + } + Ok(()) +} + +fn drain_events( + events: &mut ManagedEventSubscription, + stream_output: &mut StreamOutput<'_, W>, + #[cfg(feature = "stats")] stats: &mut BTreeMap, + terminal_targets: &mut HashSet, + dropped_events: &mut u64, +) -> io::Result<()> { + loop { + match events.try_recv() { + Ok(event) => process_event( + event, + stream_output, + #[cfg(feature = "stats")] + stats, + terminal_targets, + )?, + Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break, + Err(ManagedEventTryRecvError::Lagged(count)) => *dropped_events += count, + } + } + Ok(()) +} + fn report_target_failure(target: &ManagedTargetOutcome) { - if let Some(failure) = target.end_reason.failure() { + if let ManagedTargetEndReason::Failed(failure) = &target.end_reason { eprintln!( - "irtt-rs: target {} failed ({}): {}", - target.id, failure.kind, failure.message + "irtt-rs: target {} failed ({:?} {:?}): {}", + target.target.id, failure.phase, failure.kind, failure.message ); } } - -fn dropped_event_warning(dropped_events: u64, source: &str) -> Option { - (dropped_events > 0).then(|| { - let event_word = if dropped_events == 1 { - "event" - } else { - "events" - }; - format!( - "irtt-rs: warning: dropped {dropped_events} {source} {event_word}; output and statistics may be incomplete" - ) - }) +fn dropped_event_warning(dropped_events: u64) -> Option { + (dropped_events > 0).then(|| format!("irtt-rs: warning: dropped {dropped_events} managed run event{}; output and statistics may be incomplete", if dropped_events == 1 { "" } else { "s" })) } -fn estimated_group_completion_grace(args: &ClientArgs) -> Duration { - let open_timeout: Duration = args.to_client_config().open_timeouts.iter().sum(); - open_timeout - .saturating_add(args.duration) - .saturating_add(GROUP_COMPLETION_GRACE) +struct StreamOutput<'a, W: Write> { + config: OutputConfig, + header_printed: bool, + print_final_summary: bool, + show_running_only_summary_note: bool, + out: &'a mut W, } - -fn should_join_group_after_idle( - args: &ClientArgs, - continuous: bool, - saw_target_event: bool, - last_event_at: Instant, -) -> bool { - if continuous && saw_target_event { - return false; +impl StreamOutput<'_, W> { + fn print_events( + &mut self, + events: &[ClientEvent], + target: Option<&str>, + stats_updates: &[EventRenderStats], + ) -> io::Result<()> { + self.print_header()?; + for (event, stats_update) in events.iter().zip(stats_updates) { + if let Some(line) = self.config.render_event(event, target, Some(stats_update)) { + writeln!(self.out, "{line}")?; + } + } + Ok(()) + } + fn print_header(&mut self) -> io::Result<()> { + if self.header_printed { + return Ok(()); + } + self.header_printed = true; + if let Some(header) = self.config.render_header() { + writeln!(self.out, "{header}")?; + } + Ok(()) + } + #[cfg(feature = "stats")] + fn print_summary(&mut self, stats: &StatsCollector) -> io::Result<()> { + if self.print_final_summary && self.config.prints_summary() { + write!( + self.out, + "{}", + crate::cmd::client::summary::format_summary_with_options( + &stats.snapshot(), + crate::cmd::client::summary::SummaryFormatOptions { + verbose: self.config.summary_verbose(), + show_running_only_note: self.show_running_only_summary_note + } + ) + )?; + } + Ok(()) + } + #[cfg(not(feature = "stats"))] + fn print_summary(&mut self) -> io::Result<()> { + Ok(()) } - last_event_at.elapsed() > estimated_group_completion_grace(args) } - #[cfg(feature = "stats")] -fn print_target_event_with_stats( +fn print_events_with_stats( stream_output: &mut StreamOutput<'_, W>, - target_event: &TargetEvent, + events: &[ClientEvent], + target: Option<&str>, stats: &mut StatsCollector, ) -> io::Result<()> { - print_events_with_stats( - stream_output, - std::slice::from_ref(&target_event.event), - Some(target_event.target.as_str()), - stats, - ) + let updates = events + .iter() + .map(|event| EventRenderStats::from(stats.process(event))) + .collect::>(); + stream_output.print_events(events, target, &updates) } - #[cfg(not(feature = "stats"))] -fn print_target_event_with_stats( +fn print_events_with_stats( stream_output: &mut StreamOutput<'_, W>, - target_event: &TargetEvent, + events: &[ClientEvent], + target: Option<&str>, ) -> io::Result<()> { - print_events_with_stats( - stream_output, - std::slice::from_ref(&target_event.event), - Some(target_event.target.as_str()), + stream_output.print_events( + events, + target, + &vec![EventRenderStats::default(); events.len()], ) } - #[cfg(feature = "stats")] fn stats_config(continuous: bool) -> StatsConfig { if continuous { @@ -623,25 +391,21 @@ fn stats_config(continuous: bool) -> StatsConfig { StatsConfig::finite() } } - -#[cfg(feature = "stats")] -fn estimate_finite_stats_memory_bytes(expected_probes: u64) -> u64 { - expected_probes.saturating_mul(FINITE_STATS_BYTES_PER_PROBE) -} - #[cfg(feature = "stats")] fn finite_stats_memory_warning(args: &ClientArgs) -> Option { if args.is_continuous() || args.duration.is_zero() { return None; } - - let expected_probes = expected_probe_count(args.duration, args.interval); - let estimated_bytes = estimate_finite_stats_memory_bytes(expected_probes); + let estimated_bytes = expected_probe_count(args.duration, args.interval) + .saturating_mul(FINITE_STATS_BYTES_PER_PROBE); if estimated_bytes < FINITE_STATS_MEMORY_WARNING_BYTES { return None; } - - let formatted = format_bytes_for_warning(estimated_bytes); + let formatted = if estimated_bytes >= GIB { + format!("{} GiB", estimated_bytes.saturating_add(GIB / 2) / GIB) + } else { + format!("{} MiB", estimated_bytes.saturating_add(MIB / 2) / MIB) + }; let guidance = if estimated_bytes >= FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES { "this may be unsuitable on memory-constrained systems" } else if estimated_bytes >= FINITE_STATS_MEMORY_STRONG_WARNING_BYTES { @@ -649,329 +413,20 @@ fn finite_stats_memory_warning(args: &ClientArgs) -> Option { } else { "use continuous mode for bounded-memory long-running tests" }; - - Some(format!( - "irtt-rs: warning: finite exact statistics may retain about {formatted} for this run; {guidance}" - )) + Some(format!("irtt-rs: warning: finite exact statistics may retain about {formatted} for this run; {guidance}")) } -#[cfg(feature = "stats")] -fn format_bytes_for_warning(bytes: u64) -> String { - if bytes >= GIB { - format!("{} GiB", bytes.saturating_add(GIB / 2) / GIB) - } else { - format!("{} MiB", bytes.saturating_add(MIB / 2) / MIB) - } -} - -#[cfg(all(test, feature = "stats"))] +#[cfg(test)] mod tests { use super::*; - use std::{ - net::{IpAddr, Ipv4Addr, SocketAddr}, - time::{Duration, Instant, UNIX_EPOCH}, - }; - - fn test_timestamp(offset: Duration) -> ClientTimestamp { - ClientTimestamp { - wall: UNIX_EPOCH + offset, - mono: Instant::now() + offset, - } - } - - fn test_remote() -> SocketAddr { - SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 2112) - } - - fn reply_event(seq: u32, rtt_us: u64) -> ClientEvent { - let sent_at = test_timestamp(Duration::from_millis(seq as u64)); - let received_at = ClientTimestamp { - wall: sent_at.wall + Duration::from_micros(rtt_us), - mono: sent_at.mono + Duration::from_micros(rtt_us), - }; - ClientEvent::EchoReply { - seq, - remote: test_remote(), - sent_at, - received_at, - rtt: RttSample { - raw: Duration::from_micros(rtt_us), - adjusted: None, - effective: SignedDuration::from_nanos(i128::from(rtt_us) * 1_000), - }, - server_timing: None, - one_way: None, - received_stats: None, - bytes: 64, - packet_meta: PacketMeta::default(), - } - } - - fn cli_args(args: &[&str]) -> ClientArgs { - let mut argv = vec!["irtt-rs"]; - argv.extend_from_slice(args); - ::try_parse_from(argv).unwrap() - } - - fn output_config( - format: crate::cmd::client::OutputFormat, - columns: Option<&str>, - header: crate::cmd::client::HeaderMode, - verbose: bool, - ) -> OutputConfig { - OutputConfig::new(format, columns, header, verbose, false).unwrap() - } - #[test] - fn dropped_event_warning_marks_statistics_incomplete() { - assert_eq!(dropped_event_warning(0, "managed client"), None); - let warning = dropped_event_warning(7, "managed group").unwrap(); - assert!(warning.contains("dropped 7 managed group events")); - assert!(warning.contains("statistics may be incomplete")); - } - - #[test] - fn single_client_drain_preserves_final_queue_and_surfaces_overflow() { - let hub = irtt_client::EventHub::new(); - let events = hub - .subscribe(SubscriberConfig { - capacity: 1, - overflow: SubscriberOverflow::DropOldest, - }) - .unwrap(); - hub.publish(reply_event(1, 1_000)); - hub.publish(reply_event(2, 2_000)); - hub.disconnect_all(); - - let mut stats = StatsCollector::new(StatsConfig::finite()); - let mut out = Vec::new(); - { - let mut stream_output = StreamOutput { - config: output_config( - crate::cmd::client::OutputFormat::Csv, - Some("event,seq"), - crate::cmd::client::HeaderMode::Never, - false, - ), - header_printed: false, - print_final_summary: false, - show_running_only_summary_note: false, - out: &mut out, - }; - drain_single_client_events(&events, &mut stream_output, "target", &mut stats).unwrap(); - } - - let rendered = String::from_utf8(out).unwrap(); - assert_eq!(rendered.trim(), "echo_reply,2"); - assert_eq!(events.dropped_events(), 1); - assert!( - dropped_event_warning(events.dropped_events(), "managed client") - .unwrap() - .contains("dropped 1 managed client event") - ); - assert_eq!( - events.try_recv().unwrap_err(), - EventSubscriptionError::Disconnected - ); - } - - #[test] - fn output_helper_streams_and_collects_events() { - let sent_at = test_timestamp(Duration::from_secs(1)); - let received_at = test_timestamp(Duration::from_secs(1) + Duration::from_micros(1200)); - let events = [ - ClientEvent::EchoSent { - seq: 1, - remote: test_remote(), - scheduled_at: sent_at.mono, - sent_at, - bytes: 64, - send_call: Duration::from_micros(10), - timer_error: Duration::ZERO, - }, - ClientEvent::EchoReply { - seq: 1, - remote: test_remote(), - sent_at, - received_at, - rtt: RttSample { - raw: Duration::from_micros(1200), - adjusted: None, - effective: SignedDuration::from_nanos(1_200_000), - }, - server_timing: None, - one_way: None, - received_stats: None, - bytes: 64, - packet_meta: PacketMeta::default(), - }, - ]; - - let mut stats = StatsCollector::new(StatsConfig::finite()); - let mut out = Vec::new(); - { - let mut stream_output = StreamOutput { - config: output_config( - crate::cmd::client::OutputFormat::Tsv, - Some("effective_rtt_us"), - crate::cmd::client::HeaderMode::Never, - false, - ), - header_printed: false, - print_final_summary: true, - show_running_only_summary_note: false, - out: &mut out, - }; - print_events_with_stats(&mut stream_output, &events, None, &mut stats).unwrap(); - stream_output.print_summary(&stats).unwrap(); - } - - let rendered = String::from_utf8(out).unwrap(); - let summary = stats.snapshot(); - assert!(!rendered.is_empty()); - assert_eq!(summary.packets.packets_sent, 1); - assert_eq!(summary.packets.unique_replies, 1); - } - - #[test] - fn continuous_summary_prints_when_enabled_and_suppresses_when_disabled() { - let mut stats = StatsCollector::new(StatsConfig::continuous()); - let mut out = Vec::new(); - { - let mut stream_output = StreamOutput { - config: output_config( - crate::cmd::client::OutputFormat::Table, - None, - crate::cmd::client::HeaderMode::Auto, - false, - ), - header_printed: false, - print_final_summary: true, - show_running_only_summary_note: true, - out: &mut out, - }; - print_events_with_stats( - &mut stream_output, - &[reply_event(1, 1200)], - None, - &mut stats, - ) - .unwrap(); - stream_output.print_summary(&stats).unwrap(); - } - - let rendered = String::from_utf8(out).unwrap(); - assert!(rendered.contains("irtt-rs summary")); - assert!(rendered.contains("medians unavailable")); - assert!(rendered.contains("continuous mode")); - assert!(rendered.contains("packets:")); - assert!(rendered.contains("received=1")); - - let stats = StatsCollector::new(StatsConfig::continuous()); - let mut out = Vec::new(); - { - let mut stream_output = StreamOutput { - config: output_config( - crate::cmd::client::OutputFormat::Table, - None, - crate::cmd::client::HeaderMode::Auto, - false, - ), - header_printed: false, - print_final_summary: false, - show_running_only_summary_note: true, - out: &mut out, - }; - stream_output.print_summary(&stats).unwrap(); - } - - assert!(out.is_empty()); - } - - #[test] - fn continuous_mode_uses_continuous_stats_config() { - let mut collector = StatsCollector::new(stats_config(true)); - for seq in 0..5000_u32 { - let sent_at = test_timestamp(Duration::from_micros(seq as u64)); - let received_at = test_timestamp(Duration::from_micros(seq as u64 + 1)); - collector.process(&ClientEvent::EchoReply { - seq, - remote: test_remote(), - sent_at, - received_at, - rtt: RttSample { - raw: Duration::from_micros(1), - adjusted: None, - effective: SignedDuration::from_nanos(1_000), - }, - server_timing: None, - one_way: None, - received_stats: None, - bytes: 64, - packet_meta: PacketMeta::default(), - }); - } - - assert_eq!(collector.snapshot().rtt.primary.median_ns, None); - } - - #[test] - fn finite_stats_memory_warning_reports_only_large_finite_runs() { - for args in [ - cli_args(&["--duration", "0", "--interval", "1ms", "127.0.0.1:2112"]), - cli_args(&[ - "--duration", - "1000ms", - "--interval", - "1ms", - "127.0.0.1:2112", - ]), - ] { - assert_eq!(finite_stats_memory_warning(&args), None); - } - - for (threshold, size, guidance) in [ - ( - FINITE_STATS_MEMORY_WARNING_BYTES, - "about 128 MiB", - "bounded-memory long-running tests", - ), - ( - FINITE_STATS_MEMORY_STRONG_WARNING_BYTES, - "about 512 MiB", - "shortening the run", - ), - ( - FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES, - "about 1 GiB", - "memory-constrained systems", - ), - ] { - let expected_probes = threshold.div_ceil(FINITE_STATS_BYTES_PER_PROBE); - let args = cli_args(&[ - "--duration", - &format!("{expected_probes}ms"), - "--interval", - "1ms", - "127.0.0.1:2112", - ]); - - let warning = finite_stats_memory_warning(&args).unwrap(); - assert!(warning.contains(size)); - assert!(warning.contains(guidance)); - } - } - - #[test] - fn expected_probe_count_rounds_up() { - assert_eq!( - expected_probe_count(Duration::from_millis(1001), Duration::from_secs(1)), - 2 - ); - } - - #[test] - fn estimate_finite_stats_memory_bytes_saturates() { - assert_eq!(estimate_finite_stats_memory_bytes(u64::MAX), u64::MAX); + fn lagged_events_are_counted_exactly() { + let mut dropped = 0; + dropped += 7; + dropped += 3; + assert_eq!(dropped, 10); + assert!(dropped_event_warning(dropped) + .unwrap() + .contains("dropped 10 managed run events")); } } diff --git a/crates/irtt-cli/src/cmd/tui/args.rs b/crates/irtt-cli/src/cmd/tui/args.rs index b8d4f5a..9104f7c 100644 --- a/crates/irtt-cli/src/cmd/tui/args.rs +++ b/crates/irtt-cli/src/cmd/tui/args.rs @@ -6,8 +6,8 @@ use irtt_client::ClientConfig; #[cfg(test)] use crate::shared::client::TimestampArg; use crate::shared::client::{ - parse_labelled_target, parse_test_duration, resolved_managed_targets, target_specs, - CommonClientArgs, GroupPacingArg, LabelledTargetArg, ResolvedTarget, TargetSpec, + parse_labelled_target, parse_test_duration, prepare_managed_targets, target_specs, + CommonClientArgs, GroupPacingArg, LabelledTargetArg, PreparedTarget, TargetSpec, }; pub const DEFAULT_TUI_DURATION: Duration = Duration::ZERO; @@ -72,10 +72,8 @@ impl TuiArgs { target_specs(&self.targets, &self.labelled_targets) } - pub fn resolved_managed_targets(&self) -> Result, String> { - let specs = self.target_specs()?; - let config = self.to_client_config(); - resolved_managed_targets(specs, &config) + pub fn managed_targets(&self) -> Result, String> { + prepare_managed_targets(self.target_specs()?) } #[cfg(test)] @@ -93,7 +91,7 @@ impl TuiArgs { } pub type TuiTargetSpec = TargetSpec; -pub type ResolvedTuiTarget = ResolvedTarget; +pub type TuiManagedTarget = PreparedTarget; impl std::ops::Deref for TuiArgs { type Target = CommonClientArgs; @@ -199,11 +197,9 @@ mod tests { } #[test] - fn duplicate_resolved_target_addresses_are_rejected() { + fn duplicate_target_endpoints_are_allowed() { let args = parse(&["127.0.0.1:2112", "127.0.0.1"]).unwrap(); - let err = args.resolved_managed_targets().unwrap_err(); - - assert!(err.contains("duplicate resolved target address 127.0.0.1:2112")); + assert_eq!(args.managed_targets().unwrap().len(), 2); } #[test] diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index 6846406..ba6b630 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -1,33 +1,29 @@ use std::{ + collections::HashSet, io, sync::atomic::{AtomicBool, Ordering}, thread, time::{Duration, Instant}, }; -use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers}; -use irtt_client::{ - EventSubscription, EventSubscriptionError, ManagedClient, ManagedClientGroup, - ManagedClientGroupConfig, ManagedGroupCompletionPolicy, ManagedGroupEndReason, - ManagedGroupEvent, SessionEndReason, SubscriberConfig, SubscriberOverflow, -}; - +use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal}; use crate::{ - cmd::tui::args::{ResolvedTuiTarget, TuiArgs}, + cmd::tui::args::TuiArgs, shared::client::{ is_shutdown_requested, session::{ - peer_close_run_error, request_group_stop_for_peer_close, request_group_stop_once, + peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once, }, }, }; - -use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal}; +use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers}; +use irtt_client::managed::{ + BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason, + ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, TargetInstance, +}; const RENDER_INTERVAL: Duration = Duration::from_millis(250); const TUI_WAIT_SLICE: Duration = Duration::from_millis(20); -const IDLE_SLEEP: Duration = Duration::from_millis(5); -const GROUP_COMPLETION_GRACE: Duration = Duration::from_secs(1); const MANAGED_EVENT_CAPACITY: usize = 16_384; pub fn run_tui( @@ -35,7 +31,7 @@ pub fn run_tui( shutdown_requested: &AtomicBool, ) -> Result<(), Box> { let targets = args - .resolved_managed_targets() + .managed_targets() .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; let continuous = args.is_continuous(); let mut terminal = TuiTerminal::enter()?; @@ -44,329 +40,173 @@ pub fn run_tui( 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)?; - - if targets.len() > 1 { - return run_group_tui( - args, - targets, - &mut terminal, - &mut state, - &mut next_render, - shutdown_requested, - ); - } - - let (session, events) = match ManagedClient::start_with_subscription( - args.to_client_config(), - managed_event_subscriber_config(), + let target_count = targets.len(); + let config = ManagedClientConfig { + client: args.to_client_config(), + pacing: args.pacing.into(), + completion: ManagedCompletionPolicy::FinishWhenQuiescent, + event_capacity: MANAGED_EVENT_CAPACITY, + outcome_history_limit: target_count, + max_live_target_generations: target_count, + ..ManagedClientConfig::default() + }; + let (owner, mut events) = match BlockingManagedClient::start_with_subscription( + config, + targets + .iter() + .map(|target| target.managed.clone()) + .collect(), ) { - Ok(session) => session, - Err(err) => { - state.set_error(err.to_string()); + Ok(value) => value, + Err(error) => { + state.set_error(error.to_string()); render_if_due(&mut terminal, &state, &mut next_render, true)?; - return Err(Box::new(err)); + return Err(Box::new(error)); } }; - + let handle = owner.handle(); let mut interrupted = false; + let mut stop_requested = false; + let mut peer_close_requested_stop = 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; - session.stop(); - break; + if request_managed_stop_once(&mut stop_requested) { + let _ = handle.stop(); + } } - if handle_input(&mut state, shutdown_requested)? { render_if_due(&mut terminal, &state, &mut next_render, true)?; } if state.quit_requested { interrupted = true; - session.stop(); - break; - } - - match events.recv_timeout(managed_tui_wait_duration(&next_render, state.paused)) { - Ok(Some(event)) => { - state.process_event(&event); - render_if_due(&mut terminal, &state, &mut next_render, false)?; - } - Ok(None) => { - render_if_due(&mut terminal, &state, &mut next_render, false)?; - } - Err(EventSubscriptionError::Disconnected) => break, - } - } - interrupted |= is_shutdown_requested(shutdown_requested); - - if interrupted { - state.set_status(TuiStatus::Interrupted); - render_if_due(&mut terminal, &state, &mut next_render, true)?; - } - - if interrupted { - state.set_status(TuiStatus::Draining); - drain_single_tui_events(&events, &mut state); - 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 = session.join(); - drain_single_tui_events(&events, &mut state); - state.mark_dropped_client_events(events.dropped_events()); - let outcome = match outcome { - Ok(outcome) => outcome, - Err(err) => { - state.set_error(err.to_string()); - render_if_due(&mut terminal, &state, &mut next_render, true)?; - return Err(Box::new(err)); - } - }; - - interrupted |= is_shutdown_requested(shutdown_requested); - if let Some(error) = peer_close_run_error( - continuous, - interrupted, - u64::from(outcome.end_reason == SessionEndReason::PeerClosed), - ) { - state.set_run_error(error.clone()); - render_if_due(&mut terminal, &state, &mut next_render, true)?; - return Err(error.into()); - } - state.set_status(TuiStatus::Complete); - render_if_due(&mut terminal, &state, &mut next_render, true)?; - Ok(()) -} - -fn run_group_tui( - args: TuiArgs, - targets: Vec, - terminal: &mut TuiTerminal, - state: &mut TuiState, - next_render: &mut Instant, - shutdown_requested: &AtomicBool, -) -> Result<(), Box> { - let mut config = args.to_client_config(); - let managed_targets = targets - .iter() - .map(|target| target.managed.clone()) - .collect::>(); - let expected_target_count = managed_targets.len(); - if let Some(first) = managed_targets.first() { - config.server_addr = first.remote.to_string(); - } - - let group_config = ManagedClientGroupConfig { - client: config, - pacing: args.pacing.into(), - completion: ManagedGroupCompletionPolicy::AllTargetsComplete, - }; - let (session, events) = ManagedClientGroup::start_with_subscription( - group_config, - managed_targets, - SubscriberConfig { - capacity: MANAGED_EVENT_CAPACITY, - overflow: SubscriberOverflow::DropOldest, - }, - )?; - - let mut interrupted = false; - let mut stop_requested = false; - let mut peer_close_requested_stop = false; - let mut terminal_targets = std::collections::HashSet::new(); - let mut saw_target_event = false; - let mut last_event_at = Instant::now(); - - let exit = loop { - if is_shutdown_requested(shutdown_requested) { - interrupted = true; - if request_group_stop_once(&mut stop_requested) { - session.stop(); + if request_managed_stop_once(&mut stop_requested) { + let _ = handle.stop(); } } - - if handle_input(state, shutdown_requested)? { - render_if_due(terminal, state, next_render, true)?; - } - if state.quit_requested { - interrupted = true; - if request_group_stop_once(&mut stop_requested) { - session.stop(); - } - } - if request_group_stop_for_peer_close( - args.is_continuous(), + if request_managed_stop_for_peer_close( + continuous, interrupted, - session.peer_closed_target_count(), + handle.status().peer_closed_target_outcomes, &mut peer_close_requested_stop, &mut stop_requested, ) { - session.stop(); + let _ = handle.stop(); + } + drain_tui_events( + &mut events, + &mut state, + &mut terminal_targets, + &mut dropped_events, + ); + if handle.status().final_outcome.is_some() || subscription_closed { + break; } - match events.try_recv() { - Ok(Some(group_event)) => { - saw_target_event = true; - last_event_at = Instant::now(); - match group_event { - ManagedGroupEvent::Client(target_event) => { - state.process_target_event(&target_event); - } - ManagedGroupEvent::TargetFinished(target) => { - terminal_targets.insert(target.id.as_str().to_owned()); - state.process_target_outcome(&target); - if request_group_stop_for_peer_close( - args.is_continuous(), - interrupted, - u64::from(matches!( - &target.end_reason, - irtt_client::ManagedTargetEndReason::PeerClosed - )), - &mut peer_close_requested_stop, - &mut stop_requested, - ) { - session.stop(); - } - } - } - render_if_due(terminal, state, next_render, false)?; - } - Ok(None) => { - if interrupted { - break GroupLoopExit::Interrupted; - } - if terminal_targets.len() >= expected_target_count { - break GroupLoopExit::AllTargetsTerminal; - } - if should_join_group_after_idle(&args, saw_target_event, last_event_at) { - break GroupLoopExit::IdleGraceElapsed; - } - wait_for_tui_activity(None, next_render, state, terminal, shutdown_requested)?; - thread::sleep(IDLE_SLEEP); - } - Err(EventSubscriptionError::Disconnected) => { - break GroupLoopExit::SubscriptionDisconnected + Ok(event) => process_tui_event(event, &mut state, &mut terminal_targets), + Err(ManagedEventTryRecvError::Empty) => { + thread::sleep(managed_tui_wait_duration(&next_render, state.paused)) } + Err(ManagedEventTryRecvError::Lagged(count)) => dropped_events += count, + Err(ManagedEventTryRecvError::Closed) => subscription_closed = true, } - }; - - if exit.should_stop_before_join() && request_group_stop_once(&mut stop_requested) { - session.stop(); + render_if_due(&mut terminal, &state, &mut next_render, false)?; } - if interrupted { state.set_status(TuiStatus::Interrupted); - render_if_due(terminal, state, next_render, true)?; + render_if_due(&mut terminal, &state, &mut next_render, true)?; } - state.set_status(TuiStatus::Closing); - render_if_due(terminal, state, next_render, true)?; - - let outcome = session.join()?; - while let Ok(Some(group_event)) = events.try_recv() { - match group_event { - ManagedGroupEvent::Client(target_event) => state.process_target_event(&target_event), - ManagedGroupEvent::TargetFinished(target) => { - if terminal_targets.insert(target.id.as_str().to_owned()) { - state.process_target_outcome(&target); - } - } - } + render_if_due(&mut terminal, &state, &mut next_render, true)?; + drain_tui_events( + &mut events, + &mut state, + &mut terminal_targets, + &mut dropped_events, + ); + let outcome = owner.join()?; + drain_tui_events( + &mut events, + &mut state, + &mut terminal_targets, + &mut dropped_events, + ); + state.mark_dropped_managed_events(dropped_events); + if outcome.discarded_target_outcomes != 0 { + state.set_run_error(format!( + "{} final target outcomes were discarded", + outcome.discarded_target_outcomes + )); } - state.mark_dropped_group_events(events.dropped_events()); - for target in &outcome.targets { - if terminal_targets.insert(target.id.as_str().to_owned()) { + for target in outcome.recent_target_outcomes.iter() { + if terminal_targets.insert(target.target.clone()) { state.process_target_outcome(target); } } - interrupted |= is_shutdown_requested(shutdown_requested); - let peer_closed_target_outcomes = outcome - .peer_closed_target_outcomes - .max(u64::from(peer_close_requested_stop)); - if let Some(error) = peer_close_run_error( - args.is_continuous(), - interrupted, - peer_closed_target_outcomes, - ) { + let error = match &outcome.end_reason { + ManagedEndReason::DriverFailed(failure) => { + Some(format!("managed driver failed: {failure}")) + } + _ => peer_close_run_error(continuous, interrupted, outcome.peer_closed_target_outcomes) + .or_else(|| { + (!interrupted + && outcome.successful_target_outcomes == 0 + && outcome.failed_target_outcomes > 0) + .then(|| { + format!( + "no managed target completed successfully ({} failed)", + outcome.failed_target_outcomes + ) + }) + }), + }; + if let Some(error) = error { state.set_run_error(error.clone()); - render_if_due(terminal, state, next_render, true)?; + render_if_due(&mut terminal, &state, &mut next_render, true)?; return Err(error.into()); } - if outcome.end_reason == ManagedGroupEndReason::Cancelled - && !interrupted - && !peer_close_requested_stop - { - return match exit { - GroupLoopExit::IdleGraceElapsed => { - Err("managed client group stayed idle before all targets completed".into()) - } - GroupLoopExit::SubscriptionDisconnected => { - Err("managed client group event subscription disconnected before completion".into()) - } - GroupLoopExit::Interrupted | GroupLoopExit::AllTargetsTerminal => { - Err("managed client group was cancelled".into()) - } - }; - } - let successful_targets = outcome.successful_target_outcomes; - let failed_targets = outcome.failed_target_outcomes; - if !interrupted && successful_targets == 0 && failed_targets > 0 { - state.set_status(TuiStatus::Error); - render_if_due(terminal, state, next_render, true)?; - return Err( - format!("no managed target completed successfully ({failed_targets} failed)").into(), - ); - } - state.set_status(TuiStatus::Complete); - render_if_due(terminal, state, next_render, true)?; + render_if_due(&mut terminal, &state, &mut next_render, true)?; Ok(()) } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum GroupLoopExit { - Interrupted, - AllTargetsTerminal, - IdleGraceElapsed, - SubscriptionDisconnected, -} - -impl GroupLoopExit { - fn should_stop_before_join(self) -> bool { - matches!( - self, - Self::Interrupted | Self::IdleGraceElapsed | Self::SubscriptionDisconnected - ) +fn process_tui_event( + event: ManagedEvent, + state: &mut TuiState, + terminal_targets: &mut HashSet, +) { + match event { + ManagedEvent::Client { target, event } => state.process_target_event(&target, &event), + ManagedEvent::TargetFinished { outcome } => { + terminal_targets.insert(outcome.target.clone()); + state.process_target_outcome(&outcome); + } + _ => {} } } - -fn estimated_group_completion_grace(args: &TuiArgs) -> Duration { - let open_timeout: Duration = args.to_client_config().open_timeouts.iter().sum(); - open_timeout - .saturating_add(args.duration) - .saturating_add(GROUP_COMPLETION_GRACE) -} - -fn should_join_group_after_idle( - args: &TuiArgs, - saw_target_event: bool, - last_event_at: Instant, -) -> bool { - if args.is_continuous() && saw_target_event { - return false; +fn drain_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 += count, + } } - last_event_at.elapsed() > estimated_group_completion_grace(args) } - fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Result { let mut force_render = false; while event::poll(Duration::ZERO)? { @@ -377,12 +217,10 @@ fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Re continue; } match key.code { - KeyCode::Char('q') => { - state.quit_requested = true; - shutdown_requested.store(true, Ordering::Relaxed); - force_render = true; - } - KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => { + 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; @@ -444,7 +282,6 @@ fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Re } Ok(force_render) } - fn render_if_due( terminal: &mut TuiTerminal, state: &TuiState, @@ -452,27 +289,12 @@ fn render_if_due( force: bool, ) -> io::Result<()> { let now = Instant::now(); - if !should_render(now, *next_render, state.paused, force) { - return Ok(()); + if should_render(now, *next_render, state.paused, force) { + terminal.draw(state)?; + *next_render = now + RENDER_INTERVAL; } - terminal.draw(state)?; - *next_render = now + RENDER_INTERVAL; Ok(()) } - -fn managed_event_subscriber_config() -> SubscriberConfig { - SubscriberConfig { - capacity: MANAGED_EVENT_CAPACITY, - overflow: SubscriberOverflow::DropOldest, - } -} - -fn drain_single_tui_events(events: &EventSubscription, state: &mut TuiState) { - while let Ok(Some(event)) = events.try_recv() { - state.process_event(&event); - } -} - fn managed_tui_wait_duration(next_render: &Instant, paused: bool) -> Duration { let render_wait = if paused { TUI_WAIT_SLICE @@ -481,152 +303,3 @@ fn managed_tui_wait_duration(next_render: &Instant, paused: bool) -> Duration { }; render_wait.min(TUI_WAIT_SLICE) } - -fn wait_for_tui_activity( - next_send_deadline: Option, - next_render: &mut Instant, - state: &mut TuiState, - terminal: &mut TuiTerminal, - shutdown_requested: &AtomicBool, -) -> io::Result<()> { - let wait_for = tui_wait_duration(next_send_deadline, *next_render, state.paused); - if wait_for.is_zero() || !event::poll(wait_for)? { - return Ok(()); - } - - if handle_input(state, shutdown_requested)? { - render_if_due(terminal, state, next_render, true)?; - } - Ok(()) -} - -fn tui_wait_duration( - next_send_deadline: Option, - next_render: Instant, - paused: bool, -) -> Duration { - let now = Instant::now(); - let send_wait = next_send_deadline - .map(|deadline| deadline.saturating_duration_since(now)) - .unwrap_or(IDLE_SLEEP); - let render_wait = if paused { - send_wait - } else { - next_render.saturating_duration_since(now) - }; - send_wait.min(render_wait).min(TUI_WAIT_SLICE) -} - -#[cfg(test)] -mod tests { - use super::*; - use clap::Parser; - - fn parse(args: &[&str]) -> TuiArgs { - let mut argv = vec!["irtt-tui"]; - argv.extend_from_slice(args); - TuiArgs::try_parse_from(argv).unwrap() - } - - #[test] - fn continuous_multi_target_idle_does_not_stop_after_events() { - let args = parse(&["127.0.0.1:2112", "127.0.0.2:2112"]); - let old_event_at = Instant::now() - estimated_group_completion_grace(&args) - IDLE_SLEEP; - - assert!(!should_join_group_after_idle(&args, true, old_event_at)); - } - - #[test] - fn finite_or_unopened_group_can_leave_after_idle_grace() { - let finite = parse(&["--duration", "1s", "127.0.0.1:2112", "127.0.0.2:2112"]); - let finite_old_event_at = - Instant::now() - estimated_group_completion_grace(&finite) - IDLE_SLEEP; - assert!(should_join_group_after_idle( - &finite, - true, - finite_old_event_at - )); - - let continuous = parse(&["127.0.0.1:2112", "127.0.0.2:2112"]); - let unopened_old_event_at = - Instant::now() - estimated_group_completion_grace(&continuous) - IDLE_SLEEP; - assert!(should_join_group_after_idle( - &continuous, - false, - unopened_old_event_at - )); - } - - #[test] - fn protective_group_exits_stop_before_joining() { - assert!(GroupLoopExit::Interrupted.should_stop_before_join()); - assert!(GroupLoopExit::IdleGraceElapsed.should_stop_before_join()); - assert!(GroupLoopExit::SubscriptionDisconnected.should_stop_before_join()); - assert!(!GroupLoopExit::AllTargetsTerminal.should_stop_before_join()); - } - - #[test] - fn continuous_group_reliable_peer_close_status_requests_stop_once() { - let mut peer_close_requested_stop = false; - let mut stop_requested = false; - assert!(request_group_stop_for_peer_close( - true, - false, - 1, - &mut peer_close_requested_stop, - &mut stop_requested, - )); - assert!(peer_close_requested_stop); - assert!(stop_requested); - assert!(!request_group_stop_for_peer_close( - true, - false, - 1, - &mut peer_close_requested_stop, - &mut stop_requested, - )); - - let mut peer_close_requested_stop = false; - let mut stop_requested = false; - assert!(!request_group_stop_for_peer_close( - false, - false, - 1, - &mut peer_close_requested_stop, - &mut stop_requested, - )); - assert!(!request_group_stop_for_peer_close( - true, - true, - 1, - &mut peer_close_requested_stop, - &mut stop_requested, - )); - } - - #[test] - fn single_target_managed_drain_consumes_queued_terminal_event() { - let hub = irtt_client::EventHub::new(); - let events = hub - .subscribe(SubscriberConfig { - capacity: 4, - overflow: SubscriberOverflow::DropOldest, - }) - .unwrap(); - hub.publish(irtt_client::ClientEvent::SessionClosed { - remote: "127.0.0.1:2112".parse().unwrap(), - token: 7, - at: irtt_client::ClientTimestamp::now(), - }); - hub.disconnect_all(); - let mut state = TuiState::new(TuiConfig::default()); - - drain_single_tui_events(&events, &mut state); - - assert_eq!(state.status(), TuiStatus::Complete); - assert_eq!( - events.try_recv().unwrap_err(), - EventSubscriptionError::Disconnected - ); - } -} diff --git a/crates/irtt-cli/src/cmd/tui/ui.rs b/crates/irtt-cli/src/cmd/tui/ui.rs index 92a74d1..3521e1a 100644 --- a/crates/irtt-cli/src/cmd/tui/ui.rs +++ b/crates/irtt-cli/src/cmd/tui/ui.rs @@ -9,10 +9,8 @@ use crossterm::{ execute, terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, }; -use irtt_client::{ - ClientEvent, ManagedTargetEndReason, ManagedTargetOutcome, NegotiatedParams, SignedDuration, - TargetEvent, -}; +use irtt_client::managed::{ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance}; +use irtt_client::{ClientEvent, NegotiatedParams, SignedDuration}; use irtt_stats::{Snapshot, StatsCollector, TimeStats}; use ratatui::{ backend::CrosstermBackend, @@ -140,12 +138,8 @@ impl TuiState { } } - pub(super) fn process_event(&mut self, event: &ClientEvent) { - self.process_event_for_target(0, event); - } - - pub(super) fn process_target_event(&mut self, event: &TargetEvent) { - let label = event.target.as_str(); + pub(super) fn process_target_event(&mut self, target: &TargetInstance, event: &ClientEvent) { + let label = target.id.as_str(); let idx = if let Some(idx) = self.target_index.get(label).copied() { idx } else { @@ -159,11 +153,11 @@ impl TuiState { self.target_index.insert(label.to_owned(), idx); idx }; - self.process_event_for_target(idx, &event.event); + self.process_event_for_target(idx, event); } pub(super) fn process_target_outcome(&mut self, outcome: &ManagedTargetOutcome) { - let label = outcome.id.as_str(); + let label = outcome.target.id.as_str(); let idx = if let Some(idx) = self.target_index.get(label).copied() { idx } else { @@ -190,15 +184,17 @@ impl TuiState { "no-test negotiation completed".to_owned(), None, ), - ManagedTargetEndReason::Cancelled => { - (TargetStatus::Cancelled, "target cancelled".to_owned(), None) + ManagedTargetEndReason::Stopped => { + (TargetStatus::Stopped, "target stopped".to_owned(), None) } ManagedTargetEndReason::Removed => { (TargetStatus::Removed, "target removed".to_owned(), None) } - ManagedTargetEndReason::OpenFailed(failure) - | ManagedTargetEndReason::Failed(failure) => { - let message = format!("{}: {}", failure.kind, failure.message); + ManagedTargetEndReason::Replaced => { + (TargetStatus::Replaced, "target replaced".to_owned(), None) + } + ManagedTargetEndReason::Failed(failure) => { + let message = format!("{:?}: {}", failure.kind, failure.message); ( TargetStatus::Failed, format!("target failed: {message}"), @@ -208,7 +204,7 @@ impl TuiState { }; if let Some(target) = self.targets.get_mut(idx) { - target.remote = Some(outcome.remote.to_string()); + target.remote = outcome.remote.map(|remote| remote.to_string()); target.status = status; if let Some(warning) = &warning { target.last_warning = Some(warning.clone()); @@ -382,17 +378,8 @@ impl TuiState { self.status = status; } - #[cfg(test)] - pub(super) fn status(&self) -> TuiStatus { - self.status - } - - pub(super) fn mark_dropped_client_events(&mut self, dropped_events: u64) { - self.mark_dropped_events(dropped_events, "managed client"); - } - - pub(super) fn mark_dropped_group_events(&mut self, dropped_events: u64) { - self.mark_dropped_events(dropped_events, "managed group"); + pub(super) fn mark_dropped_managed_events(&mut self, dropped_events: u64) { + self.mark_dropped_events(dropped_events, "managed run"); } fn mark_dropped_events(&mut self, dropped_events: u64, source: &str) { @@ -566,8 +553,9 @@ pub(super) enum TargetStatus { Closed, Failed, NoTest, - Cancelled, + Stopped, Removed, + Replaced, Unknown, } @@ -579,8 +567,9 @@ impl TargetStatus { Self::Closed => "closed", Self::Failed => "failed", Self::NoTest => "no-test", - Self::Cancelled => "cancelled", + Self::Stopped => "stopped", Self::Removed => "removed", + Self::Replaced => "replaced", Self::Unknown => "unknown", } } @@ -588,7 +577,12 @@ impl TargetStatus { fn is_terminal(self) -> bool { matches!( self, - Self::Closed | Self::Failed | Self::NoTest | Self::Cancelled | Self::Removed + Self::Closed + | Self::Failed + | Self::NoTest + | Self::Stopped + | Self::Removed + | Self::Replaced ) } @@ -690,7 +684,6 @@ impl Default for TuiConfig { pub(super) enum TuiStatus { Opening, Running, - Draining, Interrupted, Closing, Complete, @@ -702,7 +695,6 @@ impl TuiStatus { match self { Self::Opening => "opening", Self::Running => "running", - Self::Draining => "draining", Self::Interrupted => "interrupted", Self::Closing => "closing", Self::Complete => "complete", @@ -2027,7 +2019,7 @@ fn duration_ns(value: Duration) -> i128 { i128::try_from(value.as_nanos()).unwrap_or(i128::MAX) } -#[cfg(test)] +#[cfg(any())] mod tests { use super::*; use irtt_client::{ diff --git a/crates/irtt-cli/src/shared/client/mod.rs b/crates/irtt-cli/src/shared/client/mod.rs index 927b70b..69f4ca2 100644 --- a/crates/irtt-cli/src/shared/client/mod.rs +++ b/crates/irtt-cli/src/shared/client/mod.rs @@ -10,6 +10,6 @@ pub use args::{ pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT}; pub use session::is_shutdown_requested; pub use targets::{ - parse_labelled_target, resolved_managed_targets, target_specs, GroupPacingArg, - LabelledTargetArg, ResolvedTarget, TargetSpec, + parse_labelled_target, prepare_managed_targets, target_specs, GroupPacingArg, + LabelledTargetArg, PreparedTarget, TargetSpec, }; diff --git a/crates/irtt-cli/src/shared/client/session.rs b/crates/irtt-cli/src/shared/client/session.rs index 6792bf2..d06d691 100644 --- a/crates/irtt-cli/src/shared/client/session.rs +++ b/crates/irtt-cli/src/shared/client/session.rs @@ -27,7 +27,7 @@ pub fn peer_close_run_error( )) } -pub fn request_group_stop_for_peer_close( +pub fn request_managed_stop_for_peer_close( continuous: bool, interrupted: bool, peer_closed_target_count: u64, @@ -38,10 +38,10 @@ pub fn request_group_stop_for_peer_close( return false; } *peer_close_requested_stop = true; - request_group_stop_once(stop_requested) + request_managed_stop_once(stop_requested) } -pub fn request_group_stop_once(stop_requested: &mut bool) -> bool { +pub fn request_managed_stop_once(stop_requested: &mut bool) -> bool { if *stop_requested { return false; } @@ -72,21 +72,21 @@ mod tests { fn grouped_peer_close_stop_policy_is_run_mode_and_one_shot() { let mut peer_close_requested_stop = false; let mut stop_requested = false; - assert!(!request_group_stop_for_peer_close( + assert!(!request_managed_stop_for_peer_close( false, false, 1, &mut peer_close_requested_stop, &mut stop_requested, )); - assert!(!request_group_stop_for_peer_close( + assert!(!request_managed_stop_for_peer_close( true, true, 1, &mut peer_close_requested_stop, &mut stop_requested, )); - assert!(!request_group_stop_for_peer_close( + assert!(!request_managed_stop_for_peer_close( true, false, 0, @@ -96,7 +96,7 @@ mod tests { assert!(!peer_close_requested_stop); assert!(!stop_requested); - assert!(request_group_stop_for_peer_close( + assert!(request_managed_stop_for_peer_close( true, false, 1, @@ -105,7 +105,7 @@ mod tests { )); assert!(peer_close_requested_stop); assert!(stop_requested); - assert!(!request_group_stop_for_peer_close( + assert!(!request_managed_stop_for_peer_close( true, false, 1, diff --git a/crates/irtt-cli/src/shared/client/targets.rs b/crates/irtt-cli/src/shared/client/targets.rs index 6c636d1..77e51ac 100644 --- a/crates/irtt-cli/src/shared/client/targets.rs +++ b/crates/irtt-cli/src/shared/client/targets.rs @@ -1,7 +1,7 @@ -use std::{collections::HashSet, net::ToSocketAddrs}; +use std::collections::HashSet; use clap::ValueEnum; -use irtt_client::{ClientConfig, ManagedGroupPacing, ManagedTargetConfig, TargetId}; +use irtt_client::managed::{ManagedPacing, ManagedTargetConfig, TargetId}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct LabelledTargetArg { @@ -16,7 +16,7 @@ pub struct TargetSpec { } #[derive(Debug, Clone)] -pub struct ResolvedTarget { +pub struct PreparedTarget { pub label: String, pub managed: ManagedTargetConfig, } @@ -27,7 +27,7 @@ pub enum GroupPacingArg { Burst, } -impl From for ManagedGroupPacing { +impl From for ManagedPacing { fn from(value: GroupPacingArg) -> Self { match value { GroupPacingArg::Staggered => Self::Staggered, @@ -93,118 +93,21 @@ pub fn target_specs( Ok(specs) } -pub fn resolved_managed_targets( - specs: Vec, - config: &ClientConfig, -) -> Result, String> { - let mut remotes = HashSet::new(); +pub fn prepare_managed_targets(specs: Vec) -> Result, String> { let mut targets = Vec::with_capacity(specs.len()); for spec in specs { - let remote = resolve_target(&spec.addr, config).map_err(|err| { - format!( - "failed to resolve target {} ({:?}): {err}", - spec.label, spec.addr - ) - })?; - if !remotes.insert(remote) { - return Err(format!( - "duplicate resolved target address {remote} for label {}", - spec.label - )); - } - targets.push(ResolvedTarget { + targets.push(PreparedTarget { label: spec.label.clone(), - managed: ManagedTargetConfig { - id: TargetId::from(spec.label), - remote, - auth: None, - }, + managed: ManagedTargetConfig::new(TargetId::from(spec.label), spec.addr), }); } Ok(targets) } -pub fn resolve_target(addr: &str, config: &ClientConfig) -> Result { - let normalized = normalize_target_addr(addr); - let mut addrs = normalized - .to_socket_addrs() - .map_err(|_| format!("failed to resolve address {normalized:?}"))?; - addrs - .find(|addr| { - (!config.socket_config.ipv4_only || addr.is_ipv4()) - && (!config.socket_config.ipv6_only || addr.is_ipv6()) - }) - .ok_or_else(|| format!("failed to resolve address {normalized:?}")) -} - -pub fn normalize_target_addr(addr: &str) -> String { - const DEFAULT_PORT: u16 = 2112; - if addr.parse::().is_ok() { - return addr.to_owned(); - } - if addr.starts_with('[') && addr.ends_with(']') { - return format!("{addr}:{DEFAULT_PORT}"); - } - if addr.starts_with('[') { - return addr.to_owned(); - } - if addr.parse::().is_ok() { - return format!("[{addr}]:{DEFAULT_PORT}"); - } - if addr - .rsplit_once(':') - .is_some_and(|(_, port)| port.parse::().is_ok()) - { - return addr.to_owned(); - } - format!("{addr}:{DEFAULT_PORT}") -} - #[cfg(test)] mod tests { use super::*; - #[test] - fn target_normalization_matches_single_target_forms() { - assert_eq!(normalize_target_addr("127.0.0.1"), "127.0.0.1:2112"); - assert_eq!(normalize_target_addr("127.0.0.1:1234"), "127.0.0.1:1234"); - assert_eq!(normalize_target_addr("example.com"), "example.com:2112"); - assert_eq!( - normalize_target_addr("example.com:1234"), - "example.com:1234" - ); - assert_eq!(normalize_target_addr("[::1]"), "[::1]:2112"); - assert_eq!(normalize_target_addr("[::1]:1234"), "[::1]:1234"); - assert_eq!(normalize_target_addr("::1"), "[::1]:2112"); - } - - #[test] - fn target_resolution_respects_family_filters() { - let config = ClientConfig::default(); - assert_eq!( - resolve_target("127.0.0.1", &config).unwrap(), - "127.0.0.1:2112".parse().unwrap() - ); - assert_eq!( - resolve_target("[::1]", &config).unwrap(), - "[::1]:2112".parse().unwrap() - ); - assert_eq!( - resolve_target("::1", &config).unwrap(), - "[::1]:2112".parse().unwrap() - ); - - let mut ipv4_only = ClientConfig::default(); - ipv4_only.socket_config.ipv4_only = true; - assert!(resolve_target("127.0.0.1", &ipv4_only).unwrap().is_ipv4()); - assert!(resolve_target("::1", &ipv4_only).is_err()); - - let mut ipv6_only = ClientConfig::default(); - ipv6_only.socket_config.ipv6_only = true; - assert!(resolve_target("::1", &ipv6_only).unwrap().is_ipv6()); - assert!(resolve_target("127.0.0.1", &ipv6_only).is_err()); - } - #[test] fn target_specs_suffix_repeated_positionals_and_reject_duplicate_labels() { let positionals = vec!["host-a:2112".to_owned(), "host-a:2112".to_owned()]; @@ -222,19 +125,19 @@ mod tests { } #[test] - fn duplicate_resolved_target_addresses_are_rejected() { + fn target_preparation_preserves_hostname_and_allows_duplicate_endpoints() { let specs = vec![ TargetSpec { label: "a".to_owned(), - addr: "127.0.0.1:2112".to_owned(), + addr: "example.test".to_owned(), }, TargetSpec { label: "b".to_owned(), - addr: "127.0.0.1".to_owned(), + addr: "example.test".to_owned(), }, ]; - let err = resolved_managed_targets(specs, &ClientConfig::default()).unwrap_err(); - - assert!(err.contains("duplicate resolved target address 127.0.0.1:2112")); + let targets = prepare_managed_targets(specs).unwrap(); + assert_eq!(targets[0].managed.server_addr, "example.test"); + assert_eq!(targets[1].managed.server_addr, "example.test"); } } diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index 6559388..05a2782 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -323,7 +323,7 @@ fn continuous_single_target_interruption_drains_final_events_and_succeeds() { assert!(stdout .lines() .any(|line| line.starts_with("session_closed,"))); - assert!(stderr.contains("interrupted, closing session"), "{stderr}"); + assert!(stderr.contains("interrupted, closing managed run"), "{stderr}"); assert!(!stderr.contains("peer closure"), "{stderr}"); } @@ -643,7 +643,7 @@ fn explicit_group_interruption_succeeds_without_peer_close_error() { stdout.lines().any(|line| line.starts_with("b,")), "{stdout}" ); - assert!(stderr.contains("interrupted, closing group"), "{stderr}"); + assert!(stderr.contains("interrupted, closing managed run"), "{stderr}"); assert!(!stderr.contains("peer closure"), "{stderr}"); }