From 01c10e2e0d8ac111d211b9777ae6d6f9f69d12c0 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 18 May 2026 01:34:04 +0700 Subject: [PATCH] docs: document client public API --- crates/irtt-client/src/client.rs | 39 ++++ crates/irtt-client/src/config.rs | 96 ++++++++++ crates/irtt-client/src/event.rs | 167 +++++++++++++++++- .../irtt-client/src/managed/cancellation.rs | 11 ++ crates/irtt-client/src/managed/hub.rs | 24 +++ crates/irtt-client/src/managed/runner.rs | 43 +++++ crates/irtt-client/src/session.rs | 8 + crates/irtt-client/src/timing.rs | 8 + 8 files changed, 395 insertions(+), 1 deletion(-) diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 58e46e3..e07adb3 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -32,6 +32,12 @@ use crate::{ const MAX_OPEN_PACKET_SIZE: usize = 512; const MIN_RECV_BUFFER_SIZE: usize = 2048; +/// Low-level synchronous IRTT client. +/// +/// `Client` exposes the protocol steps directly: connect a UDP socket, open a +/// session, send probes, receive replies, poll timeouts, and close. Callers +/// that do not need to own this loop can use [`ManagedClient`](crate::ManagedClient) +/// instead. #[derive(Debug)] pub struct Client { config: ClientConfig, @@ -45,6 +51,11 @@ pub struct Client { } impl Client { + /// Resolve the configured server and create a connected UDP socket. + /// + /// This validates local configuration and prepares the open request, but it + /// does not contact the server. Call [`open`](Self::open) to perform the + /// IRTT open exchange. pub fn connect(config: ClientConfig) -> Result { validate_open_timeouts(&config.open_timeouts)?; if config.max_pending_probes == 0 { @@ -73,6 +84,11 @@ impl Client { }) } + /// Perform the IRTT open exchange. + /// + /// On success, returns the negotiated open outcome and transitions the + /// client into either an open probe session or completed no-test state. + /// Open attempts use [`ClientConfig::open_timeouts`]. pub fn open(&mut self) -> Result { match self.phase { ClientPhase::Connected => {} @@ -127,6 +143,10 @@ impl Client { } } + /// Send a close request and emit a [`ClientEvent::SessionClosed`] event. + /// + /// The close event means the client has sent its close packet and stopped + /// tracking the session locally; it is not a server acknowledgement. pub fn close(&mut self) -> Result, ClientError> { let token = match self.phase { ClientPhase::Open { token } => token, @@ -153,6 +173,8 @@ impl Client { }]) } + /// Return the monotonic deadline for the next probe send, if another probe + /// is scheduled. pub fn next_send_deadline(&self) -> Option { let session = self.session.as_ref()?; if session.sending_done { @@ -161,14 +183,24 @@ impl Client { Some(session.next_send_at) } + /// Return the local timeout used to classify pending probes as lost. pub fn probe_timeout(&self) -> Duration { self.config.probe_timeout } + /// Send one echo probe if the negotiated run is still active. + /// + /// Returns an `EchoSent` event when a probe is sent. Returns an empty event + /// list when the run duration has elapsed and no further probe should be + /// sent. pub fn send_probe(&mut self) -> Result, ClientError> { self.send_probe_inner(None) } + /// Receive and classify at most one datagram from the socket. + /// + /// Returns an empty event list when the socket read would block or times + /// out. Malformed or unrelated datagrams are reported as warning events. pub fn recv_once(&mut self) -> Result, ClientError> { self.recv_once_inner(None) } @@ -315,6 +347,8 @@ impl Client { self.process_echo_reply(reply, packet_len, now, datagram.meta) } + /// Receive and classify datagrams until a receive produces no events or the + /// receive budget is exhausted. pub fn recv_available(&mut self, budget: RecvBudget) -> Result, ClientError> { let mut all_events = Vec::new(); for _ in 0..budget.max_packets { @@ -366,6 +400,11 @@ impl Client { Ok(events) } + /// Return whether the current run has completed. + /// + /// A normal run is complete once no more probes will be sent and all + /// pending probes have either replied or timed out. No-test and closed + /// sessions are also considered complete. pub fn is_run_complete(&self) -> bool { let Some(session) = self.session.as_ref() else { return matches!( diff --git a/crates/irtt-client/src/config.rs b/crates/irtt-client/src/config.rs index d53254f..93bdc5c 100644 --- a/crates/irtt-client/src/config.rs +++ b/crates/irtt-client/src/config.rs @@ -19,23 +19,83 @@ pub(crate) const MIN_OPEN_TIMEOUT: Duration = Duration::from_millis(200); pub(crate) const DEFAULT_PROBE_TIMEOUT: Duration = Duration::from_secs(4); pub(crate) const DEFAULT_MAX_PENDING: usize = 4096; +/// Configuration for opening and running an IRTT client session. +/// +/// This type describes both the protocol parameters sent in the IRTT open +/// request and the local client behavior used to drive the UDP socket. Values +/// that are negotiated by the server are available after opening the session +/// through [`NegotiatedParams`](crate::NegotiatedParams). #[derive(Debug, Clone, PartialEq, Eq)] pub struct ClientConfig { + /// Remote server name or address. + /// + /// If no port is present, the default IRTT port 2112 is used. IPv6 + /// literals may be supplied either bracketed or unbracketed when the + /// default port should be used. pub server_addr: String, + /// Requested run duration. + /// + /// `Some(duration)` requests a finite test and must be greater than zero. + /// `None` requests continuous mode and is encoded on the wire as a zero + /// duration. Use [`RunMode::NoTest`] when the caller wants negotiation only + /// without sending probes. pub duration: Option, + /// Requested spacing between probe sends. + /// + /// The interval is encoded as nanoseconds in the open request and must be + /// greater than zero. The server may return a different interval depending + /// on the negotiated policy and server restrictions. pub interval: Duration, + /// Requested echo packet payload length, in bytes. + /// + /// The value must fit within the UDP payload limit after protocol overhead. + /// A server may reduce the requested length during negotiation. pub length: u32, + /// Requested server receive-statistics fields in echo replies. pub received_stats: ReceivedStats, + /// Requested timestamp placement in echo replies. pub stamp_at: StampAt, + /// Requested server clock sources for timestamp fields. pub clock: Clock, + /// Requested DSCP codepoint. + /// + /// This is the six-bit DSCP value, not the full traffic-class byte, and + /// must be less than or equal to [`MAX_DSCP_CODEPOINT`]. pub dscp: u8, + /// Optional HMAC key used to authenticate IRTT packets. + /// + /// When present, open, echo, and close packets are encoded and decoded with + /// HMAC authentication. The peer must be configured with the same key. pub hmac_key: Option>, + /// Optional server payload fill request. + /// + /// `None` leaves server fill behavior unspecified. `Some(value)` requests + /// a non-empty server fill mode/value and must not exceed + /// [`MAX_SERVER_FILL_BYTES`] bytes when UTF-8 encoded. pub server_fill: Option, + /// Per-attempt receive timeouts used while opening the session. + /// + /// The client sends an open request for each entry until a valid open reply + /// is received. The list must not be empty, and each timeout must be at + /// least 200 ms. pub open_timeouts: Vec, + /// Whether opening the session should start a probe test or perform a + /// negotiation-only no-test exchange. pub run_mode: RunMode, + /// Policy for server changes to negotiable protocol parameters. pub negotiation_policy: NegotiationPolicy, + /// Local UDP socket configuration. pub socket_config: SocketConfig, + /// Time after sending a probe before the client reports it as lost. + /// + /// This timeout is local client behavior; it is not negotiated with the + /// server. It must be greater than zero. pub probe_timeout: Duration, + /// Maximum number of probes tracked as pending/timed-out/completed. + /// + /// This bounds memory used for reply classification and must be greater + /// than zero. A very small value can reject sends when replies are still + /// outstanding. pub max_pending_probes: usize, } @@ -62,29 +122,65 @@ impl Default for ClientConfig { } } +/// Local UDP socket options used by [`ClientConfig`]. +/// +/// These settings affect how the client binds, resolves, and receives from the +/// socket. They do not change the IRTT protocol parameters negotiated with the +/// server. #[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct SocketConfig { + /// Local address to bind before connecting the UDP socket. + /// + /// `None` binds to an unspecified address with an ephemeral port matching + /// the selected remote address family. pub bind_addr: Option, + /// Optional IPv4 TTL or IPv6 hop limit applied to sent packets. + /// + /// `None` leaves the platform default unchanged. Values must fit in the + /// platform socket option range; [`MAX_TTL`] is the public configuration + /// bound used by this crate. pub ttl: Option, + /// Restrict name resolution to IPv4 addresses. pub ipv4_only: bool, + /// Restrict name resolution to IPv6 addresses and set IPV6_V6ONLY for IPv6 + /// sockets. pub ipv6_only: bool, + /// Socket read timeout used after the session is open. + /// + /// `None` leaves reads blocking for APIs that perform a single receive. + /// Managed sessions may replace `None` or long timeouts with a short + /// timeout so cooperative cancellation can be observed promptly. pub recv_timeout: Option, } +/// How strictly to handle server-side negotiation restrictions. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum NegotiationPolicy { + /// Reject any negotiated parameter that is more restrictive or different + /// than requested. Strict, + /// Accept documented server restrictions and report them in + /// [`NegotiatedParams`](crate::NegotiatedParams). Loose, } +/// Mode requested during the IRTT open exchange. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RunMode { + /// Open a session and send echo probes for the negotiated run duration. Normal, + /// Complete negotiation without running the echo probe test. NoTest, } +/// Bound for a single receive-drain operation. +/// +/// This is used by lower-level callers that drive [`Client`](crate::Client) +/// directly and want to cap how many datagrams are processed before returning +/// to their own event loop. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct RecvBudget { + /// Maximum number of datagrams to process before returning. pub max_packets: usize, } diff --git a/crates/irtt-client/src/event.rs b/crates/irtt-client/src/event.rs index 2e0834f..705062b 100644 --- a/crates/irtt-client/src/event.rs +++ b/crates/irtt-client/src/event.rs @@ -5,105 +5,231 @@ use std::{ use crate::{session::NegotiatedParams, timing::ClientTimestamp}; +/// Result of the IRTT open exchange. +/// +/// The lower-level [`Client`](crate::Client) API returns this before any echo +/// probes are driven. Managed sessions publish the contained lifecycle event to +/// subscribers immediately after a successful open. #[derive(Debug, Clone, PartialEq, Eq)] pub enum OpenOutcome { + /// A normal probe session was opened. Started { + /// Resolved remote socket address used by the UDP socket. remote: SocketAddr, + /// Session token assigned by the server and used on echo/close packets. token: u64, + /// Parameters returned by the server after negotiation. negotiated: NegotiatedParams, + /// Lifecycle event corresponding to this outcome. event: ClientEvent, }, + /// A no-test open exchange completed without starting a probe session. NoTestCompleted { + /// Resolved remote socket address used by the UDP socket. remote: SocketAddr, + /// Parameters returned by the server after negotiation. negotiated: NegotiatedParams, + /// Lifecycle event corresponding to this outcome. event: ClientEvent, }, } +/// Event emitted by an IRTT client session. +/// +/// Events form the public lifecycle and measurement stream for callers. UIs +/// usually display lifecycle, warning, and per-packet events directly; summary +/// reporting should aggregate `EchoSent`, `EchoReply`, `EchoLoss`, +/// `LateReply`, and `DuplicateReply` events or pass the stream to +/// `irtt-stats`. #[derive(Debug, Clone, PartialEq, Eq)] pub enum ClientEvent { + /// A normal session has opened and is ready to send echo probes. SessionStarted { + /// Resolved remote socket address. remote: SocketAddr, + /// Session token assigned by the server. token: u64, + /// Negotiated protocol parameters for this session. negotiated: NegotiatedParams, + /// Client timestamp when the open reply was accepted. at: ClientTimestamp, }, + /// A negotiation-only no-test exchange completed. + /// + /// No `EchoSent`, `EchoReply`, `EchoLoss`, or `SessionClosed` events are + /// expected for this open outcome. NoTestCompleted { + /// Resolved remote socket address. remote: SocketAddr, + /// Negotiated protocol parameters returned by the server. negotiated: NegotiatedParams, + /// Client timestamp when the open reply was accepted. at: ClientTimestamp, }, + /// The client sent a close request and considers the session closed. + /// + /// This is a local lifecycle event, not an acknowledgement from the server. + /// Managed sessions disconnect event subscribers after the worker exits, + /// while leaving already queued events available to drain. SessionClosed { + /// Resolved remote socket address. remote: SocketAddr, + /// Session token that was closed. token: u64, + /// Client timestamp when the close request was sent. at: ClientTimestamp, }, + /// An echo request was sent. + /// + /// This records local scheduling and send-call timing. It does not imply a + /// reply will arrive; callers should pair it with `EchoReply`, `EchoLoss`, + /// `LateReply`, or `DuplicateReply` by sequence number when displaying + /// packet lifecycle state. EchoSent { + /// Wire sequence number sent in the echo request. seq: u32, + /// Resolved remote socket address. remote: SocketAddr, + /// Monotonic deadline at which the probe was scheduled to be sent. scheduled_at: Instant, + /// Client wall/monotonic timestamp captured for the send. sent_at: ClientTimestamp, + /// Number of bytes passed to the UDP socket. bytes: usize, + /// Elapsed time spent in the socket send call. send_call: Duration, + /// Absolute difference between the scheduled send time and actual send + /// timestamp. timer_error: Duration, }, + /// First in-window reply for a pending echo request. + /// + /// This is the primary successful measurement event. RTT and optional + /// server timing/statistics fields should generally be aggregated from this + /// variant. EchoReply { + /// Wire sequence number from the echo reply. seq: u32, + /// Resolved remote socket address. remote: SocketAddr, + /// Client timestamp recorded when the matching request was sent. sent_at: ClientTimestamp, + /// Client timestamp recorded when the reply was received. received_at: ClientTimestamp, + /// Round-trip timing sample for this reply. rtt: RttSample, + /// Optional server-reported timestamps and processing time. server_timing: Option, + /// Optional one-way delay values derived from client/server wall clocks. one_way: Option, + /// Optional server-reported received-statistics sample. received_stats: Option, + /// UDP datagram size received from the socket. bytes: usize, + /// Receive-side metadata observed outside the IRTT wire payload. packet_meta: PacketMeta, }, + /// A sent echo request exceeded the local probe timeout before a reply was + /// matched. + /// + /// A later datagram for the same sequence may still arrive and will be + /// reported as [`ClientEvent::LateReply`] rather than changing this event. EchoLoss { + /// Wire sequence number that timed out. seq: u32, + /// Client timestamp recorded when the request was sent. sent_at: ClientTimestamp, + /// Monotonic deadline at which the probe was declared lost. timeout_at: std::time::Instant, }, + /// Additional reply for a sequence that was already completed. + /// + /// Duplicates are not primary RTT samples. They are useful for diagnostics + /// and packet counters, but should not be aggregated as successful first + /// replies. DuplicateReply { + /// Wire sequence number from the duplicate reply. seq: u32, + /// Resolved remote socket address. remote: SocketAddr, + /// Client timestamp recorded when the duplicate was received. received_at: ClientTimestamp, + /// UDP datagram size received from the socket. bytes: usize, }, + /// Reply for a sequence that was no longer pending. + /// + /// This commonly means the probe had already been reported as lost, or the + /// reply arrived out of the client's current tracking window. When the + /// original send metadata is still retained, RTT and one-way timing are + /// included; otherwise those fields are `None`. LateReply { + /// Wire sequence number from the late reply. seq: u32, + /// Highest sequence number accepted as an in-window first reply so far. highest_seen: u32, + /// Resolved remote socket address. remote: SocketAddr, + /// Original client send timestamp, when still retained. sent_at: Option, + /// Client timestamp recorded when the late reply was received. received_at: ClientTimestamp, + /// RTT sample computed when the original send timestamp is available. rtt: Option, + /// Optional server-reported timestamps and processing time. server_timing: Option, + /// Optional one-way delay values derived from client/server wall clocks. one_way: Option, + /// Optional server-reported received-statistics sample. received_stats: Option, + /// UDP datagram size received from the socket. bytes: usize, + /// Receive-side metadata observed outside the IRTT wire payload. packet_meta: PacketMeta, }, + /// Non-fatal condition observed while receiving or classifying packets. + /// + /// Warning events are diagnostic. They do not close the session by + /// themselves and are not measurement samples. Warning { + /// Machine-readable warning category. kind: WarningKind, + /// Human-readable diagnostic text. message: String, + /// Client timestamp when the warning was emitted. at: ClientTimestamp, }, } +/// Category for a non-fatal [`ClientEvent::Warning`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[non_exhaustive] pub enum WarningKind { + /// A received datagram could not be decoded as a relevant IRTT packet. MalformedOrUnrelatedPacket, + /// A decoded packet used a token that does not match the open session. WrongToken, + /// A decoded reply could not be matched to tracked probe state. UntrackedReply, } +/// Round-trip timing for one echo reply. +/// +/// `raw` is the client-observed elapsed time from sending the echo request to +/// receiving the reply and is therefore non-negative. When the server reports +/// enough monotonic timing to compute processing time, `adjusted` subtracts +/// that server processing from `raw`. `effective` is the value used by reporting +/// and statistics: adjusted when available, otherwise raw converted to a signed +/// value. +/// +/// Adjusted and effective RTT values are signed. They may be negative if the +/// reported server processing time is greater than the raw client-observed RTT. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct RttSample { /// Client-observed RTT from send to receive. @@ -117,34 +243,48 @@ pub struct RttSample { pub effective: SignedDuration, } +/// Signed nanosecond-backed duration used for derived timing values. +/// +/// This type preserves negative values that can arise after subtracting server +/// processing time or comparing wall-clock timestamps from different hosts. It +/// should not be used to represent raw elapsed wall-clock or monotonic time, +/// which is non-negative and represented by [`Duration`]. #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub struct SignedDuration { ns: i128, } impl SignedDuration { + /// Construct a signed duration from nanoseconds. pub const fn from_nanos(ns: i128) -> Self { Self { ns } } + /// Convert a non-negative [`Duration`] into a signed duration. + /// + /// Durations larger than `i128::MAX` nanoseconds saturate to `i128::MAX`. pub fn from_duration(duration: Duration) -> Self { Self { ns: i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX), } } + /// Return the duration in nanoseconds. pub const fn as_nanos(self) -> i128 { self.ns } + /// Return the duration in whole microseconds, truncated toward zero. pub const fn as_micros(self) -> i128 { self.ns / 1_000 } + /// Return the duration in whole milliseconds, truncated toward zero. pub const fn as_millis(self) -> i128 { self.ns / 1_000_000 } + /// Return whether this duration is less than zero. pub const fn is_negative(self) -> bool { self.ns < 0 } @@ -156,18 +296,35 @@ impl From for SignedDuration { } } +/// Server-reported timing fields decoded from an echo reply. +/// +/// These values are optional because the negotiated timestamp mode, clock mode, +/// server implementation, or packet contents may omit individual fields. +/// Timestamp values are nanoseconds in the server's reported wall-clock or +/// monotonic clock domain. `None` means unavailable; `Some(0)` means the server +/// reported a zero timestamp. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct ServerTiming { + /// Server wall-clock receive timestamp, in nanoseconds. pub receive_wall_ns: Option, + /// Server monotonic receive timestamp, in nanoseconds. pub receive_mono_ns: Option, + /// Server wall-clock send timestamp, in nanoseconds. pub send_wall_ns: Option, + /// Server monotonic send timestamp, in nanoseconds. pub send_mono_ns: Option, + /// Server wall-clock midpoint timestamp, in nanoseconds. pub midpoint_wall_ns: Option, + /// Server monotonic midpoint timestamp, in nanoseconds. pub midpoint_mono_ns: Option, + /// Server processing time computed from monotonic receive/send timestamps. + /// + /// `None` means the required monotonic timestamps were unavailable or + /// unusable. `Some(Duration::ZERO)` means the computed processing interval + /// was available and zero. pub processing: Option, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] /// Directional one-way delay samples computed from wall-clock timestamps. /// /// Each directional field is an `Option`. `None` means the @@ -178,6 +335,7 @@ pub struct ServerTiming { /// Negative computed values are preserved as `Some` instead of being clamped to /// zero or dropped. A negative one-way delay usually indicates wall-clock skew /// between the client and server. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct OneWayDelaySample { /// Client-to-server delay. /// @@ -191,9 +349,16 @@ pub struct OneWayDelaySample { pub server_to_client: Option, } +/// Server-reported receive counters decoded from an echo reply. +/// +/// These fields are optional because the negotiated received-statistics mode or +/// server implementation may omit them. `None` means unavailable; `Some(0)` +/// means the server reported the value and it was zero. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct ReceivedStatsSample { + /// Server-reported count of received packets, when available. pub count: Option, + /// Server-reported receive window value, when available. pub window: Option, } diff --git a/crates/irtt-client/src/managed/cancellation.rs b/crates/irtt-client/src/managed/cancellation.rs index 66b27f0..0362bbc 100644 --- a/crates/irtt-client/src/managed/cancellation.rs +++ b/crates/irtt-client/src/managed/cancellation.rs @@ -3,20 +3,31 @@ use std::sync::{ Arc, }; +/// Cloneable cooperative cancellation flag. +/// +/// Managed sessions use this internally to ask their worker loop to stop. It is +/// exported for callers that want the same simple cancellation primitive in +/// code built around `irtt-client`. #[derive(Debug, Clone, Default)] pub struct CancellationToken { cancelled: Arc, } impl CancellationToken { + /// Create a token in the non-cancelled state. pub fn new() -> Self { Self::default() } + /// Mark the token as cancelled. + /// + /// Cancellation is idempotent and is visible through all clones of the + /// token. pub fn cancel(&self) { self.cancelled.store(true, Ordering::SeqCst); } + /// Return whether cancellation has been requested. #[must_use] pub fn is_cancelled(&self) -> bool { self.cancelled.load(Ordering::SeqCst) diff --git a/crates/irtt-client/src/managed/hub.rs b/crates/irtt-client/src/managed/hub.rs index e2c02f4..30b11cc 100644 --- a/crates/irtt-client/src/managed/hub.rs +++ b/crates/irtt-client/src/managed/hub.rs @@ -8,9 +8,17 @@ use std::{ use crate::{error::EventSubscriptionError, event::ClientEvent}; +/// Configuration for one managed event subscriber. +/// +/// Each subscriber has an independent bounded queue. The queue bounds memory +/// growth when event production is faster than that subscriber's consumer. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct SubscriberConfig { + /// Maximum number of events buffered for this subscriber. + /// + /// Must be greater than zero. pub capacity: usize, + /// Policy applied when publishing to a full subscriber queue. pub overflow: SubscriberOverflow, } @@ -37,6 +45,11 @@ pub enum SubscriberOverflow { Disconnect, } +/// Publish/subscribe fan-out for [`ClientEvent`] values. +/// +/// `EventHub` is used by managed sessions internally and is also exported for +/// callers that want the same bounded subscription behavior around their own +/// event producer. Publishing clones each event once per subscriber. #[derive(Debug, Clone)] pub struct EventHub { inner: Arc, @@ -75,6 +88,7 @@ struct SubscriberState { } impl EventHub { + /// Create an empty event hub with no subscribers. pub fn new() -> Self { Self { inner: Arc::new(HubInner { @@ -84,6 +98,9 @@ impl EventHub { } } + /// Register a new subscriber with its own bounded queue. + /// + /// Returns an error when `config.capacity` is zero. pub fn subscribe( &self, config: SubscriberConfig, @@ -116,6 +133,11 @@ impl EventHub { }) } + /// Publish an event to all currently connected subscribers. + /// + /// Slow subscribers are handled according to their + /// [`SubscriberOverflow`] policy. Publishing does not block waiting for a + /// subscriber to consume queued events. pub fn publish(&self, event: ClientEvent) { let subscribers: Vec<(u64, Arc)> = self .inner @@ -149,6 +171,8 @@ impl EventHub { } } + /// Disconnect all subscribers after leaving their already queued events + /// available to drain. pub fn disconnect_all(&self) { let subscribers: Vec> = self .inner diff --git a/crates/irtt-client/src/managed/runner.rs b/crates/irtt-client/src/managed/runner.rs index c5d5217..97b0d48 100644 --- a/crates/irtt-client/src/managed/runner.rs +++ b/crates/irtt-client/src/managed/runner.rs @@ -21,6 +21,12 @@ const MANAGED_FINAL_DRAIN: Duration = Duration::from_millis(100); const IDLE_SLEEP: Duration = Duration::from_millis(1); const MAX_SLEEP: Duration = Duration::from_millis(20); +/// Entry point for running a client session on a worker thread. +/// +/// The managed API owns the lower-level [`Client`](crate::Client) loop: it +/// opens the session, sends probes at the negotiated interval, receives +/// datagrams, publishes [`ClientEvent`] values, and closes the session when the +/// run completes or is cancelled. #[derive(Debug)] pub struct ManagedClient; @@ -46,26 +52,49 @@ pub struct ManagedClientSession { #[must_use = "managed session outcomes contain completion status and counters"] #[derive(Debug, Clone, PartialEq, Eq)] pub struct SessionOutcome { + /// Why the managed loop stopped. pub end_reason: SessionEndReason, + /// Number of echo requests sent by the client. pub packets_sent: u64, + /// Number of first in-window echo replies received. pub replies_received: u64, + /// Number of duplicate reply events emitted. pub duplicates: u64, + /// Number of late reply events emitted. pub late: u64, + /// Number of warning events emitted. pub warning_events: u64, } +/// Reason a managed client session ended. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SessionEndReason { + /// The negotiated finite test duration completed and the client closed the + /// session normally. TestComplete, + /// Cancellation was requested through [`ManagedClientSession::stop`] or by + /// dropping the session handle. Cancelled, + /// The session was opened in [`RunMode::NoTest`](crate::RunMode) and + /// completed after negotiation. NoTestComplete, } impl ManagedClient { + /// Start a managed session without creating an initial event subscription. + /// + /// The returned [`ManagedClientSession`] can still create subscriptions via + /// [`ManagedClientSession::subscribe`], but events emitted before a + /// subscriber is registered are not replayed. pub fn start(config: ClientConfig) -> Result { Self::start_inner(config, None).map(|(session, _)| session) } + /// Start a managed session and subscribe to events before the worker runs. + /// + /// The initial subscription receives the open lifecycle event and + /// subsequent session events, subject to its queue capacity and overflow + /// policy. pub fn start_with_subscription( config: ClientConfig, subscriber_config: SubscriberConfig, @@ -114,14 +143,28 @@ impl ManagedClient { } impl ManagedClientSession { + /// Add another event subscriber to this running managed session. + /// + /// The subscription receives only events published after it is registered. pub fn subscribe(&self, config: SubscriberConfig) -> Result { self.hub.subscribe(config) } + /// Request cooperative cancellation of the managed session. + /// + /// Cancellation is observed by the worker loop between socket receive, + /// timeout polling, and send scheduling steps. Call [`join`](Self::join) to + /// wait for the worker and obtain the final [`SessionOutcome`]. pub fn stop(&self) { self.cancellation.cancel(); } + /// Wait for the managed worker thread to finish. + /// + /// On success, returns lifecycle counters for the completed session. If the + /// worker panicked, this returns [`ClientError::WorkerPanicked`]. Joining + /// also disconnects all event subscriptions while leaving already queued + /// events available to drain. pub fn join(mut self) -> Result { let worker = self .worker diff --git a/crates/irtt-client/src/session.rs b/crates/irtt-client/src/session.rs index 790ec1f..258b5e0 100644 --- a/crates/irtt-client/src/session.rs +++ b/crates/irtt-client/src/session.rs @@ -8,9 +8,16 @@ use crate::{ probe::{CompletedSet, PendingMap, TimedOutMap}, }; +/// Protocol parameters accepted for a session. +/// +/// `params` contains the server-returned values that the client will use for +/// echo packets. `restrictions` records accepted differences from the request +/// when loose negotiation is enabled. #[derive(Debug, Clone, PartialEq, Eq)] pub struct NegotiatedParams { + /// Server-returned protocol parameters used for the session. pub params: Params, + /// Accepted server restrictions or parameter changes. pub restrictions: Vec, } @@ -57,6 +64,7 @@ pub enum NegotiationRestriction { } impl NegotiationRestriction { + /// Return a human-readable description of this negotiated restriction. pub fn message(&self) -> String { self.to_string() } diff --git a/crates/irtt-client/src/timing.rs b/crates/irtt-client/src/timing.rs index ec145e9..865e275 100644 --- a/crates/irtt-client/src/timing.rs +++ b/crates/irtt-client/src/timing.rs @@ -1,12 +1,20 @@ use std::time::{Instant, SystemTime}; +/// Client-side timestamp captured in both wall-clock and monotonic time. +/// +/// Wall-clock time is used for protocol timestamp comparison and one-way delay +/// calculations. Monotonic time is used for elapsed client-side timing such as +/// RTT, scheduling, and timeouts. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct ClientTimestamp { + /// System wall-clock time. pub wall: SystemTime, + /// Monotonic timestamp sampled alongside [`wall`](Self::wall). pub mono: Instant, } impl ClientTimestamp { + /// Capture the current wall-clock and monotonic timestamps. pub fn now() -> Self { Self { wall: SystemTime::now(),