diff --git a/crates/irtt-cli/src/cmd/client/args.rs b/crates/irtt-cli/src/cmd/client/args.rs new file mode 100644 index 0000000..18016f9 --- /dev/null +++ b/crates/irtt-cli/src/cmd/client/args.rs @@ -0,0 +1,136 @@ +use std::time::Duration; + +use clap::{Parser, ValueEnum}; +use irtt_client::ClientConfig; + +use crate::shared::client::{parse_test_duration, CommonClientArgs, TimestampArg}; + +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. + pub server: String, + + #[arg( + long, + default_value = "10s", + value_parser = parse_test_duration, + help = "Test duration; use 0 for continuous mode", + long_help = "Test duration; use 0 for continuous mode.\n\nFinite runs retain exact statistics for final summaries. Continuous mode uses bounded-memory running statistics and prints a final summary only when interrupted." + )] + pub duration: Duration, + + #[command(flatten)] + pub common: CommonClientArgs, + + /// Output format. + #[arg(long, value_enum, default_value_t = OutputMode::Human)] + pub output: OutputMode, + + /// Include extra fields in human output. + #[arg(long)] + pub verbose: bool, +} + +impl ClientArgs { + pub fn to_client_config(&self) -> ClientConfig { + self.common.to_client_config(&self.server, self.duration) + } + + pub fn is_continuous(&self) -> bool { + self.duration == Duration::ZERO + } + + pub fn timestamp_mode(&self) -> TimestampArg { + self.common.tstamp + } +} + +impl std::ops::Deref for ClientArgs { + type Target = CommonClientArgs; + + fn deref(&self) -> &Self::Target { + &self.common + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum OutputMode { + /// Readable terminal output with a final summary. + Human, + /// Parseable full event fields. + Machine, + /// Simple key=value-ish event stream. + Simple, + /// RTT microseconds only. + RttUs, +} + +impl OutputMode { + pub fn prints_summary(self) -> bool { + matches!(self, Self::Human) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::{CommandFactory, Parser}; + use irtt_client::NegotiationPolicy; + use irtt_proto::{Clock, ReceivedStats, StampAt}; + + fn parse(args: &[&str]) -> Result { + let mut argv = vec!["irtt-cli"]; + argv.extend_from_slice(args); + ClientArgs::try_parse_from(argv) + } + + #[test] + fn client_parser_keeps_finite_default_duration() { + let args = parse(&["127.0.0.1:2112"]).unwrap(); + + assert_eq!(args.server, "127.0.0.1:2112"); + assert_eq!(args.duration, DEFAULT_CLIENT_DURATION); + assert_eq!(args.interval, Duration::from_secs(1)); + assert_eq!(args.output, OutputMode::Human); + assert!(!args.is_continuous()); + assert_eq!(args.timestamp_mode(), TimestampArg::Both); + + let config = args.to_client_config(); + assert_eq!(config.duration, Some(DEFAULT_CLIENT_DURATION)); + assert_eq!(config.received_stats, ReceivedStats::Both); + assert_eq!(config.stamp_at, StampAt::Both); + assert_eq!(config.clock, Clock::Both); + assert_eq!(config.negotiation_policy, NegotiationPolicy::Strict); + } + + #[test] + fn client_parser_accepts_stream_outputs_and_rejects_tui_output() { + for output in ["human", "simple", "machine", "rtt-us"] { + assert!(parse(&["--output", output, "127.0.0.1:2112"]).is_ok()); + } + assert!(parse(&["--output", "tui", "127.0.0.1:2112"]).is_err()); + } + + #[test] + fn client_help_lists_shared_protocol_options() { + let help = ClientArgs::command().render_help().to_string(); + assert!(help.contains("--output ")); + assert!(help.contains("--tstamp ")); + assert!(help.contains("--stats ")); + assert!(help.contains("--sfill ")); + assert!(help.contains("--dscp <0..=63>")); + assert!(help.contains("--ttl <1..=255>")); + assert!(help.contains("--loose")); + } + + #[test] + fn output_mode_summary_policy_is_human_only() { + assert!(OutputMode::Human.prints_summary()); + assert!(!OutputMode::Simple.prints_summary()); + assert!(!OutputMode::Machine.prints_summary()); + assert!(!OutputMode::RttUs.prints_summary()); + } +} diff --git a/crates/irtt-cli/src/cmd/client/output.rs b/crates/irtt-cli/src/cmd/client/output.rs new file mode 100644 index 0000000..34963f9 --- /dev/null +++ b/crates/irtt-cli/src/cmd/client/output.rs @@ -0,0 +1,656 @@ +use std::{ + fmt::Write as _, + net::SocketAddr, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use irtt_client::{ + ClientEvent, NegotiatedParams, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample, + ServerTiming, SignedDuration, WarningKind, +}; + +use super::args::OutputMode; + +pub fn format_event(event: &ClientEvent, mode: OutputMode) -> Option { + if matches!(event, ClientEvent::EchoSent { .. }) { + return None; + } + match mode { + OutputMode::RttUs => format_rtt_us(event), + OutputMode::Human => Some(format_human_event(event, None)), + OutputMode::Machine => Some(format_machine(event)), + OutputMode::Simple => Some(format_simple(event)), + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct HumanEventStats { + pub contributed_sample: bool, + pub ipdv_pairs: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct HumanIpdvPair { + pub previous_seq: u32, + pub current_seq: u32, + pub rtt_ipdv: Duration, + pub send_ipdv: Option, + pub receive_ipdv: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct HumanOutputOptions { + pub verbose: bool, +} + +#[cfg(feature = "stats")] +impl From for HumanEventStats { + fn from(value: irtt_stats::EventStatsUpdate) -> Self { + Self { + contributed_sample: value.contributed_sample, + ipdv_pairs: value.ipdv_pairs.into_iter().map(Into::into).collect(), + } + } +} + +#[cfg(feature = "stats")] +impl From for HumanIpdvPair { + fn from(value: irtt_stats::IpdvPairUpdate) -> Self { + Self { + previous_seq: value.previous_seq, + current_seq: value.current_seq, + rtt_ipdv: value.rtt_ipdv, + send_ipdv: value.send_ipdv, + receive_ipdv: value.receive_ipdv, + } + } +} + +pub fn format_human_event(event: &ClientEvent, stats: Option) -> String { + format_human_event_with_options(event, stats, HumanOutputOptions::default()) +} + +pub fn format_human_event_with_options( + event: &ClientEvent, + stats: Option, + options: HumanOutputOptions, +) -> String { + let stats = stats.as_ref(); + + match event { + ClientEvent::SessionStarted { remote, token, .. } => { + format!("session started remote={remote} token={token:#x}") + } + ClientEvent::NoTestCompleted { remote, .. } => { + format!("no-test completed remote={remote}") + } + ClientEvent::SessionClosed { remote, token, .. } => { + format!("session closed remote={remote} token={token:#x}") + } + ClientEvent::EchoSent { .. } => String::new(), + ClientEvent::EchoReply { + seq, + rtt, + server_timing, + one_way, + received_stats, + .. + } => { + let mut out = format!("seq={seq}"); + write!(out, " rtt={}", format_signed_duration(rtt.effective)).unwrap(); + write_human_one_way(&mut out, *one_way); + write!(out, " ipdv={}", format_human_ipdv(stats, *seq)).unwrap(); + if let Some(processing) = server_timing.and_then(|timing| timing.processing) { + write!(out, " proc={}", format_duration(processing)).unwrap(); + } + if options.verbose { + write_human_received_stats(&mut out, *received_stats); + } + out + } + ClientEvent::EchoLoss { seq, .. } => { + format!("loss seq={seq}") + } + ClientEvent::DuplicateReply { seq, remote, .. } => { + format!("duplicate seq={seq} remote={remote}") + } + ClientEvent::LateReply { + seq, + highest_seen, + remote, + rtt, + one_way, + received_stats, + .. + } => { + let mut out = format!("late seq={seq} highest_seen={highest_seen} remote={remote}",); + if let Some(rtt) = rtt { + write!(out, " rtt={}", format_signed_duration(rtt.effective)).unwrap(); + write_human_one_way(&mut out, *one_way); + write!(out, " ipdv={}", format_human_ipdv(stats, *seq)).unwrap(); + } + write_human_received_stats(&mut out, *received_stats); + out + } + ClientEvent::Warning { kind, message, .. } => { + format!("warning kind={} message={message}", warning_kind(*kind)) + } + } +} + +fn format_rtt_us(event: &ClientEvent) -> Option { + match event { + ClientEvent::EchoReply { rtt, .. } => Some(signed_duration_us(rtt.effective).to_string()), + _ => None, + } +} + +fn format_machine(event: &ClientEvent) -> String { + let mut out = String::new(); + match event { + ClientEvent::SessionStarted { + remote, + token, + negotiated, + at, + } => { + write_common(&mut out, "session_started"); + write_remote(&mut out, *remote); + write_token(&mut out, *token); + write_wall(&mut out, "event_wall_ns", at.wall); + write_negotiated(&mut out, negotiated); + } + ClientEvent::NoTestCompleted { + remote, + negotiated, + at, + } => { + write_common(&mut out, "no_test_completed"); + write_remote(&mut out, *remote); + write_wall(&mut out, "event_wall_ns", at.wall); + write_negotiated(&mut out, negotiated); + } + ClientEvent::SessionClosed { remote, token, at } => { + write_common(&mut out, "session_closed"); + write_remote(&mut out, *remote); + write_token(&mut out, *token); + write_wall(&mut out, "event_wall_ns", at.wall); + } + ClientEvent::EchoSent { .. } => {} + ClientEvent::EchoReply { + seq, + remote, + sent_at, + received_at, + rtt, + server_timing, + one_way, + received_stats, + bytes: _, + packet_meta, + } => { + write_common(&mut out, "echo_reply"); + write_seq(&mut out, *seq); + write_remote(&mut out, *remote); + write_wall(&mut out, "client_send_wall_ns", sent_at.wall); + write_wall(&mut out, "client_receive_wall_ns", received_at.wall); + write_rtt(&mut out, rtt); + write_server_timing(&mut out, *server_timing); + write_one_way(&mut out, *one_way); + write_received_stats(&mut out, *received_stats); + write_packet_meta(&mut out, *packet_meta); + } + ClientEvent::EchoLoss { seq, sent_at, .. } => { + write_common(&mut out, "loss"); + write_seq(&mut out, *seq); + write_wall(&mut out, "client_send_wall_ns", sent_at.wall); + out.push_str(" warning=loss"); + } + ClientEvent::DuplicateReply { + seq, + remote, + received_at, + bytes: _, + } => { + write_common(&mut out, "duplicate"); + write_seq(&mut out, *seq); + write_remote(&mut out, *remote); + write_wall(&mut out, "client_receive_wall_ns", received_at.wall); + out.push_str(" warning=duplicate"); + } + ClientEvent::LateReply { + seq, + highest_seen, + remote, + sent_at, + received_at, + rtt, + server_timing, + one_way, + received_stats, + bytes: _, + packet_meta, + } => { + write_common(&mut out, "late"); + write_seq(&mut out, *seq); + write_remote(&mut out, *remote); + write!(out, " highest_seen={highest_seen}").unwrap(); + if let Some(sent_at) = sent_at { + write_wall(&mut out, "client_send_wall_ns", sent_at.wall); + } + write_wall(&mut out, "client_receive_wall_ns", received_at.wall); + if let Some(rtt) = rtt { + write_rtt(&mut out, rtt); + } + write_server_timing(&mut out, *server_timing); + write_one_way(&mut out, *one_way); + write_received_stats(&mut out, *received_stats); + write_packet_meta(&mut out, *packet_meta); + out.push_str(" warning=late"); + } + ClientEvent::Warning { kind, message, at } => { + write_common(&mut out, "warning"); + write_wall(&mut out, "at", at.wall); + write!( + out, + " warning_kind={} message={}", + warning_kind(*kind), + escape_value(message) + ) + .unwrap(); + } + } + out +} + +fn format_simple(event: &ClientEvent) -> String { + match event { + ClientEvent::SessionStarted { remote, token, .. } => { + format!("session started remote={remote} token={token:#x}") + } + ClientEvent::NoTestCompleted { remote, .. } => { + format!("no-test completed remote={remote}") + } + ClientEvent::SessionClosed { remote, token, .. } => { + format!("session closed remote={remote} token={token:#x}") + } + ClientEvent::EchoSent { .. } => String::new(), + ClientEvent::EchoReply { + seq, + remote, + rtt, + server_timing, + bytes: _, + .. + } => { + let mut out = format!( + "reply seq={seq} remote={remote} rtt_us={}", + signed_duration_us(rtt.effective) + ); + if rtt.adjusted.is_some() { + write!(out, " raw_rtt_us={}", duration_us(rtt.raw)).unwrap(); + } + if let Some(processing) = server_timing.and_then(|timing| timing.processing) { + write!(out, " server_processing_us={}", duration_us(processing)).unwrap(); + } + out + } + ClientEvent::EchoLoss { seq, .. } => { + format!("loss seq={seq}") + } + ClientEvent::DuplicateReply { seq, remote, .. } => { + format!("duplicate seq={seq} remote={remote}") + } + ClientEvent::LateReply { + seq, + highest_seen, + remote, + rtt, + .. + } => { + let mut out = format!("late seq={seq} highest_seen={highest_seen} remote={remote}",); + if let Some(rtt) = rtt { + write!(out, " rtt_us={}", signed_duration_us(rtt.effective)).unwrap(); + } + out + } + ClientEvent::Warning { kind, message, .. } => { + format!("warning kind={} message={message}", warning_kind(*kind)) + } + } +} + +fn write_common(out: &mut String, event: &str) { + write!(out, "event={event}").unwrap(); +} + +fn write_seq(out: &mut String, seq: u32) { + write!(out, " seq={seq}").unwrap(); +} + +fn write_remote(out: &mut String, remote: SocketAddr) { + write!(out, " remote={remote}").unwrap(); +} + +fn write_token(out: &mut String, token: u64) { + write!(out, " token={token:#x}").unwrap(); +} + +fn write_wall(out: &mut String, key: &str, wall: SystemTime) { + if let Ok(duration) = wall.duration_since(UNIX_EPOCH) { + write!(out, " {key}={}", duration.as_nanos()).unwrap(); + } +} + +fn write_negotiated(out: &mut String, negotiated: &NegotiatedParams) { + write!( + out, + " duration_ns={} interval_ns={} payload_length={}", + negotiated.params.duration_ns, negotiated.params.interval_ns, negotiated.params.length + ) + .unwrap(); +} + +fn write_rtt(out: &mut String, rtt: &RttSample) { + write!( + out, + " raw_rtt_us={} effective_rtt_us={}", + duration_us(rtt.raw), + signed_duration_us(rtt.effective) + ) + .unwrap(); + if let Some(adjusted) = rtt.adjusted { + write!(out, " adjusted_rtt_us={}", signed_duration_us(adjusted)).unwrap(); + } +} + +fn write_server_timing(out: &mut String, timing: Option) { + if let Some(timing) = timing { + write_optional_i64(out, "server_receive_wall_ns", timing.receive_wall_ns); + write_optional_i64(out, "server_receive_mono_ns", timing.receive_mono_ns); + write_optional_i64(out, "server_send_wall_ns", timing.send_wall_ns); + write_optional_i64(out, "server_send_mono_ns", timing.send_mono_ns); + write_optional_i64(out, "server_midpoint_wall_ns", timing.midpoint_wall_ns); + write_optional_i64(out, "server_midpoint_mono_ns", timing.midpoint_mono_ns); + if let Some(processing) = timing.processing { + write!(out, " server_processing_us={}", duration_us(processing)).unwrap(); + } + } +} + +fn write_one_way(out: &mut String, one_way: Option) { + if let Some(one_way) = one_way { + if let Some(value) = one_way.client_to_server { + write!(out, " client_to_server_us={}", signed_duration_us(value)).unwrap(); + } + if let Some(value) = one_way.server_to_client { + write!(out, " server_to_client_us={}", signed_duration_us(value)).unwrap(); + } + } +} + +fn write_received_stats(out: &mut String, stats: Option) { + if let Some(stats) = stats { + if let Some(count) = stats.count { + write!(out, " server_received_count={count}").unwrap(); + } + if let Some(window) = stats.window { + write!(out, " server_received_window={window:#x}").unwrap(); + } + } +} + +fn write_packet_meta(out: &mut String, meta: PacketMeta) { + write_optional_u8(out, "traffic_class", meta.traffic_class); + write_optional_u8(out, "dscp", meta.dscp); + write_optional_u8(out, "ecn", meta.ecn); + match meta.kernel_rx_timestamp { + Some(timestamp) => write_wall(out, "kernel_rx_ns", timestamp), + None => write!(out, " kernel_rx_ns=none").unwrap(), + } +} + +fn write_human_one_way(out: &mut String, one_way: Option) { + match one_way { + Some(one_way) => { + write!( + out, + " rd={} sd={}", + format_optional_signed_duration(one_way.server_to_client), + format_optional_signed_duration(one_way.client_to_server) + ) + .unwrap(); + } + None => out.push_str(" rd=n/a sd=n/a"), + } +} + +fn write_human_received_stats(out: &mut String, stats: Option) { + if let Some(stats) = stats { + if let Some(count) = stats.count { + write!(out, " server_received={count}").unwrap(); + } + if let Some(window) = stats.window { + write!(out, " server_window={window:#x}").unwrap(); + } + } +} + +fn format_human_ipdv(stats: Option<&HumanEventStats>, seq: u32) -> String { + let Some(stats) = stats else { + return "n/a".to_owned(); + }; + + let pair = stats + .ipdv_pairs + .iter() + .find(|pair| pair.current_seq == seq) + .or_else(|| { + stats + .ipdv_pairs + .iter() + .find(|pair| pair.previous_seq == seq) + }); + + pair.map(|pair| format_duration(pair.rtt_ipdv)) + .unwrap_or_else(|| "n/a".to_owned()) +} + +fn write_optional_u8(out: &mut String, key: &str, value: Option) { + match value { + Some(value) => write!(out, " {key}={value}").unwrap(), + None => write!(out, " {key}=none").unwrap(), + } +} + +fn write_optional_i64(out: &mut String, key: &str, value: Option) { + if let Some(value) = value { + write!(out, " {key}={value}").unwrap(); + } +} + +fn duration_us(duration: Duration) -> u128 { + duration.as_micros() +} + +fn signed_duration_us(duration: SignedDuration) -> i128 { + duration.as_micros() +} + +fn format_optional_signed_duration(duration: Option) -> String { + duration + .map(format_signed_duration) + .unwrap_or_else(|| "n/a".to_owned()) +} + +fn format_duration(duration: Duration) -> String { + format_ns(duration.as_nanos() as f64) +} + +fn format_signed_duration(duration: SignedDuration) -> String { + format_signed_ns(duration.as_nanos() as f64) +} + +fn format_signed_ns(ns: f64) -> String { + if ns < 0.0 { + format!("-{}", format_ns(-ns)) + } else { + format_ns(ns) + } +} + +fn format_ns(ns: f64) -> String { + if ns < 1_000.0 { + format!("{ns:.0}ns") + } else if ns < 1_000_000.0 { + format!("{:.1}µs", ns / 1_000.0) + } else if ns < 1_000_000_000.0 { + format!("{:.1}ms", ns / 1_000_000.0) + } else { + format!("{:.3}s", ns / 1_000_000_000.0) + } +} + +fn warning_kind(kind: WarningKind) -> &'static str { + match kind { + WarningKind::MalformedOrUnrelatedPacket => "malformed_or_unrelated_packet", + WarningKind::WrongToken => "wrong_token", + WarningKind::UntrackedReply => "untracked_reply", + _ => "unknown", + } +} + +fn escape_value(value: &str) -> String { + value + .replace('\\', "\\\\") + .replace(' ', "\\s") + .replace('\n', "\\n") + .replace('\r', "\\r") + .replace('\t', "\\t") +} + +#[cfg(test)] +mod tests { + use super::*; + use irtt_client::{ClientTimestamp, OneWayDelaySample, PacketMeta, ReceivedStatsSample}; + use std::{ + net::{IpAddr, Ipv4Addr, SocketAddr}, + time::{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() -> ClientEvent { + ClientEvent::EchoReply { + seq: 7, + remote: test_remote(), + sent_at: test_timestamp(Duration::from_secs(1)), + received_at: test_timestamp(Duration::from_secs(1) + Duration::from_micros(1500)), + rtt: RttSample { + raw: Duration::from_micros(1500), + adjusted: Some(SignedDuration::from_nanos(1_200_000)), + effective: SignedDuration::from_nanos(1_200_000), + }, + server_timing: Some(ServerTiming { + receive_wall_ns: Some(1_000), + receive_mono_ns: Some(2_000), + send_wall_ns: Some(301_000), + send_mono_ns: Some(302_000), + midpoint_wall_ns: None, + midpoint_mono_ns: None, + processing: Some(Duration::from_micros(300)), + }), + one_way: Some(OneWayDelaySample { + client_to_server: Some(SignedDuration::from_nanos(400_000)), + server_to_client: Some(SignedDuration::from_nanos(500_000)), + }), + received_stats: Some(ReceivedStatsSample { + count: Some(9), + window: Some(0x7), + }), + bytes: 64, + packet_meta: PacketMeta::default(), + } + } + + #[test] + fn rtt_us_prints_only_effective_reply_rtt() { + assert_eq!( + format_event(&reply_event(), OutputMode::RttUs), + Some("1200".to_owned()) + ); + assert_eq!( + format_event( + &ClientEvent::Warning { + kind: WarningKind::WrongToken, + message: "bad".to_owned(), + at: ClientTimestamp::now() + }, + OutputMode::RttUs + ), + None + ); + } + + #[test] + fn machine_prints_stable_key_value_fields() { + let line = format_event(&reply_event(), OutputMode::Machine).unwrap(); + assert!(line.starts_with("event=echo_reply ")); + assert!(line.contains("seq=7")); + assert!(line.contains("remote=127.0.0.1:2112")); + assert!(line.contains("client_send_wall_ns=1000000000")); + assert!(line.contains("client_receive_wall_ns=1001500000")); + assert!(line.contains("raw_rtt_us=1500")); + assert!(line.contains("adjusted_rtt_us=1200")); + assert!(line.contains("effective_rtt_us=1200")); + assert!(line.contains("server_processing_us=300")); + assert!(line.contains("server_received_count=9")); + } + + #[test] + fn simple_and_human_use_readable_reply_lines() { + assert_eq!( + format_event(&reply_event(), OutputMode::Simple), + Some( + "reply seq=7 remote=127.0.0.1:2112 rtt_us=1200 raw_rtt_us=1500 server_processing_us=300" + .to_owned() + ) + ); + + let human = format_event(&reply_event(), OutputMode::Human).unwrap(); + assert!(human.starts_with("seq=7 rtt=1.2ms")); + assert!(human.contains("rd=500.0µs")); + assert!(human.contains("sd=400.0µs")); + assert!(human.contains("ipdv=n/a")); + assert!(human.contains("proc=300.0µs")); + assert!(!human.contains("rtt_us=")); + } + + #[test] + fn echo_sent_is_not_formatted_for_stream_outputs() { + let ts = test_timestamp(Duration::from_secs(1)); + let event = ClientEvent::EchoSent { + seq: 1, + remote: test_remote(), + scheduled_at: ts.mono, + sent_at: ts, + bytes: 64, + send_call: Duration::from_micros(10), + timer_error: Duration::ZERO, + }; + + assert!(format_event(&event, OutputMode::RttUs).is_none()); + assert!(format_event(&event, OutputMode::Human).is_none()); + assert!(format_event(&event, OutputMode::Machine).is_none()); + assert!(format_event(&event, OutputMode::Simple).is_none()); + } +} diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs new file mode 100644 index 0000000..a68210e --- /dev/null +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -0,0 +1,485 @@ +use std::{ + io::{self, Write}, + sync::atomic::AtomicBool, +}; + +use irtt_client::ClientEvent; +#[cfg(all(test, feature = "stats"))] +use irtt_client::{ClientTimestamp, PacketMeta, RttSample, SignedDuration}; + +use super::{ + args::{ClientArgs, OutputMode}, + output::{format_event, format_human_event_with_options, HumanOutputOptions}, +}; +#[cfg(feature = "stats")] +use crate::shared::client::expected_probe_count; +use crate::shared::client::{ + is_shutdown_requested, session::should_print_final_summary, ClientSession, +}; + +#[cfg(feature = "stats")] +use irtt_stats::{StatsCollector, StatsConfig}; + +#[cfg(feature = "stats")] +const FINITE_STATS_BYTES_PER_PROBE: u64 = 500; +#[cfg(feature = "stats")] +const MIB: u64 = 1024 * 1024; +#[cfg(feature = "stats")] +const GIB: u64 = 1024 * MIB; +#[cfg(feature = "stats")] +const FINITE_STATS_MEMORY_WARNING_BYTES: u64 = 128 * MIB; +#[cfg(feature = "stats")] +const FINITE_STATS_MEMORY_STRONG_WARNING_BYTES: u64 = 512 * MIB; +#[cfg(feature = "stats")] +const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB; + +pub fn run_stream( + args: ClientArgs, + shutdown_requested: &AtomicBool, +) -> Result<(), Box> { + let mode = args.output; + let continuous = args.is_continuous(); + #[cfg(feature = "stats")] + if let Some(warning) = finite_stats_memory_warning(&args) { + eprintln!("{warning}"); + } + + 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 { + mode, + human_options: HumanOutputOptions { + verbose: args.verbose, + }, + 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) { + return Ok(()); + } + + let mut session = ClientSession::connect(args.to_client_config(), continuous)?; + + if is_shutdown_requested(shutdown_requested) { + return Ok(()); + } + + let events = session.open()?; + stream_output.print_events(&events)?; + + let mut interrupted = false; + while session.should_continue(shutdown_requested) { + let events = session.step(shutdown_requested)?; + stream_output.print_events(&events)?; + + if is_shutdown_requested(shutdown_requested) { + interrupted = true; + break; + } + + session.sleep_until_next_send(); + } + interrupted |= is_shutdown_requested(shutdown_requested); + + if interrupted { + eprintln!("interrupted, closing session..."); + } + + if session.should_drain_final(interrupted) { + session.drain_final(|events| { + let _ = stream_output.print_events(events); + })?; + } + + let events = session.poll_timeouts()?; + stream_output.print_events(&events)?; + + let events = session.close()?; + stream_output.print_events(&events)?; + 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; + stream_output.print_summary()?; + stream_output.out.flush()?; + Ok(()) +} + +struct StreamOutput<'a, W: Write> { + mode: OutputMode, + human_options: HumanOutputOptions, + 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<()> { + for event in events { + self.print_event(event)?; + } + 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 = crate::cmd::client::HumanEventStats::default(); + + let line = + if self.mode == OutputMode::Human && !matches!(event, ClientEvent::EchoSent { .. }) { + #[cfg(feature = "stats")] + let human_stats = stats_update.into(); + #[cfg(not(feature = "stats"))] + let human_stats = stats_update; + + Some(format_human_event_with_options( + event, + Some(human_stats), + self.human_options, + )) + } else { + format_event(event, self.mode) + }; + + if let Some(line) = line { + writeln!(self.out, "{line}")?; + } + Ok(()) + } + + fn print_summary(&mut self) -> io::Result<()> { + if !self.print_final_summary || !self.mode.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.human_options.verbose, + show_running_only_note: self.show_running_only_summary_note, + }, + ) + )?; + } + + Ok(()) + } +} + +#[cfg(feature = "stats")] +fn stats_config(continuous: bool) -> StatsConfig { + if continuous { + StatsConfig::continuous() + } else { + 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); + if estimated_bytes < FINITE_STATS_MEMORY_WARNING_BYTES { + return None; + } + + let formatted = format_bytes_for_warning(estimated_bytes); + 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 { + "consider shortening the run, increasing the interval, or using continuous mode" + } 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}" + )) +} + +#[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"))] +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() + } + + #[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 { + mode: OutputMode::RttUs, + human_options: HumanOutputOptions::default(), + 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(); + } + + let rendered = String::from_utf8(out).unwrap(); + let summary = stats.snapshot(); + assert_eq!(rendered, "1200\n"); + 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 { + mode: OutputMode::Human, + human_options: HumanOutputOptions::default(), + 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(); + } + + 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 mut stats = StatsCollector::new(StatsConfig::continuous()); + let mut out = Vec::new(); + { + let mut stream_output = StreamOutput { + mode: OutputMode::Human, + human_options: HumanOutputOptions::default(), + print_final_summary: false, + show_running_only_summary_note: true, + out: &mut out, + stats: &mut stats, + }; + stream_output.print_summary().unwrap(); + } + + assert!(out.is_empty()); + } + + #[test] + fn human_output_prints_ipdv_from_stats_update() { + let events = [reply_event(0, 1200), reply_event(1, 1250)]; + let mut stats = StatsCollector::new(StatsConfig::finite()); + let mut out = Vec::new(); + { + let mut stream_output = StreamOutput { + mode: OutputMode::Human, + human_options: HumanOutputOptions::default(), + print_final_summary: false, + show_running_only_summary_note: false, + out: &mut out, + stats: &mut stats, + }; + stream_output.print_events(&events).unwrap(); + } + + let rendered = String::from_utf8(out).unwrap(); + let mut lines = rendered.lines(); + assert!(lines.next().unwrap().contains("ipdv=n/a")); + assert!(lines.next().unwrap().contains("ipdv=50.0µs")); + } + + #[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); + } +} diff --git a/crates/irtt-cli/src/summary.rs b/crates/irtt-cli/src/cmd/client/summary.rs similarity index 100% rename from crates/irtt-cli/src/summary.rs rename to crates/irtt-cli/src/cmd/client/summary.rs diff --git a/crates/irtt-cli/src/shared/client/session.rs b/crates/irtt-cli/src/shared/client/session.rs new file mode 100644 index 0000000..116ee37 --- /dev/null +++ b/crates/irtt-cli/src/shared/client/session.rs @@ -0,0 +1,162 @@ +use std::{ + sync::atomic::{AtomicBool, Ordering}, + thread, + time::{Duration, Instant}, +}; + +use irtt_client::{Client, ClientConfig, ClientEvent, OpenOutcome, RecvBudget}; + +pub const RECV_BUDGET: RecvBudget = RecvBudget { max_packets: 16 }; +const MAX_FINAL_DRAIN: Duration = Duration::from_secs(30); +const IDLE_SLEEP: Duration = Duration::from_millis(5); +const MAX_SLEEP: Duration = Duration::from_millis(20); + +pub struct ClientSession { + client: Client, + continuous: bool, +} + +impl ClientSession { + pub fn connect( + config: ClientConfig, + continuous: bool, + ) -> Result { + Ok(Self { + client: Client::connect(config)?, + continuous, + }) + } + + pub fn open(&mut self) -> Result, irtt_client::ClientError> { + Ok(vec![open_event(&self.client.open()?).clone()]) + } + + pub fn step( + &mut self, + shutdown_requested: &AtomicBool, + ) -> Result, irtt_client::ClientError> { + let mut events = Vec::new(); + if should_send_probe(self.client.next_send_deadline(), shutdown_requested) { + events.extend(self.client.send_probe()?); + } + events.extend(self.client.recv_available(RECV_BUDGET)?); + events.extend(self.client.poll_timeouts()?); + Ok(events) + } + + pub fn should_continue(&self, shutdown_requested: &AtomicBool) -> bool { + !is_shutdown_requested(shutdown_requested) + && (self.continuous || self.client.next_send_deadline().is_some()) + } + + pub fn should_drain_final(&self, interrupted: bool) -> bool { + should_drain_final(self.continuous, interrupted) + } + + pub fn drain_final(&mut self, mut on_events: F) -> Result<(), irtt_client::ClientError> + where + F: FnMut(&[ClientEvent]), + { + let deadline = Instant::now() + final_drain_duration(self.client.probe_timeout()); + loop { + let mut received = false; + + let events = self.client.recv_available(RECV_BUDGET)?; + received |= !events.is_empty(); + on_events(&events); + + let events = self.client.poll_timeouts()?; + received |= !events.is_empty(); + on_events(&events); + + if self.client.is_run_complete() || Instant::now() >= deadline { + break; + } + + if !received { + thread::sleep(IDLE_SLEEP); + } + } + Ok(()) + } + + pub fn poll_timeouts(&mut self) -> Result, irtt_client::ClientError> { + self.client.poll_timeouts() + } + + pub fn close(&mut self) -> Result, irtt_client::ClientError> { + self.client.close() + } + + pub fn next_send_deadline(&self) -> Option { + self.client.next_send_deadline() + } + + pub fn probe_timeout(&self) -> Duration { + self.client.probe_timeout() + } + + pub fn sleep_until_next_send(&self) { + let sleep_for = match self.next_send_deadline() { + Some(deadline) => deadline + .saturating_duration_since(Instant::now()) + .min(MAX_SLEEP), + None => Duration::from_millis(1), + }; + if !sleep_for.is_zero() { + thread::sleep(sleep_for); + } + } +} + +pub fn is_shutdown_requested(shutdown_requested: &AtomicBool) -> bool { + shutdown_requested.load(Ordering::Relaxed) +} + +pub fn should_drain_final(continuous: bool, interrupted: bool) -> bool { + !continuous || interrupted +} + +pub fn should_print_final_summary(continuous: bool, interrupted: bool) -> bool { + !continuous || interrupted +} + +pub fn final_drain_duration(probe_timeout: Duration) -> Duration { + probe_timeout.min(MAX_FINAL_DRAIN) +} + +fn should_send_probe(next_send_deadline: Option, shutdown_requested: &AtomicBool) -> bool { + !is_shutdown_requested(shutdown_requested) + && next_send_deadline.is_some_and(|deadline| deadline <= Instant::now()) +} + +fn open_event(outcome: &OpenOutcome) -> &ClientEvent { + match outcome { + OpenOutcome::Started { event, .. } | OpenOutcome::NoTestCompleted { event, .. } => event, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn final_drain_uses_capped_probe_timeout() { + assert_eq!( + final_drain_duration(Duration::from_secs(4)), + Duration::from_secs(4) + ); + assert_eq!( + final_drain_duration(Duration::from_secs(60)), + Duration::from_secs(30) + ); + } + + #[test] + fn should_drain_final_for_finite_or_interrupted_runs() { + assert!(should_drain_final(false, false)); + assert!(should_drain_final(false, true)); + assert!(should_drain_final(true, true)); + assert!(!should_drain_final(true, false)); + } +}