From 41cd2a9589e0337d961c09d4865cedb8ad46d1c3 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Jul 2026 00:28:17 +0700 Subject: [PATCH] fix(managed): skip missed probe slots Advance delayed managed schedules to the latest due cadence slot and then directly to the first future deadline. Apply the same absolute-cadence behavior to the single runner and group burst/staggered pacing, while leaving the manual Client send API caller-paced. Bound receive waits by the next send deadline so 10 ms intervals are not hidden behind the managed socket timeout. Add deterministic delayed-slot, finite-end, burst, staggered, and receive-bound regressions. --- crates/irtt-client/src/client.rs | 82 +++++++++++- .../tests/probes_replies/probe_schedule.rs | 123 ++++++++++++++++++ crates/irtt-client/src/managed/group.rs | 115 ++++++++++++---- crates/irtt-client/src/managed/runner.rs | 24 +--- crates/irtt-client/src/runtime.rs | 75 +++++++++-- 5 files changed, 365 insertions(+), 54 deletions(-) diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 5e8e24e..9d2f6c7 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -163,6 +163,29 @@ impl Client { /// Returns an empty event list when the socket read would block or times /// out. Malformed or unrelated datagrams are reported as warning events. pub fn recv_once(&mut self) -> Result, ClientError> { + self.socket + .set_read_timeout(self.runtime.config().socket_config.recv_timeout)?; + self.recv_once_inner() + } + + pub(crate) fn recv_once_until( + &mut self, + deadline: Option, + max_wait: Duration, + ) -> Result, 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, ClientError> { let datagram = match recv_datagram(&self.socket, &mut self.recv_buffer) { Ok(datagram) => datagram, Err(err) @@ -186,9 +209,11 @@ impl Client { /// Receive and classify datagrams until a receive produces no events or the /// receive budget is exhausted. pub fn recv_available(&mut self, budget: RecvBudget) -> Result, ClientError> { + self.socket + .set_read_timeout(self.runtime.config().socket_config.recv_timeout)?; let mut all_events = Vec::new(); for _ in 0..budget.max_packets { - let events = self.recv_once()?; + let events = self.recv_once_inner()?; if events.is_empty() { break; } @@ -243,6 +268,13 @@ impl Client { self.runtime.packets_sent() } + pub(crate) fn send_managed_probe( + &mut self, + scheduled_at: Instant, + ) -> Result, ClientError> { + self.send_managed_probe_inner(scheduled_at, None) + } + fn send_probe_inner( &mut self, override_ts: Option, @@ -260,6 +292,26 @@ impl Client { }) }) } + + fn send_managed_probe_inner( + &mut self, + scheduled_at: Instant, + override_ts: Option, + ) -> Result, ClientError> { + let socket = &self.socket; + self.runtime + .send_probe_for_deadline(override_ts, scheduled_at, |packet| { + let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); + let send_call_start = Instant::now(); + let bytes = socket.send(packet)?; + let send_call = send_call_start.elapsed(); + Ok(SendProbeResult { + sent_at, + bytes, + send_call, + }) + }) + } } #[cfg(test)] @@ -267,6 +319,34 @@ impl Client { fn send_probe_at(&mut self, ts: ClientTimestamp) -> Result, ClientError> { self.send_probe_inner(Some(ts)) } + + fn send_managed_probe_at( + &mut self, + scheduled_at: Instant, + ts: ClientTimestamp, + ) -> Result, ClientError> { + self.send_managed_probe_inner(scheduled_at, Some(ts)) + } +} + +fn bounded_receive_timeout( + deadline: Option, + configured_timeout: Option, + max_wait: Duration, + now: Instant, +) -> Option { + if max_wait.is_zero() { + return None; + } + let mut timeout = configured_timeout.unwrap_or(max_wait).min(max_wait); + if let Some(deadline) = deadline { + let remaining = deadline.checked_duration_since(now)?; + if remaining.is_zero() { + return None; + } + timeout = timeout.min(remaining); + } + (!timeout.is_zero()).then_some(timeout) } #[cfg(test)] diff --git a/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs b/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs index 5ee1e98..a02c26c 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs @@ -101,6 +101,129 @@ fn send_probe_respects_finite_duration_exclusive_end() { server.join(); } +#[test] +fn managed_probe_skips_missed_schedule_slots() { + let interval = Duration::from_millis(10); + let params = Params { + protocol_version: 1, + duration_ns: 0, + interval_ns: 10_000_000, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::default() + }; + let server = echo_server(params); + let config = ClientConfig { + duration: None, + interval, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + + let first_deadline = client.next_send_deadline().unwrap(); + let delayed_send = ClientTimestamp { + mono: first_deadline + Duration::from_millis(45), + wall: SystemTime::now(), + }; + let events = client + .send_managed_probe_at(first_deadline, delayed_send) + .unwrap(); + + assert_eq!(events.len(), 1); + assert!(matches!( + events[0], + ClientEvent::EchoSent { + seq: 0, + scheduled_at, + sent_at, + .. + } if scheduled_at == first_deadline + Duration::from_millis(40) + && sent_at == delayed_send + )); + assert_eq!( + client.next_send_deadline(), + Some(first_deadline + Duration::from_millis(50)) + ); + + client.close().unwrap(); + server.join(); +} + +#[test] +fn managed_probe_skip_preserves_finite_run_end() { + let interval = Duration::from_millis(10); + let duration = Duration::from_millis(50); + let params = Params { + protocol_version: 1, + duration_ns: 50_000_000, + interval_ns: 10_000_000, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::default() + }; + let server = echo_server(params); + let config = ClientConfig { + duration: Some(duration), + interval, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + + let first_deadline = client.next_send_deadline().unwrap(); + let delayed_send = ClientTimestamp { + mono: first_deadline + Duration::from_millis(45), + wall: SystemTime::now(), + }; + let first = client + .send_managed_probe_at(first_deadline, delayed_send) + .unwrap(); + let second = client + .send_managed_probe_at( + first_deadline + duration, + ClientTimestamp { + mono: first_deadline + Duration::from_millis(46), + wall: SystemTime::now(), + }, + ) + .unwrap(); + + assert_eq!(first.len(), 1); + assert!(second.is_empty()); + assert!(client.next_send_deadline().is_none()); + + client.close().unwrap(); + server.join(); +} + +#[test] +fn receive_wait_is_bounded_by_next_ten_millisecond_deadline() { + let now = Instant::now(); + let deadline = now + Duration::from_millis(10); + + assert_eq!( + bounded_receive_timeout( + Some(deadline), + Some(Duration::from_millis(20)), + Duration::from_millis(20), + now, + ), + Some(Duration::from_millis(10)) + ); + assert_eq!( + bounded_receive_timeout( + Some(now), + Some(Duration::from_millis(20)), + Duration::from_millis(20), + now, + ), + None + ); +} + #[test] fn connect_rejects_invalid_probe_limits() { for config in [ diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 3734983..ec33e11 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -13,7 +13,7 @@ use crate::{ error::ClientError, event::{ClientEvent, OpenOutcome}, receive::recv_datagram_from, - runtime::{params_from_config, SendProbeResult, SessionRuntime}, + runtime::{advance_cadence, params_from_config, SendProbeResult, SessionRuntime}, socket::{bind_unconnected_udp_socket, validate_open_timeouts}, socket_options::apply_dscp_to_socket, timing::ClientTimestamp, @@ -149,6 +149,10 @@ pub struct TargetEvent { /// /// Per-session events are followed by exactly one [`TargetFinished`](Self::TargetFinished) /// event when that target incarnation reaches a terminal state. +// ClientEvent intentionally owns its structured measurement payload. EventHub +// queues are explicitly bounded, so preserving the direct event API does not +// create unbounded retention. +#[allow(clippy::large_enum_variant)] #[derive(Debug, Clone, PartialEq, Eq)] pub enum ManagedGroupEvent { /// A protocol lifecycle, measurement, or warning event from one target. @@ -589,6 +593,8 @@ struct TargetState { final_reason: Option, } +type ScheduledTarget = (Arc>, Instant); + #[derive(Debug)] enum TargetStatus { Opening { @@ -751,17 +757,17 @@ fn run_group_scheduler( let active = active_targets(&shared); pacing.reconcile(&active, now); - if pacing.send_due(&active, now, config.client.interval) { + if pacing.send_due(&active, now) { match config.pacing { ManagedGroupPacing::Staggered => { if let Some((target, scheduled_at)) = - pacing.next_staggered(&active, config.client.interval) + pacing.next_staggered(&active, config.client.interval, now)? { send_echo_to_target(&socket, &shared, target, scheduled_at); } } ManagedGroupPacing::Burst => { - if let Some(scheduled_at) = pacing.next_burst(config.client.interval) { + if let Some(scheduled_at) = pacing.next_burst(config.client.interval, now)? { for target in active { send_echo_to_target(&socket, &shared, target, scheduled_at); } @@ -1227,7 +1233,7 @@ fn send_echo_to_target( let remote = target.remote; let result = target .runtime - .send_probe_for_deadline(scheduled_at, |packet| { + .send_probe_for_deadline(None, scheduled_at, |packet| { let sent_at = ClientTimestamp::now(); let send_call_start = Instant::now(); let bytes = socket.send_to(packet, remote)?; @@ -1449,12 +1455,7 @@ impl PacingRuntime { } } - fn send_due( - &self, - active: &[Arc>], - now: Instant, - _interval: Duration, - ) -> bool { + fn send_due(&self, active: &[Arc>], now: Instant) -> bool { if active.is_empty() { return false; } @@ -1470,31 +1471,49 @@ impl PacingRuntime { &mut self, active: &[Arc>], interval: Duration, - ) -> Option<(Arc>, Instant)> { - let scheduled_at = self.next_slot_at?; + now: Instant, + ) -> Result, ClientError> { + let Some(next_slot_at) = self.next_slot_at else { + return Ok(None); + }; if active.is_empty() { self.next_slot_at = None; - return None; + return Ok(None); } + let (scheduled_at, next_slot_at, first_index) = + advance_staggered_slot(next_slot_at, interval, active.len(), self.slot_index, now)?; + self.next_slot_at = Some(next_slot_at); + for offset in 0..active.len() { - let index = (self.slot_index + offset) % active.len(); + let index = (first_index + offset) % active.len(); if !target_is_due_for_slot(&active[index], scheduled_at) { continue; } let target = active[index].clone(); self.slot_index = (index + 1) % active.len(); - self.next_slot_at = Some(scheduled_at + divide_duration(interval, active.len())); - return Some((target, scheduled_at)); + return Ok(Some((target, scheduled_at))); } - self.next_slot_at = active.iter().filter_map(target_next_send_deadline).min(); - None + self.slot_index = (first_index + 1) % active.len(); + if let Some(target_deadline) = active.iter().filter_map(target_next_send_deadline).min() { + self.next_slot_at = Some(self.next_slot_at.map_or(target_deadline, |group_deadline| { + group_deadline.min(target_deadline) + })); + } + Ok(None) } - fn next_burst(&mut self, interval: Duration) -> Option { - let scheduled_at = self.next_burst_at?; - self.next_burst_at = Some(scheduled_at + interval); - Some(scheduled_at) + fn next_burst( + &mut self, + interval: Duration, + now: Instant, + ) -> Result, ClientError> { + let Some(next_burst_at) = self.next_burst_at else { + return Ok(None); + }; + let (scheduled_at, next_burst_at, _) = advance_cadence(next_burst_at, interval, now)?; + self.next_burst_at = Some(next_burst_at); + Ok(Some(scheduled_at)) } fn next_wakeup(&self) -> Option { @@ -1505,6 +1524,23 @@ impl PacingRuntime { } } +fn advance_staggered_slot( + next_slot_at: Instant, + interval: Duration, + active_len: usize, + slot_index: usize, + now: Instant, +) -> Result<(Instant, Instant, usize), ClientError> { + debug_assert!(active_len > 0); + let spacing = divide_duration(interval, active_len); + let (scheduled_at, following_slot_at, skipped_slots) = + advance_cadence(next_slot_at, spacing, now)?; + let skipped_mod = usize::try_from(skipped_slots % active_len as u128) + .expect("staggered slot remainder fits usize"); + let first_index = (slot_index + skipped_mod) % active_len; + Ok((scheduled_at, following_slot_at, first_index)) +} + fn target_is_due_for_slot(target: &Arc>, scheduled_at: Instant) -> bool { target_next_send_deadline(target).is_some_and(|deadline| deadline <= scheduled_at) } @@ -2262,6 +2298,39 @@ mod tests { ); } + #[test] + fn delayed_burst_pacing_emits_one_current_slot_and_advances_to_future() { + let interval = Duration::from_millis(10); + let first_slot = Instant::now(); + let delayed_now = first_slot + Duration::from_millis(45); + let mut pacing = PacingRuntime::new(ManagedGroupPacing::Burst); + pacing.next_burst_at = Some(first_slot); + + let scheduled_at = pacing.next_burst(interval, delayed_now).unwrap().unwrap(); + + assert_eq!(scheduled_at, first_slot + Duration::from_millis(40)); + assert_eq!( + pacing.next_wakeup(), + Some(first_slot + Duration::from_millis(50)) + ); + assert!(pacing.next_wakeup().unwrap() > delayed_now); + } + + #[test] + fn delayed_staggered_pacing_skips_historical_slots() { + let first_slot = Instant::now(); + let delayed_now = first_slot + Duration::from_millis(35); + + let (scheduled_at, next_slot_at, target_index) = + advance_staggered_slot(first_slot, Duration::from_millis(30), 3, 0, delayed_now) + .unwrap(); + + assert_eq!(scheduled_at, first_slot + Duration::from_millis(30)); + assert_eq!(next_slot_at, first_slot + Duration::from_millis(40)); + assert_eq!(target_index, 0); + assert!(next_slot_at > delayed_now); + } + #[test] fn staggered_pacing_does_not_burst_active_targets() { let duration = Duration::from_millis(170); diff --git a/crates/irtt-client/src/managed/runner.rs b/crates/irtt-client/src/managed/runner.rs index 7c5c47d..24ad6e8 100644 --- a/crates/irtt-client/src/managed/runner.rs +++ b/crates/irtt-client/src/managed/runner.rs @@ -19,7 +19,6 @@ 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); -const MAX_SLEEP: Duration = Duration::from_millis(20); /// Entry point for running a client session on a worker thread. /// @@ -232,12 +231,9 @@ fn run_client( break; } - let now = Instant::now(); - if client - .next_send_deadline() - .is_some_and(|deadline| deadline <= now) - { - let events = client.send_probe()?; + 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; } @@ -245,7 +241,7 @@ fn run_client( publish_events( &hub, &mut counters, - client.recv_available(MANAGED_RECV_BUDGET)?, + client.recv_once_until(send_deadline, MANAGED_RECV_TIMEOUT)?, ); if client.is_peer_closed() { break; @@ -255,8 +251,6 @@ fn run_client( if client.is_run_complete() { break; } - - sleep_until_next_wakeup(client.next_send_deadline()); } if !cancelled { @@ -340,16 +334,6 @@ fn publish_events(hub: &EventHub, counters: &mut OutcomeCounters, events: Vec) { - let sleep_for = deadline - .and_then(|deadline| deadline.checked_duration_since(Instant::now())) - .map(|duration| duration.min(MAX_SLEEP)) - .unwrap_or(IDLE_SLEEP); - if sleep_for > Duration::ZERO { - thread::sleep(sleep_for); - } -} - #[derive(Debug, Default)] struct OutcomeCounters { replies_received: u64, diff --git a/crates/irtt-client/src/runtime.rs b/crates/irtt-client/src/runtime.rs index 41d819a..0640100 100644 --- a/crates/irtt-client/src/runtime.rs +++ b/crates/irtt-client/src/runtime.rs @@ -152,24 +152,26 @@ impl SessionRuntime { where F: FnOnce(&[u8]) -> Result, { - self.send_probe_inner(override_ts, None, send) + self.send_probe_inner(override_ts, None, false, send) } pub(crate) fn send_probe_for_deadline( &mut self, + override_ts: Option, scheduled_at: Instant, send: F, ) -> Result, ClientError> where F: FnOnce(&[u8]) -> Result, { - self.send_probe_inner(None, Some(scheduled_at), send) + self.send_probe_inner(override_ts, Some(scheduled_at), true, send) } fn send_probe_inner( &mut self, override_ts: Option, scheduled_at_override: Option, + skip_missed_slots: bool, send: F, ) -> Result, ClientError> where @@ -218,7 +220,6 @@ impl SessionRuntime { }; let packet = encode_echo_request(&request, self.config.hmac_key.as_deref())?; let send_result = send(&packet)?; - let timer_error = instant_abs_diff(send_result.sent_at.mono, scheduled_at); let session = self .session @@ -251,17 +252,23 @@ impl SessionRuntime { .expect("negotiated must exist when Open"); let interval_ns = u64::try_from(negotiated.params.interval_ns) .expect("validated positive negotiated interval"); + let interval = Duration::from_nanos(interval_ns); + let (scheduled_at, next_send_at) = if skip_missed_slots { + let (scheduled_at, next_send_at, _) = + advance_cadence(scheduled_at, interval, send_result.sent_at.mono)?; + (scheduled_at, next_send_at) + } else { + ( + scheduled_at, + next_probe_deadline(session.start_mono, interval_ns, session.packets_sent)?, + ) + }; + let timer_error = instant_abs_diff(send_result.sent_at.mono, scheduled_at); let session = self .session .as_mut() .expect("session must exist when phase is Open"); - session.next_send_at = if let Some(scheduled_at) = scheduled_at_override { - scheduled_at - .checked_add(Duration::from_nanos(interval_ns)) - .ok_or(ClientError::DurationOverflow)? - } else { - next_probe_deadline(session.start_mono, interval_ns, session.packets_sent)? - }; + session.next_send_at = next_send_at; if let Some(end) = session.end_mono { if session.next_send_at >= end { @@ -684,6 +691,54 @@ fn next_probe_deadline( .ok_or(ClientError::DurationOverflow) } +pub(crate) fn advance_cadence( + deadline: Instant, + interval: Duration, + now: Instant, +) -> Result<(Instant, Instant, u128), ClientError> { + if interval.is_zero() { + return Err(ClientError::InvalidConfig { + reason: "probe interval must be greater than zero".to_owned(), + }); + } + + let elapsed_slots = now + .checked_duration_since(deadline) + .map_or(0, |elapsed| elapsed.as_nanos() / interval.as_nanos()); + let scheduled_offset = duration_from_nanos( + interval + .as_nanos() + .checked_mul(elapsed_slots) + .ok_or(ClientError::DurationOverflow)?, + )?; + let next_offset = duration_from_nanos( + interval + .as_nanos() + .checked_mul( + elapsed_slots + .checked_add(1) + .ok_or(ClientError::DurationOverflow)?, + ) + .ok_or(ClientError::DurationOverflow)?, + )?; + let scheduled_at = deadline + .checked_add(scheduled_offset) + .ok_or(ClientError::DurationOverflow)?; + let next_at = deadline + .checked_add(next_offset) + .ok_or(ClientError::DurationOverflow)?; + Ok((scheduled_at, next_at, elapsed_slots)) +} + +fn duration_from_nanos(nanos: u128) -> Result { + const NANOS_PER_SECOND: u128 = 1_000_000_000; + let seconds = + u64::try_from(nanos / NANOS_PER_SECOND).map_err(|_| ClientError::DurationOverflow)?; + let subsec_nanos = u32::try_from(nanos % NANOS_PER_SECOND) + .expect("nanosecond remainder is always less than one second"); + Ok(Duration::new(seconds, subsec_nanos)) +} + pub(crate) fn sequence_is_after(candidate: u32, current: u32) -> bool { candidate != current && candidate.wrapping_sub(current) < (1 << 31) }