refactor(client): remove redundant timestamp parameter from poll_timeouts

Use internal monotonic time by default and introduce poll_timeouts_at for callers requiring a specific sampled Instant. This simplifies the API for standard usage while maintaining flexibility for external event loops.
This commit is contained in:
Nils Andreas Svee committed 2026-05-11 09:33:55 +07:00
1 parent d8b580a19c
commit 1a04f44bb5
10 files changed
+37 -31

No files matched your search

+8 -4
View File
@@ -13,7 +13,11 @@ use clap::Parser;
use irtt_cli::{ use irtt_cli::{
format_event, format_human_event_with_options, CliArgs, HumanOutputOptions, OutputMode, format_event, format_human_event_with_options, CliArgs, HumanOutputOptions, OutputMode,
}; };
use irtt_client::{Client, ClientEvent, ClientTimestamp, OpenOutcome, RecvBudget}; use irtt_client::{Client, ClientEvent, OpenOutcome, RecvBudget};
#[cfg(test)]
use irtt_client::ClientTimestamp;
#[cfg(feature = "stats")] #[cfg(feature = "stats")]
use irtt_stats::{StatsCollector, StatsConfig}; use irtt_stats::{StatsCollector, StatsConfig};
@@ -85,7 +89,7 @@ fn run(args: CliArgs, shutdown_requested: &AtomicBool) -> Result<(), Box<dyn std
let events = client.recv_available(RECV_BUDGET)?; let events = client.recv_available(RECV_BUDGET)?;
output.print_events(&events)?; output.print_events(&events)?;
let events = client.poll_timeouts(ClientTimestamp::now())?; let events = client.poll_timeouts()?;
output.print_events(&events)?; output.print_events(&events)?;
if is_shutdown_requested(shutdown_requested) { if is_shutdown_requested(shutdown_requested) {
@@ -105,7 +109,7 @@ fn run(args: CliArgs, shutdown_requested: &AtomicBool) -> Result<(), Box<dyn std
drain_final_replies(&mut client, &mut output)?; drain_final_replies(&mut client, &mut output)?;
} }
let events = client.poll_timeouts(ClientTimestamp::now())?; let events = client.poll_timeouts()?;
output.print_events(&events)?; output.print_events(&events)?;
let events = client.close()?; let events = client.close()?;
@@ -154,7 +158,7 @@ fn drain_final_replies<W: Write>(
printed |= !events.is_empty(); printed |= !events.is_empty();
output.print_events(&events)?; output.print_events(&events)?;
let events = client.poll_timeouts(ClientTimestamp::now())?; let events = client.poll_timeouts()?;
printed |= !events.is_empty(); printed |= !events.is_empty();
output.print_events(&events)?; output.print_events(&events)?;
+14 -2
View File
@@ -341,7 +341,19 @@ impl Client {
recv_buffer_size(self.config.hmac_key.is_some(), self.negotiated.as_ref()) recv_buffer_size(self.config.hmac_key.is_some(), self.negotiated.as_ref())
} }
pub fn poll_timeouts(&mut self, now: ClientTimestamp) -> Result<Vec<ClientEvent>, ClientError> { /// Polls for probes that have timed out as of the current monotonic time.
pub fn poll_timeouts(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
self.poll_timeouts_at(Instant::now())
}
/// Polls for probes that have timed out as of `now`.
///
/// This is useful for callers that drive `Client` from their own event loop and
/// want timeout decisions to use the same sampled `Instant` as their scheduling
/// logic.
///
/// `now` is monotonic time only; wall-clock time is not used for timeout expiry.
pub fn poll_timeouts_at(&mut self, now: Instant) -> Result<Vec<ClientEvent>, ClientError> {
match self.phase { match self.phase {
ClientPhase::Open { .. } => {} ClientPhase::Open { .. } => {}
ClientPhase::Closed => return Err(ClientError::AlreadyClosed), ClientPhase::Closed => return Err(ClientError::AlreadyClosed),
@@ -354,7 +366,7 @@ impl Client {
.as_mut() .as_mut()
.expect("session must exist when phase is Open"); .expect("session must exist when phase is Open");
let expired = session.pending.drain_expired(now.mono); let expired = session.pending.drain_expired(now);
let mut events = Vec::with_capacity(expired.len()); let mut events = Vec::with_capacity(expired.len());
for probe in expired { for probe in expired {
events.push(ClientEvent::EchoLoss { events.push(ClientEvent::EchoLoss {
+1 -1
View File
@@ -85,7 +85,7 @@ fn close_clears_timed_out_metadata() {
client.send_probe().unwrap(); client.send_probe().unwrap();
thread::sleep(Duration::from_millis(60)); thread::sleep(Duration::from_millis(60));
client.poll_timeouts(ClientTimestamp::now()).unwrap(); client.poll_timeouts().unwrap();
assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1); assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1);
client.close().unwrap(); client.close().unwrap();
@@ -896,11 +896,11 @@ fn poll_timeouts_emits_echo_loss() {
client.send_probe().unwrap(); client.send_probe().unwrap();
client.send_probe().unwrap(); client.send_probe().unwrap();
let no_loss = client.poll_timeouts(ClientTimestamp::now()).unwrap(); let no_loss = client.poll_timeouts().unwrap();
assert!(no_loss.is_empty()); assert!(no_loss.is_empty());
thread::sleep(Duration::from_millis(150)); thread::sleep(Duration::from_millis(150));
let events = client.poll_timeouts(ClientTimestamp::now()).unwrap(); let events = client.poll_timeouts().unwrap();
assert_eq!(events.len(), 2); assert_eq!(events.len(), 2);
for event in &events { for event in &events {
assert!(matches!(event, ClientEvent::EchoLoss { .. })); assert!(matches!(event, ClientEvent::EchoLoss { .. }));
@@ -925,7 +925,7 @@ fn poll_timeouts_removes_expired_pending() {
client.send_probe().unwrap(); client.send_probe().unwrap();
thread::sleep(Duration::from_millis(150)); thread::sleep(Duration::from_millis(150));
client.poll_timeouts(ClientTimestamp::now()).unwrap(); client.poll_timeouts().unwrap();
let session = client.session.as_ref().unwrap(); let session = client.session.as_ref().unwrap();
assert_eq!(session.pending.len(), 0); assert_eq!(session.pending.len(), 0);
@@ -971,7 +971,7 @@ fn late_reply_after_timeout_preserves_measurement_metadata() {
client.send_probe().unwrap(); client.send_probe().unwrap();
thread::sleep(Duration::from_millis(60)); thread::sleep(Duration::from_millis(60));
let losses = client.poll_timeouts(ClientTimestamp::now()).unwrap(); let losses = client.poll_timeouts().unwrap();
assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. })); assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. }));
assert_eq!(client.session.as_ref().unwrap().pending.len(), 0); assert_eq!(client.session.as_ref().unwrap().pending.len(), 0);
assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1); assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1);
@@ -1050,7 +1050,7 @@ fn late_reply_metadata_is_unavailable_without_ancillary() {
client.send_probe().unwrap(); client.send_probe().unwrap();
thread::sleep(Duration::from_millis(60)); thread::sleep(Duration::from_millis(60));
let losses = client.poll_timeouts(ClientTimestamp::now()).unwrap(); let losses = client.poll_timeouts().unwrap();
assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. })); assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. }));
let late = client.recv_once().unwrap(); let late = client.recv_once().unwrap();
@@ -1099,7 +1099,7 @@ fn late_reply_metadata_propagates_observed_dscp_with_ancillary() {
client.send_probe().unwrap(); client.send_probe().unwrap();
thread::sleep(Duration::from_millis(60)); thread::sleep(Duration::from_millis(60));
let losses = client.poll_timeouts(ClientTimestamp::now()).unwrap(); let losses = client.poll_timeouts().unwrap();
assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. })); assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. }));
let events = client.recv_once().unwrap(); let events = client.recv_once().unwrap();
@@ -1617,7 +1617,7 @@ fn process_received_packet_uses_supplied_receive_metadata_for_late_reply() {
client.send_probe_at(ClientTimestamp::now()).unwrap(); client.send_probe_at(ClientTimestamp::now()).unwrap();
thread::sleep(Duration::from_millis(60)); thread::sleep(Duration::from_millis(60));
let losses = client.poll_timeouts(ClientTimestamp::now()).unwrap(); let losses = client.poll_timeouts().unwrap();
assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. })); assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. }));
let recv_ts = ClientTimestamp::now(); let recv_ts = ClientTimestamp::now();
+3 -12
View File
@@ -7,7 +7,6 @@ use crate::{
config::{ClientConfig, RecvBudget}, config::{ClientConfig, RecvBudget},
error::ClientError, error::ClientError,
event::{ClientEvent, OpenOutcome}, event::{ClientEvent, OpenOutcome},
timing::ClientTimestamp,
Client, Client,
}; };
@@ -182,11 +181,7 @@ fn run_client(
&mut counters, &mut counters,
client.recv_available(MANAGED_RECV_BUDGET)?, client.recv_available(MANAGED_RECV_BUDGET)?,
); );
publish_events( publish_events(&hub, &mut counters, client.poll_timeouts()?);
&hub,
&mut counters,
client.poll_timeouts(ClientTimestamp::now())?,
);
break; break;
} }
@@ -205,11 +200,7 @@ fn run_client(
&mut counters, &mut counters,
client.recv_available(MANAGED_RECV_BUDGET)?, client.recv_available(MANAGED_RECV_BUDGET)?,
); );
publish_events( publish_events(&hub, &mut counters, client.poll_timeouts()?);
&hub,
&mut counters,
client.poll_timeouts(ClientTimestamp::now())?,
);
if client.is_run_complete() { if client.is_run_complete() {
break; break;
@@ -267,7 +258,7 @@ fn drain_final_late_replies(
published |= !events.is_empty(); published |= !events.is_empty();
publish_events(hub, counters, events); publish_events(hub, counters, events);
let events = client.poll_timeouts(ClientTimestamp::now())?; let events = client.poll_timeouts()?;
published |= !events.is_empty(); published |= !events.is_empty();
publish_events(hub, counters, events); publish_events(hub, counters, events);
+1 -1
View File
@@ -3,7 +3,7 @@ mod support;
use std::time::Duration; use std::time::Duration;
use irtt_client::{ use irtt_client::{
Client, ClientConfig, ClientError, ClientEvent, ClientTimestamp, OpenOutcome, SocketConfig, Client, ClientConfig, ClientError, ClientEvent, OpenOutcome, SocketConfig,
WarningKind, WarningKind,
}; };
use irtt_proto::{echo_packet_len, ProtoError, TimestampFields}; use irtt_proto::{echo_packet_len, ProtoError, TimestampFields};
+1 -1
View File
@@ -1,6 +1,6 @@
mod support; mod support;
use irtt_client::{Client, ClientConfig, ClientEvent, ClientTimestamp, NegotiationPolicy}; use irtt_client::{Client, ClientConfig, ClientEvent, NegotiationPolicy};
use irtt_proto::{echo_header_len, echo_packet_len, Params, TimestampFields}; use irtt_proto::{echo_header_len, echo_packet_len, Params, TimestampFields};
use support::{config_for_params, default_params, run_one_probe, BackendPeer, ServerObservation}; use support::{config_for_params, default_params, run_one_probe, BackendPeer, ServerObservation};
@@ -1,6 +1,5 @@
mod support; mod support;
use irtt_client::ClientTimestamp;
use std::time::Duration; use std::time::Duration;
use irtt_client::{ use irtt_client::{
+1 -1
View File
@@ -1,7 +1,7 @@
mod support; mod support;
use irtt_client::{ use irtt_client::{
Client, ClientConfig, ClientError, ClientEvent, ClientTimestamp, NegotiationPolicy, Client, ClientConfig, ClientError, ClientEvent, NegotiationPolicy,
}; };
use irtt_proto::{echo_packet_len, Params, TimestampFields}; use irtt_proto::{echo_packet_len, Params, TimestampFields};
+1 -1
View File
@@ -11,7 +11,7 @@ use std::{
}; };
use irtt_client::{ use irtt_client::{
Client, ClientConfig, ClientEvent, ClientTimestamp, NegotiatedParams, OpenOutcome, SocketConfig, Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, SocketConfig,
}; };
use irtt_proto::{ use irtt_proto::{
compute_hmac_in_place, echo_packet_len, flags, verify_hmac, Clock, Params, ProtoError, compute_hmac_in_place, echo_packet_len, flags, verify_hmac, Clock, Params, ProtoError,