Lochnair-Labs/irtt-rs
2
0 Fork 0
Code Issues 1 Pull Requests 4 Actions 4 Packages Projects Releases 6 Wiki Activity

refactor(managed): remove legacy runtime

Browse Source
This commit is contained in:
Nils Andreas Svee committed 2026-08-09 20:58:36 +07:00
1 parent 1a9e8f28aa
commit c599c84715
18 files changed
+36 -7847
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/irtt-cli/src/cmd/client/run.rs
+2 -2
View File
@@ -122,7 +122,7 @@ pub fn run_stream(
if is_shutdown_requested(shutdown_requested) {
interrupted = true;
if request_managed_stop_once(&mut stop_requested) {
let _ = handle.stop();
drop(handle.stop());
}
}
if request_managed_stop_for_peer_close(
@@ -132,7 +132,7 @@ pub fn run_stream(
&mut peer_close_requested_stop,
&mut stop_requested,
) {
let _ = handle.stop();
drop(handle.stop());
}
drain_events(
&mut events,
crates/irtt-cli/src/cmd/tui/run.rs
+3 -3
View File
@@ -80,7 +80,7 @@ pub fn run_tui(
if is_shutdown_requested(shutdown_requested) {
interrupted = true;
if request_managed_stop_once(&mut stop_requested) {
let _ = handle.stop();
drop(handle.stop());
}
}
if handle_input(&mut state, shutdown_requested)? {
@@ -89,7 +89,7 @@ pub fn run_tui(
if state.quit_requested {
interrupted = true;
if request_managed_stop_once(&mut stop_requested) {
let _ = handle.stop();
drop(handle.stop());
}
}
if request_managed_stop_for_peer_close(
@@ -99,7 +99,7 @@ pub fn run_tui(
&mut peer_close_requested_stop,
&mut stop_requested,
) {
let _ = handle.stop();
drop(handle.stop());
}
drain_tui_events(
&mut events,
crates/irtt-cli/src/cmd/tui/ui.rs
+2 -2
View File
@@ -2131,8 +2131,8 @@ mod tests {
&state.targets[0]
}
fn target_event(label: &str, event: ClientEvent) -> TargetEvent {
TargetEvent {
fn target_event(label: &str, event: ClientEvent) -> UiEventScope {
UiEventScope {
target: TargetId::from(label),
event,
}
crates/irtt-cli/src/shared/client/session.rs
+2 -2
View File
@@ -30,11 +30,11 @@ pub fn peer_close_run_error(
pub fn request_managed_stop_for_peer_close(
continuous: bool,
interrupted: bool,
peer_closed_target_count: u64,
peer_closed_target_outcomes: u64,
peer_close_requested_stop: &mut bool,
stop_requested: &mut bool,
) -> bool {
if !continuous || interrupted || peer_closed_target_count == 0 {
if !continuous || interrupted || peer_closed_target_outcomes == 0 {
return false;
}
*peer_close_requested_stop = true;
crates/irtt-cli/tests/client_multi_target.rs
+8 -2
View File
@@ -323,7 +323,10 @@ fn continuous_single_target_interruption_drains_final_events_and_succeeds() {
assert!(stdout
.lines()
.any(|line| line.starts_with("session_closed,")));
assert!(stderr.contains("interrupted, closing managed run"), "{stderr}");
assert!(
stderr.contains("interrupted, closing managed run"),
"{stderr}"
);
assert!(!stderr.contains("peer closure"), "{stderr}");
}
@@ -643,7 +646,10 @@ fn explicit_group_interruption_succeeds_without_peer_close_error() {
stdout.lines().any(|line| line.starts_with("b,")),
"{stdout}"
);
assert!(stderr.contains("interrupted, closing managed run"), "{stderr}");
assert!(
stderr.contains("interrupted, closing managed run"),
"{stderr}"
);
assert!(!stderr.contains("peer closure"), "{stderr}");
}
crates/irtt-client/src/client.rs
+7 -26
View File
@@ -31,7 +31,10 @@ use schedule::{instant_abs_diff, ProbeSchedule};
#[derive(Debug, Clone, Copy)]
enum ProbeScheduleMode {
CallerPaced,
Managed { scheduled_at: Instant },
#[cfg(test)]
Managed {
scheduled_at: Instant,
},
}
#[derive(Debug)]
@@ -250,23 +253,6 @@ impl Client {
self.recv_once_inner()
}
pub(crate) fn recv_once_until(
&mut self,
deadline: Option<Instant>,
max_wait: Duration,
) -> Result<Vec<ClientEvent>, ClientError> {
let Some(timeout) = bounded_receive_timeout(
deadline,
self.runtime.config().socket_config.recv_timeout,
max_wait,
Instant::now(),
) else {
return Ok(Vec::new());
};
self.socket.set_read_timeout(Some(timeout))?;
self.recv_once_inner()
}
fn recv_once_inner(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
let datagram = match recv_datagram(&self.socket, &mut self.recv_buffer) {
Ok(datagram) => datagram,
@@ -353,14 +339,7 @@ impl Client {
self.runtime.is_peer_closed()
}
pub(crate) fn has_timed_out_metadata(&self) -> bool {
self.runtime.has_timed_out_metadata()
}
pub(crate) fn packets_sent(&self) -> u64 {
self.runtime.packets_sent()
}
#[cfg(test)]
pub(crate) fn send_managed_probe(
&mut self,
scheduled_at: Instant,
@@ -400,6 +379,7 @@ impl Client {
ProbeScheduleMode::CallerPaced => {
schedule.preflight_caller_commit(machine_preflight.next_packets_sent)?
}
#[cfg(test)]
ProbeScheduleMode::Managed { scheduled_at } => {
schedule.preflight_managed_commit(scheduled_at, permission_at)?
}
@@ -763,6 +743,7 @@ impl Client {
}
}
#[cfg(test)]
fn bounded_receive_timeout(
deadline: Option<Instant>,
configured_timeout: Option<Duration>,
crates/irtt-client/src/error.rs
-23
View File
@@ -186,27 +186,4 @@ pub enum ClientError {
/// The reported reason names a value that must be changed before retrying.
#[error("invalid configuration: {reason}")]
InvalidConfig { reason: String },
/// A managed-session worker thread panicked before producing an outcome.
///
/// This is a runtime failure from joining the managed session, not a
/// protocol response from the server.
#[error("managed client worker thread panicked")]
WorkerPanicked,
}
/// Failure reported while setting up or receiving from an event subscription.
///
/// Zero-capacity subscriber queues are rejected when subscriptions are created.
/// Once a subscription exists, this error type reports delivery failures observed
/// by [`EventSubscription`](crate::EventSubscription).
#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
pub enum EventSubscriptionError {
/// The subscription can no longer receive new events.
///
/// Already queued events are delivered first. After the queue is drained,
/// `recv`, `recv_timeout`, and `try_recv` return this error when the hub
/// disconnects the subscriber, the hub is shut down, or the subscription is
/// disconnected by its overflow policy.
#[error("event subscription is disconnected")]
Disconnected,
}
crates/irtt-client/src/lib.rs
+6 -44
View File
@@ -13,39 +13,11 @@
//! also signed when the required wall-clock timestamps are available; negative
//! values usually indicate client/server clock skew.
//!
//! A managed session can drive the socket loop on a worker thread and publish
//! events through a subscription:
//!
//! ```no_run
//! use std::time::Duration;
//!
//! use irtt_client::{ClientConfig, ClientEvent, ManagedClient, SubscriberConfig};
//!
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let config = ClientConfig {
//! server_addr: "127.0.0.1:2112".to_owned(),
//! duration: Some(Duration::from_secs(10)),
//! interval: Duration::from_secs(1),
//! ..ClientConfig::default()
//! };
//!
//! let (session, events) =
//! ManagedClient::start_with_subscription(config, SubscriberConfig::default())?;
//!
//! while let Ok(event) = events.recv() {
//! match event {
//! ClientEvent::EchoReply { seq, rtt, .. } => {
//! println!("seq={seq} effective_rtt_us={}", rtt.effective.as_micros());
//! }
//! ClientEvent::SessionClosed { .. } => break,
//! _ => {}
//! }
//! }
//!
//! let _outcome = session.join()?;
//! # Ok(())
//! # }
//! ```
//! [`Client`] is the runtime-free low-level blocking adapter. With the
//! `tokio` feature, [`AsyncClient`] provides the corresponding Tokio adapter;
//! [`managed`] contains the unified managed task/handle implementation and
//! [`managed::BlockingManagedClient`] owns its dedicated runtime thread for
//! synchronous managed callers.
//!
#![cfg_attr(
not(all(target_os = "linux", feature = "ancillary")),
@@ -75,20 +47,10 @@ pub use config::{
ClientAuthConfig, ClientConfig, NegotiationPolicy, RecvBudget, RunMode, SocketConfig,
MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_TTL, MAX_UDP_PAYLOAD_LENGTH,
};
pub use error::{ClientError, EventSubscriptionError};
pub use error::ClientError;
pub use event::{
ClientEvent, OneWayDelaySample, OpenOutcome, PacketMeta, ReceivedStatsSample, RttSample,
ServerTiming, SignedDuration, WarningKind,
};
pub use managed::legacy::{
CancellationToken, EventHub, EventSubscription, ManagedClient, ManagedClientGroup,
ManagedClientGroupConfig, ManagedClientGroupSession, ManagedClientSession,
ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedGroupOutcome,
ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure,
ManagedTargetFailureKind, ManagedTargetOutcome, SessionEndReason, SessionOutcome,
SubscriberConfig, SubscriberOverflow, TargetEvent, TargetEventSubscription,
MANAGED_GROUP_OUTCOME_HISTORY_LIMIT,
};
pub use managed::TargetId;
pub use session::{NegotiatedParams, NegotiationRestriction};
pub use timing::ClientTimestamp;
crates/irtt-client/src/managed/cancellation.rs
-52
View File
@@ -1,52 +0,0 @@
use std::sync::{
atomic::{AtomicBool, Ordering},
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<AtomicBool>,
}
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)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
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());
}
}
crates/irtt-client/src/managed/group.rs
-6104
View File
File diff suppressed because it is too large. Load diff
crates/irtt-client/src/managed/hub.rs
-606
View File
@@ -1,606 +0,0 @@
use std::{
collections::{HashMap, VecDeque},
sync::{
atomic::{AtomicU64, Ordering},
Arc, Condvar, Mutex, Weak,
},
time::{Duration, Instant},
};
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,
}
impl Default for SubscriberConfig {
fn default() -> Self {
Self {
capacity: 1024,
overflow: SubscriberOverflow::DropNewest,
}
}
}
/// 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,
}
/// Publish/subscribe fan-out for managed event 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. Once
/// sealed with [`disconnect_all`](Self::disconnect_all), the hub cannot accept
/// new subscribers or publish more events.
#[derive(Debug)]
pub struct EventHub<T = ClientEvent> {
inner: Arc<HubInner<T>>,
}
#[derive(Debug)]
struct HubInner<T> {
next_id: AtomicU64,
state: Mutex<HubState<T>>,
}
#[derive(Debug)]
struct HubState<T> {
subscribers: HashMap<u64, Arc<SubscriberInner<T>>>,
sealed: bool,
}
/// 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. Use [`dropped_events`](Self::dropped_events) to
/// detect incomplete consumption. Dropping the handle unregisters the
/// subscriber.
#[must_use = "dropping the subscription unregisters it"]
#[derive(Debug)]
pub struct EventSubscription<T = ClientEvent> {
id: u64,
hub: Weak<HubInner<T>>,
inner: Arc<SubscriberInner<T>>,
}
#[derive(Debug)]
struct SubscriberInner<T> {
state: Mutex<SubscriberState<T>>,
available: Condvar,
config: SubscriberConfig,
}
#[derive(Debug)]
struct SubscriberState<T> {
queue: VecDeque<T>,
connected: bool,
dropped_events: u64,
}
impl<T> Clone for EventHub<T> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T> EventHub<T> {
/// Create an empty event hub with no subscribers.
pub fn new() -> Self {
Self {
inner: Arc::new(HubInner {
next_id: AtomicU64::new(1),
state: Mutex::new(HubState {
subscribers: HashMap::new(),
sealed: false,
}),
}),
}
}
/// Register a new subscriber with its own bounded queue.
///
/// Returns an error when `config.capacity` is zero.
pub fn subscribe(
&self,
config: SubscriberConfig,
) -> Result<EventSubscription<T>, crate::ClientError> {
if config.capacity == 0 {
return Err(crate::ClientError::InvalidConfig {
reason: "subscriber capacity must be greater than zero".to_owned(),
});
}
let mut hub_state = self.inner.state.lock().expect("event hub mutex poisoned");
if hub_state.sealed {
return Err(crate::ClientError::AlreadyClosed);
}
let id = self.inner.next_id.fetch_add(1, Ordering::Relaxed);
let inner = Arc::new(SubscriberInner {
state: Mutex::new(SubscriberState {
queue: VecDeque::with_capacity(config.capacity),
connected: true,
dropped_events: 0,
}),
available: Condvar::new(),
config,
});
hub_state.subscribers.insert(id, inner.clone());
Ok(EventSubscription {
id,
hub: Arc::downgrade(&self.inner),
inner,
})
}
/// Permanently seal the hub and disconnect all subscribers.
///
/// Already queued events remain available to drain. Future subscriptions
/// fail with [`crate::ClientError::AlreadyClosed`].
pub fn disconnect_all(&self) {
let subscribers: Vec<Arc<SubscriberInner<T>>> = {
let mut state = self.inner.state.lock().expect("event hub mutex poisoned");
state.sealed = true;
state
.subscribers
.drain()
.map(|(_, subscriber)| subscriber)
.collect()
};
for subscriber in subscribers {
subscriber.disconnect();
}
}
#[cfg(test)]
fn subscriber_count(&self) -> usize {
self.inner
.state
.lock()
.expect("event hub mutex poisoned")
.subscribers
.len()
}
}
impl<T: Clone> EventHub<T> {
/// 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: T) {
let subscribers: Vec<(u64, Arc<SubscriberInner<T>>)> = self
.inner
.state
.lock()
.expect("event hub mutex poisoned")
.subscribers
.iter()
.map(|(id, subscriber)| (*id, subscriber.clone()))
.collect();
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);
}
}
if disconnected.is_empty() {
return;
}
let mut subscribers = self.inner.state.lock().expect("event hub mutex poisoned");
for id in disconnected {
subscribers.subscribers.remove(&id);
}
}
}
impl<T> Default for EventHub<T> {
fn default() -> Self {
Self::new()
}
}
impl<T> EventSubscription<T> {
/// 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<T, EventSubscriptionError> {
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
loop {
if let Some(event) = state.queue.pop_front() {
return Ok(event);
}
if !state.connected {
return Err(EventSubscriptionError::Disconnected);
}
state = self
.inner
.available
.wait(state)
.expect("subscriber mutex poisoned");
}
}
/// Wait up to `timeout` for the next queued event.
///
/// Returns `Ok(None)` when the timeout elapses while the subscription
/// remains connected. If the subscription is disconnected, this returns
/// [`EventSubscriptionError::Disconnected`] after any already queued events
/// have been drained.
pub fn recv_timeout(&self, timeout: Duration) -> Result<Option<T>, EventSubscriptionError> {
let started_at = Instant::now();
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
loop {
if let Some(event) = state.queue.pop_front() {
return Ok(Some(event));
}
if !state.connected {
return Err(EventSubscriptionError::Disconnected);
}
let Some(wait_for) = timeout.checked_sub(started_at.elapsed()) else {
return Ok(None);
};
if wait_for.is_zero() {
return Ok(None);
}
let (next_state, _) = self
.inner
.available
.wait_timeout(state, wait_for)
.expect("subscriber mutex poisoned");
state = next_state;
}
}
/// 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<T>, EventSubscriptionError> {
let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
if let Some(event) = state.queue.pop_front() {
return Ok(Some(event));
}
if !state.connected {
return Err(EventSubscriptionError::Disconnected);
}
Ok(None)
}
/// Return the number of events discarded by this subscription's overflow
/// policy.
///
/// The count is monotonic for the lifetime of the subscription and remains
/// available after the hub is sealed or the subscription disconnects.
pub fn dropped_events(&self) -> u64 {
self.inner
.state
.lock()
.expect("subscriber mutex poisoned")
.dropped_events
}
}
impl<T> Drop for EventSubscription<T> {
fn drop(&mut self) {
self.inner.disconnect();
if let Some(hub) = self.hub.upgrade() {
hub.state
.lock()
.expect("event hub mutex poisoned")
.subscribers
.remove(&self.id);
}
}
}
impl<T> SubscriberInner<T> {
fn publish(&self, event: T) -> bool {
let mut state = self.state.lock().expect("subscriber mutex poisoned");
if !state.connected {
return false;
}
if state.queue.len() < self.config.capacity {
state.queue.push_back(event);
self.available.notify_one();
return true;
}
match self.config.overflow {
SubscriberOverflow::DropNewest => {
state.dropped_events = state.dropped_events.saturating_add(1);
true
}
SubscriberOverflow::DropOldest => {
state.queue.pop_front();
state.queue.push_back(event);
state.dropped_events = state.dropped_events.saturating_add(1);
self.available.notify_one();
true
}
SubscriberOverflow::Disconnect => {
let discarded = u64::try_from(state.queue.len())
.unwrap_or(u64::MAX)
.saturating_add(1);
state.dropped_events = state.dropped_events.saturating_add(discarded);
state.queue.clear();
state.connected = false;
self.available.notify_all();
false
}
}
}
fn disconnect(&self) {
let mut state = self.state.lock().expect("subscriber mutex poisoned");
state.connected = false;
self.available.notify_all();
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{ClientTimestamp, WarningKind};
fn event(n: usize) -> ClientEvent {
ClientEvent::Warning {
kind: WarningKind::UntrackedReply,
message: format!("event-{n}"),
at: ClientTimestamp::now(),
}
}
fn event_message(event: ClientEvent) -> String {
match event {
ClientEvent::Warning { message, .. } => message,
other => panic!("unexpected event: {other:?}"),
}
}
fn assert_next(
sub: &EventSubscription,
expected: Result<Option<&str>, EventSubscriptionError>,
) {
match (sub.try_recv(), expected) {
(Ok(Some(event)), Ok(Some(message))) => assert_eq!(event_message(event), message),
(Ok(None), Ok(None)) => {}
(Err(actual), Err(expected)) => assert_eq!(actual, expected),
(actual, expected) => panic!("expected {expected:?}, got {actual:?}"),
}
}
#[test]
fn publishes_to_multiple_subscribers() {
let hub = EventHub::new();
let a = hub.subscribe(SubscriberConfig::default()).unwrap();
let b = hub.subscribe(SubscriberConfig::default()).unwrap();
hub.publish(event(1));
assert_eq!(event_message(a.try_recv().unwrap().unwrap()), "event-1");
assert_eq!(event_message(b.try_recv().unwrap().unwrap()), "event-1");
}
#[test]
fn full_subscriber_overflow_policy_is_applied() {
let cases = [
(
SubscriberOverflow::DropNewest,
Ok(Some("event-1")),
Ok(None),
1,
),
(
SubscriberOverflow::DropOldest,
Ok(Some("event-2")),
Ok(None),
1,
),
(
SubscriberOverflow::Disconnect,
Err(EventSubscriptionError::Disconnected),
Err(EventSubscriptionError::Disconnected),
2,
),
];
for (overflow, first_expected, second_expected, expected_drops) in cases {
let hub = EventHub::new();
let sub = hub
.subscribe(SubscriberConfig {
capacity: 1,
overflow,
})
.unwrap();
hub.publish(event(1));
hub.publish(event(2));
assert_next(&sub, first_expected);
assert_next(&sub, second_expected);
assert_eq!(sub.dropped_events(), expected_drops);
}
}
#[test]
fn drop_oldest_counter_accumulates_and_survives_hub_seal() {
let hub = EventHub::new();
let sub = hub
.subscribe(SubscriberConfig {
capacity: 2,
overflow: SubscriberOverflow::DropOldest,
})
.unwrap();
for n in 1..=5 {
hub.publish(event(n));
}
hub.disconnect_all();
assert_eq!(sub.dropped_events(), 3);
assert_next(&sub, Ok(Some("event-4")));
assert_next(&sub, Ok(Some("event-5")));
assert_next(&sub, Err(EventSubscriptionError::Disconnected));
assert_eq!(sub.dropped_events(), 3);
}
#[test]
fn full_subscriber_does_not_prevent_other_delivery() {
let hub = EventHub::new();
let slow = hub
.subscribe(SubscriberConfig {
capacity: 1,
overflow: SubscriberOverflow::DropNewest,
})
.unwrap();
let fast = hub
.subscribe(SubscriberConfig {
capacity: 4,
overflow: SubscriberOverflow::DropNewest,
})
.unwrap();
hub.publish(event(1));
hub.publish(event(2));
assert_eq!(event_message(slow.try_recv().unwrap().unwrap()), "event-1");
assert_eq!(event_message(fast.try_recv().unwrap().unwrap()), "event-1");
assert_eq!(event_message(fast.try_recv().unwrap().unwrap()), "event-2");
}
#[test]
fn subscribing_after_events_does_not_replay() {
let hub = EventHub::new();
hub.publish(event(1));
let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
assert!(sub.try_recv().unwrap().is_none());
}
#[test]
fn dropping_subscription_unregisters_from_hub() {
let hub: EventHub = EventHub::new();
let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
assert_eq!(hub.subscriber_count(), 1);
drop(sub);
assert_eq!(hub.subscriber_count(), 0);
}
#[test]
fn disconnect_all_wakes_blocking_receivers() {
let hub: EventHub = EventHub::new();
let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
hub.disconnect_all();
assert_eq!(
sub.recv().unwrap_err(),
EventSubscriptionError::Disconnected
);
}
#[test]
fn timed_receive_drains_queued_events_and_observes_disconnect() {
let hub = EventHub::new();
let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
assert_eq!(sub.recv_timeout(Duration::ZERO).unwrap(), None);
hub.publish(event(1));
hub.disconnect_all();
assert_eq!(
event_message(sub.recv_timeout(Duration::from_secs(1)).unwrap().unwrap()),
"event-1"
);
assert_eq!(
sub.recv_timeout(Duration::from_secs(1)).unwrap_err(),
EventSubscriptionError::Disconnected
);
}
#[test]
fn sealed_hub_drains_queued_events_then_disconnects() {
let hub = EventHub::new();
let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
hub.publish(event(1));
hub.disconnect_all();
hub.publish(event(2));
assert_eq!(event_message(sub.recv().unwrap()), "event-1");
assert_eq!(
sub.recv().unwrap_err(),
EventSubscriptionError::Disconnected
);
}
#[test]
fn subscribing_after_hub_is_sealed_fails() {
let hub: EventHub = EventHub::new();
hub.disconnect_all();
hub.disconnect_all();
assert!(matches!(
hub.subscribe(SubscriberConfig::default()),
Err(crate::ClientError::AlreadyClosed)
));
}
#[test]
fn session_closed_event_can_be_queued() {
let hub = EventHub::new();
let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
hub.publish(ClientEvent::SessionClosed {
remote: "127.0.0.1:1".parse().unwrap(),
token: 1,
at: ClientTimestamp::now(),
});
assert!(matches!(
sub.recv().unwrap(),
ClientEvent::SessionClosed { token: 1, .. }
));
}
}
crates/irtt-client/src/managed/mod.rs
-22
View File
@@ -1,29 +1,7 @@
mod cancellation;
mod group;
mod hub;
mod runner;
mod target_id;
pub use target_id::TargetId;
/// Temporary exports for the obsolete thread-based managed implementations.
#[doc(hidden)]
pub mod legacy {
pub use super::cancellation::CancellationToken;
pub use super::group::{
ManagedClientGroup, ManagedClientGroupConfig, ManagedClientGroupSession,
ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent,
ManagedGroupOutcome, ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason,
ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetOutcome, TargetEvent,
TargetEventSubscription, MANAGED_GROUP_OUTCOME_HISTORY_LIMIT,
};
pub use super::hub::{EventHub, EventSubscription, SubscriberConfig, SubscriberOverflow};
pub use super::runner::{
ManagedClient, ManagedClientSession, SessionEndReason, SessionOutcome,
};
pub use super::TargetId;
}
#[cfg(feature = "tokio")]
mod blocking;
#[cfg(feature = "tokio")]
crates/irtt-client/src/managed/runner.rs
-883
View File
@@ -1,883 +0,0 @@
use std::{
thread::{self, JoinHandle},
time::{Duration, Instant},
};
use crate::{
config::{ClientConfig, RecvBudget},
error::ClientError,
event::{ClientEvent, OpenOutcome},
Client,
};
use super::{
cancellation::CancellationToken,
hub::{EventHub, EventSubscription, SubscriberConfig},
};
const MANAGED_RECV_TIMEOUT: Duration = Duration::from_millis(20);
const MANAGED_RECV_BUDGET: RecvBudget = RecvBudget { max_packets: 64 };
const MANAGED_FINAL_DRAIN: Duration = Duration::from_millis(100);
const IDLE_SLEEP: Duration = Duration::from_millis(1);
/// 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. Delayed workers skip missed pacing slots
/// instead of sending catch-up bursts.
#[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,
cancellation: CancellationToken,
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 {
/// 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,
/// The peer ended the session with an authenticated close-flagged reply.
PeerClosed,
/// 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<ManagedClientSession, ClientError> {
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,
) -> Result<(ManagedClientSession, EventSubscription), ClientError> {
let (session, subscription) = Self::start_inner(config, Some(subscriber_config))?;
Ok((
session,
subscription.expect("initial subscription must be present"),
))
}
fn start_inner(
mut config: ClientConfig,
subscriber_config: Option<SubscriberConfig>,
) -> Result<(ManagedClientSession, Option<EventSubscription>), ClientError> {
if config.socket_config.recv_timeout.is_none()
|| config.socket_config.recv_timeout > Some(MANAGED_RECV_TIMEOUT)
{
config.socket_config.recv_timeout = Some(MANAGED_RECV_TIMEOUT);
}
let hub = EventHub::new();
let initial_subscription = subscriber_config
.map(|config| hub.subscribe(config))
.transpose()?;
let mut client = Client::connect(config)?;
let outcome = client.open()?;
publish_open_outcome(&hub, &outcome);
let cancellation = CancellationToken::new();
let worker_hub = hub.clone();
let worker_cancellation = cancellation.clone();
let worker =
thread::spawn(move || run_client_with_cleanup(client, worker_hub, worker_cancellation));
Ok((
ManagedClientSession {
hub,
cancellation,
worker: Some(worker),
},
initial_subscription,
))
}
}
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> {
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<SessionOutcome, ClientError> {
let worker = self
.worker
.take()
.expect("ManagedClientSession invariant violated: worker handle missing before join");
match worker.join() {
Ok(outcome) => {
self.hub.disconnect_all();
outcome
}
Err(_) => {
self.hub.disconnect_all();
Err(ClientError::WorkerPanicked)
}
}
}
}
impl Drop for ManagedClientSession {
fn drop(&mut self) {
self.cancellation.cancel();
}
}
fn publish_open_outcome(hub: &EventHub, outcome: &OpenOutcome) {
match outcome {
OpenOutcome::Started { event, .. } | OpenOutcome::NoTestCompleted { event, .. } => {
hub.publish(event.clone());
}
}
}
fn run_client(
mut client: Client,
hub: EventHub,
cancellation: CancellationToken,
) -> Result<SessionOutcome, ClientError> {
if client.is_run_complete() {
return Ok(SessionOutcome {
end_reason: SessionEndReason::NoTestComplete,
packets_sent: 0,
replies_received: 0,
duplicates: 0,
late: 0,
warning_events: 0,
});
}
let mut counters = OutcomeCounters::default();
let mut cancelled = false;
loop {
if cancellation.is_cancelled() {
cancelled = true;
publish_events(
&hub,
&mut counters,
client.recv_available(MANAGED_RECV_BUDGET)?,
);
if !client.is_peer_closed() {
publish_events(&hub, &mut counters, client.poll_timeouts()?);
}
break;
}
let send_deadline = client.next_send_deadline();
if let Some(deadline) = send_deadline.filter(|deadline| *deadline <= Instant::now()) {
let events = client.send_managed_probe(deadline)?;
publish_events(&hub, &mut counters, events);
continue;
}
publish_events(
&hub,
&mut counters,
client.recv_once_until(send_deadline, MANAGED_RECV_TIMEOUT)?,
);
if client.is_peer_closed() {
break;
}
publish_events(&hub, &mut counters, client.poll_timeouts()?);
if client.is_run_complete() {
break;
}
}
if !cancelled {
drain_final_late_replies(&mut client, &hub, &mut counters)?;
}
let packets_sent = client.packets_sent();
if !client.is_peer_closed() {
let close_events = client.close()?;
publish_events(&hub, &mut counters, close_events);
}
let end_reason = if cancelled {
SessionEndReason::Cancelled
} else if client.is_peer_closed() {
SessionEndReason::PeerClosed
} else {
SessionEndReason::TestComplete
};
Ok(SessionOutcome {
end_reason,
packets_sent,
replies_received: counters.replies_received,
duplicates: counters.duplicates,
late: counters.late,
warning_events: counters.warning_events,
})
}
fn run_client_with_cleanup(
client: Client,
hub: EventHub,
cancellation: CancellationToken,
) -> Result<SessionOutcome, ClientError> {
let outcome = run_client(client, hub.clone(), cancellation);
hub.disconnect_all();
outcome
}
fn drain_final_late_replies(
client: &mut Client,
hub: &EventHub,
counters: &mut OutcomeCounters,
) -> Result<(), ClientError> {
if !client.has_timed_out_metadata() {
return Ok(());
}
let deadline = Instant::now() + MANAGED_FINAL_DRAIN;
while Instant::now() < deadline && client.has_timed_out_metadata() {
if client.is_peer_closed() {
break;
}
let mut published = false;
let events = client.recv_available(MANAGED_RECV_BUDGET)?;
published |= !events.is_empty();
publish_events(hub, counters, events);
if client.is_peer_closed() {
break;
}
let events = client.poll_timeouts()?;
published |= !events.is_empty();
publish_events(hub, counters, events);
if !published {
thread::sleep(IDLE_SLEEP);
}
}
Ok(())
}
fn publish_events(hub: &EventHub, counters: &mut OutcomeCounters, events: Vec<ClientEvent>) {
for event in events {
counters.observe(&event);
hub.publish(event);
}
}
#[derive(Debug, Default)]
struct OutcomeCounters {
replies_received: u64,
duplicates: u64,
late: u64,
warning_events: u64,
}
impl OutcomeCounters {
fn observe(&mut self, event: &ClientEvent) {
match event {
ClientEvent::EchoReply { .. } => self.replies_received += 1,
ClientEvent::DuplicateReply { .. } => self.duplicates += 1,
ClientEvent::LateReply { .. } => self.late += 1,
ClientEvent::Warning { .. } => self.warning_events += 1,
_ => {}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{config::NegotiationPolicy, EventSubscriptionError, SubscriberOverflow};
use irtt_proto::{
echo_packet_len,
flags::{self, FLAG_OPEN, FLAG_REPLY},
layout::PacketLayout,
Clock, Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION,
};
use std::{
net::{SocketAddr, UdpSocket},
sync::mpsc,
};
const TOKEN: u64 = 0x1234_5678_90ab_cdef;
struct FakeServer {
addr: SocketAddr,
done: JoinHandle<()>,
}
impl FakeServer {
fn join(self) {
self.done.join().unwrap();
}
}
fn test_echo_packet_len(hmac: bool, params: &Params) -> usize {
echo_packet_len(hmac, params)
.expect("managed runner test params must have a non-negative packet length")
}
fn test_params(duration: Option<Duration>, interval: Duration) -> Params {
Params {
protocol_version: PROTOCOL_VERSION,
duration_ns: duration.map_or(0, test_duration_ns_i64),
interval_ns: test_duration_ns_i64(interval),
length: 0,
received_stats: ReceivedStats::Both,
stamp_at: StampAt::Both,
clock: Clock::Both,
dscp: 0,
server_fill: None,
}
}
fn test_duration_ns_i64(duration: Duration) -> i64 {
i64::try_from(duration.as_nanos()).expect("test duration fits i64 nanoseconds")
}
fn config(addr: SocketAddr, duration: Option<Duration>) -> ClientConfig {
ClientConfig {
server_addr: addr.to_string(),
duration,
interval: Duration::from_millis(10),
negotiation_policy: NegotiationPolicy::Strict,
open_timeouts: vec![Duration::from_millis(200)],
probe_timeout: Duration::from_millis(40),
..ClientConfig::default()
}
}
fn start_echo_server(params: Params) -> FakeServer {
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
let addr = socket.local_addr().unwrap();
let done = thread::spawn(move || {
let (_, peer) = recv_request(&socket);
socket
.send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
.unwrap();
socket
.set_read_timeout(Some(Duration::from_millis(500)))
.unwrap();
while let Some((packet, peer)) = recv_request_timeout(&socket) {
if packet[3] & flags::FLAG_CLOSE != 0 {
break;
}
let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
let ts = TimestampFields {
recv_wall: Some(1_000_000_000),
recv_mono: Some(100_000),
send_wall: Some(1_000_000_000),
send_mono: Some(100_000),
..Default::default()
};
socket
.send_to(&echo_reply_packet(TOKEN, seq, &params, &ts), peer)
.unwrap();
}
});
FakeServer { addr, done }
}
fn start_delayed_reply_server(params: Params, delay: Duration) -> FakeServer {
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
let addr = socket.local_addr().unwrap();
let done = thread::spawn(move || {
let (_, peer) = recv_request(&socket);
socket
.send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
.unwrap();
socket
.set_read_timeout(Some(Duration::from_millis(500)))
.unwrap();
let Some((packet, peer)) = recv_request_timeout(&socket) else {
return;
};
assert_eq!(packet[3] & flags::FLAG_CLOSE, 0);
assert!(
packet.len() >= 16,
"echo request too short: {}",
packet.len()
);
let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
thread::sleep(delay);
let ts = TimestampFields {
recv_wall: Some(1_000_000_000),
recv_mono: Some(100_000),
send_wall: Some(1_001_000_000),
send_mono: Some(1_100_000),
..Default::default()
};
socket
.send_to(&echo_reply_packet(TOKEN, seq, &params, &ts), peer)
.unwrap();
while let Some((packet, _)) = recv_request_timeout(&socket) {
if packet[3] & flags::FLAG_CLOSE != 0 {
break;
}
}
});
FakeServer { addr, done }
}
fn start_peer_close_server(params: Params) -> FakeServer {
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
let addr = socket.local_addr().unwrap();
let done = thread::spawn(move || {
let (_, peer) = recv_request(&socket);
socket
.send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
.unwrap();
socket
.set_read_timeout(Some(Duration::from_millis(500)))
.unwrap();
let Some((packet, peer)) = recv_request_timeout(&socket) else {
return;
};
let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
let ts = TimestampFields {
recv_wall: Some(1_000_000_000),
recv_mono: Some(100_000),
send_wall: Some(1_000_000_000),
send_mono: Some(100_000),
..Default::default()
};
socket
.send_to(
&echo_reply_packet_with_flags(
TOKEN,
seq,
&params,
&ts,
FLAG_REPLY | flags::FLAG_CLOSE,
),
peer,
)
.unwrap();
if let Some((packet, _)) = recv_request_timeout(&socket) {
panic!(
"managed cleanup must not send any packet after peer close; flags={:?} len={}",
packet.get(3).copied(),
packet.len()
);
}
});
FakeServer { addr, done }
}
fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
let mut buf = [0_u8; 2048];
let (size, peer) = socket.recv_from(&mut buf).unwrap();
(buf[..size].to_vec(), peer)
}
fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> {
let mut buf = [0_u8; 2048];
socket
.recv_from(&mut buf)
.ok()
.map(|(size, peer)| (buf[..size].to_vec(), peer))
}
fn open_reply(flags: u8, token: u64, params: &Params) -> Vec<u8> {
let mut packet = Vec::new();
packet.extend_from_slice(&MAGIC);
packet.push(flags);
packet.extend_from_slice(&token.to_le_bytes());
packet.extend_from_slice(&params.encode());
packet
}
fn echo_reply_packet(
token: u64,
seq: u32,
params: &Params,
timestamps: &TimestampFields,
) -> Vec<u8> {
echo_reply_packet_with_flags(token, seq, params, timestamps, FLAG_REPLY)
}
fn echo_reply_packet_with_flags(
token: u64,
seq: u32,
params: &Params,
timestamps: &TimestampFields,
flags: u8,
) -> Vec<u8> {
let layout = PacketLayout::echo(false, params);
let packet_len = test_echo_packet_len(false, params);
let mut packet = Vec::with_capacity(packet_len);
packet.extend_from_slice(&MAGIC);
packet.push(flags);
packet.extend_from_slice(&token.to_le_bytes());
packet.extend_from_slice(&seq.to_le_bytes());
if layout.recv_count {
packet.extend_from_slice(&42_u32.to_le_bytes());
}
if layout.recv_window {
packet.extend_from_slice(&0x07_u64.to_le_bytes());
}
if layout.recv_wall {
packet.extend_from_slice(&timestamps.recv_wall.unwrap_or(0).to_le_bytes());
}
if layout.recv_mono {
packet.extend_from_slice(&timestamps.recv_mono.unwrap_or(0).to_le_bytes());
}
if layout.midpoint_wall {
packet.extend_from_slice(&timestamps.midpoint_wall.unwrap_or(0).to_le_bytes());
}
if layout.midpoint_mono {
packet.extend_from_slice(&timestamps.midpoint_mono.unwrap_or(0).to_le_bytes());
}
if layout.send_wall {
packet.extend_from_slice(&timestamps.send_wall.unwrap_or(0).to_le_bytes());
}
if layout.send_mono {
packet.extend_from_slice(&timestamps.send_mono.unwrap_or(0).to_le_bytes());
}
packet.resize(packet_len, 0);
packet
}
fn recv_event_with_timeout(sub: &EventSubscription) -> ClientEvent {
let deadline = Instant::now() + Duration::from_secs(2);
loop {
match sub.try_recv() {
Ok(Some(event)) => return event,
Ok(None) if Instant::now() < deadline => {
thread::sleep(Duration::from_millis(1));
}
Ok(None) => panic!("timed out waiting for managed client event"),
Err(err) => panic!("subscription ended while waiting for event: {err}"),
}
}
}
fn collect_until_closed(sub: &EventSubscription) -> Vec<ClientEvent> {
let mut events = Vec::new();
let deadline = Instant::now() + Duration::from_secs(2);
while Instant::now() < deadline {
match sub.try_recv() {
Ok(Some(event)) => {
let closed = matches!(event, ClientEvent::SessionClosed { .. });
events.push(event);
if closed {
return events;
}
}
Ok(None) => thread::sleep(Duration::from_millis(1)),
Err(_) => return events,
}
}
panic!("timed out waiting for session close event");
}
#[test]
fn stop_is_idempotent() {
let server = start_echo_server(test_params(None, Duration::from_millis(10)));
let session = ManagedClient::start(config(server.addr, None)).unwrap();
session.stop();
session.stop();
let outcome = session.join().unwrap();
server.join();
assert_eq!(outcome.end_reason, SessionEndReason::Cancelled);
}
#[test]
fn finite_managed_run_emits_session_probe_and_close_events() {
let duration = Duration::from_millis(35);
let server = start_echo_server(test_params(Some(duration), Duration::from_millis(10)));
let (session, sub) = ManagedClient::start_with_subscription(
config(server.addr, Some(duration)),
SubscriberConfig {
capacity: 16,
overflow: SubscriberOverflow::DropNewest,
},
)
.unwrap();
let events = collect_until_closed(&sub);
let outcome = session.join().unwrap();
server.join();
assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
assert!(outcome.packets_sent > 0);
assert!(outcome.replies_received > 0);
assert!(events
.iter()
.any(|event| matches!(event, ClientEvent::SessionStarted { .. })));
assert!(events
.iter()
.any(|event| matches!(event, ClientEvent::EchoReply { .. })));
assert!(events
.iter()
.any(|event| matches!(event, ClientEvent::SessionClosed { .. })));
}
#[test]
fn joining_before_consuming_preserves_queued_final_events() {
let duration = Duration::from_millis(25);
let server = start_echo_server(test_params(Some(duration), Duration::from_millis(10)));
let (session, sub) = ManagedClient::start_with_subscription(
config(server.addr, Some(duration)),
SubscriberConfig {
capacity: 32,
overflow: SubscriberOverflow::DropOldest,
},
)
.unwrap();
let outcome = session.join().unwrap();
let mut events = Vec::new();
loop {
match sub.try_recv() {
Ok(Some(event)) => events.push(event),
Ok(None) => panic!("joined managed subscription remained connected"),
Err(EventSubscriptionError::Disconnected) => break,
}
}
server.join();
assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
assert_eq!(sub.dropped_events(), 0);
assert!(matches!(
events.first(),
Some(ClientEvent::SessionStarted { .. })
));
assert!(matches!(
events.last(),
Some(ClientEvent::SessionClosed { .. })
));
}
#[test]
fn finite_managed_run_drains_late_reply_after_timeout_before_close() {
let interval = Duration::from_millis(100);
let duration = interval;
let params = test_params(Some(duration), interval);
let server = start_delayed_reply_server(params, Duration::from_millis(60));
let mut cfg = config(server.addr, Some(duration));
cfg.interval = interval;
cfg.probe_timeout = Duration::from_millis(20);
let (session, sub) = ManagedClient::start_with_subscription(
cfg,
SubscriberConfig {
capacity: 16,
overflow: SubscriberOverflow::DropNewest,
},
)
.unwrap();
let events = collect_until_closed(&sub);
let outcome = session.join().unwrap();
server.join();
assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
assert_eq!(outcome.packets_sent, 1);
assert_eq!(outcome.replies_received, 0);
assert_eq!(outcome.late, 1);
assert!(events
.iter()
.any(|event| matches!(event, ClientEvent::EchoLoss { seq: 0, .. })));
assert!(events.iter().any(|event| matches!(
event,
ClientEvent::LateReply {
seq: 0,
sent_at: Some(_),
rtt: Some(_),
..
}
)));
let late_before_close = events
.iter()
.position(|event| matches!(event, ClientEvent::LateReply { rtt: Some(_), .. }));
let close = events
.iter()
.position(|event| matches!(event, ClientEvent::SessionClosed { .. }));
let late_before_close = late_before_close.expect("missing stats-eligible LateReply");
let close = close.expect("missing SessionClosed");
assert!(late_before_close < close);
}
#[test]
fn peer_close_during_managed_run_is_successful() {
let server = start_peer_close_server(test_params(None, Duration::from_millis(10)));
let (session, sub) = ManagedClient::start_with_subscription(
config(server.addr, None),
SubscriberConfig {
capacity: 16,
overflow: SubscriberOverflow::DropNewest,
},
)
.unwrap();
let events = collect_until_closed(&sub);
let outcome = session.join().unwrap();
server.join();
assert_eq!(outcome.end_reason, SessionEndReason::PeerClosed);
assert_eq!(outcome.packets_sent, 1);
assert_eq!(outcome.replies_received, 1);
assert!(events
.iter()
.any(|event| matches!(event, ClientEvent::EchoReply { .. })));
assert!(events
.iter()
.any(|event| matches!(event, ClientEvent::SessionClosed { .. })));
}
#[test]
fn continuous_managed_run_can_be_stopped_cleanly() {
let server = start_echo_server(test_params(None, Duration::from_millis(10)));
let session = ManagedClient::start(config(server.addr, None)).unwrap();
let sub = session
.subscribe(SubscriberConfig {
capacity: 16,
overflow: SubscriberOverflow::DropNewest,
})
.unwrap();
loop {
if matches!(recv_event_with_timeout(&sub), ClientEvent::EchoReply { .. }) {
break;
}
}
session.stop();
let events = collect_until_closed(&sub);
let outcome = session.join().unwrap();
server.join();
assert_eq!(outcome.end_reason, SessionEndReason::Cancelled);
assert!(outcome.packets_sent > 0);
assert!(outcome.replies_received > 0);
assert!(events
.iter()
.any(|event| matches!(event, ClientEvent::SessionClosed { .. })));
}
#[test]
fn join_reports_worker_panic() {
let hub = EventHub::new();
let cancellation = CancellationToken::new();
let worker = thread::spawn(|| -> Result<SessionOutcome, ClientError> {
panic!("intentional managed worker panic")
});
let session = ManagedClientSession {
hub,
cancellation,
worker: Some(worker),
};
assert!(matches!(session.join(), Err(ClientError::WorkerPanicked)));
}
#[test]
fn no_test_managed_run_returns_no_test_outcome() {
use crate::RunMode;
let params = test_params(Some(Duration::from_millis(10)), Duration::from_millis(10));
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
let addr = socket.local_addr().unwrap();
let (tx, rx) = mpsc::channel();
let done = thread::spawn(move || {
let (request, peer) = recv_request(&socket);
tx.send(request[3] & flags::FLAG_CLOSE != 0).unwrap();
socket
.send_to(
&open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, &params),
peer,
)
.unwrap();
});
let mut cfg = config(addr, Some(Duration::from_millis(10)));
cfg.run_mode = RunMode::NoTest;
let session = ManagedClient::start(cfg).unwrap();
assert!(rx.recv_timeout(Duration::from_secs(2)).unwrap());
let outcome = session.join().unwrap();
done.join().unwrap();
assert_eq!(outcome.end_reason, SessionEndReason::NoTestComplete);
assert_eq!(outcome.packets_sent, 0);
}
}
crates/irtt-client/src/probe.rs
+1
View File
@@ -145,6 +145,7 @@ impl TimedOutMap {
self.insertion_order.clear();
}
#[cfg(test)]
pub fn len(&self) -> usize {
self.map.len()
}
crates/irtt-client/src/receive.rs
-32
View File
@@ -15,14 +15,6 @@ pub(crate) struct ReceivedDatagram {
pub(crate) meta: ReceiveMeta,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ReceivedDatagramFrom {
pub(crate) len: usize,
pub(crate) source: SocketAddr,
pub(crate) received_at: ClientTimestamp,
pub(crate) meta: ReceiveMeta,
}
#[cfg(not(all(target_os = "linux", feature = "ancillary")))]
pub(crate) fn recv_datagram(
socket: &UdpSocket,
@@ -38,22 +30,6 @@ pub(crate) fn recv_datagram(
})
}
#[cfg(not(all(target_os = "linux", feature = "ancillary")))]
pub(crate) fn recv_datagram_from(
socket: &UdpSocket,
buf: &mut [u8],
) -> Result<ReceivedDatagramFrom, io::Error> {
let (len, source) = socket.recv_from(buf)?;
let received_at = ClientTimestamp::now();
Ok(ReceivedDatagramFrom {
len,
source,
received_at,
meta: ReceiveMeta::default(),
})
}
#[cfg(all(target_os = "linux", feature = "ancillary"))]
pub(crate) fn recv_datagram(
socket: &UdpSocket,
@@ -88,14 +64,6 @@ pub(crate) fn try_recv_tokio_datagram(
linux::try_recv_tokio_datagram(socket, buf)
}
#[cfg(all(target_os = "linux", feature = "ancillary"))]
pub(crate) fn recv_datagram_from(
socket: &UdpSocket,
buf: &mut [u8],
) -> Result<ReceivedDatagramFrom, io::Error> {
linux::recv_datagram_from(socket, buf)
}
#[cfg(not(all(target_os = "linux", feature = "ancillary")))]
pub(crate) fn configure_receive_metadata(
_socket: &UdpSocket,
crates/irtt-client/src/receive/linux.rs
+4 -6
View File
@@ -1,17 +1,13 @@
#![allow(unsafe_code)]
use std::{
io, mem,
net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, UdpSocket},
net::{SocketAddr, UdpSocket},
os::fd::{AsRawFd, RawFd},
ptr,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use crate::{
metadata::ReceiveMeta,
receive::{ReceivedDatagram, ReceivedDatagramFrom},
timing::ClientTimestamp,
};
use crate::{metadata::ReceiveMeta, receive::ReceivedDatagram, timing::ClientTimestamp};
const CONTROL_LEN: usize = 128;
@@ -109,6 +105,7 @@ fn recv_datagram_fd(
})
}
#[cfg(any())]
pub(crate) fn recv_datagram_from(
socket: &UdpSocket,
buf: &mut [u8],
@@ -185,6 +182,7 @@ unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta {
meta
}
#[cfg(any())]
unsafe fn socket_addr_from_storage(
storage: &libc::sockaddr_storage,
len: libc::socklen_t,
crates/irtt-client/src/session/machine.rs
+1
View File
@@ -480,6 +480,7 @@ impl SessionMachine {
)
}
#[cfg(test)]
pub(crate) fn has_timed_out_metadata(&self) -> bool {
matches!(
&self.state,
crates/irtt-client/src/socket.rs
-38
View File
@@ -164,44 +164,6 @@ pub(crate) fn connect_tokio_udp_socket(
Ok(tokio::net::UdpSocket::from_std(socket)?)
}
pub(crate) fn bind_unconnected_udp_socket(
config: &SocketConfig,
family_remote: SocketAddr,
) -> Result<UdpSocket, ClientError> {
let domain = if family_remote.is_ipv4() {
Domain::IPV4
} else {
Domain::IPV6
};
let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
if config.ipv6_only && family_remote.is_ipv6() {
socket.set_only_v6(true)?;
}
let bind_addr = config.bind_addr.unwrap_or_else(|| {
if family_remote.is_ipv4() {
SocketAddr::from(([0, 0, 0, 0], 0))
} else {
SocketAddr::from(([0_u16; 8], 0))
}
});
socket.bind(&bind_addr.into())?;
let socket: UdpSocket = socket.into();
configure_receive_metadata(&socket, family_remote).map_err(|source| {
ClientError::SocketOption {
operation: "enable receive metadata",
remote: family_remote,
source,
}
})?;
if let Some(ttl) = config.ttl {
apply_ttl_to_socket(&socket, family_remote, ttl)?;
}
socket.set_read_timeout(config.recv_timeout)?;
Ok(socket)
}
#[cfg(test)]
mod tests {
use super::*;