From 03ee6e2f283b3530b382fa56ed3559499d371cda Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 9 Aug 2026 19:10:56 +0700 Subject: [PATCH] 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")]