From 03ee6e2f283b3530b382fa56ed3559499d371cda Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 9 Aug 2026 19:10:56 +0700 Subject: [PATCH 1/3] feat(managed): add blocking managed owner --- crates/irtt-client/src/managed/blocking.rs | 102 +++++++++++++++++++++ crates/irtt-client/src/managed/mod.rs | 4 + 2 files changed, 106 insertions(+) create mode 100644 crates/irtt-client/src/managed/blocking.rs diff --git a/crates/irtt-client/src/managed/blocking.rs b/crates/irtt-client/src/managed/blocking.rs new file mode 100644 index 0000000..f901121 --- /dev/null +++ b/crates/irtt-client/src/managed/blocking.rs @@ -0,0 +1,102 @@ +use std::thread::{self, JoinHandle}; + +use thiserror::Error; +use tokio::runtime::Builder; + +use super::{ + ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedConfigError, ManagedOutcome, + ManagedTargetConfig, +}; + +/// Failure to start a blocking managed client owner. +#[derive(Debug, Error)] +pub enum BlockingManagedStartError { + /// The managed task configuration was invalid. + #[error(transparent)] + Config(#[from] ManagedConfigError), + + /// The dedicated Tokio runtime could not be created. + #[error("failed to create blocking managed Tokio runtime: {source}")] + Runtime { + #[source] + source: std::io::Error, + }, + + /// The dedicated worker thread could not be spawned. + #[error("failed to spawn blocking managed worker thread: {source}")] + ThreadSpawn { + #[source] + source: std::io::Error, + }, +} + +/// Failure to obtain completion from a blocking managed client owner. +#[derive(Debug, Error, Eq, PartialEq)] +pub enum BlockingManagedJoinError { + /// The dedicated managed worker thread panicked. + #[error("blocking managed worker thread panicked")] + WorkerPanicked, +} + +/// Blocking completion owner for one unified managed Tokio task. +/// +/// This owner runs exactly one [`ManagedClient`] task on one dedicated OS thread +/// with its own current-thread Tokio runtime. Dropping a live owner requests a +/// graceful stop and synchronously joins that worker, so drop may block while +/// bounded managed cleanup completes. +#[must_use = "dropping the blocking managed owner requests stop and joins its worker"] +pub struct BlockingManagedClient { + handle: ManagedClientHandle, + worker: Option>, +} + +impl BlockingManagedClient { + /// Validate configuration, create a dedicated runtime, and start its worker. + pub fn start( + config: ManagedClientConfig, + targets: Vec, + ) -> Result { + let (task, handle) = ManagedClient::task(config, targets)?; + let runtime = Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .map_err(|source| BlockingManagedStartError::Runtime { source })?; + let worker = thread::Builder::new() + .name("irtt-managed".to_owned()) + .spawn(move || runtime.block_on(task)) + .map_err(|source| BlockingManagedStartError::ThreadSpawn { source })?; + + Ok(Self { + handle, + worker: Some(worker), + }) + } + + /// Return a cloned managed control and observation capability. + pub fn handle(&self) -> ManagedClientHandle { + self.handle.clone() + } + + /// Wait synchronously for the managed task's authoritative outcome. + pub fn join(mut self) -> Result { + self.join_worker() + } + + fn join_worker(&mut self) -> Result { + self.worker + .take() + .expect("blocking managed worker is joined at most once") + .join() + .map_err(|_| BlockingManagedJoinError::WorkerPanicked) + } +} + +impl Drop for BlockingManagedClient { + fn drop(&mut self) { + if self.worker.is_some() { + let _receipt = self.handle.stop(); + let _ = self.join_worker(); + } + } +} diff --git a/crates/irtt-client/src/managed/mod.rs b/crates/irtt-client/src/managed/mod.rs index 69869f1..5e9ac35 100644 --- a/crates/irtt-client/src/managed/mod.rs +++ b/crates/irtt-client/src/managed/mod.rs @@ -24,11 +24,15 @@ pub mod legacy { pub use super::TargetId; } +#[cfg(feature = "tokio")] +mod blocking; #[cfg(feature = "tokio")] mod model; #[cfg(feature = "tokio")] mod task; +#[cfg(feature = "tokio")] +pub use blocking::*; #[cfg(feature = "tokio")] pub use model::*; #[cfg(feature = "tokio")] From 749306394c46e69dfe15b77e0d0ba5462eb4b954 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 9 Aug 2026 19:13:47 +0700 Subject: [PATCH 2/3] 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; From 93fed6dce6f66e152ea442d2774023dc73c53a14 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 9 Aug 2026 20:02:35 +0700 Subject: [PATCH 3/3] fix(managed): bound blocking runtime shutdown --- crates/irtt-client/src/managed/blocking.rs | 52 ++++++++++- .../irtt-client/src/managed/blocking_tests.rs | 91 ++++++++++++------- 2 files changed, 107 insertions(+), 36 deletions(-) diff --git a/crates/irtt-client/src/managed/blocking.rs b/crates/irtt-client/src/managed/blocking.rs index 5c6bc31..15fbeed 100644 --- a/crates/irtt-client/src/managed/blocking.rs +++ b/crates/irtt-client/src/managed/blocking.rs @@ -1,13 +1,54 @@ -use std::thread::{self, JoinHandle}; +use std::{ + thread::{self, JoinHandle}, + time::Duration, +}; use thiserror::Error; -use tokio::runtime::Builder; +use tokio::runtime::{Builder, Runtime}; use super::{ ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedConfigError, ManagedOutcome, ManagedTargetConfig, }; +const BLOCKING_RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100); + +/// Owns the runtime used exclusively by the blocking managed worker. +pub(super) struct WorkerRuntime { + runtime: Option, +} + +impl WorkerRuntime { + pub(super) fn new(runtime: Runtime) -> Self { + Self { + runtime: Some(runtime), + } + } + + fn block_on(&self, task: F) -> F::Output { + self.runtime + .as_ref() + .expect("worker runtime is available until shutdown") + .block_on(task) + } + + /// Shut down residual runtime work without delaying managed completion indefinitely. + pub(super) fn shutdown(mut self) { + self.runtime + .take() + .expect("worker runtime is available until shutdown") + .shutdown_timeout(BLOCKING_RUNTIME_SHUTDOWN_TIMEOUT); + } +} + +impl Drop for WorkerRuntime { + fn drop(&mut self) { + if let Some(runtime) = self.runtime.take() { + runtime.shutdown_background(); + } + } +} + /// Failure to start a blocking managed client owner. #[derive(Debug, Error)] pub enum BlockingManagedStartError { @@ -62,9 +103,14 @@ impl BlockingManagedClient { .enable_time() .build() .map_err(|source| BlockingManagedStartError::Runtime { source })?; + let runtime = WorkerRuntime::new(runtime); let worker = thread::Builder::new() .name("irtt-managed".to_owned()) - .spawn(move || runtime.block_on(task)) + .spawn(move || { + let outcome = runtime.block_on(task); + runtime.shutdown(); + outcome + }) .map_err(|source| BlockingManagedStartError::ThreadSpawn { source })?; Ok(Self { diff --git a/crates/irtt-client/src/managed/blocking_tests.rs b/crates/irtt-client/src/managed/blocking_tests.rs index f6af03b..96cf767 100644 --- a/crates/irtt-client/src/managed/blocking_tests.rs +++ b/crates/irtt-client/src/managed/blocking_tests.rs @@ -1,8 +1,8 @@ use std::{ net::{SocketAddr, UdpSocket}, - sync::{Arc, Condvar, Mutex}, + sync::{mpsc::sync_channel, Arc, Condvar, Mutex}, thread::{self, JoinHandle}, - time::Duration, + time::{Duration, Instant}, }; use irtt_proto::{ @@ -12,11 +12,11 @@ use irtt_proto::{ use tokio::{runtime::Builder, time::timeout}; use super::{ - blocking::join_worker, BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, - ManagedCompletionPolicy, ManagedEndReason, ManagedLifecycle, ManagedPacing, - ManagedTargetConfig, ManagedTargetEndReason, + blocking::{join_worker, WorkerRuntime}, + BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, ManagedCompletionPolicy, + ManagedEndReason, ManagedLifecycle, ManagedPacing, ManagedTargetConfig, ManagedTargetEndReason, }; -use crate::{ClientConfig, ClientEvent}; +use crate::ClientConfig; const TOKEN: u64 = 0x1234_5678_90ab_cdef; @@ -284,6 +284,44 @@ fn worker_panic_is_a_join_error() { ); } +#[test] +fn dropping_worker_runtime_inside_async_context_does_not_panic() { + test_runtime().block_on(async { + let runtime = Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .unwrap(); + drop(WorkerRuntime::new(runtime)); + }); +} + +#[test] +fn worker_runtime_shutdown_is_bounded_with_blocking_work() { + let runtime = Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .unwrap(); + let (started_send, started_receive) = sync_channel(0); + let (release_send, release_receive) = sync_channel(0); + + runtime.block_on(async move { + tokio::task::spawn_blocking(move || { + started_send.send(()).unwrap(); + release_receive.recv().unwrap(); + }); + }); + started_receive + .recv_timeout(Duration::from_secs(2)) + .unwrap(); + + let started = Instant::now(); + WorkerRuntime::new(runtime).shutdown(); + assert!(started.elapsed() < Duration::from_secs(2)); + release_send.send(()).unwrap(); +} + #[test] fn drop_after_natural_completion_preserves_outcome() { let server = EchoServer::start(); @@ -293,34 +331,21 @@ fn drop_after_natural_completion_preserves_outcome() { ) .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_probe(); server.wait_close(); + let deadline = Instant::now() + Duration::from_secs(2); + let status = loop { + let status = handle.status(); + if status.lifecycle == ManagedLifecycle::Completed && status.final_outcome.is_some() { + break status; + } + assert!(Instant::now() < deadline, "managed worker did not complete"); + thread::sleep(Duration::from_millis(5)); + }; + assert_eq!( + status.final_outcome.as_ref().unwrap().end_reason, + ManagedEndReason::TargetsComplete + ); drop(owner); assert_eq!( handle.status().final_outcome.as_ref().unwrap().end_reason,