diff --git a/README.md b/README.md index 51d78b9..b0a9259 100644 --- a/README.md +++ b/README.md @@ -96,12 +96,27 @@ Run continuously: irtt-cli --duration 0 ``` +Probe multiple targets concurrently: + +```sh +irtt-cli host-a:2112 host-b:2112 +irtt-cli --target ams=ams.example.com:2112 --target sg=sg.example.com:2112 +irtt-cli host-a:2112 host-b:2112 --pacing burst +``` + +For positional multi-target runs, the original target string is used as the +logical target label. Repeated positional strings get a stable numeric suffix. +For `--target label=host:port`, `label` is used in the `target` output column. +Labels must be unique. `--pacing staggered|burst` selects managed group pacing +for multi-target runs; the default is `staggered`. + Use a specific output format or column selection: ```sh irtt-cli --format table irtt-cli --format jsonl irtt-cli --format csv --columns event,seq,remote,effective_rtt_us +irtt-cli host-a:2112 host-b:2112 --format csv --columns target,seq,effective_rtt_us irtt-cli --format tsv --columns effective_rtt_us --header never irtt-cli --list-columns ``` @@ -146,7 +161,7 @@ comma-separated list. Use `--list-columns` to print the available names. Useful columns include: ```text -event, seq, remote, token, rtt, rtt_us, raw_rtt_us, effective_rtt_us, +target, event, seq, remote, token, rtt, rtt_us, raw_rtt_us, effective_rtt_us, adjusted_rtt_us, rd, rd_us, sd, sd_us, ipdv, ipdv_us, proc, server_processing_us, bytes, send_call_us, timer_error_us, highest_seen, server_received, server_window, dscp, ecn, traffic_class, kernel_rx_ns, @@ -159,14 +174,16 @@ for `rd` and `rd_us`, `send_delay` and `send_delay_us` for `sd` and `sd_us`, `server_processing` for `proc`, `server_received_count` for `server_received`, and `server_received_window` for `server_window`. -The default table output favors readability with compact columns: +The default single-target table output favors readability with compact columns: `event,seq,rtt,rd,sd,ipdv,proc,message`. It omits `echo_sent` rows. Passing `--columns default` is the same as omitting `--columns`, including the default -table row filtering. Custom table column selections include all event rows. -CSV, TSV, and JSONL default to all columns for structured export; use -`--columns all` explicitly to request every column in any format. Missing table -values render as `-`; missing CSV and TSV values render as empty fields; -missing JSONL values render as `null`. +table row filtering. Multi-target defaults add `target` as the first column so +rows can be attributed to a logical target label. Custom table column selections +include all event rows. CSV, TSV, and JSONL default to all non-target columns +for single-target export and include `target` by default for multi-target +export; use `--columns all` explicitly to request every column in any format. +Missing table values render as `-`; missing CSV and TSV values render as empty +fields; missing JSONL values render as `null`. Use JSON Lines for structured event streaming: diff --git a/crates/irtt-cli/src/cmd/client/args.rs b/crates/irtt-cli/src/cmd/client/args.rs index 8b762f9..f2bc7f9 100644 --- a/crates/irtt-cli/src/cmd/client/args.rs +++ b/crates/irtt-cli/src/cmd/client/args.rs @@ -1,7 +1,7 @@ -use std::time::Duration; +use std::{collections::HashSet, net::ToSocketAddrs, time::Duration}; use clap::{Parser, ValueEnum}; -use irtt_client::ClientConfig; +use irtt_client::{ClientConfig, ManagedGroupPacing, ManagedTargetConfig, TargetId}; use crate::shared::client::{parse_test_duration, CommonClientArgs, TimestampArg}; @@ -10,9 +10,31 @@ pub const DEFAULT_CLIENT_DURATION: Duration = Duration::from_secs(10); #[derive(Debug, Clone, Parser)] #[command(name = "irtt-cli", about = "Minimal IRTT-compatible stream client")] pub struct ClientArgs { - /// Server address or host, with optional port. - #[arg(required_unless_present = "list_columns")] - pub server: Option, + /// Server address or host, with optional port. Repeat for multi-target mode. + #[arg( + value_name = "TARGET", + num_args = 0.., + long_help = "Server address or host, with optional port. Repeat the positional target for multi-target mode, for example: irtt-cli host-a:2112 host-b:2112." + )] + pub targets: Vec, + + /// Explicit labelled target, as label=host:port. Repeat for multi-target mode. + #[arg( + long = "target", + value_name = "LABEL=TARGET", + value_parser = parse_labelled_target, + long_help = "Explicit labelled target, as label=host:port. Repeat for multi-target mode. The label is used as the target column and must be unique." + )] + pub labelled_targets: Vec, + + /// Managed group pacing for multi-target mode. + #[arg( + long, + value_enum, + default_value_t = GroupPacingArg::Staggered, + long_help = "Managed group pacing for multi-target mode.\n\nstaggered spaces active targets across the probe interval. burst sends one probe to every active target back-to-back once per interval." + )] + pub pacing: GroupPacingArg, #[arg( long, @@ -68,9 +90,8 @@ pub struct ClientArgs { impl ClientArgs { pub fn to_client_config(&self) -> ClientConfig { self.common.to_client_config( - self.server - .as_deref() - .expect("server is required unless --list-columns is set"), + self.primary_target_addr() + .expect("target is required unless --list-columns is set"), self.duration, ) } @@ -82,6 +103,82 @@ impl ClientArgs { pub fn timestamp_mode(&self) -> TimestampArg { self.common.tstamp } + + pub fn target_specs(&self) -> Result, String> { + let mut specs = Vec::new(); + let mut positional_counts = std::collections::HashMap::<&str, usize>::new(); + for target in &self.targets { + let count = positional_counts.entry(target.as_str()).or_default(); + *count += 1; + let label = if *count == 1 { + target.clone() + } else { + format!("{target}#{}", *count) + }; + specs.push(CliTargetSpec { + label, + addr: target.clone(), + }); + } + + for target in &self.labelled_targets { + specs.push(CliTargetSpec { + label: target.label.clone(), + addr: target.addr.clone(), + }); + } + + if specs.is_empty() { + return Err("at least one target is required unless --list-columns is set".to_owned()); + } + + let mut labels = HashSet::new(); + for spec in &specs { + if !labels.insert(spec.label.clone()) { + return Err(format!("duplicate target label {:?}", spec.label)); + } + } + + Ok(specs) + } + + pub fn resolved_managed_targets(&self) -> Result, String> { + let specs = self.target_specs()?; + let config = self.to_client_config(); + let mut remotes = HashSet::new(); + let mut targets = Vec::with_capacity(specs.len()); + for spec in specs { + let remote = resolve_cli_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(ResolvedCliTarget { + label: spec.label.clone(), + managed: ManagedTargetConfig { + id: TargetId::from(spec.label), + remote, + auth: None, + }, + }); + } + Ok(targets) + } + + fn primary_target_addr(&self) -> Option<&str> { + self.targets.first().map(String::as_str).or_else(|| { + self.labelled_targets + .first() + .map(|target| target.addr.as_str()) + }) + } } impl std::ops::Deref for ClientArgs { @@ -104,6 +201,39 @@ pub enum OutputFormat { Jsonl, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LabelledTargetArg { + pub label: String, + pub addr: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CliTargetSpec { + pub label: String, + pub addr: String, +} + +#[derive(Debug, Clone)] +pub struct ResolvedCliTarget { + pub label: String, + pub managed: ManagedTargetConfig, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum GroupPacingArg { + Staggered, + Burst, +} + +impl From for ManagedGroupPacing { + fn from(value: GroupPacingArg) -> Self { + match value { + GroupPacingArg::Staggered => Self::Staggered, + GroupPacingArg::Burst => Self::Burst, + } + } +} + impl OutputFormat { pub fn prints_summary(self) -> bool { matches!(self, Self::Table) @@ -120,6 +250,58 @@ pub enum HeaderMode { Never, } +fn parse_labelled_target(input: &str) -> Result { + let (label, addr) = input + .split_once('=') + .ok_or_else(|| "target must use LABEL=TARGET syntax".to_owned())?; + if label.is_empty() { + return Err("target label must not be empty".to_owned()); + } + if addr.is_empty() { + return Err("target address must not be empty".to_owned()); + } + Ok(LabelledTargetArg { + label: label.to_owned(), + addr: addr.to_owned(), + }) +} + +fn resolve_cli_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:?}")) +} + +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::*; @@ -137,7 +319,9 @@ mod tests { fn client_parser_keeps_finite_default_duration() { let args = parse(&["127.0.0.1:2112"]).unwrap(); - assert_eq!(args.server.as_deref(), Some("127.0.0.1:2112")); + assert_eq!(args.targets, ["127.0.0.1:2112"]); + assert!(args.labelled_targets.is_empty()); + assert_eq!(args.pacing, GroupPacingArg::Staggered); assert_eq!(args.duration, DEFAULT_CLIENT_DURATION); assert_eq!(args.interval, Duration::from_secs(1)); assert_eq!(args.format, OutputFormat::Table); @@ -165,14 +349,84 @@ mod tests { #[test] fn list_columns_does_not_require_server() { let args = parse(&["--list-columns"]).unwrap(); - assert_eq!(args.server, None); + assert!(args.targets.is_empty()); assert!(args.list_columns); } + #[test] + fn multiple_positional_targets_parse_as_targets() { + let args = parse(&["host-a:2112", "host-b:2112"]).unwrap(); + let specs = args.target_specs().unwrap(); + + assert_eq!(specs[0].label, "host-a:2112"); + assert_eq!(specs[0].addr, "host-a:2112"); + assert_eq!(specs[1].label, "host-b:2112"); + assert_eq!(specs[1].addr, "host-b:2112"); + } + + #[test] + fn repeated_labelled_targets_parse() { + let args = parse(&[ + "--target", + "ams=ams.example.com:2112", + "--target", + "sg=sg.example.com:2112", + ]) + .unwrap(); + let specs = args.target_specs().unwrap(); + + assert_eq!(specs[0].label, "ams"); + assert_eq!(specs[0].addr, "ams.example.com:2112"); + assert_eq!(specs[1].label, "sg"); + assert_eq!(specs[1].addr, "sg.example.com:2112"); + } + + #[test] + fn duplicate_labels_are_rejected() { + let args = parse(&["host-a:2112", "--target", "host-a:2112=host-b:2112"]).unwrap(); + let err = args.target_specs().unwrap_err(); + + assert!(err.contains("duplicate target label")); + } + + #[test] + fn duplicate_positional_target_strings_get_stable_suffixes() { + let args = parse(&["host-a:2112", "host-a:2112"]).unwrap(); + let specs = args.target_specs().unwrap(); + + assert_eq!(specs[0].label, "host-a:2112"); + assert_eq!(specs[1].label, "host-a:2112#2"); + } + + #[test] + fn invalid_labelled_target_syntax_is_rejected() { + assert!(parse(&["--target", "missing-equals"]).is_err()); + assert!(parse(&["--target", "=127.0.0.1:2112"]).is_err()); + assert!(parse(&["--target", "label="]).is_err()); + } + + #[test] + fn pacing_option_accepts_supported_values() { + assert_eq!( + parse(&["--pacing", "staggered", "127.0.0.1:2112"]) + .unwrap() + .pacing, + GroupPacingArg::Staggered + ); + assert_eq!( + parse(&["--pacing", "burst", "127.0.0.1:2112"]) + .unwrap() + .pacing, + GroupPacingArg::Burst + ); + } + #[test] fn client_help_lists_shared_protocol_options() { let help = ClientArgs::command().render_help().to_string(); assert!(help.contains("--format ")); + assert!(help.contains("--target ")); + assert!(help.contains("--pacing ")); assert!(help.contains("--columns ")); assert!(help.contains("--list-columns")); assert!(help.contains("--header
")); diff --git a/crates/irtt-cli/src/cmd/client/output.rs b/crates/irtt-cli/src/cmd/client/output.rs index 63136a1..2475ae1 100644 --- a/crates/irtt-cli/src/cmd/client/output.rs +++ b/crates/irtt-cli/src/cmd/client/output.rs @@ -26,14 +26,15 @@ impl OutputConfig { columns: Option<&str>, header: HeaderMode, verbose: bool, + multi_target: bool, ) -> Result { let uses_default_columns = columns .map(|columns| columns.trim() == "default") .unwrap_or(true); let default_table_rows = format == OutputFormat::Table && uses_default_columns; let columns = match columns { - Some(columns) => parse_columns(columns, format, verbose)?, - None => default_columns(format, verbose), + Some(columns) => parse_columns(columns, format, verbose, multi_target)?, + None => default_columns(format, verbose, multi_target), }; Ok(Self { @@ -92,6 +93,7 @@ impl OutputConfig { pub(super) fn render_event( &self, event: &ClientEvent, + target: Option<&str>, stats: Option<&EventRenderStats>, ) -> Option { let row = OutputRow::from_event(event); @@ -101,6 +103,7 @@ impl OutputConfig { let context = RenderContext { stats, + target, verbose: self.verbose, }; @@ -187,6 +190,7 @@ impl From for IpdvPair { #[derive(Debug, Clone, Copy)] struct RenderContext<'a> { stats: Option<&'a EventRenderStats>, + target: Option<&'a str>, verbose: bool, } @@ -412,6 +416,7 @@ struct ReplyRow { #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Column { + Target, Event, Seq, Remote, @@ -456,6 +461,7 @@ enum Column { } const ALL_COLUMNS: &[Column] = &[ + Column::Target, Column::Event, Column::Seq, Column::Remote, @@ -502,6 +508,7 @@ const ALL_COLUMNS: &[Column] = &[ impl Column { fn name(self) -> &'static str { match self { + Self::Target => "target", Self::Event => "event", Self::Seq => "seq", Self::Remote => "remote", @@ -548,6 +555,7 @@ impl Column { fn parse(input: &str) -> Option { Some(match input { + "target" => Self::Target, "event" => Self::Event, "seq" => Self::Seq, "remote" => Self::Remote, @@ -595,6 +603,7 @@ impl Column { fn description(self) -> &'static str { match self { + Self::Target => "logical CLI target label", Self::Event => "event kind", Self::Seq => "probe sequence number", Self::Remote => "remote socket address", @@ -641,6 +650,7 @@ impl Column { fn table_width(self) -> usize { match self { + Self::Target => 18, Self::Event => 17, Self::Seq => 6, Self::Remote => 21, @@ -663,7 +673,8 @@ impl Column { fn align_right(self) -> bool { !matches!( self, - Self::Event + Self::Target + | Self::Event | Self::Remote | Self::Token | Self::Message @@ -713,7 +724,12 @@ impl CellValue { } } -fn parse_columns(input: &str, format: OutputFormat, verbose: bool) -> Result, String> { +fn parse_columns( + input: &str, + format: OutputFormat, + verbose: bool, + multi_target: bool, +) -> Result, String> { let trimmed = input.trim(); if trimmed.is_empty() { return Err("--columns requires at least one column".to_owned()); @@ -722,7 +738,7 @@ fn parse_columns(input: &str, format: OutputFormat, verbose: bool) -> Result Result Vec { +fn default_columns(format: OutputFormat, verbose: bool, multi_target: bool) -> Vec { match format { OutputFormat::Table => { let mut columns = vec![ @@ -752,6 +768,9 @@ fn default_columns(format: OutputFormat, verbose: bool) -> Vec { Column::ServerProcessing, Column::Message, ]; + if multi_target { + columns.insert(0, Column::Target); + } if verbose { columns.extend([ Column::Remote, @@ -766,7 +785,17 @@ fn default_columns(format: OutputFormat, verbose: bool) -> Vec { } columns } - OutputFormat::Csv | OutputFormat::Tsv | OutputFormat::Jsonl => ALL_COLUMNS.to_vec(), + OutputFormat::Csv | OutputFormat::Tsv | OutputFormat::Jsonl => { + if multi_target { + ALL_COLUMNS.to_vec() + } else { + ALL_COLUMNS + .iter() + .copied() + .filter(|column| *column != Column::Target) + .collect() + } + } } } @@ -851,6 +880,7 @@ fn render_jsonl_row(row: &OutputRow, columns: &[Column], context: RenderContext< fn cell_for(row: &OutputRow, column: Column, context: RenderContext<'_>) -> Option { match column { + Column::Target => context.target.map(CellValue::text), Column::Event => Some(CellValue::text(row.event_name())), Column::Seq => row_seq(row).map(|seq| CellValue::Unsigned(u128::from(seq))), Column::Remote => row_remote(row).map(|remote| CellValue::text(remote.to_string())), @@ -1198,3 +1228,101 @@ fn write_json_string_content(out: &mut String, value: &str) { } } } + +#[cfg(test)] +mod tests { + use super::*; + use irtt_client::ClientTimestamp; + use std::{ + net::{IpAddr, Ipv4Addr, SocketAddr}, + time::Instant, + }; + + fn output_config( + format: OutputFormat, + columns: Option<&str>, + multi_target: bool, + ) -> OutputConfig { + OutputConfig::new(format, columns, HeaderMode::Auto, false, multi_target).unwrap() + } + + fn remote() -> SocketAddr { + SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 2112) + } + + fn reply() -> ClientEvent { + let sent_at = ClientTimestamp { + wall: UNIX_EPOCH + Duration::from_secs(1), + mono: Instant::now(), + }; + let received_at = ClientTimestamp { + wall: sent_at.wall + Duration::from_micros(1200), + mono: sent_at.mono + Duration::from_micros(1200), + }; + ClientEvent::EchoReply { + seq: 7, + remote: 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(), + } + } + + #[test] + fn list_columns_includes_target() { + assert!(OutputConfig::list_columns().contains("target")); + } + + #[test] + fn default_table_columns_include_target_only_for_multi_target() { + let single = output_config(OutputFormat::Table, None, false) + .render_header() + .unwrap(); + let multi = output_config(OutputFormat::Table, None, true) + .render_header() + .unwrap(); + + assert!(!single.split_whitespace().any(|column| column == "target")); + assert_eq!(multi.split_whitespace().next(), Some("target")); + } + + #[test] + fn default_structured_columns_include_target_for_multi_target() { + let single = output_config(OutputFormat::Csv, None, false) + .render_header() + .unwrap(); + let multi = output_config(OutputFormat::Csv, None, true) + .render_header() + .unwrap(); + + assert!(!single.split(',').any(|column| column == "target")); + assert_eq!(multi.split(',').next(), Some("target")); + } + + #[test] + fn custom_target_column_renders_label() { + let config = output_config( + OutputFormat::Csv, + Some("target,seq,effective_rtt_us"), + false, + ); + let line = config + .render_event(&reply(), Some("ams"), Some(&EventRenderStats::default())) + .unwrap(); + + assert_eq!( + config.render_header().as_deref(), + Some("target,seq,effective_rtt_us") + ); + assert_eq!(line, "ams,7,1200"); + } +} diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index e0f6dc4..5dc9834 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -1,14 +1,22 @@ +#[cfg(feature = "stats")] +use std::collections::BTreeMap; use std::{ + collections::HashSet, io::{self, Write}, sync::atomic::AtomicBool, + thread, + time::{Duration, Instant}, }; -use irtt_client::ClientEvent; +use irtt_client::{ + ClientEvent, EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig, + ManagedGroupEndReason, SubscriberConfig, SubscriberOverflow, TargetEvent, +}; #[cfg(all(test, feature = "stats"))] use irtt_client::{ClientTimestamp, PacketMeta, RttSample, SignedDuration}; use super::{ - args::ClientArgs, + args::{ClientArgs, ResolvedCliTarget}, output::{EventRenderStats, OutputConfig}, }; #[cfg(feature = "stats")] @@ -33,6 +41,9 @@ 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); + pub fn run_stream( args: ClientArgs, shutdown_requested: &AtomicBool, @@ -46,15 +57,31 @@ pub fn run_stream( args.columns.as_deref(), args.header, args.verbose, + args.target_specs() + .map(|targets| targets.len() > 1) + .unwrap_or(false), ) .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); + } + + let target_label = targets + .first() + .map(|target| target.label.as_str()) + .expect("at least one target was validated"); let mut stdout = io::LineWriter::new(io::stdout().lock()); #[cfg(feature = "stats")] let mut stats = StatsCollector::new(stats_config(continuous)); @@ -64,8 +91,6 @@ pub fn run_stream( print_final_summary: false, show_running_only_summary_note: false, out: &mut stdout, - #[cfg(feature = "stats")] - stats: &mut stats, }; if is_shutdown_requested(shutdown_requested) { @@ -79,12 +104,18 @@ pub fn run_stream( } let events = session.open()?; - stream_output.print_events(&events)?; + #[cfg(feature = "stats")] + print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?; + #[cfg(not(feature = "stats"))] + print_events_with_stats(&mut stream_output, &events, Some(target_label))?; let mut interrupted = false; while session.should_continue(shutdown_requested) { let events = session.step(shutdown_requested)?; - stream_output.print_events(&events)?; + #[cfg(feature = "stats")] + print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?; + #[cfg(not(feature = "stats"))] + print_events_with_stats(&mut stream_output, &events, Some(target_label))?; if is_shutdown_requested(shutdown_requested) { interrupted = true; @@ -101,18 +132,31 @@ pub fn run_stream( if session.should_drain_final(interrupted) { session.drain_final(|events| { - let _ = stream_output.print_events(events); + #[cfg(feature = "stats")] + let _ = + print_events_with_stats(&mut stream_output, events, Some(target_label), &mut stats); + #[cfg(not(feature = "stats"))] + let _ = print_events_with_stats(&mut stream_output, events, Some(target_label)); })?; } let events = session.poll_timeouts()?; - stream_output.print_events(&events)?; + #[cfg(feature = "stats")] + print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?; + #[cfg(not(feature = "stats"))] + print_events_with_stats(&mut stream_output, &events, Some(target_label))?; let events = session.close()?; - stream_output.print_events(&events)?; + #[cfg(feature = "stats")] + print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?; + #[cfg(not(feature = "stats"))] + print_events_with_stats(&mut stream_output, &events, Some(target_label))?; 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()?; Ok(()) @@ -124,32 +168,29 @@ struct StreamOutput<'a, W: Write> { print_final_summary: bool, show_running_only_summary_note: bool, out: &'a mut W, - #[cfg(feature = "stats")] - stats: &'a mut StatsCollector, } impl StreamOutput<'_, W> { - fn print_events(&mut self, events: &[ClientEvent]) -> io::Result<()> { + fn print_events( + &mut self, + events: &[ClientEvent], + target: Option<&str>, + stats_updates: &[EventRenderStats], + ) -> io::Result<()> { self.print_header()?; - for event in events { - self.print_event(event)?; + 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) -> io::Result<()> { - #[cfg(feature = "stats")] - let stats_update = self.stats.process(event); - - #[cfg(not(feature = "stats"))] - let stats_update = EventRenderStats::default(); - - #[cfg(feature = "stats")] - let event_stats = EventRenderStats::from(stats_update); - #[cfg(not(feature = "stats"))] - let event_stats = stats_update; - - if let Some(line) = self.config.render_event(event, Some(&event_stats)) { + 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(()) @@ -166,28 +207,250 @@ impl StreamOutput<'_, W> { Ok(()) } - fn print_summary(&mut self) -> io::Result<()> { + #[cfg(feature = "stats")] + fn print_summary(&mut self, stats: &StatsCollector) -> io::Result<()> { if !self.print_final_summary || !self.config.prints_summary() { return Ok(()); } - #[cfg(feature = "stats")] - { - write!( - self.out, - "{}", - crate::cmd::client::summary::format_summary_with_options( - &self.stats.snapshot(), - crate::cmd::client::summary::SummaryFormatOptions { - verbose: self.config.summary_verbose(), - show_running_only_note: self.show_running_only_summary_note, - }, - ) - )?; - } + 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(), + }; + + let (session, events) = ManagedClientGroup::start_with_subscription( + group_config, + managed_targets, + SubscriberConfig { + capacity: 16_384, + overflow: SubscriberOverflow::Disconnect, + }, + )?; + + let mut stdout = io::LineWriter::new(io::stdout().lock()); + #[cfg(feature = "stats")] + let mut stats = targets + .iter() + .map(|target| { + ( + target.label.clone(), + StatsCollector::new(stats_config(continuous)), + ) + }) + .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 interrupted = false; + let mut terminal_targets = HashSet::new(); + let mut last_event_at = Instant::now(); + let mut saw_target_event = false; + + loop { + if is_shutdown_requested(shutdown_requested) { + interrupted = true; + session.stop(); + } + + match events.try_recv() { + Ok(Some(target_event)) => { + last_event_at = Instant::now(); + saw_target_event = true; + let label = target_event.target.as_str(); + if is_terminal_target_event(&target_event.event) { + terminal_targets.insert(label.to_owned()); + } + #[cfg(feature = "stats")] + { + 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)?; + } + } + 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(target_event)) = events.try_recv() { + let label = target_event.target.as_str(); + #[cfg(feature = "stats")] + { + 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)?; + } + } + + if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted { + return Err("managed client group was cancelled".into()); + } + + 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)?; + } + } + #[cfg(not(feature = "stats"))] + { + stream_output.print_summary()?; + } + stream_output.out.flush()?; + Ok(()) +} + +fn is_terminal_target_event(event: &ClientEvent) -> bool { + matches!( + event, + ClientEvent::SessionClosed { .. } | ClientEvent::NoTestCompleted { .. } + ) +} + +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) +} + +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; + } + last_event_at.elapsed() > estimated_group_completion_grace(args) +} + +#[cfg(feature = "stats")] +fn print_target_event_with_stats( + stream_output: &mut StreamOutput<'_, W>, + target_event: &TargetEvent, + 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, + ) +} + +#[cfg(not(feature = "stats"))] +fn print_target_event_with_stats( + stream_output: &mut StreamOutput<'_, W>, + target_event: &TargetEvent, +) -> io::Result<()> { + print_events_with_stats( + stream_output, + std::slice::from_ref(&target_event.event), + Some(target_event.target.as_str()), + ) } #[cfg(feature = "stats")] @@ -294,7 +557,7 @@ mod tests { header: crate::cmd::client::HeaderMode, verbose: bool, ) -> OutputConfig { - OutputConfig::new(format, columns, header, verbose).unwrap() + OutputConfig::new(format, columns, header, verbose, false).unwrap() } #[test] @@ -343,10 +606,9 @@ mod tests { print_final_summary: true, show_running_only_summary_note: false, out: &mut out, - stats: &mut stats, }; - stream_output.print_events(&events).unwrap(); - stream_output.print_summary().unwrap(); + 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(); @@ -372,10 +634,15 @@ mod tests { print_final_summary: true, show_running_only_summary_note: true, out: &mut out, - stats: &mut stats, }; - stream_output.print_event(&reply_event(1, 1200)).unwrap(); - stream_output.print_summary().unwrap(); + 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(); @@ -385,7 +652,7 @@ mod tests { assert!(rendered.contains("packets:")); assert!(rendered.contains("received=1")); - let mut stats = StatsCollector::new(StatsConfig::continuous()); + let stats = StatsCollector::new(StatsConfig::continuous()); let mut out = Vec::new(); { let mut stream_output = StreamOutput { @@ -399,9 +666,8 @@ mod tests { print_final_summary: false, show_running_only_summary_note: true, out: &mut out, - stats: &mut stats, }; - stream_output.print_summary().unwrap(); + stream_output.print_summary(&stats).unwrap(); } assert!(out.is_empty()); diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs new file mode 100644 index 0000000..4c32305 --- /dev/null +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -0,0 +1,189 @@ +#![cfg(feature = "client")] + +use std::{ + net::{SocketAddr, UdpSocket}, + process::Command, + thread, + thread::JoinHandle, + time::Duration, +}; + +use irtt_proto::{ + echo_packet_len, + flags::{FLAG_CLOSE, FLAG_OPEN, FLAG_REPLY}, + layout::PacketLayout, + Clock, Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION, +}; + +const TOKEN: u64 = 0x1234_5678_90ab_cdef; + +struct FakeServer { + addr: SocketAddr, + done: JoinHandle<()>, +} + +impl FakeServer { + fn join(self) { + self.done.join().unwrap(); + } +} + +#[test] +fn multi_target_csv_emits_rows_for_both_labels() { + let params = test_params(Some(Duration::from_millis(40)), Duration::from_millis(10)); + let a = start_echo_server(params.clone()); + let b = start_echo_server(params); + + let output = Command::new(env!("CARGO_BIN_EXE_irtt-cli")) + .args([ + "--duration", + "40ms", + "--interval", + "10ms", + "--format", + "csv", + "--columns", + "target,seq,effective_rtt_us", + "--header", + "never", + "--target", + &format!("a={}", a.addr), + "--target", + &format!("b={}", b.addr), + ]) + .output() + .unwrap(); + + a.join(); + b.join(); + + assert!( + output.status.success(), + "status={:?}\nstderr={}", + output.status.code(), + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + assert!( + stdout.lines().any(|line| line.starts_with("a,")), + "{stdout}" + ); + assert!( + stdout.lines().any(|line| line.starts_with("b,")), + "{stdout}" + ); +} + +fn start_echo_server(params: Params) -> FakeServer { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let done = thread::spawn(move || { + let (_, peer) = recv_request(&socket); + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) + .unwrap(); + + loop { + let (packet, peer) = recv_request(&socket); + let flags = packet[3]; + if flags & FLAG_CLOSE != 0 { + break; + } + let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); + socket + .send_to( + &echo_reply_packet( + TOKEN, + seq, + ¶ms, + &TimestampFields::default(), + FLAG_REPLY, + ), + peer, + ) + .unwrap(); + } + }); + FakeServer { addr, done } +} + +fn recv_request(socket: &UdpSocket) -> (Vec, SocketAddr) { + let mut buf = [0_u8; 2048]; + let (size, peer) = socket.recv_from(&mut buf).unwrap(); + (buf[..size].to_vec(), peer) +} + +fn open_reply(flags: u8, token: u64, params: &Params) -> Vec { + let mut packet = Vec::new(); + packet.extend_from_slice(&MAGIC); + packet.push(flags); + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(¶ms.encode()); + packet +} + +fn echo_reply_packet( + token: u64, + seq: u32, + params: &Params, + timestamps: &TimestampFields, + flags: u8, +) -> Vec { + let layout = PacketLayout::echo(false, params); + let packet_len = echo_packet_len(false, params).unwrap(); + let mut packet = Vec::with_capacity(packet_len); + + packet.extend_from_slice(&MAGIC); + packet.push(flags); + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(&seq.to_le_bytes()); + + if layout.recv_count { + packet.extend_from_slice(&42_u32.to_le_bytes()); + } + if layout.recv_window { + packet.extend_from_slice(&0_u64.to_le_bytes()); + } + if layout.recv_wall { + packet.extend_from_slice(×tamps.recv_wall.unwrap_or(0).to_le_bytes()); + } + if layout.recv_mono { + packet.extend_from_slice(×tamps.recv_mono.unwrap_or(0).to_le_bytes()); + } + if layout.midpoint_wall { + packet.extend_from_slice(×tamps.midpoint_wall.unwrap_or(0).to_le_bytes()); + } + if layout.midpoint_mono { + packet.extend_from_slice(×tamps.midpoint_mono.unwrap_or(0).to_le_bytes()); + } + if layout.send_wall { + packet.extend_from_slice(×tamps.send_wall.unwrap_or(0).to_le_bytes()); + } + if layout.send_mono { + packet.extend_from_slice(×tamps.send_mono.unwrap_or(0).to_le_bytes()); + } + + packet.resize(packet_len, 0); + packet +} + +fn test_params(duration: Option, interval: Duration) -> Params { + Params { + protocol_version: PROTOCOL_VERSION, + duration_ns: duration.map_or(0, duration_ns_i64), + interval_ns: duration_ns_i64(interval), + length: 0, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + dscp: 0, + server_fill: None, + } +} + +fn duration_ns_i64(duration: Duration) -> i64 { + i64::try_from(duration.as_nanos()).unwrap() +}