diff --git a/README.md b/README.md index 263425e..ed70cd9 100644 --- a/README.md +++ b/README.md @@ -173,73 +173,51 @@ Add it from a local checkout: ```toml [dependencies] -irtt-client = { path = "crates/irtt-client" } +irtt-client = { path = "crates/irtt-client", features = ["tokio"] } ``` -Minimal managed-client example: +Minimal managed-client example for a synchronous frontend: ```rust use std::time::Duration; -use irtt_client::{ - ClientConfig, ClientEvent, ManagedClient, SubscriberConfig, +use irtt_client::managed::{ + BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, + ManagedTargetConfig, TargetId, }; +use irtt_client::ClientConfig; fn main() -> Result<(), Box> { - let config = ClientConfig { - server_addr: "netperf-eu.bufferbloat.net:2112".to_owned(), - duration: Some(Duration::from_secs(10)), - interval: Duration::from_secs(1), - ..ClientConfig::default() + let config = ManagedClientConfig { + client: ClientConfig { + duration: Some(Duration::from_secs(10)), + interval: Duration::from_secs(1), + ..ClientConfig::default() + }, + completion: ManagedCompletionPolicy::FinishWhenQuiescent, + ..ManagedClientConfig::default() }; - - let (session, events) = - ManagedClient::start_with_subscription( - config, - SubscriberConfig::default(), - )?; - - while let Ok(event) = events.recv() { - match event { - ClientEvent::EchoReply { seq, rtt, .. } => { - println!( - "seq={seq} rtt_us={}", - rtt.effective.as_micros() - ); - } - ClientEvent::SessionClosed { .. } => break, - _ => {} - } - } - - let outcome = session.join()?; + let targets = vec![ManagedTargetConfig::new( + TargetId::from("edge"), + "netperf-eu.bufferbloat.net:2112", + )]; + let owner = BlockingManagedClient::start(config, targets)?; + let outcome = owner.join()?; println!("session ended: {:?}", outcome.end_reason); Ok(()) } ``` -The client library emits structured events for session lifecycle, sent probes, successful replies, loss, duplicates, late replies, warnings, and shutdown. +For presentation events, use `BlockingManagedClient::start_with_subscription`, +`ManagedEvent`, and `ManagedEventTryRecvError`, while consulting the durable +`handle.status()` and final outcome for authoritative state. Presentation +events are bounded and lossy, not reliable history. Dynamic desired-target +updates are available through `ManagedClientHandle::update_targets`. -For shared-socket multi-target integrations, `ManagedClientGroup` publishes -`ManagedGroupEvent::TargetFinished` exactly once for every terminal target -incarnation, including open/runtime failures, removal, cancellation, no-test -completion, and peer closure. Consumers can use `ManagedTargetEndReason` and -`ManagedTargetFailureKind` for outcome and exit decisions without parsing -diagnostic text. - -`ManagedGroupCompletionPolicy::AllTargetsComplete` supports finite static -groups. `ExplicitCancellation` supports long-lived controllers; replacing the -desired set with an empty vector removes current targets but leaves the group -idle and ready for a later `update_targets` call. `join()` reports aggregate -outcome counts and retains only the 256 most recent target outcomes, so -long-running target churn remains bounded. The aggregate -`peer_closed_target_outcomes` count includes peer closures omitted from that -recent snapshot. - -Managed event subscriptions remain bounded and nonblocking. When using a -dropping overflow policy, check `EventSubscription::dropped_events()` before -trusting a complete statistical summary. +`BlockingManagedClient` owns a dedicated Tokio runtime and thread for +synchronous callers. Callers that already own Tokio can drive +`ManagedClientTask` directly. ## Binaries and features diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index 64319a1..6f144be 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -26,8 +26,8 @@ use crate::shared::client::expected_probe_count; use crate::shared::client::{ is_shutdown_requested, session::{ - peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once, - should_print_final_summary, + drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close, + request_managed_stop_once, should_print_final_summary, ManagedDrainState, }, }; #[cfg(feature = "stats")] @@ -116,7 +116,6 @@ pub fn run_stream( let mut dropped_events = 0_u64; let mut interrupted = false; let mut stop_requested = false; - let mut peer_close_requested_stop = false; let mut subscription_closed = false; loop { if is_shutdown_requested(shutdown_requested) { @@ -129,12 +128,11 @@ pub fn run_stream( continuous, interrupted, handle.status().peer_closed_target_outcomes, - &mut peer_close_requested_stop, &mut stop_requested, ) { drop(handle.stop()); } - drain_events( + let drain_state = drain_events( &mut events, &mut stream_output, #[cfg(feature = "stats")] @@ -142,35 +140,20 @@ pub fn run_stream( &mut terminal_targets, &mut dropped_events, )?; + match drain_state { + ManagedDrainState::Empty => thread::sleep(MANAGED_EVENT_WAIT_SLICE), + ManagedDrainState::Closed => subscription_closed = true, + ManagedDrainState::BudgetExhausted => {} + } if handle.status().final_outcome.is_some() || subscription_closed { break; } - match events.try_recv() { - Ok(event) => process_event( - event, - &mut stream_output, - #[cfg(feature = "stats")] - &mut stats, - &mut terminal_targets, - )?, - Err(ManagedEventTryRecvError::Empty) => thread::sleep(MANAGED_EVENT_WAIT_SLICE), - Err(ManagedEventTryRecvError::Lagged(count)) => dropped_events += count, - Err(ManagedEventTryRecvError::Closed) => subscription_closed = true, - } } if interrupted { eprintln!("interrupted, closing managed run..."); } - drain_events( - &mut events, - &mut stream_output, - #[cfg(feature = "stats")] - &mut stats, - &mut terminal_targets, - &mut dropped_events, - )?; let outcome = owner.join()?; - drain_events( + drain_final_events( &mut events, &mut stream_output, #[cfg(feature = "stats")] @@ -275,6 +258,24 @@ fn drain_events( #[cfg(feature = "stats")] stats: &mut BTreeMap, terminal_targets: &mut HashSet, dropped_events: &mut u64, +) -> io::Result { + drain_managed_events(events, dropped_events, |event| { + process_event( + event, + stream_output, + #[cfg(feature = "stats")] + stats, + terminal_targets, + ) + }) +} + +fn drain_final_events( + events: &mut ManagedEventSubscription, + stream_output: &mut StreamOutput<'_, W>, + #[cfg(feature = "stats")] stats: &mut BTreeMap, + terminal_targets: &mut HashSet, + dropped_events: &mut u64, ) -> io::Result<()> { loop { match events.try_recv() { @@ -286,19 +287,35 @@ fn drain_events( terminal_targets, )?, Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break, - Err(ManagedEventTryRecvError::Lagged(count)) => *dropped_events += count, + Err(ManagedEventTryRecvError::Lagged(count)) => { + *dropped_events = dropped_events.saturating_add(count); + } } } Ok(()) } fn report_target_failure(target: &ManagedTargetOutcome) { + for message in target_failure_messages(target) { + eprintln!("{message}"); + } +} + +fn target_failure_messages(target: &ManagedTargetOutcome) -> Vec { + let mut messages = Vec::with_capacity(2); if let ManagedTargetEndReason::Failed(failure) = &target.end_reason { - eprintln!( + messages.push(format!( "irtt-rs: target {} failed ({:?} {:?}): {}", target.target.id, failure.phase, failure.kind, failure.message - ); + )); } + if let Some(failure) = &target.cleanup_failure { + messages.push(format!( + "irtt-rs: target {} cleanup failed ({:?} {:?}): {}", + target.target.id, failure.phase, failure.kind, failure.message + )); + } + messages } fn dropped_event_warning(dropped_events: u64) -> Option { (dropped_events > 0).then(|| format!("irtt-rs: warning: dropped {dropped_events} managed run event{}; output and statistics may be incomplete", if dropped_events == 1 { "" } else { "s" })) @@ -419,14 +436,55 @@ fn finite_stats_memory_warning(args: &ClientArgs) -> Option { #[cfg(test)] mod tests { use super::*; + use irtt_client::managed::{ + ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetFailurePhase, TargetId, + }; + use std::sync::Arc; + #[test] - fn lagged_events_are_counted_exactly() { - let mut dropped = 0; - dropped += 7; - dropped += 3; + fn lagged_events_are_counted_saturatingly() { + let mut dropped = 0_u64; + dropped = dropped.saturating_add(7); + dropped = dropped.saturating_add(3); assert_eq!(dropped, 10); + assert_eq!(u64::MAX.saturating_add(1), u64::MAX); assert!(dropped_event_warning(dropped) .unwrap() .contains("dropped 10 managed run events")); } + + #[test] + fn target_failure_diagnostics_include_primary_and_cleanup_failures() { + let target = ManagedTargetOutcome { + target: TargetInstance { + id: TargetId::from("edge"), + generation: 1, + }, + server_addr: Arc::from("127.0.0.1:2112"), + remote: None, + end_reason: ManagedTargetEndReason::Failed(ManagedTargetFailure { + phase: ManagedTargetFailurePhase::Receiving, + kind: ManagedTargetFailureKind::Protocol, + message: "primary failure".into(), + }), + packets_sent: 0, + replies_received: 0, + duplicates: 0, + late: 0, + warning_events: 0, + cleanup_failure: Some(ManagedTargetFailure { + phase: ManagedTargetFailurePhase::Closing, + kind: ManagedTargetFailureKind::Socket, + message: "cleanup failure".into(), + }), + }; + + assert_eq!( + target_failure_messages(&target), + [ + "irtt-rs: target edge failed (Receiving Protocol): primary failure", + "irtt-rs: target edge cleanup failed (Closing Socket): cleanup failure", + ] + ); + } } diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index 5b08d88..3b8cf35 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -12,7 +12,8 @@ use crate::{ shared::client::{ is_shutdown_requested, session::{ - peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once, + drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close, + request_managed_stop_once, ManagedDrainState, }, }, }; @@ -72,7 +73,6 @@ pub fn run_tui( let handle = owner.handle(); let mut interrupted = false; let mut stop_requested = false; - let mut peer_close_requested_stop = false; let mut terminal_targets = HashSet::new(); let mut dropped_events = 0; let mut subscription_closed = false; @@ -96,28 +96,26 @@ pub fn run_tui( continuous, interrupted, handle.status().peer_closed_target_outcomes, - &mut peer_close_requested_stop, &mut stop_requested, ) { drop(handle.stop()); } - drain_tui_events( + let drain_state = drain_tui_events( &mut events, &mut state, &mut terminal_targets, &mut dropped_events, ); + match drain_state { + ManagedDrainState::Empty => { + thread::sleep(managed_tui_wait_duration(&next_render, state.paused)); + } + ManagedDrainState::Closed => subscription_closed = true, + ManagedDrainState::BudgetExhausted => {} + } if handle.status().final_outcome.is_some() || subscription_closed { break; } - match events.try_recv() { - Ok(event) => process_tui_event(event, &mut state, &mut terminal_targets), - Err(ManagedEventTryRecvError::Empty) => { - thread::sleep(managed_tui_wait_duration(&next_render, state.paused)) - } - Err(ManagedEventTryRecvError::Lagged(count)) => dropped_events += count, - Err(ManagedEventTryRecvError::Closed) => subscription_closed = true, - } render_if_due(&mut terminal, &state, &mut next_render, false)?; } if interrupted { @@ -126,14 +124,8 @@ pub fn run_tui( } state.set_status(TuiStatus::Closing); render_if_due(&mut terminal, &state, &mut next_render, true)?; - drain_tui_events( - &mut events, - &mut state, - &mut terminal_targets, - &mut dropped_events, - ); let outcome = owner.join()?; - drain_tui_events( + drain_final_tui_events( &mut events, &mut state, &mut terminal_targets, @@ -198,12 +190,27 @@ fn drain_tui_events( state: &mut TuiState, terminal_targets: &mut HashSet, dropped_events: &mut u64, +) -> ManagedDrainState { + drain_managed_events(events, dropped_events, |event| { + process_tui_event(event, state, terminal_targets); + Ok::<(), std::convert::Infallible>(()) + }) + .expect("processing TUI managed events is infallible") +} + +fn drain_final_tui_events( + events: &mut ManagedEventSubscription, + state: &mut TuiState, + terminal_targets: &mut HashSet, + dropped_events: &mut u64, ) { loop { match events.try_recv() { Ok(event) => process_tui_event(event, state, terminal_targets), Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break, - Err(ManagedEventTryRecvError::Lagged(count)) => *dropped_events += count, + Err(ManagedEventTryRecvError::Lagged(count)) => { + *dropped_events = dropped_events.saturating_add(count); + } } } } diff --git a/crates/irtt-cli/src/cmd/tui/ui.rs b/crates/irtt-cli/src/cmd/tui/ui.rs index dc1c64f..5f325e4 100644 --- a/crates/irtt-cli/src/cmd/tui/ui.rs +++ b/crates/irtt-cli/src/cmd/tui/ui.rs @@ -170,7 +170,7 @@ impl TuiState { idx }; - let (status, recent, warning) = match &outcome.end_reason { + let (status, recent, primary_warning) = match &outcome.end_reason { ManagedTargetEndReason::TestComplete => { (TargetStatus::Closed, "test completed".to_owned(), None) } @@ -203,6 +203,19 @@ impl TuiState { } }; + let cleanup_warning = outcome.cleanup_failure.as_ref().map(|failure| { + format!( + "cleanup failed ({:?} {:?}): {}", + failure.phase, failure.kind, failure.message + ) + }); + let warning = match (primary_warning, cleanup_warning.as_ref()) { + (Some(primary), Some(cleanup)) => Some(format!("{primary}; {cleanup}")), + (Some(primary), None) => Some(primary), + (None, Some(cleanup)) => Some(cleanup.clone()), + (None, None) => None, + }; + if let Some(target) = self.targets.get_mut(idx) { target.remote = outcome.remote.map(|remote| remote.to_string()); target.status = status; @@ -214,6 +227,9 @@ impl TuiState { self.last_warning = Some(warning); } self.push_event(format!("{label}: {recent}")); + if let Some(cleanup_warning) = cleanup_warning { + self.push_event(format!("{label}: {cleanup_warning}")); + } if self .targets @@ -2019,15 +2035,18 @@ fn duration_ns(value: Duration) -> i128 { i128::try_from(value.as_nanos()).unwrap_or(i128::MAX) } -#[cfg(any())] +#[cfg(test)] mod tests { use super::*; + use irtt_client::managed::{ + ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetFailurePhase, TargetId, + }; use irtt_client::{ - ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, ServerTiming, TargetId, - WarningKind, + ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, ServerTiming, WarningKind, }; use std::{ net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::Arc, time::UNIX_EPOCH, }; @@ -2131,23 +2150,25 @@ mod tests { &state.targets[0] } - fn target_event(label: &str, event: ClientEvent) -> UiEventScope { - UiEventScope { - target: TargetId::from(label), - event, + fn target_instance(label: &str) -> TargetInstance { + TargetInstance { + id: TargetId::from(label), + generation: 1, } } fn target_outcome(label: &str, end_reason: ManagedTargetEndReason) -> ManagedTargetOutcome { ManagedTargetOutcome { - id: TargetId::from(label), - remote: remote(), + target: target_instance(label), + server_addr: Arc::from("127.0.0.1:2112"), + remote: Some(remote()), end_reason, packets_sent: 0, replies_received: 0, duplicates: 0, late: 0, warning_events: 0, + cleanup_failure: None, } } @@ -2165,7 +2186,8 @@ mod tests { #[test] fn session_started_sets_remote_session_and_running_status() { let mut state = TuiState::default(); - state.process_event(&ClientEvent::SessionStarted { + let target = target_instance("target"); + let event = ClientEvent::SessionStarted { remote: remote(), token: 0xabc, negotiated: NegotiatedParams { @@ -2173,7 +2195,8 @@ mod tests { restrictions: Vec::new(), }, at: ts(Duration::ZERO), - }); + }; + state.process_target_event(&target, &event); let target = primary_target(&state); assert_eq!(target.remote.as_deref(), Some("127.0.0.1:2112")); @@ -2185,8 +2208,10 @@ mod tests { #[test] fn echo_reply_appends_bounded_primary_history() { let mut state = TuiState::default(); + let target = target_instance("target"); for seq in 0..(HISTORY_LIMIT as u32 + 3) { - state.process_event(&reply(seq, i128::from(seq) * 1_000)); + let event = reply(seq, i128::from(seq) * 1_000); + state.process_target_event(&target, &event); } let target = primary_target(&state); @@ -2205,13 +2230,15 @@ mod tests { #[test] fn duplicate_and_late_replies_do_not_append_primary_history() { let mut state = TuiState::default(); - state.process_event(&ClientEvent::DuplicateReply { + let target = target_instance("target"); + let duplicate = ClientEvent::DuplicateReply { seq: 7, remote: remote(), received_at: ts(Duration::from_secs(1)), bytes: 64, - }); - state.process_event(&ClientEvent::LateReply { + }; + state.process_target_event(&target, &duplicate); + let late = ClientEvent::LateReply { seq: 8, highest_seen: 9, remote: remote(), @@ -2223,7 +2250,8 @@ mod tests { received_stats: None, bytes: 64, packet_meta: PacketMeta::default(), - }); + }; + state.process_target_event(&target, &late); let target = primary_target(&state); assert!(target.graph_history.is_empty()); @@ -2250,7 +2278,9 @@ mod tests { let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]); - state.process_target_event(&target_event("b", reply(11, 2_500_000))); + let target = target_instance("b"); + let event = reply(11, 2_500_000); + state.process_target_event(&target, &event); assert!(state.targets[0].graph_history.is_empty()); assert_eq!(state.targets[0].stats.snapshot().packets.unique_replies, 0); @@ -2268,31 +2298,29 @@ mod tests { let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]); - state.process_target_event(&target_event( - "a", - ClientEvent::DuplicateReply { - seq: 7, - remote: remote(), - received_at: ts(Duration::from_secs(1)), - bytes: 64, - }, - )); - state.process_target_event(&target_event( - "b", - ClientEvent::LateReply { - seq: 8, - highest_seen: 9, - remote: remote(), - sent_at: Some(ts(Duration::from_secs(1))), - received_at: ts(Duration::from_secs(2)), - rtt: Some(rtt(2_000_000)), - server_timing: None, - one_way: None, - received_stats: None, - bytes: 64, - packet_meta: PacketMeta::default(), - }, - )); + let a = target_instance("a"); + let duplicate = ClientEvent::DuplicateReply { + seq: 7, + remote: remote(), + received_at: ts(Duration::from_secs(1)), + bytes: 64, + }; + state.process_target_event(&a, &duplicate); + let b = target_instance("b"); + let late = ClientEvent::LateReply { + seq: 8, + highest_seen: 9, + remote: remote(), + sent_at: Some(ts(Duration::from_secs(1))), + received_at: ts(Duration::from_secs(2)), + rtt: Some(rtt(2_000_000)), + server_timing: None, + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: PacketMeta::default(), + }; + state.process_target_event(&b, &late); assert!(state.targets[0].graph_history.is_empty()); assert!(state.targets[1].graph_history.is_empty()); @@ -2306,33 +2334,29 @@ mod tests { let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]); - state.process_target_event(&target_event( - "a", - ClientEvent::EchoLoss { - seq: 3, - sent_at: ts(Duration::from_millis(3)), - timeout_at: Instant::now(), + let a = target_instance("a"); + let loss = ClientEvent::EchoLoss { + seq: 3, + sent_at: ts(Duration::from_millis(3)), + timeout_at: Instant::now(), + }; + state.process_target_event(&a, &loss); + let b = target_instance("b"); + let warning = ClientEvent::Warning { + kind: WarningKind::WrongToken, + message: "wrong token".to_owned(), + at: ts(Duration::ZERO), + }; + state.process_target_event(&b, &warning); + let no_test = ClientEvent::NoTestCompleted { + remote: remote(), + negotiated: NegotiatedParams { + params: irtt_proto::Params::default(), + restrictions: Vec::new(), }, - )); - state.process_target_event(&target_event( - "b", - ClientEvent::Warning { - kind: WarningKind::WrongToken, - message: "wrong token".to_owned(), - at: ts(Duration::ZERO), - }, - )); - state.process_target_event(&target_event( - "a", - ClientEvent::NoTestCompleted { - remote: remote(), - negotiated: NegotiatedParams { - params: irtt_proto::Params::default(), - restrictions: Vec::new(), - }, - at: ts(Duration::ZERO), - }, - )); + at: ts(Duration::ZERO), + }; + state.process_target_event(&a, &no_test); assert_eq!(state.targets[0].stats.snapshot().events.loss_events, 1); assert_eq!(state.targets[0].status, TargetStatus::NoTest); @@ -2349,30 +2373,28 @@ mod tests { let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]); - state.process_target_event(&target_event( - "a", - ClientEvent::NoTestCompleted { - remote: remote(), - negotiated: NegotiatedParams { - params: irtt_proto::Params::default(), - restrictions: Vec::new(), - }, - at: ts(Duration::ZERO), + let a = target_instance("a"); + let no_test = ClientEvent::NoTestCompleted { + remote: remote(), + negotiated: NegotiatedParams { + params: irtt_proto::Params::default(), + restrictions: Vec::new(), }, - )); + at: ts(Duration::ZERO), + }; + state.process_target_event(&a, &no_test); assert_eq!(state.targets[0].status, TargetStatus::NoTest); assert_eq!(state.targets[1].status, TargetStatus::Opening); assert_eq!(state.status, TuiStatus::Running); - state.process_target_event(&target_event( - "b", - ClientEvent::SessionClosed { - remote: remote(), - token: 0xabc, - at: ts(Duration::ZERO), - }, - )); + let b = target_instance("b"); + let closed = ClientEvent::SessionClosed { + remote: remote(), + token: 0xabc, + at: ts(Duration::ZERO), + }; + state.process_target_event(&b, &closed); assert_eq!(state.targets[1].status, TargetStatus::Closed); assert_eq!(state.status, TuiStatus::Complete); @@ -2382,14 +2404,15 @@ mod tests { fn terminal_failure_marks_the_correct_target_failed() { let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]); - let failure = irtt_client::ManagedTargetFailure { - kind: irtt_client::ManagedTargetFailureKind::OpeningTimeout, - message: "all open requests timed out".to_owned(), + let failure = ManagedTargetFailure { + phase: ManagedTargetFailurePhase::Opening, + kind: ManagedTargetFailureKind::Timeout, + message: "all open requests timed out".into(), }; state.process_target_outcome(&target_outcome( "b", - ManagedTargetEndReason::OpenFailed(failure), + ManagedTargetEndReason::Failed(failure), )); assert_eq!(state.targets[0].status, TargetStatus::Opening); @@ -2397,7 +2420,7 @@ mod tests { assert!(state.targets[1] .last_warning .as_deref() - .is_some_and(|warning| warning.contains("opening timeout"))); + .is_some_and(|warning| warning.contains("Timeout"))); } #[test] @@ -2407,9 +2430,10 @@ mod tests { state.process_target_outcome(&target_outcome("a", ManagedTargetEndReason::PeerClosed)); state.process_target_outcome(&target_outcome( "b", - ManagedTargetEndReason::Failed(irtt_client::ManagedTargetFailure { - kind: irtt_client::ManagedTargetFailureKind::RuntimeProtocol, - message: "pending probe limit exceeded".to_owned(), + ManagedTargetEndReason::Failed(ManagedTargetFailure { + phase: ManagedTargetFailurePhase::Receiving, + kind: ManagedTargetFailureKind::Protocol, + message: "pending probe limit exceeded".into(), }), )); @@ -2422,6 +2446,79 @@ mod tests { .is_some_and(|warning| warning.contains("pending probe limit"))); } + #[test] + fn cleanup_failure_warns_without_changing_successful_target_status() { + let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]); + let mut outcome = target_outcome("a", ManagedTargetEndReason::TestComplete); + outcome.remote = None; + outcome.cleanup_failure = Some(ManagedTargetFailure { + phase: ManagedTargetFailurePhase::Closing, + kind: ManagedTargetFailureKind::Socket, + message: "close datagram failed".into(), + }); + + state.process_target_outcome(&outcome); + + assert_eq!(state.targets[0].status, TargetStatus::Closed); + assert_eq!(state.targets[0].remote, None); + assert!(state.targets[0] + .last_warning + .as_deref() + .is_some_and(|warning| warning.contains("cleanup failed (Closing Socket)"))); + assert!(state + .recent_events + .iter() + .any(|event| event.contains("cleanup failed (Closing Socket)"))); + } + + #[test] + fn primary_and_cleanup_failures_are_both_visible() { + let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]); + let mut outcome = target_outcome( + "a", + ManagedTargetEndReason::Failed(ManagedTargetFailure { + phase: ManagedTargetFailurePhase::Receiving, + kind: ManagedTargetFailureKind::Protocol, + message: "primary failure".into(), + }), + ); + outcome.cleanup_failure = Some(ManagedTargetFailure { + phase: ManagedTargetFailurePhase::Closing, + kind: ManagedTargetFailureKind::Socket, + message: "cleanup failure".into(), + }); + + state.process_target_outcome(&outcome); + + assert_eq!(state.targets[0].status, TargetStatus::Failed); + assert!(state.targets[0] + .last_warning + .as_deref() + .is_some_and(|warning| warning.contains("primary failure") + && warning.contains("cleanup failure"))); + assert!(state + .recent_events + .iter() + .any(|event| event.contains("target failed: Protocol: primary failure"))); + assert!(state + .recent_events + .iter() + .any(|event| event.contains("cleanup failed (Closing Socket): cleanup failure"))); + } + + #[test] + fn managed_outcomes_map_stopped_removed_and_replaced_statuses() { + for (reason, expected) in [ + (ManagedTargetEndReason::Stopped, TargetStatus::Stopped), + (ManagedTargetEndReason::Removed, TargetStatus::Removed), + (ManagedTargetEndReason::Replaced, TargetStatus::Replaced), + ] { + let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]); + state.process_target_outcome(&target_outcome("a", reason)); + assert_eq!(state.targets[0].status, expected); + } + } + #[test] fn run_error_preserves_peer_closed_target_status() { let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]); @@ -2446,9 +2543,10 @@ mod tests { for label in ["a", "b"] { state.process_target_outcome(&target_outcome( label, - ManagedTargetEndReason::OpenFailed(irtt_client::ManagedTargetFailure { - kind: irtt_client::ManagedTargetFailureKind::OpeningProtocol, - message: "server returned a zero connection token".to_owned(), + ManagedTargetEndReason::Failed(ManagedTargetFailure { + phase: ManagedTargetFailurePhase::Opening, + kind: ManagedTargetFailureKind::Protocol, + message: "server returned a zero connection token".into(), }), )); } @@ -2465,8 +2563,12 @@ mod tests { let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]); - state.process_target_event(&target_event("a", reply(1, 1_000_000))); - state.process_target_event(&target_event("b", reply(2, 2_000_000))); + let a = target_instance("a"); + let a_reply = reply(1, 1_000_000); + state.process_target_event(&a, &a_reply); + let b = target_instance("b"); + let b_reply = reply(2, 2_000_000); + state.process_target_event(&b, &b_reply); state.clear_visible_history(); assert!(state.targets[0].graph_history.is_empty()); @@ -2480,11 +2582,13 @@ mod tests { #[test] fn warning_updates_recent_events_and_last_warning() { let mut state = TuiState::default(); - state.process_event(&ClientEvent::Warning { + let target = target_instance("target"); + let event = ClientEvent::Warning { kind: WarningKind::WrongToken, message: "wrong token".to_owned(), at: ts(Duration::ZERO), - }); + }; + state.process_target_event(&target, &event); assert_eq!( state.last_warning.as_deref(), @@ -2501,7 +2605,7 @@ mod tests { fn dropped_events_mark_tui_statistics_incomplete() { let mut state = TuiState::default(); - state.mark_dropped_group_events(9); + state.mark_dropped_managed_events(9); assert_eq!(state.dropped_events, 9); assert!(state @@ -2511,20 +2615,20 @@ mod tests { assert!(state .recent_events .back() - .is_some_and(|event| event.contains("dropped 9 managed group events"))); + .is_some_and(|event| event.contains("dropped 9 managed run events"))); } #[test] fn dropped_single_client_events_are_visible() { let mut state = TuiState::default(); - state.mark_dropped_client_events(3); + state.mark_dropped_managed_events(3); assert_eq!(state.dropped_events, 3); assert!(state .recent_events .back() - .is_some_and(|event| event.contains("dropped 3 managed client events"))); + .is_some_and(|event| event.contains("dropped 3 managed run events"))); } #[test] @@ -2803,7 +2907,9 @@ mod tests { #[test] fn sample_details_preserve_signed_values() { let mut state = TuiState::default(); - state.process_event(&reply(4, -1_250_000)); + let target = target_instance("target"); + let event = reply(4, -1_250_000); + state.process_target_event(&target, &event); let target = primary_target(&state); let sample = target.last_sample.unwrap(); diff --git a/crates/irtt-cli/src/shared/client/session.rs b/crates/irtt-cli/src/shared/client/session.rs index 67393fe..1c2c381 100644 --- a/crates/irtt-cli/src/shared/client/session.rs +++ b/crates/irtt-cli/src/shared/client/session.rs @@ -1,5 +1,40 @@ use std::sync::atomic::{AtomicBool, Ordering}; +#[cfg(feature = "client-runtime")] +use irtt_client::managed::{ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError}; + +#[cfg(feature = "client-runtime")] +pub(crate) const MANAGED_EVENT_WORK_BUDGET: usize = 128; + +#[cfg(feature = "client-runtime")] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ManagedDrainState { + Empty, + Closed, + BudgetExhausted, +} + +/// Drain a bounded amount of lossy presentation work before returning to the +/// frontend control loop. +#[cfg(feature = "client-runtime")] +pub(crate) fn drain_managed_events( + events: &mut ManagedEventSubscription, + dropped_events: &mut u64, + mut process: impl FnMut(ManagedEvent) -> Result<(), E>, +) -> Result { + for _ in 0..MANAGED_EVENT_WORK_BUDGET { + match events.try_recv() { + Ok(event) => process(event)?, + Err(ManagedEventTryRecvError::Lagged(count)) => { + *dropped_events = dropped_events.saturating_add(count); + } + Err(ManagedEventTryRecvError::Empty) => return Ok(ManagedDrainState::Empty), + Err(ManagedEventTryRecvError::Closed) => return Ok(ManagedDrainState::Closed), + } + } + Ok(ManagedDrainState::BudgetExhausted) +} + pub fn is_shutdown_requested(shutdown_requested: &AtomicBool) -> bool { shutdown_requested.load(Ordering::Relaxed) } @@ -31,13 +66,11 @@ pub fn request_managed_stop_for_peer_close( continuous: bool, interrupted: bool, peer_closed_target_outcomes: u64, - peer_close_requested_stop: &mut bool, stop_requested: &mut bool, ) -> bool { if !continuous || interrupted || peer_closed_target_outcomes == 0 { return false; } - *peer_close_requested_stop = true; request_managed_stop_once(stop_requested) } @@ -70,46 +103,38 @@ mod tests { #[test] fn grouped_peer_close_stop_policy_is_run_mode_and_one_shot() { - let mut peer_close_requested_stop = false; let mut stop_requested = false; assert!(!request_managed_stop_for_peer_close( false, false, 1, - &mut peer_close_requested_stop, &mut stop_requested, )); assert!(!request_managed_stop_for_peer_close( true, true, 1, - &mut peer_close_requested_stop, &mut stop_requested, )); assert!(!request_managed_stop_for_peer_close( true, false, 0, - &mut peer_close_requested_stop, &mut stop_requested, )); - assert!(!peer_close_requested_stop); assert!(!stop_requested); assert!(request_managed_stop_for_peer_close( true, false, 1, - &mut peer_close_requested_stop, &mut stop_requested, )); - assert!(peer_close_requested_stop); assert!(stop_requested); assert!(!request_managed_stop_for_peer_close( true, false, 1, - &mut peer_close_requested_stop, &mut stop_requested, )); } diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index a900688..7ed52c9 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -94,8 +94,8 @@ use crate::{ /// /// `Client` exposes the protocol steps directly: connect a UDP socket, open a /// session, send probes, receive replies, poll timeouts, and close. Callers -/// that do not need to own this loop can use [`ManagedClient`](crate::ManagedClient) -/// instead. +/// that do not need to own this loop can use the unified managed API in +/// [`crate::managed`] when the `tokio` feature is enabled. #[derive(Debug)] pub struct Client { runtime: SessionMachine, diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index 742352e..13e1c17 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -14,10 +14,9 @@ //! values usually indicate client/server clock skew. //! //! [`Client`] is the runtime-free low-level blocking adapter. With the -//! `tokio` feature, [`AsyncClient`] provides the corresponding Tokio adapter; -//! [`managed`] contains the unified managed task/handle implementation and -//! [`managed::BlockingManagedClient`] owns its dedicated runtime thread for -//! synchronous managed callers. +//! `tokio` feature, `AsyncClient` provides the corresponding Tokio adapter; +//! [`managed`] contains the unified managed task/handle implementation, +//! including the dedicated-runtime blocking frontend for synchronous callers. //! #![cfg_attr( not(all(target_os = "linux", feature = "ancillary")), diff --git a/crates/irtt-client/src/receive/linux.rs b/crates/irtt-client/src/receive/linux.rs index 502e784..dd3384b 100644 --- a/crates/irtt-client/src/receive/linux.rs +++ b/crates/irtt-client/src/receive/linux.rs @@ -105,62 +105,6 @@ fn recv_datagram_fd( }) } -#[cfg(any())] -pub(crate) fn recv_datagram_from( - socket: &UdpSocket, - buf: &mut [u8], -) -> Result { - let mut control = ControlBuffer::new(); - let mut name: libc::sockaddr_storage = unsafe { mem::zeroed() }; - let mut iov = libc::iovec { - iov_base: buf.as_mut_ptr().cast(), - iov_len: buf.len(), - }; - let mut msg: libc::msghdr = unsafe { mem::zeroed() }; - msg.msg_name = (&mut name as *mut libc::sockaddr_storage).cast(); - msg.msg_namelen = mem::size_of::() as libc::socklen_t; - msg.msg_iov = &mut iov; - msg.msg_iovlen = 1; - msg.msg_control = control.as_mut_ptr().cast(); - msg.msg_controllen = control.len(); - - let len = unsafe { - // SAFETY: `msg` points to one writable iovec backed by `buf`, a - // writable source-address buffer, and a writable control buffer. All - // buffers live until `recvmsg` returns. The socket file descriptor is - // borrowed from a valid `UdpSocket`. - libc::recvmsg(socket.as_raw_fd(), &mut msg, 0) - }; - if len < 0 { - return Err(io::Error::last_os_error()); - } - let len = usize::try_from(len).map_err(|_| { - io::Error::new( - io::ErrorKind::InvalidData, - "recvmsg returned an unrepresentable datagram length", - ) - })?; - let received_at = ClientTimestamp::now(); - let source = unsafe { - // SAFETY: `name` and `msg_namelen` were initialized by a successful - // `recvmsg` call above. - socket_addr_from_storage(&name, msg.msg_namelen)? - }; - let meta = unsafe { - // SAFETY: `msg` was initialized by a successful `recvmsg` call. The - // parser only reads cmsghdr entries within `msg_controllen` and ignores - // short or unrelated control messages. - parse_receive_meta(&msg) - }; - - Ok(ReceivedDatagramFrom { - len, - source, - received_at, - meta, - }) -} - unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta { let mut meta = ReceiveMeta::default(); let mut cmsg = libc::CMSG_FIRSTHDR(msg); @@ -182,53 +126,6 @@ unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta { meta } -#[cfg(any())] -unsafe fn socket_addr_from_storage( - storage: &libc::sockaddr_storage, - len: libc::socklen_t, -) -> io::Result { - match i32::from(storage.ss_family) { - libc::AF_INET => { - if usize::try_from(len).unwrap_or(0) < mem::size_of::() { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "short IPv4 source address from recvmsg", - )); - } - let addr = ptr::read_unaligned( - (storage as *const libc::sockaddr_storage).cast::(), - ); - let ip = Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr)); - let port = u16::from_be(addr.sin_port); - Ok(SocketAddr::V4(SocketAddrV4::new(ip, port))) - } - libc::AF_INET6 => { - if usize::try_from(len).unwrap_or(0) < mem::size_of::() { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "short IPv6 source address from recvmsg", - )); - } - let addr = ptr::read_unaligned( - (storage as *const libc::sockaddr_storage).cast::(), - ); - let ip = Ipv6Addr::from(addr.sin6_addr.s6_addr); - let port = u16::from_be(addr.sin6_port); - let flowinfo = u32::from_be(addr.sin6_flowinfo); - Ok(SocketAddr::V6(SocketAddrV6::new( - ip, - port, - flowinfo, - addr.sin6_scope_id, - ))) - } - family => Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("unsupported source address family from recvmsg: {family}"), - )), - } -} - fn is_traffic_class_cmsg(cmsg: &libc::cmsghdr) -> bool { (cmsg.cmsg_level == libc::IPPROTO_IP && cmsg.cmsg_type == libc::IP_TOS) || (cmsg.cmsg_level == libc::IPPROTO_IPV6 && cmsg.cmsg_type == libc::IPV6_TCLASS)