From 6af4663b6e773209e6be853f80f76e178376a8c9 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Jul 2026 14:42:02 +0700 Subject: [PATCH] fix(cli): drive single-target runs with ManagedClient Replace the manual single-target ClientSession loops in the stream and TUI adapters with ManagedClient::start_with_subscription so the existing worker owns opening, deadline-aware pacing, receive handling, timeout polling, draining, and close. Use bounded subscription waits for interruption and input polling, drain queued events before and after join, and report per-subscription drops without changing output, statistics, labels, summaries, warning, or peer-close policy. Add process and unit regressions for 10 ms cadence, interruption, final-event draining, TUI terminal state, and overflow visibility. --- crates/irtt-cli/src/cmd/client/run.rs | 212 ++++++++--- crates/irtt-cli/src/cmd/tui/run.rs | 136 ++++--- crates/irtt-cli/src/cmd/tui/ui.rs | 43 ++- crates/irtt-cli/src/shared/client/mod.rs | 2 +- crates/irtt-cli/src/shared/client/session.rs | 365 +------------------ crates/irtt-cli/tests/client_multi_target.rs | 152 ++++++++ crates/irtt-client/src/error.rs | 6 +- crates/irtt-client/src/managed/hub.rs | 53 +++ crates/irtt-client/src/managed/runner.rs | 38 +- 9 files changed, 530 insertions(+), 477 deletions(-) diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index 33c62d8..4d47efb 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -9,9 +9,10 @@ use std::{ }; use irtt_client::{ - ClientEvent, EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig, - ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedTargetOutcome, - SubscriberConfig, SubscriberOverflow, TargetEvent, + ClientEvent, EventSubscriptionError, ManagedClient, ManagedClientGroup, + ManagedClientGroupConfig, ManagedGroupCompletionPolicy, ManagedGroupEndReason, + ManagedGroupEvent, ManagedTargetOutcome, SessionEndReason, SubscriberConfig, + SubscriberOverflow, TargetEvent, }; #[cfg(all(test, feature = "stats"))] use irtt_client::{ClientTimestamp, PacketMeta, RttSample, SignedDuration}; @@ -28,7 +29,6 @@ use crate::shared::client::{ peer_close_run_error, request_group_stop_once, should_print_final_summary, should_stop_group_on_peer_close, }, - ClientSession, }; #[cfg(feature = "stats")] @@ -49,6 +49,8 @@ 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; pub fn run_stream( args: ClientArgs, @@ -103,66 +105,71 @@ pub fn run_stream( return Ok(()); } - let mut session = ClientSession::connect(args.to_client_config(), continuous)?; - - if is_shutdown_requested(shutdown_requested) { - return Ok(()); - } - - let events = session.open()?; - #[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 (session, events) = ManagedClient::start_with_subscription( + args.to_client_config(), + managed_event_subscriber_config(), + )?; let mut interrupted = false; - while session.should_continue(shutdown_requested) { - let events = session.step(shutdown_requested)?; - #[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))?; - + loop { if is_shutdown_requested(shutdown_requested) { interrupted = true; + session.stop(); break; } - session.sleep_until_next_send(); + 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..."); - } - - if session.should_drain_final(interrupted) { - session.drain_final(|events| { + drain_single_client_events( + &events, + &mut stream_output, + target_label, #[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)); - })?; + &mut stats, + )?; } - let events = session.poll_timeouts()?; - #[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 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}"); + } - let events = session.close()?; - #[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))?; interrupted |= is_shutdown_requested(shutdown_requested); - let peer_close_error = peer_close_run_error( - continuous, - interrupted, - if session.is_peer_closed() { 1 } else { 0 }, - ); + 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; @@ -171,12 +178,59 @@ pub fn run_stream( #[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, @@ -300,7 +354,7 @@ fn run_group_stream( group_config, managed_targets, SubscriberConfig { - capacity: 16_384, + 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 @@ -427,7 +481,7 @@ fn run_group_stream( } } } - if let Some(warning) = dropped_event_warning(events.dropped_events()) { + if let Some(warning) = dropped_event_warning(events.dropped_events(), "managed group") { eprintln!("{warning}"); } @@ -492,10 +546,15 @@ fn report_target_failure(target: &ManagedTargetOutcome) { } } -fn dropped_event_warning(dropped_events: u64) -> Option { +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} managed group events; output and statistics may be incomplete" + "irtt-rs: warning: dropped {dropped_events} {source} {event_word}; output and statistics may be incomplete" ) }) } @@ -654,12 +713,57 @@ mod tests { #[test] fn dropped_event_warning_marks_statistics_incomplete() { - assert_eq!(dropped_event_warning(0), None); - let warning = dropped_event_warning(7).unwrap(); + 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)); diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index 1a633d9..99d8b77 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -7,9 +7,9 @@ use std::{ use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers}; use irtt_client::{ - EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig, - ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, SubscriberConfig, - SubscriberOverflow, + EventSubscription, EventSubscriptionError, ManagedClient, ManagedClientGroup, + ManagedClientGroupConfig, ManagedGroupCompletionPolicy, ManagedGroupEndReason, + ManagedGroupEvent, SessionEndReason, SubscriberConfig, SubscriberOverflow, }; use crate::{ @@ -17,7 +17,6 @@ use crate::{ shared::client::{ is_shutdown_requested, session::{peer_close_run_error, request_group_stop_once, should_stop_group_on_peer_close}, - ClientSession, }, }; @@ -27,6 +26,7 @@ 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( args: TuiArgs, @@ -61,7 +61,10 @@ pub fn run_tui( ); } - let mut session = match ClientSession::connect(args.to_client_config(), continuous) { + let (session, events) = match ManagedClient::start_with_subscription( + args.to_client_config(), + managed_event_subscriber_config(), + ) { Ok(session) => session, Err(err) => { state.set_error(err.to_string()); @@ -70,43 +73,33 @@ pub fn run_tui( } }; - if is_shutdown_requested(shutdown_requested) { - return Ok(()); - } - - let events = session.open()?; - state.process_events(&events); - render_if_due(&mut terminal, &state, &mut next_render, true)?; - let mut interrupted = false; - // Keep this in lockstep with run_stream: send due probes, drain available - // replies, poll timeouts, sleep toward the next absolute send deadline, - // then perform the same final drain, timeout poll, and close sequence. - while session.should_continue(shutdown_requested) { + loop { + if is_shutdown_requested(shutdown_requested) { + interrupted = true; + session.stop(); + break; + } + 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; } - let events = session.step(shutdown_requested)?; - state.process_events(&events); - - if is_shutdown_requested(shutdown_requested) { - interrupted = true; - 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, } - - render_if_due(&mut terminal, &state, &mut next_render, false)?; - wait_for_tui_activity( - session.next_send_deadline(), - &mut next_render, - &mut state, - &mut terminal, - shutdown_requested, - )?; } interrupted |= is_shutdown_requested(shutdown_requested); @@ -115,29 +108,32 @@ pub fn run_tui( render_if_due(&mut terminal, &state, &mut next_render, true)?; } - if session.should_drain_final(interrupted) { + if interrupted { state.set_status(TuiStatus::Draining); - let mut drain_render = Instant::now(); - session.drain_final(|events| { - state.process_events(events); - let _ = render_if_due(&mut terminal, &state, &mut drain_render, false); - })?; + drain_single_tui_events(&events, &mut state); render_if_due(&mut terminal, &state, &mut next_render, true)?; } - let events = session.poll_timeouts()?; - state.process_events(&events); - state.set_status(TuiStatus::Closing); render_if_due(&mut terminal, &state, &mut next_render, true)?; - let events = session.close()?; - state.process_events(&events); + 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, - if session.is_peer_closed() { 1 } else { 0 }, + u64::from(outcome.end_reason == SessionEndReason::PeerClosed), ) { state.set_run_error(error.clone()); render_if_due(&mut terminal, &state, &mut next_render, true)?; @@ -175,7 +171,7 @@ fn run_group_tui( group_config, managed_targets, SubscriberConfig { - capacity: 16_384, + capacity: MANAGED_EVENT_CAPACITY, overflow: SubscriberOverflow::DropOldest, }, )?; @@ -272,7 +268,7 @@ fn run_group_tui( } } } - state.mark_dropped_events(events.dropped_events()); + state.mark_dropped_group_events(events.dropped_events()); for target in &outcome.targets { if terminal_targets.insert(target.id.as_str().to_owned()) { state.process_target_outcome(target); @@ -451,6 +447,28 @@ fn render_if_due( 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 + } else { + next_render.saturating_duration_since(Instant::now()) + }; + render_wait.min(TUI_WAIT_SLICE) +} + fn wait_for_tui_activity( next_send_deadline: Option, next_render: &mut Instant, @@ -552,4 +570,30 @@ mod tests { &irtt_client::ManagedTargetEndReason::PeerClosed )); } + + #[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 bcaabee..92a74d1 100644 --- a/crates/irtt-cli/src/cmd/tui/ui.rs +++ b/crates/irtt-cli/src/cmd/tui/ui.rs @@ -140,12 +140,6 @@ impl TuiState { } } - pub(super) fn process_events(&mut self, events: &[ClientEvent]) { - for event in events { - self.process_event(event); - } - } - pub(super) fn process_event(&mut self, event: &ClientEvent) { self.process_event_for_target(0, event); } @@ -388,13 +382,31 @@ impl TuiState { self.status = status; } - pub(super) fn mark_dropped_events(&mut self, dropped_events: u64) { + #[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"); + } + + fn mark_dropped_events(&mut self, dropped_events: u64, source: &str) { if dropped_events == 0 { return; } self.dropped_events = dropped_events; + let event_word = if dropped_events == 1 { + "event" + } else { + "events" + }; let warning = - format!("dropped {dropped_events} managed group events; statistics may be incomplete"); + format!("dropped {dropped_events} {source} {event_word}; statistics may be incomplete"); self.last_warning = Some(warning.clone()); self.push_event(format!("warning {warning}")); } @@ -2497,7 +2509,7 @@ mod tests { fn dropped_events_mark_tui_statistics_incomplete() { let mut state = TuiState::default(); - state.mark_dropped_events(9); + state.mark_dropped_group_events(9); assert_eq!(state.dropped_events, 9); assert!(state @@ -2510,6 +2522,19 @@ mod tests { .is_some_and(|event| event.contains("dropped 9 managed group events"))); } + #[test] + fn dropped_single_client_events_are_visible() { + let mut state = TuiState::default(); + + state.mark_dropped_client_events(3); + + assert_eq!(state.dropped_events, 3); + assert!(state + .recent_events + .back() + .is_some_and(|event| event.contains("dropped 3 managed client events"))); + } + #[test] fn recent_event_buffer_stays_bounded() { let mut state = TuiState::default(); diff --git a/crates/irtt-cli/src/shared/client/mod.rs b/crates/irtt-cli/src/shared/client/mod.rs index 297de9d..927b70b 100644 --- a/crates/irtt-cli/src/shared/client/mod.rs +++ b/crates/irtt-cli/src/shared/client/mod.rs @@ -8,7 +8,7 @@ pub use args::{ ClockArg, CommonClientArgs, ReceivedStatsArg, TimestampArg, }; pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT}; -pub use session::{final_drain_duration, is_shutdown_requested, ClientSession, RECV_BUDGET}; +pub use session::is_shutdown_requested; pub use targets::{ parse_labelled_target, resolved_managed_targets, target_specs, GroupPacingArg, LabelledTargetArg, ResolvedTarget, TargetSpec, diff --git a/crates/irtt-cli/src/shared/client/session.rs b/crates/irtt-cli/src/shared/client/session.rs index 444173f..79bce61 100644 --- a/crates/irtt-cli/src/shared/client/session.rs +++ b/crates/irtt-cli/src/shared/client/session.rs @@ -1,146 +1,11 @@ -use std::{ - sync::atomic::{AtomicBool, Ordering}, - thread, - time::{Duration, Instant}, -}; +use std::sync::atomic::{AtomicBool, Ordering}; -use irtt_client::{ - Client, ClientConfig, ClientEvent, ManagedTargetEndReason, 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)?); - if self.client.is_peer_closed() { - return Ok(events); - } - events.extend(self.client.poll_timeouts()?); - Ok(events) - } - - pub fn should_continue(&self, shutdown_requested: &AtomicBool) -> bool { - !is_shutdown_requested(shutdown_requested) - && !self.client.is_peer_closed() - && (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 { - if self.client.is_peer_closed() { - break; - } - - let mut received = false; - - let events = self.client.recv_available(RECV_BUDGET)?; - received |= !events.is_empty(); - on_events(&events); - - if self.client.is_peer_closed() { - break; - } - - 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> { - if self.client.is_peer_closed() { - return Ok(vec![]); - } - self.client.poll_timeouts() - } - - pub fn close(&mut self) -> Result, irtt_client::ClientError> { - if self.client.is_peer_closed() { - return Ok(vec![]); - } - self.client.close() - } - - pub fn is_peer_closed(&self) -> bool { - self.client.is_peer_closed() - } - - 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); - } - } -} +use irtt_client::ManagedTargetEndReason; 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 } @@ -180,67 +45,9 @@ pub fn request_group_stop_once(stop_requested: &mut bool) -> bool { true } -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::*; - use irtt_proto::{ - echo_packet_len, - flags::{self, FLAG_OPEN, FLAG_REPLY}, - layout::PacketLayout, - Clock, Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION, - }; - use std::{ - net::{SocketAddr, UdpSocket}, - thread::JoinHandle, - }; - - const TOKEN: u64 = 0x1234_5678_90ab_cdef; - - struct FakeServer { - addr: SocketAddr, - done: JoinHandle<()>, - } - - impl FakeServer { - fn join(self) { - self.done.join().unwrap(); - } - } - - #[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)); - } #[test] fn peer_close_run_error_is_run_mode_and_interruption_aware() { @@ -284,172 +91,4 @@ mod tests { assert!(request_group_stop_once(&mut stop_requested)); assert!(!request_group_stop_once(&mut stop_requested)); } - - #[test] - fn peer_close_followed_by_session_cleanup_is_successful() { - let params = test_params(None, Duration::from_millis(10)); - let server = start_peer_close_server(params); - let mut session = ClientSession::connect(config(server.addr, None), true).unwrap(); - let shutdown_requested = AtomicBool::new(false); - - assert!(matches!( - session.open().unwrap().as_slice(), - [ClientEvent::SessionStarted { .. }] - )); - - let events = session.step(&shutdown_requested).unwrap(); - assert!(events - .iter() - .any(|event| matches!(event, ClientEvent::EchoReply { .. }))); - assert!(events - .iter() - .any(|event| matches!(event, ClientEvent::SessionClosed { .. }))); - assert!(!session.should_continue(&shutdown_requested)); - assert!(session.is_peer_closed()); - - session.drain_final(|_| {}).unwrap(); - assert!(session.poll_timeouts().unwrap().is_empty()); - assert!(session.close().unwrap().is_empty()); - server.join(); - } - - fn config(addr: SocketAddr, duration: Option) -> ClientConfig { - ClientConfig { - server_addr: addr.to_string(), - duration, - interval: Duration::from_millis(10), - open_timeouts: vec![Duration::from_millis(200)], - probe_timeout: Duration::from_millis(20), - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..ClientConfig::default() - } - } - - fn test_params(duration: Option, interval: Duration) -> Params { - Params { - protocol_version: PROTOCOL_VERSION, - duration_ns: duration.map_or(0, test_duration_ns_i64), - interval_ns: test_duration_ns_i64(interval), - length: 0, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - dscp: 0, - server_fill: None, - } - } - - fn test_duration_ns_i64(duration: Duration) -> i64 { - i64::try_from(duration.as_nanos()).expect("test duration fits i64 nanoseconds") - } - - fn start_peer_close_server(params: Params) -> FakeServer { - let socket = UdpSocket::bind("127.0.0.1:0").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(); - - let (packet, peer) = recv_request(&socket); - 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 | flags::FLAG_CLOSE, - ), - peer, - ) - .unwrap(); - - socket - .set_read_timeout(Some(Duration::from_millis(250))) - .unwrap(); - if let Some((packet, _)) = recv_request_timeout(&socket) { - panic!( - "session cleanup must not send any packet after peer close; flags={:?} len={}", - packet.get(3).copied(), - packet.len() - ); - } - }); - 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 recv_request_timeout(socket: &UdpSocket) -> Option<(Vec, SocketAddr)> { - let mut buf = [0_u8; 2048]; - socket - .recv_from(&mut buf) - .ok() - .map(|(size, peer)| (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) - .expect("session test params must have a non-negative packet length"); - 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 - } } diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index 62701cc..6559388 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -48,6 +48,19 @@ impl InterruptibleFakeServer { } } +struct ObservedFakeServer { + addr: SocketAddr, + probes: mpsc::Receiver, + done: JoinHandle<()>, +} + +impl ObservedFakeServer { + fn join(self) -> Vec { + self.done.join().unwrap(); + self.probes.try_iter().collect() + } +} + #[test] fn list_columns_succeeds_without_target() { let output = Command::new(env!("CARGO_BIN_EXE_irtt-cli")) @@ -143,6 +156,56 @@ fn single_target_custom_target_column_renders_positional_label() { ); } +#[test] +fn single_target_ten_millisecond_cadence_uses_managed_deadlines() { + let duration = Duration::from_millis(90); + let interval = Duration::from_millis(10); + let server = start_observed_echo_server(test_params(Some(duration), interval)); + let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); + command.args([ + "--duration", + "90ms", + "--interval", + "10ms", + "--format", + "csv", + "--columns", + "event,seq", + "--header", + "never", + &server.addr.to_string(), + ]); + + let output = run_with_timeout(command, Duration::from_secs(3)); + let probe_times = server.join(); + + assert!( + output.status.success(), + "status={:?}\nstderr={}", + output.status.code(), + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + let sent_rows = stdout + .lines() + .filter(|line| line.starts_with("echo_sent,")) + .count(); + assert!( + (7..=11).contains(&sent_rows), + "expected about 9 probes at 10 ms cadence, got {sent_rows}\n{stdout}" + ); + assert_eq!(sent_rows, probe_times.len(), "{stdout}"); + assert!( + !probe_times + .windows(3) + .any(|window| window[2].duration_since(window[0]) < Duration::from_millis(5)), + "managed pacing emitted a catch-up burst: {probe_times:?}" + ); + assert!(stdout + .lines() + .any(|line| line.starts_with("session_closed,"))); +} + #[test] fn finite_single_target_peer_close_exits_successfully() { let server = start_peer_close_server(test_params( @@ -213,6 +276,57 @@ fn continuous_single_target_peer_close_exits_nonzero() { assert!(stderr.contains("peer closure"), "{stderr}"); } +#[cfg(unix)] +#[test] +fn continuous_single_target_interruption_drains_final_events_and_succeeds() { + let params = test_params(None, Duration::from_millis(10)); + let server = start_interruptible_echo_server(params); + let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); + command.args([ + "--duration", + "0s", + "--interval", + "10ms", + "--format", + "csv", + "--columns", + "event,seq", + "--header", + "never", + &server.addr.to_string(), + ]); + + let child = command + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + server.wait_until_opened(); + thread::sleep(Duration::from_millis(50)); + let signal_status = Command::new("kill") + .args(["-INT", &child.id().to_string()]) + .status() + .unwrap(); + assert!(signal_status.success()); + let output = wait_with_timeout(child, Duration::from_secs(2)); + server.join(); + + assert!( + output.status.success(), + "status={:?}\nstderr={}", + output.status.code(), + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!(stdout.lines().any(|line| line.starts_with("echo_reply,"))); + assert!(stdout + .lines() + .any(|line| line.starts_with("session_closed,"))); + assert!(stderr.contains("interrupted, closing session"), "{stderr}"); + assert!(!stderr.contains("peer closure"), "{stderr}"); +} + #[test] fn multi_target_csv_emits_rows_for_both_labels() { let params = test_params(Some(Duration::from_millis(40)), Duration::from_millis(10)); @@ -658,6 +772,44 @@ fn start_echo_server_inner( FakeServer { addr, done } } +fn start_observed_echo_server(params: Params) -> ObservedFakeServer { + 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 (probe_tx, probes) = mpsc::channel(); + 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; + } + probe_tx.send(Instant::now()).unwrap(); + 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(); + } + }); + ObservedFakeServer { addr, probes, done } +} + fn start_interruptible_echo_server(params: Params) -> InterruptibleFakeServer { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket diff --git a/crates/irtt-client/src/error.rs b/crates/irtt-client/src/error.rs index 164d247..ace8630 100644 --- a/crates/irtt-client/src/error.rs +++ b/crates/irtt-client/src/error.rs @@ -169,9 +169,9 @@ pub enum EventSubscriptionError { /// The subscription can no longer receive new events. /// /// Already queued events are delivered first. After the queue is drained, - /// `recv` and `try_recv` return this error when the hub disconnects the - /// subscriber, the hub is shut down, or the subscription is disconnected by - /// its overflow policy. + /// `recv`, `recv_timeout`, and `try_recv` return this error when the hub + /// disconnects the subscriber, the hub is shut down, or the subscription is + /// disconnected by its overflow policy. #[error("event subscription is disconnected")] Disconnected, } diff --git a/crates/irtt-client/src/managed/hub.rs b/crates/irtt-client/src/managed/hub.rs index 8f52c05..55cf3f3 100644 --- a/crates/irtt-client/src/managed/hub.rs +++ b/crates/irtt-client/src/managed/hub.rs @@ -4,6 +4,7 @@ use std::{ atomic::{AtomicU64, Ordering}, Arc, Condvar, Mutex, Weak, }, + time::{Duration, Instant}, }; use crate::{error::EventSubscriptionError, event::ClientEvent}; @@ -254,6 +255,39 @@ impl EventSubscription { } } + /// Wait up to `timeout` for the next queued event. + /// + /// Returns `Ok(None)` when the timeout elapses while the subscription + /// remains connected. If the subscription is disconnected, this returns + /// [`EventSubscriptionError::Disconnected`] after any already queued events + /// have been drained. + pub fn recv_timeout(&self, timeout: Duration) -> Result, EventSubscriptionError> { + let started_at = Instant::now(); + let mut state = self.inner.state.lock().expect("subscriber mutex poisoned"); + loop { + if let Some(event) = state.queue.pop_front() { + return Ok(Some(event)); + } + if !state.connected { + return Err(EventSubscriptionError::Disconnected); + } + + let Some(wait_for) = timeout.checked_sub(started_at.elapsed()) else { + return Ok(None); + }; + if wait_for.is_zero() { + return Ok(None); + } + + let (next_state, _) = self + .inner + .available + .wait_timeout(state, wait_for) + .expect("subscriber mutex poisoned"); + state = next_state; + } + } + /// Try to receive one queued event without blocking. /// /// Returns `Ok(None)` when the subscription is still connected but no event @@ -507,6 +541,25 @@ mod tests { ); } + #[test] + fn timed_receive_drains_queued_events_and_observes_disconnect() { + let hub = EventHub::new(); + let sub = hub.subscribe(SubscriberConfig::default()).unwrap(); + + assert_eq!(sub.recv_timeout(Duration::ZERO).unwrap(), None); + hub.publish(event(1)); + hub.disconnect_all(); + + assert_eq!( + event_message(sub.recv_timeout(Duration::from_secs(1)).unwrap().unwrap()), + "event-1" + ); + assert_eq!( + sub.recv_timeout(Duration::from_secs(1)).unwrap_err(), + EventSubscriptionError::Disconnected + ); + } + #[test] fn sealed_hub_drains_queued_events_then_disconnects() { let hub = EventHub::new(); diff --git a/crates/irtt-client/src/managed/runner.rs b/crates/irtt-client/src/managed/runner.rs index 86b0edb..f45848d 100644 --- a/crates/irtt-client/src/managed/runner.rs +++ b/crates/irtt-client/src/managed/runner.rs @@ -358,7 +358,7 @@ impl OutcomeCounters { #[cfg(test)] mod tests { use super::*; - use crate::{config::NegotiationPolicy, SubscriberOverflow}; + use crate::{config::NegotiationPolicy, EventSubscriptionError, SubscriberOverflow}; use irtt_proto::{ echo_packet_len, flags::{self, FLAG_OPEN, FLAG_REPLY}, @@ -687,6 +687,42 @@ mod tests { .any(|event| matches!(event, ClientEvent::SessionClosed { .. }))); } + #[test] + fn joining_before_consuming_preserves_queued_final_events() { + let duration = Duration::from_millis(25); + let server = start_echo_server(test_params(Some(duration), Duration::from_millis(10))); + let (session, sub) = ManagedClient::start_with_subscription( + config(server.addr, Some(duration)), + SubscriberConfig { + capacity: 32, + overflow: SubscriberOverflow::DropOldest, + }, + ) + .unwrap(); + + let outcome = session.join().unwrap(); + let mut events = Vec::new(); + loop { + match sub.try_recv() { + Ok(Some(event)) => events.push(event), + Ok(None) => panic!("joined managed subscription remained connected"), + Err(EventSubscriptionError::Disconnected) => break, + } + } + server.join(); + + assert_eq!(outcome.end_reason, SessionEndReason::TestComplete); + assert_eq!(sub.dropped_events(), 0); + assert!(matches!( + events.first(), + Some(ClientEvent::SessionStarted { .. }) + )); + assert!(matches!( + events.last(), + Some(ClientEvent::SessionClosed { .. }) + )); + } + #[test] fn finite_managed_run_drains_late_reply_after_timeout_before_close() { let duration = Duration::from_millis(1);