docs: document client public API

This commit is contained in:
Nils Andreas Svee committed 2026-05-18 01:34:04 +07:00
1 parent 8fd0b803e9
commit 01c10e2e0d
8 files changed
+395 -1

No files matched your search

+39
View File
@@ -32,6 +32,12 @@ use crate::{
const MAX_OPEN_PACKET_SIZE: usize = 512; const MAX_OPEN_PACKET_SIZE: usize = 512;
const MIN_RECV_BUFFER_SIZE: usize = 2048; 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)] #[derive(Debug)]
pub struct Client { pub struct Client {
config: ClientConfig, config: ClientConfig,
@@ -45,6 +51,11 @@ pub struct Client {
} }
impl 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<Self, ClientError> { pub fn connect(config: ClientConfig) -> Result<Self, ClientError> {
validate_open_timeouts(&config.open_timeouts)?; validate_open_timeouts(&config.open_timeouts)?;
if config.max_pending_probes == 0 { 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<OpenOutcome, ClientError> { pub fn open(&mut self) -> Result<OpenOutcome, ClientError> {
match self.phase { match self.phase {
ClientPhase::Connected => {} 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<Vec<ClientEvent>, ClientError> { pub fn close(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
let token = match self.phase { let token = match self.phase {
ClientPhase::Open { token } => token, 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<Instant> { pub fn next_send_deadline(&self) -> Option<Instant> {
let session = self.session.as_ref()?; let session = self.session.as_ref()?;
if session.sending_done { if session.sending_done {
@@ -161,14 +183,24 @@ impl Client {
Some(session.next_send_at) Some(session.next_send_at)
} }
/// Return the local timeout used to classify pending probes as lost.
pub fn probe_timeout(&self) -> Duration { pub fn probe_timeout(&self) -> Duration {
self.config.probe_timeout 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<Vec<ClientEvent>, ClientError> { pub fn send_probe(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
self.send_probe_inner(None) 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<Vec<ClientEvent>, ClientError> { pub fn recv_once(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
self.recv_once_inner(None) self.recv_once_inner(None)
} }
@@ -315,6 +347,8 @@ impl Client {
self.process_echo_reply(reply, packet_len, now, datagram.meta) 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<Vec<ClientEvent>, ClientError> { pub fn recv_available(&mut self, budget: RecvBudget) -> Result<Vec<ClientEvent>, ClientError> {
let mut all_events = Vec::new(); let mut all_events = Vec::new();
for _ in 0..budget.max_packets { for _ in 0..budget.max_packets {
@@ -366,6 +400,11 @@ impl Client {
Ok(events) 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 { pub fn is_run_complete(&self) -> bool {
let Some(session) = self.session.as_ref() else { let Some(session) = self.session.as_ref() else {
return matches!( return matches!(
+96
View File
@@ -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_PROBE_TIMEOUT: Duration = Duration::from_secs(4);
pub(crate) const DEFAULT_MAX_PENDING: usize = 4096; 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)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientConfig { 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, 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<Duration>, pub duration: Option<Duration>,
/// 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, 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, pub length: u32,
/// Requested server receive-statistics fields in echo replies.
pub received_stats: ReceivedStats, pub received_stats: ReceivedStats,
/// Requested timestamp placement in echo replies.
pub stamp_at: StampAt, pub stamp_at: StampAt,
/// Requested server clock sources for timestamp fields.
pub clock: Clock, 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, 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<Vec<u8>>, pub hmac_key: Option<Vec<u8>>,
/// 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<String>, pub server_fill: Option<String>,
/// 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<Duration>, pub open_timeouts: Vec<Duration>,
/// Whether opening the session should start a probe test or perform a
/// negotiation-only no-test exchange.
pub run_mode: RunMode, pub run_mode: RunMode,
/// Policy for server changes to negotiable protocol parameters.
pub negotiation_policy: NegotiationPolicy, pub negotiation_policy: NegotiationPolicy,
/// Local UDP socket configuration.
pub socket_config: SocketConfig, 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, 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, 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)] #[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct SocketConfig { 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<SocketAddr>, pub bind_addr: Option<SocketAddr>,
/// 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<u32>, pub ttl: Option<u32>,
/// Restrict name resolution to IPv4 addresses.
pub ipv4_only: bool, pub ipv4_only: bool,
/// Restrict name resolution to IPv6 addresses and set IPV6_V6ONLY for IPv6
/// sockets.
pub ipv6_only: bool, 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<Duration>, pub recv_timeout: Option<Duration>,
} }
/// How strictly to handle server-side negotiation restrictions.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NegotiationPolicy { pub enum NegotiationPolicy {
/// Reject any negotiated parameter that is more restrictive or different
/// than requested.
Strict, Strict,
/// Accept documented server restrictions and report them in
/// [`NegotiatedParams`](crate::NegotiatedParams).
Loose, Loose,
} }
/// Mode requested during the IRTT open exchange.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunMode { pub enum RunMode {
/// Open a session and send echo probes for the negotiated run duration.
Normal, Normal,
/// Complete negotiation without running the echo probe test.
NoTest, 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)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RecvBudget { pub struct RecvBudget {
/// Maximum number of datagrams to process before returning.
pub max_packets: usize, pub max_packets: usize,
} }
+166 -1
View File
@@ -5,105 +5,231 @@ use std::{
use crate::{session::NegotiatedParams, timing::ClientTimestamp}; 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)] #[derive(Debug, Clone, PartialEq, Eq)]
pub enum OpenOutcome { pub enum OpenOutcome {
/// A normal probe session was opened.
Started { Started {
/// Resolved remote socket address used by the UDP socket.
remote: SocketAddr, remote: SocketAddr,
/// Session token assigned by the server and used on echo/close packets.
token: u64, token: u64,
/// Parameters returned by the server after negotiation.
negotiated: NegotiatedParams, negotiated: NegotiatedParams,
/// Lifecycle event corresponding to this outcome.
event: ClientEvent, event: ClientEvent,
}, },
/// A no-test open exchange completed without starting a probe session.
NoTestCompleted { NoTestCompleted {
/// Resolved remote socket address used by the UDP socket.
remote: SocketAddr, remote: SocketAddr,
/// Parameters returned by the server after negotiation.
negotiated: NegotiatedParams, negotiated: NegotiatedParams,
/// Lifecycle event corresponding to this outcome.
event: ClientEvent, 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)] #[derive(Debug, Clone, PartialEq, Eq)]
pub enum ClientEvent { pub enum ClientEvent {
/// A normal session has opened and is ready to send echo probes.
SessionStarted { SessionStarted {
/// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Session token assigned by the server.
token: u64, token: u64,
/// Negotiated protocol parameters for this session.
negotiated: NegotiatedParams, negotiated: NegotiatedParams,
/// Client timestamp when the open reply was accepted.
at: ClientTimestamp, at: ClientTimestamp,
}, },
/// A negotiation-only no-test exchange completed.
///
/// No `EchoSent`, `EchoReply`, `EchoLoss`, or `SessionClosed` events are
/// expected for this open outcome.
NoTestCompleted { NoTestCompleted {
/// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Negotiated protocol parameters returned by the server.
negotiated: NegotiatedParams, negotiated: NegotiatedParams,
/// Client timestamp when the open reply was accepted.
at: ClientTimestamp, 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 { SessionClosed {
/// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Session token that was closed.
token: u64, token: u64,
/// Client timestamp when the close request was sent.
at: ClientTimestamp, 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 { EchoSent {
/// Wire sequence number sent in the echo request.
seq: u32, seq: u32,
/// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Monotonic deadline at which the probe was scheduled to be sent.
scheduled_at: Instant, scheduled_at: Instant,
/// Client wall/monotonic timestamp captured for the send.
sent_at: ClientTimestamp, sent_at: ClientTimestamp,
/// Number of bytes passed to the UDP socket.
bytes: usize, bytes: usize,
/// Elapsed time spent in the socket send call.
send_call: Duration, send_call: Duration,
/// Absolute difference between the scheduled send time and actual send
/// timestamp.
timer_error: Duration, 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 { EchoReply {
/// Wire sequence number from the echo reply.
seq: u32, seq: u32,
/// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Client timestamp recorded when the matching request was sent.
sent_at: ClientTimestamp, sent_at: ClientTimestamp,
/// Client timestamp recorded when the reply was received.
received_at: ClientTimestamp, received_at: ClientTimestamp,
/// Round-trip timing sample for this reply.
rtt: RttSample, rtt: RttSample,
/// Optional server-reported timestamps and processing time.
server_timing: Option<ServerTiming>, server_timing: Option<ServerTiming>,
/// Optional one-way delay values derived from client/server wall clocks.
one_way: Option<OneWayDelaySample>, one_way: Option<OneWayDelaySample>,
/// Optional server-reported received-statistics sample.
received_stats: Option<ReceivedStatsSample>, received_stats: Option<ReceivedStatsSample>,
/// UDP datagram size received from the socket.
bytes: usize, bytes: usize,
/// Receive-side metadata observed outside the IRTT wire payload.
packet_meta: PacketMeta, 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 { EchoLoss {
/// Wire sequence number that timed out.
seq: u32, seq: u32,
/// Client timestamp recorded when the request was sent.
sent_at: ClientTimestamp, sent_at: ClientTimestamp,
/// Monotonic deadline at which the probe was declared lost.
timeout_at: std::time::Instant, 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 { DuplicateReply {
/// Wire sequence number from the duplicate reply.
seq: u32, seq: u32,
/// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Client timestamp recorded when the duplicate was received.
received_at: ClientTimestamp, received_at: ClientTimestamp,
/// UDP datagram size received from the socket.
bytes: usize, 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 { LateReply {
/// Wire sequence number from the late reply.
seq: u32, seq: u32,
/// Highest sequence number accepted as an in-window first reply so far.
highest_seen: u32, highest_seen: u32,
/// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Original client send timestamp, when still retained.
sent_at: Option<ClientTimestamp>, sent_at: Option<ClientTimestamp>,
/// Client timestamp recorded when the late reply was received.
received_at: ClientTimestamp, received_at: ClientTimestamp,
/// RTT sample computed when the original send timestamp is available.
rtt: Option<RttSample>, rtt: Option<RttSample>,
/// Optional server-reported timestamps and processing time.
server_timing: Option<ServerTiming>, server_timing: Option<ServerTiming>,
/// Optional one-way delay values derived from client/server wall clocks.
one_way: Option<OneWayDelaySample>, one_way: Option<OneWayDelaySample>,
/// Optional server-reported received-statistics sample.
received_stats: Option<ReceivedStatsSample>, received_stats: Option<ReceivedStatsSample>,
/// UDP datagram size received from the socket.
bytes: usize, bytes: usize,
/// Receive-side metadata observed outside the IRTT wire payload.
packet_meta: PacketMeta, 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 { Warning {
/// Machine-readable warning category.
kind: WarningKind, kind: WarningKind,
/// Human-readable diagnostic text.
message: String, message: String,
/// Client timestamp when the warning was emitted.
at: ClientTimestamp, at: ClientTimestamp,
}, },
} }
/// Category for a non-fatal [`ClientEvent::Warning`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive] #[non_exhaustive]
pub enum WarningKind { pub enum WarningKind {
/// A received datagram could not be decoded as a relevant IRTT packet.
MalformedOrUnrelatedPacket, MalformedOrUnrelatedPacket,
/// A decoded packet used a token that does not match the open session.
WrongToken, WrongToken,
/// A decoded reply could not be matched to tracked probe state.
UntrackedReply, 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)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RttSample { pub struct RttSample {
/// Client-observed RTT from send to receive. /// Client-observed RTT from send to receive.
@@ -117,34 +243,48 @@ pub struct RttSample {
pub effective: SignedDuration, 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)] #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct SignedDuration { pub struct SignedDuration {
ns: i128, ns: i128,
} }
impl SignedDuration { impl SignedDuration {
/// Construct a signed duration from nanoseconds.
pub const fn from_nanos(ns: i128) -> Self { pub const fn from_nanos(ns: i128) -> Self {
Self { ns } 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 { pub fn from_duration(duration: Duration) -> Self {
Self { Self {
ns: i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX), ns: i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX),
} }
} }
/// Return the duration in nanoseconds.
pub const fn as_nanos(self) -> i128 { pub const fn as_nanos(self) -> i128 {
self.ns self.ns
} }
/// Return the duration in whole microseconds, truncated toward zero.
pub const fn as_micros(self) -> i128 { pub const fn as_micros(self) -> i128 {
self.ns / 1_000 self.ns / 1_000
} }
/// Return the duration in whole milliseconds, truncated toward zero.
pub const fn as_millis(self) -> i128 { pub const fn as_millis(self) -> i128 {
self.ns / 1_000_000 self.ns / 1_000_000
} }
/// Return whether this duration is less than zero.
pub const fn is_negative(self) -> bool { pub const fn is_negative(self) -> bool {
self.ns < 0 self.ns < 0
} }
@@ -156,18 +296,35 @@ impl From<Duration> 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)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ServerTiming { pub struct ServerTiming {
/// Server wall-clock receive timestamp, in nanoseconds.
pub receive_wall_ns: Option<i64>, pub receive_wall_ns: Option<i64>,
/// Server monotonic receive timestamp, in nanoseconds.
pub receive_mono_ns: Option<i64>, pub receive_mono_ns: Option<i64>,
/// Server wall-clock send timestamp, in nanoseconds.
pub send_wall_ns: Option<i64>, pub send_wall_ns: Option<i64>,
/// Server monotonic send timestamp, in nanoseconds.
pub send_mono_ns: Option<i64>, pub send_mono_ns: Option<i64>,
/// Server wall-clock midpoint timestamp, in nanoseconds.
pub midpoint_wall_ns: Option<i64>, pub midpoint_wall_ns: Option<i64>,
/// Server monotonic midpoint timestamp, in nanoseconds.
pub midpoint_mono_ns: Option<i64>, pub midpoint_mono_ns: Option<i64>,
/// 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<Duration>, pub processing: Option<Duration>,
} }
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Directional one-way delay samples computed from wall-clock timestamps. /// Directional one-way delay samples computed from wall-clock timestamps.
/// ///
/// Each directional field is an `Option<SignedDuration>`. `None` means the /// Each directional field is an `Option<SignedDuration>`. `None` means the
@@ -178,6 +335,7 @@ pub struct ServerTiming {
/// Negative computed values are preserved as `Some` instead of being clamped to /// 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 /// zero or dropped. A negative one-way delay usually indicates wall-clock skew
/// between the client and server. /// between the client and server.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct OneWayDelaySample { pub struct OneWayDelaySample {
/// Client-to-server delay. /// Client-to-server delay.
/// ///
@@ -191,9 +349,16 @@ pub struct OneWayDelaySample {
pub server_to_client: Option<SignedDuration>, pub server_to_client: Option<SignedDuration>,
} }
/// 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)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReceivedStatsSample { pub struct ReceivedStatsSample {
/// Server-reported count of received packets, when available.
pub count: Option<u32>, pub count: Option<u32>,
/// Server-reported receive window value, when available.
pub window: Option<u64>, pub window: Option<u64>,
} }
@@ -3,20 +3,31 @@ use std::sync::{
Arc, 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)] #[derive(Debug, Clone, Default)]
pub struct CancellationToken { pub struct CancellationToken {
cancelled: Arc<AtomicBool>, cancelled: Arc<AtomicBool>,
} }
impl CancellationToken { impl CancellationToken {
/// Create a token in the non-cancelled state.
pub fn new() -> Self { pub fn new() -> Self {
Self::default() Self::default()
} }
/// Mark the token as cancelled.
///
/// Cancellation is idempotent and is visible through all clones of the
/// token.
pub fn cancel(&self) { pub fn cancel(&self) {
self.cancelled.store(true, Ordering::SeqCst); self.cancelled.store(true, Ordering::SeqCst);
} }
/// Return whether cancellation has been requested.
#[must_use] #[must_use]
pub fn is_cancelled(&self) -> bool { pub fn is_cancelled(&self) -> bool {
self.cancelled.load(Ordering::SeqCst) self.cancelled.load(Ordering::SeqCst)
+24
View File
@@ -8,9 +8,17 @@ use std::{
use crate::{error::EventSubscriptionError, event::ClientEvent}; 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)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SubscriberConfig { pub struct SubscriberConfig {
/// Maximum number of events buffered for this subscriber.
///
/// Must be greater than zero.
pub capacity: usize, pub capacity: usize,
/// Policy applied when publishing to a full subscriber queue.
pub overflow: SubscriberOverflow, pub overflow: SubscriberOverflow,
} }
@@ -37,6 +45,11 @@ pub enum SubscriberOverflow {
Disconnect, 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)] #[derive(Debug, Clone)]
pub struct EventHub { pub struct EventHub {
inner: Arc<HubInner>, inner: Arc<HubInner>,
@@ -75,6 +88,7 @@ struct SubscriberState {
} }
impl EventHub { impl EventHub {
/// Create an empty event hub with no subscribers.
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
inner: Arc::new(HubInner { 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( pub fn subscribe(
&self, &self,
config: SubscriberConfig, 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) { pub fn publish(&self, event: ClientEvent) {
let subscribers: Vec<(u64, Arc<SubscriberInner>)> = self let subscribers: Vec<(u64, Arc<SubscriberInner>)> = self
.inner .inner
@@ -149,6 +171,8 @@ impl EventHub {
} }
} }
/// Disconnect all subscribers after leaving their already queued events
/// available to drain.
pub fn disconnect_all(&self) { pub fn disconnect_all(&self) {
let subscribers: Vec<Arc<SubscriberInner>> = self let subscribers: Vec<Arc<SubscriberInner>> = self
.inner .inner
+43
View File
@@ -21,6 +21,12 @@ const MANAGED_FINAL_DRAIN: Duration = Duration::from_millis(100);
const IDLE_SLEEP: Duration = Duration::from_millis(1); const IDLE_SLEEP: Duration = Duration::from_millis(1);
const MAX_SLEEP: Duration = Duration::from_millis(20); 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)] #[derive(Debug)]
pub struct ManagedClient; pub struct ManagedClient;
@@ -46,26 +52,49 @@ pub struct ManagedClientSession {
#[must_use = "managed session outcomes contain completion status and counters"] #[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 {
/// Why the managed loop stopped.
pub end_reason: SessionEndReason, pub end_reason: SessionEndReason,
/// Number of echo requests sent by the client.
pub packets_sent: u64, pub packets_sent: u64,
/// Number of first in-window echo replies received.
pub replies_received: u64, pub replies_received: u64,
/// Number of duplicate reply events emitted.
pub duplicates: u64, pub duplicates: u64,
/// Number of late reply events emitted.
pub late: u64, pub late: u64,
/// Number of warning events emitted.
pub warning_events: u64, pub warning_events: u64,
} }
/// Reason a managed client session ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionEndReason { pub enum SessionEndReason {
/// The negotiated finite test duration completed and the client closed the
/// session normally.
TestComplete, TestComplete,
/// Cancellation was requested through [`ManagedClientSession::stop`] or by
/// dropping the session handle.
Cancelled, Cancelled,
/// The session was opened in [`RunMode::NoTest`](crate::RunMode) and
/// completed after negotiation.
NoTestComplete, NoTestComplete,
} }
impl ManagedClient { 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<ManagedClientSession, ClientError> { pub fn start(config: ClientConfig) -> Result<ManagedClientSession, ClientError> {
Self::start_inner(config, None).map(|(session, _)| session) 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( pub fn start_with_subscription(
config: ClientConfig, config: ClientConfig,
subscriber_config: SubscriberConfig, subscriber_config: SubscriberConfig,
@@ -114,14 +143,28 @@ impl ManagedClient {
} }
impl ManagedClientSession { 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<EventSubscription, ClientError> { pub fn subscribe(&self, config: SubscriberConfig) -> Result<EventSubscription, ClientError> {
self.hub.subscribe(config) 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) { pub fn stop(&self) {
self.cancellation.cancel(); 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<SessionOutcome, ClientError> { pub fn join(mut self) -> Result<SessionOutcome, ClientError> {
let worker = self let worker = self
.worker .worker
+8
View File
@@ -8,9 +8,16 @@ use crate::{
probe::{CompletedSet, PendingMap, TimedOutMap}, 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)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct NegotiatedParams { pub struct NegotiatedParams {
/// Server-returned protocol parameters used for the session.
pub params: Params, pub params: Params,
/// Accepted server restrictions or parameter changes.
pub restrictions: Vec<NegotiationRestriction>, pub restrictions: Vec<NegotiationRestriction>,
} }
@@ -57,6 +64,7 @@ pub enum NegotiationRestriction {
} }
impl NegotiationRestriction { impl NegotiationRestriction {
/// Return a human-readable description of this negotiated restriction.
pub fn message(&self) -> String { pub fn message(&self) -> String {
self.to_string() self.to_string()
} }
+8
View File
@@ -1,12 +1,20 @@
use std::time::{Instant, SystemTime}; 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)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ClientTimestamp { pub struct ClientTimestamp {
/// System wall-clock time.
pub wall: SystemTime, pub wall: SystemTime,
/// Monotonic timestamp sampled alongside [`wall`](Self::wall).
pub mono: Instant, pub mono: Instant,
} }
impl ClientTimestamp { impl ClientTimestamp {
/// Capture the current wall-clock and monotonic timestamps.
pub fn now() -> Self { pub fn now() -> Self {
Self { Self {
wall: SystemTime::now(), wall: SystemTime::now(),