chore(client): polish managed runner ergonomics

This commit is contained in:
Nils Andreas Svee committed 2026-05-10 20:56:43 +07:00
1 parent afb20b6625
commit be2f43b07a
3 files changed
+47 -2

No files matched your search

@@ -17,6 +17,7 @@ impl CancellationToken {
self.cancelled.store(true, Ordering::SeqCst);
}
#[must_use]
pub fn is_cancelled(&self) -> bool {
self.cancelled.load(Ordering::SeqCst)
}
@@ -32,12 +33,14 @@ mod tests {
}
#[test]
fn cancel_is_visible_through_clones() {
fn cancel_is_idempotent_and_visible_through_clones() {
let token = CancellationToken::new();
let clone = token.clone();
token.cancel();
token.cancel();
assert!(token.is_cancelled());
assert!(clone.is_cancelled());
}
}
+26
View File
@@ -23,10 +23,17 @@ impl Default for SubscriberConfig {
}
}
/// Overflow behavior for a bounded event subscription queue.
///
/// These policies are applied independently per subscriber when that
/// subscriber's queue is full.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SubscriberOverflow {
/// Leave the existing queue unchanged and discard the newly published event.
DropNewest,
/// Remove the oldest queued event and enqueue the newly published event.
DropOldest,
/// Disconnect the subscriber and clear its queue.
Disconnect,
}
@@ -41,6 +48,12 @@ struct HubInner {
subscribers: Mutex<HashMap<u64, Arc<SubscriberInner>>>,
}
/// Handle for receiving managed client events.
///
/// Each subscription has its own bounded queue configured by
/// [`SubscriberConfig`]. The configured [`SubscriberOverflow`] policy applies
/// when that queue is full. Dropping the handle unregisters the subscriber.
#[must_use = "dropping the subscription unregisters it"]
#[derive(Debug)]
pub struct EventSubscription {
id: u64,
@@ -115,6 +128,8 @@ impl EventHub {
let mut disconnected = Vec::new();
for (id, subscriber) in subscribers {
// Events are cloned per subscriber; bounded queues keep slow
// consumers from causing unbounded memory growth.
if !subscriber.publish(event.clone()) {
disconnected.push(id);
}
@@ -166,6 +181,11 @@ impl Default for EventHub {
}
impl EventSubscription {
/// Block until the next queued event is available.
///
/// If the subscription is disconnected, this returns
/// [`EventSubscriptionError::Disconnected`] after any already queued events
/// have been drained.
pub fn recv(&self) -> Result<ClientEvent, EventSubscriptionError> {
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
loop {
@@ -183,6 +203,12 @@ impl EventSubscription {
}
}
/// Try to receive one queued event without blocking.
///
/// Returns `Ok(None)` when the subscription is still connected but no event
/// is queued. If the subscription is disconnected, this returns
/// [`EventSubscriptionError::Disconnected`] after any already queued events
/// have been drained.
pub fn try_recv(&self) -> Result<Option<ClientEvent>, EventSubscriptionError> {
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
if let Some(event) = state.queue.pop_front() {
+17 -1
View File
@@ -25,6 +25,13 @@ const MAX_SLEEP: Duration = Duration::from_millis(20);
#[derive(Debug)]
pub struct ManagedClient;
/// Running managed client session.
///
/// [`ManagedClientSession::join`] waits for the worker and returns the session
/// outcome or client error. Dropping the session requests cooperative
/// cancellation; callers that need the outcome should explicitly join instead
/// of relying on drop.
#[must_use = "dropping the session cancels the managed client; call join() to wait for completion"]
#[derive(Debug)]
pub struct ManagedClientSession {
hub: EventHub,
@@ -32,6 +39,12 @@ pub struct ManagedClientSession {
worker: Option<JoinHandle<Result<SessionOutcome, ClientError>>>,
}
/// Outcome returned by a completed managed client session.
///
/// These are runner-level lifecycle counters, not statistical summaries. Use
/// `irtt-stats` with emitted `ClientEvent` values for RTT, loss, IPDV, and
/// related summaries.
#[must_use = "managed session outcomes contain completion status and counters"]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionOutcome {
pub end_reason: SessionEndReason,
@@ -111,7 +124,10 @@ impl ManagedClientSession {
}
pub fn join(mut self) -> Result<SessionOutcome, ClientError> {
let worker = self.worker.take().expect("worker handle must be present");
let worker = self
.worker
.take()
.expect("ManagedClientSession invariant violated: worker handle missing before join");
match worker.join() {
Ok(outcome) => {
self.hub.disconnect_all();