chore(client): polish managed runner ergonomics
This commit is contained in:
3 files changed
+47
-2
No files matched your search
@@ -17,6 +17,7 @@ impl CancellationToken {
|
|||||||
self.cancelled.store(true, Ordering::SeqCst);
|
self.cancelled.store(true, Ordering::SeqCst);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[must_use]
|
||||||
pub fn is_cancelled(&self) -> bool {
|
pub fn is_cancelled(&self) -> bool {
|
||||||
self.cancelled.load(Ordering::SeqCst)
|
self.cancelled.load(Ordering::SeqCst)
|
||||||
}
|
}
|
||||||
@@ -32,12 +33,14 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn cancel_is_visible_through_clones() {
|
fn cancel_is_idempotent_and_visible_through_clones() {
|
||||||
let token = CancellationToken::new();
|
let token = CancellationToken::new();
|
||||||
let clone = token.clone();
|
let clone = token.clone();
|
||||||
|
|
||||||
|
token.cancel();
|
||||||
token.cancel();
|
token.cancel();
|
||||||
|
|
||||||
|
assert!(token.is_cancelled());
|
||||||
assert!(clone.is_cancelled());
|
assert!(clone.is_cancelled());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -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)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
pub enum SubscriberOverflow {
|
pub enum SubscriberOverflow {
|
||||||
|
/// Leave the existing queue unchanged and discard the newly published event.
|
||||||
DropNewest,
|
DropNewest,
|
||||||
|
/// Remove the oldest queued event and enqueue the newly published event.
|
||||||
DropOldest,
|
DropOldest,
|
||||||
|
/// Disconnect the subscriber and clear its queue.
|
||||||
Disconnect,
|
Disconnect,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -41,6 +48,12 @@ struct HubInner {
|
|||||||
subscribers: Mutex<HashMap<u64, Arc<SubscriberInner>>>,
|
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)]
|
#[derive(Debug)]
|
||||||
pub struct EventSubscription {
|
pub struct EventSubscription {
|
||||||
id: u64,
|
id: u64,
|
||||||
@@ -115,6 +128,8 @@ impl EventHub {
|
|||||||
|
|
||||||
let mut disconnected = Vec::new();
|
let mut disconnected = Vec::new();
|
||||||
for (id, subscriber) in subscribers {
|
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()) {
|
if !subscriber.publish(event.clone()) {
|
||||||
disconnected.push(id);
|
disconnected.push(id);
|
||||||
}
|
}
|
||||||
@@ -166,6 +181,11 @@ impl Default for EventHub {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl EventSubscription {
|
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> {
|
pub fn recv(&self) -> Result<ClientEvent, EventSubscriptionError> {
|
||||||
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
|
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
|
||||||
loop {
|
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> {
|
pub fn try_recv(&self) -> Result<Option<ClientEvent>, EventSubscriptionError> {
|
||||||
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
|
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
|
||||||
if let Some(event) = state.queue.pop_front() {
|
if let Some(event) = state.queue.pop_front() {
|
||||||
|
|||||||
@@ -25,6 +25,13 @@ const MAX_SLEEP: Duration = Duration::from_millis(20);
|
|||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct ManagedClient;
|
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)]
|
#[derive(Debug)]
|
||||||
pub struct ManagedClientSession {
|
pub struct ManagedClientSession {
|
||||||
hub: EventHub,
|
hub: EventHub,
|
||||||
@@ -32,6 +39,12 @@ pub struct ManagedClientSession {
|
|||||||
worker: Option<JoinHandle<Result<SessionOutcome, ClientError>>>,
|
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)]
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||||
pub struct SessionOutcome {
|
pub struct SessionOutcome {
|
||||||
pub end_reason: SessionEndReason,
|
pub end_reason: SessionEndReason,
|
||||||
@@ -111,7 +124,10 @@ impl ManagedClientSession {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn join(mut self) -> Result<SessionOutcome, ClientError> {
|
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() {
|
match worker.join() {
|
||||||
Ok(outcome) => {
|
Ok(outcome) => {
|
||||||
self.hub.disconnect_all();
|
self.hub.disconnect_all();
|
||||||
|
|||||||
Reference in new issue
Block a user