From 749306394c46e69dfe15b77e0d0ba5462eb4b954 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 9 Aug 2026 19:13:47 +0700 Subject: [PATCH] test(managed): verify blocking owner lifecycle --- crates/irtt-client/src/managed/blocking.rs | 18 +- .../irtt-client/src/managed/blocking_tests.rs | 330 ++++++++++++++++++ crates/irtt-client/src/managed/mod.rs | 3 + 3 files changed, 346 insertions(+), 5 deletions(-) create mode 100644 crates/irtt-client/src/managed/blocking_tests.rs diff --git a/crates/irtt-client/src/managed/blocking.rs b/crates/irtt-client/src/managed/blocking.rs index f901121..5c6bc31 100644 --- a/crates/irtt-client/src/managed/blocking.rs +++ b/crates/irtt-client/src/managed/blocking.rs @@ -84,14 +84,22 @@ impl BlockingManagedClient { } fn join_worker(&mut self) -> Result { - self.worker - .take() - .expect("blocking managed worker is joined at most once") - .join() - .map_err(|_| BlockingManagedJoinError::WorkerPanicked) + join_worker( + self.worker + .take() + .expect("blocking managed worker is joined at most once"), + ) } } +pub(super) fn join_worker( + worker: JoinHandle, +) -> Result { + worker + .join() + .map_err(|_| BlockingManagedJoinError::WorkerPanicked) +} + impl Drop for BlockingManagedClient { fn drop(&mut self) { if self.worker.is_some() { diff --git a/crates/irtt-client/src/managed/blocking_tests.rs b/crates/irtt-client/src/managed/blocking_tests.rs new file mode 100644 index 0000000..f6af03b --- /dev/null +++ b/crates/irtt-client/src/managed/blocking_tests.rs @@ -0,0 +1,330 @@ +use std::{ + net::{SocketAddr, UdpSocket}, + sync::{Arc, Condvar, Mutex}, + thread::{self, JoinHandle}, + time::Duration, +}; + +use irtt_proto::{ + decode_close_request, decode_echo_request, decode_open_request, encode_echo_reply, + encode_open_reply, flags, EchoReply, OpenReply, ReceivedStats, StampAt, TimestampFields, +}; +use tokio::{runtime::Builder, time::timeout}; + +use super::{ + blocking::join_worker, BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, + ManagedCompletionPolicy, ManagedEndReason, ManagedLifecycle, ManagedPacing, + ManagedTargetConfig, ManagedTargetEndReason, +}; +use crate::{ClientConfig, ClientEvent}; + +const TOKEN: u64 = 0x1234_5678_90ab_cdef; + +struct Signal { + value: Mutex, + ready: Condvar, +} + +impl Signal { + fn set(&self) { + *self.value.lock().unwrap() = true; + self.ready.notify_all(); + } + + fn wait(&self) { + let value = self.value.lock().unwrap(); + let (value, timeout) = self + .ready + .wait_timeout_while(value, Duration::from_secs(2), |value| !*value) + .unwrap(); + assert!(*value && !timeout.timed_out()); + } +} + +struct EchoServer { + addr: SocketAddr, + probe: Arc, + close: Arc, + worker: JoinHandle<()>, +} + +impl EchoServer { + fn start() -> Self { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let probe = Arc::new(Signal { + value: Mutex::new(false), + ready: Condvar::new(), + }); + let close = Arc::new(Signal { + value: Mutex::new(false), + ready: Condvar::new(), + }); + let worker_probe = Arc::clone(&probe); + let worker_close = Arc::clone(&close); + let worker = thread::spawn(move || { + let mut buffer = [0_u8; 8192]; + let (open_len, peer) = socket.recv_from(&mut buffer).unwrap(); + let request = decode_open_request(&buffer[..open_len], None).unwrap(); + let negotiated = request.params; + let reply = encode_open_reply( + &OpenReply { + flags: flags::FLAG_OPEN | flags::FLAG_REPLY, + token: TOKEN, + params: negotiated.clone(), + }, + None, + ) + .unwrap(); + socket.send_to(&reply, peer).unwrap(); + + loop { + let (len, packet_peer) = socket.recv_from(&mut buffer).unwrap(); + let packet = &buffer[..len]; + if let Ok(probe) = decode_echo_request(packet, &negotiated, None) { + worker_probe.set(); + let reply = encode_echo_reply( + &EchoReply { + flags: flags::FLAG_REPLY, + token: TOKEN, + sequence: probe.sequence, + recv_count: None, + recv_window: None, + timestamps: TimestampFields::default(), + payload: Vec::new(), + }, + &negotiated, + None, + ) + .unwrap(); + socket.send_to(&reply, packet_peer).unwrap(); + } else if decode_close_request(packet, None).is_ok() { + worker_close.set(); + return; + } + } + }); + Self { + addr, + probe, + close, + worker, + } + } + + fn wait_probe(&self) { + self.probe.wait(); + } + + fn wait_close(&self) { + self.close.wait(); + } + + fn finish(self) { + self.worker.join().unwrap(); + } +} + +fn test_runtime() -> tokio::runtime::Runtime { + Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .unwrap() +} + +fn config(completion: ManagedCompletionPolicy) -> ManagedClientConfig { + ManagedClientConfig { + client: ClientConfig { + duration: Some(Duration::from_millis(120)), + interval: Duration::from_millis(20), + probe_timeout: Duration::from_millis(40), + open_timeouts: vec![Duration::from_millis(200)], + received_stats: ReceivedStats::None, + stamp_at: StampAt::None, + ..ClientConfig::default() + }, + pacing: ManagedPacing::Burst, + completion, + final_drain: Duration::from_millis(5), + ..ManagedClientConfig::default() + } +} + +fn target(server: &EchoServer) -> ManagedTargetConfig { + ManagedTargetConfig::new("target", server.addr.to_string()) +} + +fn assert_stopped(outcome: &super::ManagedOutcome) { + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); +} + +#[test] +fn starts_without_caller_runtime_and_finishes_naturally() { + let server = EchoServer::start(); + let owner = BlockingManagedClient::start( + config(ManagedCompletionPolicy::FinishWhenQuiescent), + vec![target(&server)], + ) + .unwrap(); + + server.wait_probe(); + let outcome = owner.join().unwrap(); + assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete); + assert_eq!(outcome.recent_target_outcomes.len(), 1); + assert_eq!( + outcome.recent_target_outcomes[0].end_reason, + ManagedTargetEndReason::TestComplete + ); + server.wait_close(); + server.finish(); +} + +#[test] +fn external_handle_stop_ends_with_graceful_close() { + let server = EchoServer::start(); + let owner = BlockingManagedClient::start( + config(ManagedCompletionPolicy::ExplicitStop), + vec![target(&server)], + ) + .unwrap(); + let handle = owner.handle(); + + server.wait_probe(); + test_runtime().block_on(async { + timeout(Duration::from_secs(2), handle.stop()) + .await + .unwrap(); + }); + let outcome = owner.join().unwrap(); + assert_stopped(&outcome); + server.wait_close(); + server.finish(); +} + +#[test] +fn dynamic_update_reaches_background_worker() { + let server = EchoServer::start(); + let owner = BlockingManagedClient::start(config(ManagedCompletionPolicy::ExplicitStop), vec![]) + .unwrap(); + let handle = owner.handle(); + + let acknowledgement = test_runtime().block_on(async { + timeout( + Duration::from_secs(2), + handle.update_targets(vec![target(&server)]).unwrap(), + ) + .await + .unwrap() + .unwrap() + }); + assert_eq!(acknowledgement.sequence, 1); + server.wait_probe(); + test_runtime().block_on(async { + timeout(Duration::from_secs(2), owner.handle().stop()) + .await + .unwrap(); + }); + assert_stopped(&owner.join().unwrap()); + server.wait_close(); + server.finish(); +} + +#[test] +fn dropping_external_handles_does_not_stop_owner() { + let server = EchoServer::start(); + let owner = BlockingManagedClient::start( + config(ManagedCompletionPolicy::FinishWhenQuiescent), + vec![target(&server)], + ) + .unwrap(); + drop(owner.handle()); + drop(owner.handle()); + + server.wait_probe(); + assert_eq!( + owner.join().unwrap().end_reason, + ManagedEndReason::TargetsComplete + ); + server.wait_close(); + server.finish(); +} + +#[test] +fn dropping_owner_stops_and_joins_worker() { + let server = EchoServer::start(); + let owner = BlockingManagedClient::start( + config(ManagedCompletionPolicy::ExplicitStop), + vec![target(&server)], + ) + .unwrap(); + let handle = owner.handle(); + + server.wait_probe(); + drop(owner); + let status = handle.status(); + assert_eq!(status.lifecycle, ManagedLifecycle::Completed); + assert_eq!( + status.final_outcome.as_ref().unwrap().end_reason, + ManagedEndReason::StopRequested + ); + server.wait_close(); + server.finish(); +} + +#[test] +fn worker_panic_is_a_join_error() { + let worker = thread::spawn(|| -> super::ManagedOutcome { panic!("test worker panic") }); + assert_eq!( + join_worker(worker), + Err(BlockingManagedJoinError::WorkerPanicked) + ); +} + +#[test] +fn drop_after_natural_completion_preserves_outcome() { + let server = EchoServer::start(); + let owner = BlockingManagedClient::start( + config(ManagedCompletionPolicy::FinishWhenQuiescent), + vec![target(&server)], + ) + .unwrap(); + let handle = owner.handle(); + let mut events = handle.subscribe().unwrap(); + + test_runtime().block_on(async { + loop { + if let super::ManagedEvent::Client { + event: ClientEvent::EchoSent { .. }, + .. + } = timeout(Duration::from_secs(2), events.recv()) + .await + .unwrap() + .unwrap() + { + break; + } + } + loop { + if let super::ManagedEvent::Completed { outcome } = + timeout(Duration::from_secs(2), events.recv()) + .await + .unwrap() + .unwrap() + { + assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete); + break; + } + } + }); + server.wait_close(); + drop(owner); + assert_eq!( + handle.status().final_outcome.as_ref().unwrap().end_reason, + ManagedEndReason::TargetsComplete + ); + server.finish(); +} diff --git a/crates/irtt-client/src/managed/mod.rs b/crates/irtt-client/src/managed/mod.rs index 5e9ac35..9005366 100644 --- a/crates/irtt-client/src/managed/mod.rs +++ b/crates/irtt-client/src/managed/mod.rs @@ -37,3 +37,6 @@ pub use blocking::*; pub use model::*; #[cfg(feature = "tokio")] pub use task::*; + +#[cfg(all(feature = "tokio", test))] +mod blocking_tests;