From 9769a1b49eb22d24a80d39c5245a469ded4f1750 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 01:00:50 +0700 Subject: [PATCH 01/10] refactor(client): index pending probe deadlines --- crates/irtt-client/src/probe.rs | 210 ++++++++++++++++++++-- crates/irtt-client/src/session/machine.rs | 28 +++ 2 files changed, 220 insertions(+), 18 deletions(-) diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 682bcfb..f39fa37 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -1,5 +1,5 @@ use std::{ - collections::{HashMap, HashSet, TryReserveError, VecDeque}, + collections::{BTreeSet, HashMap, HashSet, TryReserveError, VecDeque}, time::Instant, }; @@ -12,9 +12,33 @@ pub(crate) struct PendingProbe { pub timeout_at: Instant, } +#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] +struct DeadlineKey { + timeout_at: Instant, + sent_at: Instant, + wire_seq: u32, +} + +impl From<&PendingProbe> for DeadlineKey { + fn from(probe: &PendingProbe) -> Self { + Self { + timeout_at: probe.timeout_at, + sent_at: probe.sent_at.mono, + wire_seq: probe.wire_seq, + } + } +} + +#[derive(Debug)] +pub(crate) struct ExpiredBatch { + pub probes: Vec, + pub more_due: bool, +} + #[derive(Debug)] pub(crate) struct PendingMap { map: HashMap, + deadlines: BTreeSet, max_capacity: usize, } @@ -22,6 +46,7 @@ impl PendingMap { pub fn new(max_capacity: usize) -> Self { Self { map: HashMap::new(), + deadlines: BTreeSet::new(), max_capacity, } } @@ -42,43 +67,75 @@ impl PendingMap { } pub fn commit_insert(&mut self, probe: PendingProbe) { + let inserted = self.deadlines.insert(DeadlineKey::from(&probe)); + debug_assert!(inserted, "pending deadline index already contained probe"); let replaced = self.map.insert(probe.wire_seq, probe); debug_assert!(replaced.is_none(), "preflight rejected pending collision"); + self.assert_index_consistent(); } pub fn remove(&mut self, wire_seq: u32) -> Option { - self.map.remove(&wire_seq) + let probe = self.map.remove(&wire_seq)?; + let removed = self.deadlines.remove(&DeadlineKey::from(&probe)); + debug_assert!(removed, "pending deadline index lost removed probe"); + self.assert_index_consistent(); + Some(probe) } pub fn drain_expired(&mut self, now: Instant) -> Vec { - let expired_keys: Vec = self - .map - .iter() - .filter(|(_, probe)| probe.timeout_at <= now) - .map(|(key, _)| *key) - .collect(); - let mut expired = Vec::with_capacity(expired_keys.len()); - for key in expired_keys { - if let Some(probe) = self.map.remove(&key) { - expired.push(probe); + let batch = self.drain_expired_bounded(now, usize::MAX); + debug_assert!(!batch.more_due); + batch.probes + } + + pub(crate) fn drain_expired_bounded(&mut self, now: Instant, limit: usize) -> ExpiredBatch { + let mut probes = Vec::with_capacity(limit.min(self.map.len())); + while probes.len() < limit { + let Some(key) = self.deadlines.first().copied() else { + break; + }; + if key.timeout_at > now { + break; } + + let removed_key = self + .deadlines + .take(&key) + .expect("first pending deadline remains present"); + debug_assert_eq!(removed_key, key); + let probe = self + .map + .remove(&key.wire_seq) + .expect("pending deadline always has a pending probe"); + debug_assert_eq!(DeadlineKey::from(&probe), key); + probes.push(probe); } - expired.sort_by_key(|p| p.sent_at.mono); - expired + self.assert_index_consistent(); + let more_due = self + .deadlines + .first() + .is_some_and(|key| key.timeout_at <= now); + ExpiredBatch { probes, more_due } } pub fn len(&self) -> usize { self.map.len() } - #[cfg(feature = "tokio")] pub fn next_timeout_deadline(&self) -> Option { - self.map.values().map(|probe| probe.timeout_at).min() + self.deadlines.first().map(|key| key.timeout_at) } - #[cfg(feature = "tokio")] pub fn latest_timeout_deadline(&self) -> Option { - self.map.values().map(|probe| probe.timeout_at).max() + self.deadlines.last().map(|key| key.timeout_at) + } + + fn assert_index_consistent(&self) { + debug_assert_eq!( + self.deadlines.len(), + self.map.len(), + "pending deadline index cardinality diverged" + ); } #[cfg(test)] @@ -90,6 +147,11 @@ impl PendingMap { pub fn capacity(&self) -> usize { self.map.capacity() } + + #[cfg(test)] + pub fn deadline_index_len(&self) -> usize { + self.deadlines.len() + } } fn pending_allocation_failed(source: TryReserveError) -> ClientError { @@ -244,6 +306,11 @@ mod tests { } } + fn insert(map: &mut PendingMap, probe: PendingProbe) { + map.preflight_insert(probe.wire_seq).unwrap(); + map.commit_insert(probe); + } + #[test] fn pending_map_rejects_over_capacity() { let mut map = PendingMap::new(2); @@ -303,6 +370,113 @@ mod tests { )); } + #[test] + fn pending_map_indexes_deadline_extrema_and_removals() { + let mut map = PendingMap::new(4); + let now = Instant::now(); + insert(&mut map, pending(1, now + Duration::from_secs(3))); + insert(&mut map, pending(2, now + Duration::from_secs(1))); + insert(&mut map, pending(3, now + Duration::from_secs(2))); + + assert_eq!( + map.next_timeout_deadline(), + Some(now + Duration::from_secs(1)) + ); + assert_eq!( + map.latest_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!(map.deadline_index_len(), map.len()); + + assert!(map.remove(3).is_some()); + assert_eq!( + map.next_timeout_deadline(), + Some(now + Duration::from_secs(1)) + ); + assert_eq!( + map.latest_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!(map.deadline_index_len(), map.len()); + + assert!(map.remove(2).is_some()); + assert_eq!( + map.next_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!( + map.latest_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!(map.deadline_index_len(), map.len()); + + assert!(map.remove(1).is_some()); + assert_eq!(map.next_timeout_deadline(), None); + assert_eq!(map.latest_timeout_deadline(), None); + assert_eq!(map.deadline_index_len(), map.len()); + } + + #[test] + fn pending_map_bounded_expiration_matches_exhaustive_order() { + let now = Instant::now(); + let probes = [ + pending(4, now + Duration::from_secs(4)), + pending(2, now + Duration::from_secs(2)), + pending(1, now + Duration::from_secs(1)), + pending(3, now + Duration::from_secs(3)), + ]; + let mut bounded = PendingMap::new(probes.len()); + let mut exhaustive = PendingMap::new(probes.len()); + for probe in probes.iter().cloned() { + insert(&mut bounded, probe.clone()); + insert(&mut exhaustive, probe); + } + + let empty = bounded.drain_expired_bounded(now + Duration::from_secs(4), 0); + assert!(empty.probes.is_empty()); + assert!(empty.more_due); + assert_eq!(bounded.deadline_index_len(), bounded.len()); + + let mut actual = Vec::new(); + loop { + let batch = bounded.drain_expired_bounded(now + Duration::from_secs(4), 2); + actual.extend(batch.probes.into_iter().map(|probe| probe.wire_seq)); + assert_eq!(bounded.deadline_index_len(), bounded.len()); + if !batch.more_due { + break; + } + } + let expected = exhaustive + .drain_expired(now + Duration::from_secs(4)) + .into_iter() + .map(|probe| probe.wire_seq) + .collect::>(); + + assert_eq!(actual, expected); + assert_eq!(actual, vec![1, 2, 3, 4]); + assert_eq!(bounded.deadline_index_len(), 0); + } + + #[test] + fn pending_map_sequence_reuse_leaves_no_stale_deadline() { + let mut map = PendingMap::new(1); + let now = Instant::now(); + insert(&mut map, pending(7, now)); + assert!(map.remove(7).is_some()); + insert(&mut map, pending(7, now + Duration::from_secs(1))); + + assert!(map.drain_expired(now).is_empty()); + assert_eq!(map.deadline_index_len(), map.len()); + assert_eq!( + map.drain_expired(now + Duration::from_secs(1)) + .into_iter() + .map(|probe| probe.wire_seq) + .collect::>(), + vec![7] + ); + assert_eq!(map.deadline_index_len(), map.len()); + } + #[test] fn bounded_probe_tracking_evicts_oldest_entries() { let mut set = CompletedSet::new(3); diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 7d3d8f9..a180909 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -1803,6 +1803,34 @@ mod tests { assert_eq!(active(&machine).pending.len(), 0); } + #[test] + fn exhaustive_timeout_polling_returns_every_due_loss() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let now = Instant::now(); + for seq in [2, 0, 1] { + let sent_at = timestamp(now - Duration::from_secs(u64::from(seq) + 1)); + let session = active_mut(&mut machine); + session.pending.preflight_insert(seq).unwrap(); + session.pending.commit_insert(PendingProbe { + wire_seq: seq, + sent_at, + timeout_at: now, + }); + } + + let events = machine.poll_timeouts_at(now).unwrap(); + assert!(matches!( + events.as_slice(), + [ + ClientEvent::EchoLoss { seq: 2, .. }, + ClientEvent::EchoLoss { seq: 1, .. }, + ClientEvent::EchoLoss { seq: 0, .. }, + ] + )); + assert_eq!(active(&machine).pending.len(), 0); + assert_eq!(active(&machine).timed_out.len(), 3); + } + #[test] fn wrapping_sequence_from_max_to_zero_remains_valid() { let mut machine = open_machine(2, Duration::from_secs(1)); -- 2.54.0 From a67331a2974cddcee2328a8357a6cfc914db2702 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 01:05:41 +0700 Subject: [PATCH 02/10] fix(managed): bound timeout work per poll --- crates/irtt-client/src/async_client.rs | 10 +- crates/irtt-client/src/managed/task.rs | 219 ++++++++++++++++++---- crates/irtt-client/src/probe.rs | 14 +- crates/irtt-client/src/session/machine.rs | 25 ++- 4 files changed, 217 insertions(+), 51 deletions(-) diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index ecee6c3..ec6b67c 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -27,7 +27,7 @@ use crate::{ receive::try_recv_tokio_datagram, session::machine::{ recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest, - PreparedProbe, SessionMachine, MAX_OPEN_PACKET_SIZE, + PreparedProbe, SessionMachine, TimeoutBatch, MAX_OPEN_PACKET_SIZE, }, socket::{connect_tokio_udp_socket, resolve_remote_tokio, validate_open_timeouts}, socket_options::{apply_dscp_to_tokio_socket, clear_dscp_on_tokio_socket}, @@ -300,6 +300,14 @@ impl AsyncClient { self.machine.poll_timeouts_at(now) } + pub(crate) fn poll_timeouts_bounded_at( + &mut self, + now: Instant, + limit: usize, + ) -> Result { + self.machine.poll_timeouts_bounded_at(now, limit) + } + /// Send the retained close packet and commit local close exactly once. /// /// DSCP is cleared only after write readiness and immediately before the diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 88a2b95..8bb2d07 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -37,6 +37,7 @@ use super::{ const TARGET_WORK_BUDGET: usize = 128; const COMMAND_WORK_BUDGET: usize = 32; +const TIMEOUT_WORK_BUDGET: usize = 128; const POST_DEADLINE_RECEIVE_BUDGET: usize = TARGET_WORK_BUDGET; // Tokio 1.53 broadcast asserts this exact bound before rounding capacity to a // power of two; Tokio does not expose it as a public constant. @@ -573,6 +574,7 @@ pub struct ManagedClientTask { wake: Option, timer: Option>>, cursor: usize, + timeout_cursor: usize, scan_remaining: usize, send_cursor: usize, burst_remaining: usize, @@ -1046,9 +1048,11 @@ impl ManagedClientTask { let len = self.targets.len(); if len == 0 { self.cursor = 0; + self.timeout_cursor = 0; self.send_cursor = 0; } else { self.cursor %= len; + self.timeout_cursor %= len; self.send_cursor %= len; } self.scan_remaining = self.scan_remaining.min(len); @@ -1205,19 +1209,8 @@ impl ManagedClientTask { .next_probe_timeout_deadline() .is_some_and(|deadline| deadline <= now) { - match client.poll_timeouts_at(now) { - Ok(events) => self.publish_client_events(index, events), - Err(error) => { - return self.begin_open_session_failure( - index, - client, - ManagedTargetFailurePhase::Timing, - error, - now, - OpenSessionFailureCleanup::Close, - ); - } - } + self.targets[index].state = TargetState::Active { client }; + return true; } let received = match client.poll_recv(cx) { Poll::Pending => false, @@ -1277,35 +1270,26 @@ impl ManagedClientTask { return false; } + if client + .next_probe_timeout_deadline() + .is_some_and(|timeout| timeout <= now) + { + self.targets[index].state = TargetState::Draining { + client, + drain_started_at, + deadline, + primary_end, + cleanup_failure, + post_deadline_receives_remaining, + }; + return true; + } + #[cfg(test)] let injected_receive_failure = self.take_drain_receive_failure(); #[cfg(not(test))] let injected_receive_failure = None; let mut retained_state_changed = false; - if client - .next_probe_timeout_deadline() - .is_some_and(|timeout| timeout <= now) - { - match client.poll_timeouts_at(now) { - Ok(events) => { - self.publish_client_events(index, events); - retained_state_changed = true; - } - Err(error) => { - self.targets[index].counters.packets_sent = client.packets_sent(); - cleanup_failure.get_or_insert_with(|| { - classify_client_error(ManagedTargetFailurePhase::Timing, &error) - }); - return self.begin_close( - index, - client, - primary_end, - cleanup_failure, - now, - ); - } - } - } let received = match client.poll_recv(cx) { Poll::Pending => false, Poll::Ready(Ok(events)) => { @@ -1728,6 +1712,152 @@ impl ManagedClientTask { immediate || self.scan_remaining > 0 } + fn poll_timeout_pass(&mut self, now: Instant) -> bool { + let target_len = self.targets.len(); + if target_len == 0 { + return false; + } + + let mut due_targets = Vec::new(); + for offset in 0..target_len { + let index = (self.timeout_cursor + offset) % target_len; + if self.target_has_due_timeout(index, now) { + due_targets.push(index); + } + } + if due_targets.is_empty() { + return false; + } + + let mut remaining = TIMEOUT_WORK_BUDGET; + let mut position = 0; + let mut last_processed = None; + while remaining > 0 && !due_targets.is_empty() { + if position == due_targets.len() { + position = 0; + } + let index = due_targets[position]; + let step = self.poll_target_timeout(index, now); + if step.expired == 0 { + due_targets.remove(position); + continue; + } + + remaining -= step.expired; + last_processed = Some(index); + if step.more_due { + position += 1; + } else { + due_targets.remove(position); + } + } + if let Some(index) = last_processed { + self.timeout_cursor = (index + 1) % target_len; + } + + !due_targets.is_empty() + } + + fn target_has_due_timeout(&self, index: usize, now: Instant) -> bool { + match &self.targets[index].state { + TargetState::Active { client } => { + self.state != DriverState::Stopping + && self.effective_retirement(index).is_none() + && client + .next_probe_timeout_deadline() + .is_some_and(|deadline| deadline <= now) + } + TargetState::Draining { + client, + deadline: _deadline, + .. + } => { + #[cfg(test)] + let defer_work = self.drain_test_hook.defer_work_until_deadline && now < *_deadline; + #[cfg(not(test))] + let defer_work = false; + !defer_work + && client + .next_probe_timeout_deadline() + .is_some_and(|timeout| timeout <= now) + } + _ => false, + } + } + + fn poll_target_timeout(&mut self, index: usize, now: Instant) -> TimeoutStep { + let state = mem::replace(&mut self.targets[index].state, TargetState::Terminal); + match state { + TargetState::Active { mut client } => match client.poll_timeouts_bounded_at(now, 1) { + Ok(batch) => { + let expired = batch.events.len(); + self.publish_client_events(index, batch.events); + self.targets[index].state = TargetState::Active { client }; + TimeoutStep { + expired, + more_due: batch.more_due, + } + } + Err(error) => { + self.begin_open_session_failure( + index, + client, + ManagedTargetFailurePhase::Timing, + error, + now, + OpenSessionFailureCleanup::Close, + ); + TimeoutStep::NONE + } + }, + TargetState::Draining { + mut client, + drain_started_at, + deadline, + primary_end, + mut cleanup_failure, + post_deadline_receives_remaining, + } => match client.poll_timeouts_bounded_at(now, 1) { + Ok(batch) => { + let expired = batch.events.len(); + self.publish_client_events(index, batch.events); + let deadline = match self.drain_deadline(&client, drain_started_at) { + Some(candidate) => deadline.min(candidate), + None => { + cleanup_failure.get_or_insert_with(duration_overflow_failure); + self.begin_close(index, client, primary_end, cleanup_failure, now); + return TimeoutStep::NONE; + } + }; + self.targets[index].state = TargetState::Draining { + client, + drain_started_at, + deadline, + primary_end, + cleanup_failure, + post_deadline_receives_remaining, + }; + TimeoutStep { + expired, + more_due: batch.more_due, + } + } + Err(error) => { + self.targets[index].counters.packets_sent = client.packets_sent(); + cleanup_failure.get_or_insert_with(|| { + classify_client_error(ManagedTargetFailurePhase::Timing, &error) + }); + self.begin_close(index, client, primary_end, cleanup_failure, now); + TimeoutStep::NONE + } + }, + state => { + self.targets[index].state = state; + TimeoutStep::NONE + } + } + } + fn all_targets_terminal(&self) -> bool { self.targets .iter() @@ -1888,6 +2018,19 @@ enum SendResult { Failed { accepted: bool }, } +#[derive(Clone, Copy)] +struct TimeoutStep { + expired: usize, + more_due: bool, +} + +impl TimeoutStep { + const NONE: Self = Self { + expired: 0, + more_due: false, + }; +} + impl SendResult { fn accepted(self) -> bool { matches!( @@ -1922,6 +2065,7 @@ impl Future for ManagedClientTask { this.observe_stop(); } let now = Instant::now(); + immediate |= this.poll_timeout_pass(now); immediate |= this.poll_target_pass(cx, now); this.prune_undesired_terminal(); if this.state == DriverState::Running { @@ -2129,6 +2273,7 @@ fn build_task( wake: Some(arm_wake(wake_receiver)), timer: None, cursor: 0, + timeout_cursor: 0, scan_remaining: 0, send_cursor: 0, burst_remaining: 0, diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index f39fa37..f77fdb8 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -82,12 +82,6 @@ impl PendingMap { Some(probe) } - pub fn drain_expired(&mut self, now: Instant) -> Vec { - let batch = self.drain_expired_bounded(now, usize::MAX); - debug_assert!(!batch.more_due); - batch.probes - } - pub(crate) fn drain_expired_bounded(&mut self, now: Instant, limit: usize) -> ExpiredBatch { let mut probes = Vec::with_capacity(limit.min(self.map.len())); while probes.len() < limit { @@ -447,7 +441,8 @@ mod tests { } } let expected = exhaustive - .drain_expired(now + Duration::from_secs(4)) + .drain_expired_bounded(now + Duration::from_secs(4), usize::MAX) + .probes .into_iter() .map(|probe| probe.wire_seq) .collect::>(); @@ -465,10 +460,11 @@ mod tests { assert!(map.remove(7).is_some()); insert(&mut map, pending(7, now + Duration::from_secs(1))); - assert!(map.drain_expired(now).is_empty()); + assert!(map.drain_expired_bounded(now, usize::MAX).probes.is_empty()); assert_eq!(map.deadline_index_len(), map.len()); assert_eq!( - map.drain_expired(now + Duration::from_secs(1)) + map.drain_expired_bounded(now + Duration::from_secs(1), usize::MAX) + .probes .into_iter() .map(|probe| probe.wire_seq) .collect::>(), diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index a180909..cb3a812 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -157,6 +157,12 @@ pub(crate) struct ProbeCommit { next_packets_sent: u64, } +#[derive(Debug)] +pub(crate) struct TimeoutBatch { + pub events: Vec, + pub more_due: bool, +} + #[derive(Debug, Clone, Copy)] pub(crate) struct ProbeSent { pub(crate) seq: u32, @@ -408,11 +414,19 @@ impl SessionMachine { &mut self, now: Instant, ) -> Result, ClientError> { + Ok(self.poll_timeouts_bounded_at(now, usize::MAX)?.events) + } + + pub(crate) fn poll_timeouts_bounded_at( + &mut self, + now: Instant, + limit: usize, + ) -> Result { let session = self.open_session_mut()?; - let expired = session.pending.drain_expired(now); - let mut events = Vec::with_capacity(expired.len()); - for probe in expired { + let expired = session.pending.drain_expired_bounded(now, limit); + let mut events = Vec::with_capacity(expired.probes.len()); + for probe in expired.probes { events.push(ClientEvent::EchoLoss { seq: probe.wire_seq, sent_at: probe.sent_at, @@ -421,7 +435,10 @@ impl SessionMachine { session.timed_out.insert(probe); } - Ok(events) + Ok(TimeoutBatch { + events, + more_due: expired.more_due, + }) } pub(crate) fn prepare_close(&self) -> Result, ClientError> { -- 2.54.0 From 4632e310d7963329267d063c33a7e48cb5fd917d Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 01:13:44 +0700 Subject: [PATCH 03/10] test(managed): verify timeout budget fairness --- crates/irtt-client/src/async_client.rs | 5 + crates/irtt-client/src/async_client/tests.rs | 34 +- crates/irtt-client/src/managed/tests.rs | 404 ++++++++++++++++++- crates/irtt-client/src/probe.rs | 2 + crates/irtt-client/src/session/machine.rs | 16 +- 5 files changed, 457 insertions(+), 4 deletions(-) diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index ec6b67c..2ffba62 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -773,6 +773,11 @@ impl AsyncClient { self.socket.poll_recv_ready(cx) } + #[cfg(test)] + pub(crate) fn replace_pending_for_test(&mut self, probe: crate::probe::PendingProbe) { + self.machine.replace_pending_for_test(probe); + } + pub(crate) fn skip_missed_probe_slots_at(&mut self, now: Instant) -> Result<(), ClientError> { let Some(schedule) = self.schedule.as_mut() else { return Ok(()); diff --git a/crates/irtt-client/src/async_client/tests.rs b/crates/irtt-client/src/async_client/tests.rs index 6447c5a..779d84f 100644 --- a/crates/irtt-client/src/async_client/tests.rs +++ b/crates/irtt-client/src/async_client/tests.rs @@ -16,7 +16,10 @@ use irtt_proto::{ use tokio::runtime::{Builder, Runtime}; use super::*; -use crate::{socket_options::tokio_socket_traffic_class, Client, RunMode, SocketConfig}; +use crate::{ + probe::PendingProbe, socket_options::tokio_socket_traffic_class, Client, ClientTimestamp, + RunMode, SocketConfig, +}; const TOKEN: u64 = 0x1234_5678_90ab_cdef; @@ -708,6 +711,35 @@ fn probe_false_readiness_resamples_all_timestamps() { assert_eq!(server.finish().len(), 2); } +#[test] +fn public_timeout_polling_remains_exhaustive() { + let server = start_open_server(None, 0); + runtime().block_on(async { + let mut client = opened_client(server.addr, 0).await; + let now = Instant::now(); + for seq in 0..3 { + client.replace_pending_for_test(PendingProbe { + wire_seq: seq, + sent_at: ClientTimestamp { + mono: now - Duration::from_secs(2), + wall: SystemTime::UNIX_EPOCH, + }, + timeout_at: now - Duration::from_secs(1), + }); + } + + assert!(matches!( + client.poll_timeouts_at(now).unwrap().as_slice(), + [ + ClientEvent::EchoLoss { seq: 0, .. }, + ClientEvent::EchoLoss { seq: 1, .. }, + ClientEvent::EchoLoss { seq: 2, .. }, + ] + )); + }); + assert_eq!(server.finish().len(), 1); +} + #[test] fn successful_short_probe_commits_before_length_error() { let server = start_open_server(None, 0); diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index f767145..05e22ce 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -6,7 +6,7 @@ use std::{ sync::{Arc, Condvar, Mutex}, task::{Context, Poll, Waker}, thread::{self, JoinHandle}, - time::{Duration, Instant}, + time::{Duration, Instant, SystemTime}, }; use irtt_proto::{ @@ -16,7 +16,10 @@ use irtt_proto::{ use tokio::runtime::{Builder, Runtime}; use super::*; -use crate::{socket::resolution_call_counts, ClientAuthConfig, NegotiationPolicy, RunMode}; +use crate::{ + probe::PendingProbe, socket::resolution_call_counts, ClientAuthConfig, ClientTimestamp, + NegotiationPolicy, RunMode, +}; const TOKEN: u64 = 0x1234_5678_90ab_cdef; @@ -318,6 +321,71 @@ fn client_events( .collect() } +fn timeout_probe(seq: u32, sent_at: Instant, timeout_at: Instant) -> PendingProbe { + PendingProbe { + wire_seq: seq, + sent_at: ClientTimestamp { + mono: sent_at, + wall: SystemTime::UNIX_EPOCH, + }, + timeout_at, + } +} + +fn open_target_for_timeout_test(runtime: &Runtime, task: &mut ManagedClientTask, index: usize) { + let TargetState::Pending { client_config } = + mem::replace(&mut task.targets[index].state, TargetState::Terminal) + else { + panic!("timeout test target must start pending"); + }; + let client = runtime.block_on(async { + let mut client = AsyncClient::connect(client_config).await.unwrap(); + client.open().await.unwrap(); + client + }); + task.targets[index].desired = false; + task.install_target_state(index, TargetState::Active { client }); +} + +fn replace_pending_for_timeout_test( + task: &mut ManagedClientTask, + index: usize, + probe: PendingProbe, +) { + let client = match &mut task.targets[index].state { + TargetState::Active { client } | TargetState::Draining { client, .. } => client, + _ => panic!("timeout test target must be active or draining"), + }; + client.replace_pending_for_test(probe); +} + +fn poll_task_once(runtime: &Runtime, task: &mut Pin>) { + runtime.block_on(poll_fn(|cx| { + assert!(task.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + })); +} + +fn timeout_losses_for( + observations: &Mutex)>>, + id: &str, +) -> usize { + observations + .lock() + .unwrap() + .iter() + .filter(|(event, _)| { + matches!( + event, + ManagedEvent::Client { + target, + event: ClientEvent::EchoLoss { .. }, + } if target.id.as_ref() == id + ) + }) + .count() +} + fn probes(records: &[PacketRecord]) -> Vec { records .iter() @@ -722,6 +790,7 @@ fn pruning_preserves_target_cursor_progress() { ) .unwrap(); task.cursor = 5; + task.timeout_cursor = 5; task.send_cursor = 5; task.scan_remaining = 3; task.burst_remaining = 3; @@ -729,6 +798,7 @@ fn pruning_preserves_target_cursor_progress() { task.targets.truncate(2); task.rebase_target_cursors(); assert_eq!(task.cursor, 1); + assert_eq!(task.timeout_cursor, 1); assert_eq!(task.send_cursor, 1); assert_eq!(task.scan_remaining, 2); assert_eq!(task.burst_remaining, 2); @@ -737,6 +807,7 @@ fn pruning_preserves_target_cursor_progress() { task.targets.clear(); task.rebase_target_cursors(); assert_eq!(task.cursor, 0); + assert_eq!(task.timeout_cursor, 0); assert_eq!(task.send_cursor, 0); assert_eq!(task.scan_remaining, 0); assert_eq!(task.burst_remaining, 0); @@ -1653,6 +1724,335 @@ fn stagger_gate_tracks_active_membership() { second.finish(); } +#[test] +fn timeout_work_is_bounded_per_task_poll_and_eventually_drains() { + let server = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, handle) = + ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let now = Instant::now(); + for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 1).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!( + timeout_losses_for(&observations, "one"), + TIMEOUT_WORK_BUDGET + ); + assert!(matches!( + &task.targets[0].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_some() + )); + + drive_task_until(&runtime, &mut task, |task| { + matches!( + &task.targets[0].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_none() + ) + }); + assert_eq!( + timeout_losses_for(&observations, "one"), + TIMEOUT_WORK_BUDGET + 1 + ); + + drop(handle.stop()); + runtime.block_on(task); + server.finish(); +} + +#[test] +fn timeout_work_round_robins_simultaneous_target_backlogs() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, handle) = ManagedClient::task( + managed, + vec![target("first", first.addr), target("second", second.addr)], + ) + .unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + open_target_for_timeout_test(&runtime, &mut task, 1); + let now = Instant::now(); + let per_target = TIMEOUT_WORK_BUDGET + 16; + for index in 0..2 { + for seq in 0..u32::try_from(per_target).unwrap() { + replace_pending_for_timeout_test( + &mut task, + index, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + } + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!( + timeout_losses_for(&observations, "first"), + TIMEOUT_WORK_BUDGET / 2 + ); + assert_eq!( + timeout_losses_for(&observations, "second"), + TIMEOUT_WORK_BUDGET / 2 + ); + assert!(matches!( + &task.targets[0].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_some() + )); + assert!(matches!( + &task.targets[1].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_some() + )); + + drive_task_until(&runtime, &mut task, |task| { + task.targets.iter().all(|target| { + matches!( + &target.state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_none() + ) + }) + }); + assert_eq!(timeout_losses_for(&observations, "first"), per_target); + assert_eq!(timeout_losses_for(&observations, "second"), per_target); + + drop(handle.stop()); + runtime.block_on(task); + first.finish(); + second.finish(); +} + +#[test] +fn partial_timeout_batch_blocks_active_receive_until_loss_commits() { + let reply_gate = Arc::new(PacketGate::default()); + let server = start_server_with_gates( + ServerBehavior::Echo, + None, + None, + None, + Some(Arc::clone(&reply_gate)), + ); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, handle) = + ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let sent = match &mut task.targets[0].state { + TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(), + _ => panic!("timeout test target did not open"), + }; + assert!(matches!( + sent.as_slice(), + [ClientEvent::EchoSent { seq: 0, .. }] + )); + wait_flag(&server.probe_seen); + + let now = Instant::now(); + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe(0, now - Duration::from_secs(1), now), + ); + for seq in 1..=u32::try_from(TIMEOUT_WORK_BUDGET).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + reply_gate.release(); + wait_flag(&server.reply_sent); + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!( + timeout_losses_for(&observations, "one"), + TIMEOUT_WORK_BUDGET + ); + assert!(!client_events(&observations).iter().any(|event| { + matches!( + event, + ClientEvent::EchoLoss { seq: 0, .. } + | ClientEvent::LateReply { seq: 0, .. } + | ClientEvent::EchoReply { seq: 0, .. } + ) + })); + + drive_task_until(&runtime, &mut task, |_| { + client_events(&observations) + .iter() + .any(|event| matches!(event, ClientEvent::LateReply { seq: 0, .. })) + }); + let classified = client_events(&observations) + .into_iter() + .filter(|event| { + matches!( + event, + ClientEvent::EchoLoss { seq: 0, .. } + | ClientEvent::LateReply { seq: 0, .. } + | ClientEvent::EchoReply { seq: 0, .. } + ) + }) + .collect::>(); + assert!(matches!( + classified.as_slice(), + [ + ClientEvent::EchoLoss { seq: 0, .. }, + ClientEvent::LateReply { seq: 0, .. } + ] + )); + + drop(handle.stop()); + runtime.block_on(task); + server.finish(); +} + +#[test] +fn partial_timeout_batch_blocks_draining_receive_until_loss_commits() { + let reply_gate = Arc::new(PacketGate::default()); + let server = start_server_with_gates( + ServerBehavior::Echo, + None, + None, + None, + Some(Arc::clone(&reply_gate)), + ); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, handle) = + ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let sent = match &mut task.targets[0].state { + TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(), + _ => panic!("timeout test target did not open"), + }; + assert!(matches!( + sent.as_slice(), + [ClientEvent::EchoSent { seq: 0, .. }] + )); + wait_flag(&server.probe_seen); + + let now = Instant::now(); + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe(0, now - Duration::from_secs(1), now), + ); + for seq in 1..=u32::try_from(TIMEOUT_WORK_BUDGET).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + let TargetState::Active { client } = + mem::replace(&mut task.targets[0].state, TargetState::Terminal) + else { + panic!("timeout test target did not remain active"); + }; + assert!(task.begin_drain(0, client, ManagedTargetEndReason::Stopped, now)); + reply_gate.release(); + wait_flag(&server.reply_sent); + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!( + timeout_losses_for(&observations, "one"), + TIMEOUT_WORK_BUDGET + ); + assert!(matches!( + &task.targets[0].state, + TargetState::Draining { + primary_end: ManagedTargetEndReason::Stopped, + cleanup_failure: None, + .. + } + )); + assert!(!client_events(&observations).iter().any(|event| { + matches!( + event, + ClientEvent::EchoLoss { seq: 0, .. } + | ClientEvent::LateReply { seq: 0, .. } + | ClientEvent::EchoReply { seq: 0, .. } + ) + })); + + drive_task_until(&runtime, &mut task, |_| { + client_events(&observations) + .iter() + .any(|event| matches!(event, ClientEvent::LateReply { seq: 0, .. })) + }); + let classified = client_events(&observations) + .into_iter() + .filter(|event| { + matches!( + event, + ClientEvent::EchoLoss { seq: 0, .. } + | ClientEvent::LateReply { seq: 0, .. } + | ClientEvent::EchoReply { seq: 0, .. } + ) + }) + .collect::>(); + assert!(matches!( + classified.as_slice(), + [ + ClientEvent::EchoLoss { seq: 0, .. }, + ClientEvent::LateReply { seq: 0, .. } + ] + )); + + drop(handle.stop()); + let outcome = runtime.block_on(task); + let target = &outcome.recent_target_outcomes[0]; + assert!(matches!(target.end_reason, ManagedTargetEndReason::Stopped)); + assert!(target.cleanup_failure.is_none()); + server.finish(); +} + #[test] fn overdue_queued_reply_is_lost_then_late() { let reply_gate = Arc::new(PacketGate::default()); diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index f77fdb8..60b73a6 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -116,10 +116,12 @@ impl PendingMap { self.map.len() } + #[cfg(any(feature = "tokio", test))] pub fn next_timeout_deadline(&self) -> Option { self.deadlines.first().map(|key| key.timeout_at) } + #[cfg(any(feature = "tokio", test))] pub fn latest_timeout_deadline(&self) -> Option { self.deadlines.last().map(|key| key.timeout_at) } diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index cb3a812..57f9528 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -414,7 +414,9 @@ impl SessionMachine { &mut self, now: Instant, ) -> Result, ClientError> { - Ok(self.poll_timeouts_bounded_at(now, usize::MAX)?.events) + let batch = self.poll_timeouts_bounded_at(now, usize::MAX)?; + debug_assert!(!batch.more_due); + Ok(batch.events) } pub(crate) fn poll_timeouts_bounded_at( @@ -1036,6 +1038,18 @@ impl SessionMachine { } } +#[cfg(all(test, feature = "tokio"))] +impl SessionMachine { + pub(crate) fn replace_pending_for_test(&mut self, probe: PendingProbe) { + let session = self + .open_session_mut() + .expect("test pending probes require an open session"); + session.pending.remove(probe.wire_seq); + session.pending.preflight_insert(probe.wire_seq).unwrap(); + session.pending.commit_insert(probe); + } +} + #[cfg(test)] mod tests { use super::*; -- 2.54.0 From f65998ff707a039462125dc0f448c3dac6b655f8 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 01:56:49 +0700 Subject: [PATCH 04/10] refactor(client): make deadline indexing allocation-free --- crates/irtt-client/src/async_client.rs | 8 + crates/irtt-client/src/managed/tests.rs | 36 ++- crates/irtt-client/src/probe.rs | 274 +++++++++++++++------- crates/irtt-client/src/session/machine.rs | 11 +- 4 files changed, 238 insertions(+), 91 deletions(-) diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index 2ffba62..f693b85 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -773,6 +773,14 @@ impl AsyncClient { self.socket.poll_recv_ready(cx) } + #[cfg(test)] + pub(crate) fn remove_pending_for_test( + &mut self, + wire_seq: u32, + ) -> Option { + self.machine.remove_pending_for_test(wire_seq) + } + #[cfg(test)] pub(crate) fn replace_pending_for_test(&mut self, probe: crate::probe::PendingProbe) { self.machine.replace_pending_for_test(probe); diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 05e22ce..076ae83 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -359,6 +359,20 @@ fn replace_pending_for_timeout_test( client.replace_pending_for_test(probe); } +fn remove_pending_for_timeout_test( + task: &mut ManagedClientTask, + index: usize, + wire_seq: u32, +) -> PendingProbe { + let client = match &mut task.targets[index].state { + TargetState::Active { client } | TargetState::Draining { client, .. } => client, + _ => panic!("timeout test target must be active or draining"), + }; + client + .remove_pending_for_test(wire_seq) + .expect("timeout test pending probe must exist") +} + fn poll_task_once(runtime: &Runtime, task: &mut Pin>) { runtime.block_on(poll_fn(|cx| { assert!(task.as_mut().poll(cx).is_pending()); @@ -1880,11 +1894,7 @@ fn partial_timeout_batch_blocks_active_receive_until_loss_commits() { wait_flag(&server.probe_seen); let now = Instant::now(); - replace_pending_for_timeout_test( - &mut task, - 0, - timeout_probe(0, now - Duration::from_secs(1), now), - ); + let _ = remove_pending_for_timeout_test(&mut task, 0, 0); for seq in 1..=u32::try_from(TIMEOUT_WORK_BUDGET).unwrap() { replace_pending_for_timeout_test( &mut task, @@ -1896,6 +1906,11 @@ fn partial_timeout_batch_blocks_active_receive_until_loss_commits() { ), ); } + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe(0, now - Duration::from_secs(1), now), + ); reply_gate.release(); wait_flag(&server.reply_sent); @@ -1973,11 +1988,7 @@ fn partial_timeout_batch_blocks_draining_receive_until_loss_commits() { wait_flag(&server.probe_seen); let now = Instant::now(); - replace_pending_for_timeout_test( - &mut task, - 0, - timeout_probe(0, now - Duration::from_secs(1), now), - ); + let _ = remove_pending_for_timeout_test(&mut task, 0, 0); for seq in 1..=u32::try_from(TIMEOUT_WORK_BUDGET).unwrap() { replace_pending_for_timeout_test( &mut task, @@ -1989,6 +2000,11 @@ fn partial_timeout_batch_blocks_draining_receive_until_loss_commits() { ), ); } + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe(0, now - Duration::from_secs(1), now), + ); let TargetState::Active { client } = mem::replace(&mut task.targets[0].state, TargetState::Terminal) else { diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 60b73a6..951fb1e 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -1,5 +1,5 @@ use std::{ - collections::{BTreeSet, HashMap, HashSet, TryReserveError, VecDeque}, + collections::{HashMap, HashSet, TryReserveError, VecDeque}, time::Instant, }; @@ -12,21 +12,11 @@ pub(crate) struct PendingProbe { pub timeout_at: Instant, } -#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] -struct DeadlineKey { - timeout_at: Instant, - sent_at: Instant, - wire_seq: u32, -} - -impl From<&PendingProbe> for DeadlineKey { - fn from(probe: &PendingProbe) -> Self { - Self { - timeout_at: probe.timeout_at, - sent_at: probe.sent_at.mono, - wire_seq: probe.wire_seq, - } - } +#[derive(Debug)] +struct PendingEntry { + probe: PendingProbe, + previous: Option, + next: Option, } #[derive(Debug)] @@ -37,8 +27,9 @@ pub(crate) struct ExpiredBatch { #[derive(Debug)] pub(crate) struct PendingMap { - map: HashMap, - deadlines: BTreeSet, + map: HashMap, + first: Option, + last: Option, max_capacity: usize, } @@ -46,7 +37,8 @@ impl PendingMap { pub fn new(max_capacity: usize) -> Self { Self { map: HashMap::new(), - deadlines: BTreeSet::new(), + first: None, + last: None, max_capacity, } } @@ -67,48 +59,98 @@ impl PendingMap { } pub fn commit_insert(&mut self, probe: PendingProbe) { - let inserted = self.deadlines.insert(DeadlineKey::from(&probe)); - debug_assert!(inserted, "pending deadline index already contained probe"); - let replaced = self.map.insert(probe.wire_seq, probe); + let wire_seq = probe.wire_seq; + let previous = self.last; + if let Some(previous) = previous { + let previous = self + .map + .get_mut(&previous) + .expect("pending list tail remains present"); + // probe_timeout is fixed for an open session and committed sends + // have nondecreasing monotonic timestamps, so deadlines append. + debug_assert!(previous.probe.timeout_at <= probe.timeout_at); + debug_assert!( + previous.next.is_none(), + "pending list tail has no successor" + ); + previous.next = Some(wire_seq); + } else { + debug_assert!(self.first.is_none(), "empty pending list has no head"); + self.first = Some(wire_seq); + } + + let replaced = self.map.insert( + wire_seq, + PendingEntry { + probe, + previous, + next: None, + }, + ); debug_assert!(replaced.is_none(), "preflight rejected pending collision"); - self.assert_index_consistent(); + self.last = Some(wire_seq); + self.assert_links_consistent(); } pub fn remove(&mut self, wire_seq: u32) -> Option { - let probe = self.map.remove(&wire_seq)?; - let removed = self.deadlines.remove(&DeadlineKey::from(&probe)); - debug_assert!(removed, "pending deadline index lost removed probe"); - self.assert_index_consistent(); + let PendingEntry { + probe, + previous, + next, + } = self.map.remove(&wire_seq)?; + + if let Some(previous) = previous { + let previous = self + .map + .get_mut(&previous) + .expect("pending list predecessor remains present"); + debug_assert_eq!(previous.next, Some(wire_seq)); + previous.next = next; + } else { + debug_assert_eq!(self.first, Some(wire_seq)); + self.first = next; + } + if let Some(next) = next { + let next = self + .map + .get_mut(&next) + .expect("pending list successor remains present"); + debug_assert_eq!(next.previous, Some(wire_seq)); + next.previous = previous; + } else { + debug_assert_eq!(self.last, Some(wire_seq)); + self.last = previous; + } + + self.assert_links_consistent(); Some(probe) } pub(crate) fn drain_expired_bounded(&mut self, now: Instant, limit: usize) -> ExpiredBatch { let mut probes = Vec::with_capacity(limit.min(self.map.len())); while probes.len() < limit { - let Some(key) = self.deadlines.first().copied() else { + let Some(wire_seq) = self.first else { break; }; - if key.timeout_at > now { + let timeout_at = self + .map + .get(&wire_seq) + .expect("pending list head remains present") + .probe + .timeout_at; + if timeout_at > now { break; } - let removed_key = self - .deadlines - .take(&key) - .expect("first pending deadline remains present"); - debug_assert_eq!(removed_key, key); let probe = self - .map - .remove(&key.wire_seq) - .expect("pending deadline always has a pending probe"); - debug_assert_eq!(DeadlineKey::from(&probe), key); + .remove(wire_seq) + .expect("pending list head remains present"); probes.push(probe); } - self.assert_index_consistent(); let more_due = self - .deadlines - .first() - .is_some_and(|key| key.timeout_at <= now); + .first + .and_then(|wire_seq| self.map.get(&wire_seq)) + .is_some_and(|entry| entry.probe.timeout_at <= now); ExpiredBatch { probes, more_due } } @@ -118,20 +160,39 @@ impl PendingMap { #[cfg(any(feature = "tokio", test))] pub fn next_timeout_deadline(&self) -> Option { - self.deadlines.first().map(|key| key.timeout_at) + self.first + .and_then(|wire_seq| self.map.get(&wire_seq)) + .map(|entry| entry.probe.timeout_at) } #[cfg(any(feature = "tokio", test))] pub fn latest_timeout_deadline(&self) -> Option { - self.deadlines.last().map(|key| key.timeout_at) + self.last + .and_then(|wire_seq| self.map.get(&wire_seq)) + .map(|entry| entry.probe.timeout_at) } - fn assert_index_consistent(&self) { - debug_assert_eq!( - self.deadlines.len(), - self.map.len(), - "pending deadline index cardinality diverged" - ); + fn assert_links_consistent(&self) { + debug_assert_eq!(self.first.is_none(), self.map.is_empty()); + debug_assert_eq!(self.last.is_none(), self.map.is_empty()); + if let Some(first) = self.first { + debug_assert_eq!( + self.map + .get(&first) + .expect("pending list head remains present") + .previous, + None + ); + } + if let Some(last) = self.last { + debug_assert_eq!( + self.map + .get(&last) + .expect("pending list tail remains present") + .next, + None + ); + } } #[cfg(test)] @@ -145,8 +206,15 @@ impl PendingMap { } #[cfg(test)] - pub fn deadline_index_len(&self) -> usize { - self.deadlines.len() + pub fn linkage(&self) -> (Option, Option) { + (self.first, self.last) + } + + #[cfg(test)] + pub fn entry_links(&self, wire_seq: u32) -> Option<(Option, Option)> { + self.map + .get(&wire_seq) + .map(|entry| (entry.previous, entry.next)) } } @@ -367,12 +435,12 @@ mod tests { } #[test] - fn pending_map_indexes_deadline_extrema_and_removals() { + fn pending_map_unlinks_head_middle_and_tail_and_updates_extrema() { let mut map = PendingMap::new(4); let now = Instant::now(); - insert(&mut map, pending(1, now + Duration::from_secs(3))); - insert(&mut map, pending(2, now + Duration::from_secs(1))); - insert(&mut map, pending(3, now + Duration::from_secs(2))); + insert(&mut map, pending(1, now + Duration::from_secs(1))); + insert(&mut map, pending(2, now + Duration::from_secs(2))); + insert(&mut map, pending(3, now + Duration::from_secs(3))); assert_eq!( map.next_timeout_deadline(), @@ -382,20 +450,25 @@ mod tests { map.latest_timeout_deadline(), Some(now + Duration::from_secs(3)) ); - assert_eq!(map.deadline_index_len(), map.len()); - - assert!(map.remove(3).is_some()); - assert_eq!( - map.next_timeout_deadline(), - Some(now + Duration::from_secs(1)) - ); - assert_eq!( - map.latest_timeout_deadline(), - Some(now + Duration::from_secs(3)) - ); - assert_eq!(map.deadline_index_len(), map.len()); + assert_eq!(map.linkage(), (Some(1), Some(3))); + assert_eq!(map.entry_links(1), Some((None, Some(2)))); + assert_eq!(map.entry_links(2), Some((Some(1), Some(3)))); + assert_eq!(map.entry_links(3), Some((Some(2), None))); assert!(map.remove(2).is_some()); + assert_eq!( + map.next_timeout_deadline(), + Some(now + Duration::from_secs(1)) + ); + assert_eq!( + map.latest_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!(map.linkage(), (Some(1), Some(3))); + assert_eq!(map.entry_links(1), Some((None, Some(3)))); + assert_eq!(map.entry_links(3), Some((Some(1), None))); + + assert!(map.remove(1).is_some()); assert_eq!( map.next_timeout_deadline(), Some(now + Duration::from_secs(3)) @@ -404,22 +477,23 @@ mod tests { map.latest_timeout_deadline(), Some(now + Duration::from_secs(3)) ); - assert_eq!(map.deadline_index_len(), map.len()); + assert_eq!(map.linkage(), (Some(3), Some(3))); + assert_eq!(map.entry_links(3), Some((None, None))); - assert!(map.remove(1).is_some()); + assert!(map.remove(3).is_some()); assert_eq!(map.next_timeout_deadline(), None); assert_eq!(map.latest_timeout_deadline(), None); - assert_eq!(map.deadline_index_len(), map.len()); + assert_eq!(map.linkage(), (None, None)); } #[test] fn pending_map_bounded_expiration_matches_exhaustive_order() { let now = Instant::now(); let probes = [ - pending(4, now + Duration::from_secs(4)), - pending(2, now + Duration::from_secs(2)), pending(1, now + Duration::from_secs(1)), + pending(2, now + Duration::from_secs(2)), pending(3, now + Duration::from_secs(3)), + pending(4, now + Duration::from_secs(4)), ]; let mut bounded = PendingMap::new(probes.len()); let mut exhaustive = PendingMap::new(probes.len()); @@ -431,13 +505,12 @@ mod tests { let empty = bounded.drain_expired_bounded(now + Duration::from_secs(4), 0); assert!(empty.probes.is_empty()); assert!(empty.more_due); - assert_eq!(bounded.deadline_index_len(), bounded.len()); + assert_eq!(bounded.linkage(), (Some(1), Some(4))); let mut actual = Vec::new(); loop { let batch = bounded.drain_expired_bounded(now + Duration::from_secs(4), 2); actual.extend(batch.probes.into_iter().map(|probe| probe.wire_seq)); - assert_eq!(bounded.deadline_index_len(), bounded.len()); if !batch.more_due { break; } @@ -451,11 +524,11 @@ mod tests { assert_eq!(actual, expected); assert_eq!(actual, vec![1, 2, 3, 4]); - assert_eq!(bounded.deadline_index_len(), 0); + assert_eq!(bounded.linkage(), (None, None)); } #[test] - fn pending_map_sequence_reuse_leaves_no_stale_deadline() { + fn pending_map_sequence_reuse_leaves_no_stale_link() { let mut map = PendingMap::new(1); let now = Instant::now(); insert(&mut map, pending(7, now)); @@ -463,7 +536,8 @@ mod tests { insert(&mut map, pending(7, now + Duration::from_secs(1))); assert!(map.drain_expired_bounded(now, usize::MAX).probes.is_empty()); - assert_eq!(map.deadline_index_len(), map.len()); + assert_eq!(map.linkage(), (Some(7), Some(7))); + assert_eq!(map.entry_links(7), Some((None, None))); assert_eq!( map.drain_expired_bounded(now + Duration::from_secs(1), usize::MAX) .probes @@ -472,7 +546,49 @@ mod tests { .collect::>(), vec![7] ); - assert_eq!(map.deadline_index_len(), map.len()); + assert_eq!(map.linkage(), (None, None)); + } + + #[test] + fn pending_map_commit_uses_only_preflighted_map_capacity() { + let mut map = PendingMap::new(2); + let now = Instant::now(); + map.preflight_insert(7).unwrap(); + let capacity = map.capacity(); + + map.commit_insert(pending(7, now)); + + assert_eq!(map.capacity(), capacity); + assert_eq!(map.len(), 1); + assert_eq!(map.linkage(), (Some(7), Some(7))); + } + + #[test] + fn pending_map_mixed_removals_preserve_links() { + let mut map = PendingMap::new(4); + let now = Instant::now(); + for seq in 1..=4 { + insert( + &mut map, + pending(seq, now + Duration::from_secs(u64::from(seq))), + ); + } + + assert!(map.remove(2).is_some()); + assert_eq!(map.entry_links(1), Some((None, Some(3)))); + assert_eq!(map.entry_links(3), Some((Some(1), Some(4)))); + let expired = map.drain_expired_bounded(now + Duration::from_secs(3), usize::MAX); + assert_eq!( + expired + .probes + .into_iter() + .map(|probe| probe.wire_seq) + .collect::>(), + vec![1, 3] + ); + assert_eq!(map.linkage(), (Some(4), Some(4))); + assert!(map.remove(4).is_some()); + assert_eq!(map.linkage(), (None, None)); } #[test] diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 57f9528..7eade8b 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -1040,6 +1040,13 @@ impl SessionMachine { #[cfg(all(test, feature = "tokio"))] impl SessionMachine { + pub(crate) fn remove_pending_for_test(&mut self, wire_seq: u32) -> Option { + self.open_session_mut() + .expect("test pending probes require an open session") + .pending + .remove(wire_seq) + } + pub(crate) fn replace_pending_for_test(&mut self, probe: PendingProbe) { let session = self .open_session_mut() @@ -1838,14 +1845,14 @@ mod tests { fn exhaustive_timeout_polling_returns_every_due_loss() { let mut machine = open_machine(4, Duration::from_secs(1)); let now = Instant::now(); - for seq in [2, 0, 1] { + for seq in [2, 1, 0] { let sent_at = timestamp(now - Duration::from_secs(u64::from(seq) + 1)); let session = active_mut(&mut machine); session.pending.preflight_insert(seq).unwrap(); session.pending.commit_insert(PendingProbe { wire_seq: seq, sent_at, - timeout_at: now, + timeout_at: sent_at.mono + Duration::from_secs(1), }); } -- 2.54.0 From ec7baebe3a6ded4cbddc8863131b9ad7ad1ea051 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 01:59:13 +0700 Subject: [PATCH 05/10] fix(managed): bound timeout discovery --- crates/irtt-client/src/managed/task.rs | 59 +++++++------------------- 1 file changed, 16 insertions(+), 43 deletions(-) diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 8bb2d07..d84d5ee 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -1579,6 +1579,14 @@ impl ManagedClientTask { self.targets[index].state = state; return SendResult::NotAttempted; }; + if client + .next_probe_timeout_deadline() + .is_some_and(|deadline| deadline <= now) + { + self.targets[index].send_waiting = false; + self.targets[index].state = TargetState::Active { client }; + return SendResult::NotAttempted; + } if client .next_send_deadline() .is_none_or(|deadline| deadline > now) @@ -1718,44 +1726,17 @@ impl ManagedClientTask { return false; } - let mut due_targets = Vec::new(); - for offset in 0..target_len { - let index = (self.timeout_cursor + offset) % target_len; + let mut more_due = false; + for _ in 0..TIMEOUT_WORK_BUDGET { + let index = self.timeout_cursor; + self.timeout_cursor = (self.timeout_cursor + 1) % target_len; if self.target_has_due_timeout(index, now) { - due_targets.push(index); + let step = self.poll_target_timeout(index, now); + more_due |= step.more_due; } } - if due_targets.is_empty() { - return false; - } - let mut remaining = TIMEOUT_WORK_BUDGET; - let mut position = 0; - let mut last_processed = None; - while remaining > 0 && !due_targets.is_empty() { - if position == due_targets.len() { - position = 0; - } - let index = due_targets[position]; - let step = self.poll_target_timeout(index, now); - if step.expired == 0 { - due_targets.remove(position); - continue; - } - - remaining -= step.expired; - last_processed = Some(index); - if step.more_due { - position += 1; - } else { - due_targets.remove(position); - } - } - if let Some(index) = last_processed { - self.timeout_cursor = (index + 1) % target_len; - } - - !due_targets.is_empty() + more_due } fn target_has_due_timeout(&self, index: usize, now: Instant) -> bool { @@ -1790,11 +1771,9 @@ impl ManagedClientTask { match state { TargetState::Active { mut client } => match client.poll_timeouts_bounded_at(now, 1) { Ok(batch) => { - let expired = batch.events.len(); self.publish_client_events(index, batch.events); self.targets[index].state = TargetState::Active { client }; TimeoutStep { - expired, more_due: batch.more_due, } } @@ -1819,7 +1798,6 @@ impl ManagedClientTask { post_deadline_receives_remaining, } => match client.poll_timeouts_bounded_at(now, 1) { Ok(batch) => { - let expired = batch.events.len(); self.publish_client_events(index, batch.events); let deadline = match self.drain_deadline(&client, drain_started_at) { Some(candidate) => deadline.min(candidate), @@ -1838,7 +1816,6 @@ impl ManagedClientTask { post_deadline_receives_remaining, }; TimeoutStep { - expired, more_due: batch.more_due, } } @@ -2020,15 +1997,11 @@ enum SendResult { #[derive(Clone, Copy)] struct TimeoutStep { - expired: usize, more_due: bool, } impl TimeoutStep { - const NONE: Self = Self { - expired: 0, - more_due: false, - }; + const NONE: Self = Self { more_due: false }; } impl SendResult { -- 2.54.0 From ab9912b3d5b208de80528767b4942b31c80ad525 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 02:06:41 +0700 Subject: [PATCH 06/10] test(managed): cover timeout backlog scheduling --- crates/irtt-client/src/managed/task.rs | 8 ++ crates/irtt-client/src/managed/tests.rs | 166 ++++++++++++++++++++++++ 2 files changed, 174 insertions(+) diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index d84d5ee..37daa88 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -591,6 +591,8 @@ pub struct ManagedClientTask { stagger_observations: Option, #[cfg(test)] drain_test_hook: DrainTestHook, + #[cfg(test)] + timeout_inspections: usize, } #[cfg(test)] @@ -1730,6 +1732,10 @@ impl ManagedClientTask { for _ in 0..TIMEOUT_WORK_BUDGET { let index = self.timeout_cursor; self.timeout_cursor = (self.timeout_cursor + 1) % target_len; + #[cfg(test)] + { + self.timeout_inspections += 1; + } if self.target_has_due_timeout(index, now) { let step = self.poll_target_timeout(index, now); more_due |= step.more_due; @@ -2263,6 +2269,8 @@ fn build_task( stagger_observations: None, #[cfg(test)] drain_test_hook: DrainTestHook::default(), + #[cfg(test)] + timeout_inspections: 0, }; let handle = ManagedClientHandle { stop, diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 076ae83..4bece8b 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -1738,6 +1738,77 @@ fn stagger_gate_tracks_active_membership() { second.finish(); } +#[test] +fn timeout_discovery_inspects_at_most_one_budget_of_targets() { + let target_count = TIMEOUT_WORK_BUDGET + 1; + let mut managed = config(ManagedPacing::Burst); + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(Instant::now())); + + assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET); + assert_eq!(task.timeout_cursor, TIMEOUT_WORK_BUDGET); +} + +#[test] +fn timeout_discovery_reaches_later_target_after_budget_slice() { + let server = start_server(ServerBehavior::Echo, None); + let target_count = TIMEOUT_WORK_BUDGET + 1; + let late_index = target_count - 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| { + if index == late_index { + target("late", server.addr) + } else { + ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9") + } + }) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[..late_index] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, late_index); + let now = Instant::now(); + replace_pending_for_timeout_test( + &mut task, + late_index, + timeout_probe(0, now - Duration::from_secs(1), now), + ); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(now)); + assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET); + assert_eq!(task.timeout_cursor, late_index); + assert_eq!(timeout_losses_for(&observations, "late"), 0); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(now)); + assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET); + assert_eq!(timeout_losses_for(&observations, "late"), 1); + + let TargetState::Active { mut client } = + mem::replace(&mut task.targets[late_index].state, TargetState::Terminal) + else { + panic!("later timeout target did not remain active"); + }; + runtime.block_on(client.close()).unwrap(); + drop(task); + server.finish(); +} + #[test] fn timeout_work_is_bounded_per_task_poll_and_eventually_drains() { let server = start_server(ServerBehavior::Echo, None); @@ -1864,6 +1935,101 @@ fn timeout_work_round_robins_simultaneous_target_backlogs() { second.finish(); } +#[test] +fn budget_deferred_timeout_blocks_send_until_timeout_processing() { + let reply_gate = Arc::new(PacketGate::default()); + let server = start_server_with_gates( + ServerBehavior::Echo, + None, + None, + None, + Some(Arc::clone(&reply_gate)), + ); + let target_count = TIMEOUT_WORK_BUDGET + 1; + let late_index = target_count - 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.client.interval = Duration::from_nanos(1); + managed.client.probe_timeout = Duration::from_secs(1); + managed.client.max_pending_probes = 1; + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| { + if index == late_index { + target("late", server.addr) + } else { + ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9") + } + }) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[..late_index] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, late_index); + task.targets[late_index].desired = true; + let sent = match &mut task.targets[late_index].state { + TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(), + _ => panic!("deferred-send target did not open"), + }; + assert!(matches!( + sent.as_slice(), + [ClientEvent::EchoSent { seq: 0, .. }] + )); + wait_flag(&server.probe_seen); + + let overdue = Instant::now(); + replace_pending_for_timeout_test( + &mut task, + late_index, + timeout_probe(0, overdue - Duration::from_secs(1), overdue), + ); + task.state = DriverState::Running; + task.lifecycle = ManagedLifecycle::Running; + task.timeout_cursor = 0; + task.send_cursor = late_index; + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET); + assert_eq!(timeout_losses_for(&observations, "late"), 0); + assert!(matches!( + &task.targets[late_index].state, + TargetState::Active { client } + if client.packets_sent() == 1 + && client.next_probe_timeout_deadline() == Some(overdue) + )); + assert!(!task.targets[late_index].send_waiting); + + poll_task_once(&runtime, &mut task); + assert_eq!(timeout_losses_for(&observations, "late"), 1); + assert!(matches!( + &task.targets[late_index].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_none() + )); + + poll_task_once(&runtime, &mut task); + assert!(matches!( + &task.targets[late_index].state, + TargetState::Active { client } if client.packets_sent() == 2 + )); + + reply_gate.release(); + let TargetState::Active { mut client } = + mem::replace(&mut task.targets[late_index].state, TargetState::Terminal) + else { + panic!("deferred-send target did not remain active"); + }; + runtime.block_on(client.close()).unwrap(); + drop(task); + server.finish(); +} + #[test] fn partial_timeout_batch_blocks_active_receive_until_loss_commits() { let reply_gate = Arc::new(PacketGate::default()); -- 2.54.0 From edb634146c410b886bd5d154ffa1b1e60709be9e Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 10:25:10 +0700 Subject: [PATCH 07/10] fix(managed): retain timeout backlog fairness --- crates/irtt-client/src/managed/task.rs | 60 +++++++++++++++++++++++--- 1 file changed, 55 insertions(+), 5 deletions(-) diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 37daa88..bab8824 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -416,6 +416,12 @@ struct TargetRuntime { state: TargetState, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct TimeoutBacklogEntry { + index: usize, + generation: u64, +} + struct PlannedTarget { target: ManagedTargetConfig, client_config: ClientConfig, @@ -575,6 +581,7 @@ pub struct ManagedClientTask { timer: Option>>, cursor: usize, timeout_cursor: usize, + timeout_backlog: VecDeque, scan_remaining: usize, send_cursor: usize, burst_remaining: usize, @@ -1728,8 +1735,11 @@ impl ManagedClientTask { return false; } - let mut more_due = false; - for _ in 0..TIMEOUT_WORK_BUDGET { + let mut transitions_remaining = TIMEOUT_WORK_BUDGET; + for _ in 0..target_len.min(TIMEOUT_WORK_BUDGET) { + if transitions_remaining == 0 { + break; + } let index = self.timeout_cursor; self.timeout_cursor = (self.timeout_cursor + 1) % target_len; #[cfg(test)] @@ -1737,12 +1747,49 @@ impl ManagedClientTask { self.timeout_inspections += 1; } if self.target_has_due_timeout(index, now) { + let generation = self.targets[index].instance.generation; let step = self.poll_target_timeout(index, now); - more_due |= step.more_due; + transitions_remaining -= 1; + if step.more_due { + self.enqueue_timeout_backlog(index, generation); + } } } - more_due + while transitions_remaining > 0 { + let Some(entry) = self.timeout_backlog.pop_front() else { + break; + }; + let valid = self + .targets + .get(entry.index) + .is_some_and(|target| target.instance.generation == entry.generation) + && self.target_has_due_timeout(entry.index, now); + if !valid { + continue; + } + + let step = self.poll_target_timeout(entry.index, now); + transitions_remaining -= 1; + if step.more_due { + self.enqueue_timeout_backlog(entry.index, entry.generation); + } + } + + !self.timeout_backlog.is_empty() + } + + fn enqueue_timeout_backlog(&mut self, index: usize, generation: u64) { + if self.timeout_backlog.len() == TIMEOUT_WORK_BUDGET + || self + .timeout_backlog + .iter() + .any(|entry| entry.index == index && entry.generation == generation) + { + return; + } + self.timeout_backlog + .push_back(TimeoutBacklogEntry { index, generation }); } fn target_has_due_timeout(&self, index: usize, now: Instant) -> bool { @@ -2085,7 +2132,9 @@ impl Future for ManagedClientTask { } immediate |= this.register_stop_wake(cx); - immediate |= this.register_timer(cx); + if !immediate { + immediate |= this.register_timer(cx); + } if immediate { cx.waker().wake_by_ref(); } @@ -2253,6 +2302,7 @@ fn build_task( timer: None, cursor: 0, timeout_cursor: 0, + timeout_backlog: VecDeque::with_capacity(TIMEOUT_WORK_BUDGET), scan_remaining: 0, send_cursor: 0, burst_remaining: 0, -- 2.54.0 From 1f69c04274782044c14c87cfa9dca3fc0263a35e Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 10:30:36 +0700 Subject: [PATCH 08/10] test(managed): verify sparse timeout backlog progress --- crates/irtt-client/src/managed/task.rs | 10 +- crates/irtt-client/src/managed/tests.rs | 375 ++++++++++++++++++++++++ 2 files changed, 384 insertions(+), 1 deletion(-) diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index bab8824..f5571a0 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -600,6 +600,8 @@ pub struct ManagedClientTask { drain_test_hook: DrainTestHook, #[cfg(test)] timeout_inspections: usize, + #[cfg(test)] + deadline_inspections: usize, } #[cfg(test)] @@ -1894,7 +1896,11 @@ impl ManagedClientTask { .all(|target| matches!(target.state, TargetState::Terminal)) } - fn next_deadline(&self) -> Option { + fn next_deadline(&mut self) -> Option { + #[cfg(test)] + { + self.deadline_inspections += self.targets.len(); + } let mut non_send_deadline = None; let mut send_deadline = None; for target in &self.targets { @@ -2321,6 +2327,8 @@ fn build_task( drain_test_hook: DrainTestHook::default(), #[cfg(test)] timeout_inspections: 0, + #[cfg(test)] + deadline_inspections: 0, }; let handle = ManagedClientHandle { stop, diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 4bece8b..dc91677 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -373,6 +373,15 @@ fn remove_pending_for_timeout_test( .expect("timeout test pending probe must exist") } +fn close_timeout_test_target(runtime: &Runtime, task: &mut ManagedClientTask, index: usize) { + let (TargetState::Active { mut client } | TargetState::Draining { mut client, .. }) = + mem::replace(&mut task.targets[index].state, TargetState::Terminal) + else { + panic!("timeout test target did not remain active or draining"); + }; + runtime.block_on(client.close()).unwrap(); +} + fn poll_task_once(runtime: &Runtime, task: &mut Pin>) { runtime.block_on(poll_fn(|cx| { assert!(task.as_mut().poll(cx).is_pending()); @@ -1755,6 +1764,372 @@ fn timeout_discovery_inspects_at_most_one_budget_of_targets() { assert_eq!(task.timeout_cursor, TIMEOUT_WORK_BUDGET); } +#[test] +fn idle_timeout_discovery_inspects_one_target_once() { + let (mut task, _) = ManagedClient::task( + config(ManagedPacing::Burst), + vec![ManagedTargetConfig::new("idle", "127.0.0.1:9")], + ) + .unwrap(); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(Instant::now())); + + assert_eq!(task.timeout_inspections, 1); + assert_eq!(task.timeout_cursor, 0); +} + +#[test] +fn idle_timeout_discovery_sweeps_small_target_set_once() { + let target_count = 8; + let mut managed = config(ManagedPacing::Burst); + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| ManagedTargetConfig::new(format!("idle-{index}"), "127.0.0.1:9")) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(Instant::now())); + + assert_eq!(task.timeout_inspections, target_count); + assert_eq!(task.timeout_cursor, 0); +} + +#[test] +fn sparse_timeout_backlog_progresses_within_one_poll() { + let server = start_server(ServerBehavior::Echo, None); + let target_count = TIMEOUT_WORK_BUDGET + 1; + let late_index = target_count - 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| { + if index == late_index { + target("late", server.addr) + } else { + ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9") + } + }) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[..late_index] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, late_index); + let now = Instant::now(); + for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 64).unwrap() { + replace_pending_for_timeout_test( + &mut task, + late_index, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + + assert!(!task.poll_timeout_pass(now)); + assert!(task.poll_timeout_pass(now)); + + assert_eq!( + timeout_losses_for(&observations, "late"), + TIMEOUT_WORK_BUDGET + ); + assert_eq!(task.timeout_backlog.len(), 1); + close_timeout_test_target(&runtime, &mut task, late_index); + drop(task); + server.finish(); +} + +#[test] +fn sparse_timeout_backlog_persists_across_polls() { + let server = start_server(ServerBehavior::Echo, None); + let target_count = (2 * TIMEOUT_WORK_BUDGET) + 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = std::iter::once(target("sparse", server.addr)) + .chain( + (1..target_count) + .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")), + ) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[1..] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let now = Instant::now(); + for seq in 0..u32::try_from((2 * TIMEOUT_WORK_BUDGET) + 64).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + + assert!(task.poll_timeout_pass(now)); + assert_eq!( + timeout_losses_for(&observations, "sparse"), + TIMEOUT_WORK_BUDGET + ); + assert_eq!(task.timeout_backlog.len(), 1); + assert_eq!(task.timeout_cursor, TIMEOUT_WORK_BUDGET); + + assert!(task.poll_timeout_pass(now)); + assert_eq!( + timeout_losses_for(&observations, "sparse"), + 2 * TIMEOUT_WORK_BUDGET + ); + assert_eq!(task.timeout_backlog.len(), 1); + assert_eq!(task.timeout_cursor, 2 * TIMEOUT_WORK_BUDGET); + close_timeout_test_target(&runtime, &mut task, 0); + drop(task); + server.finish(); +} + +#[test] +fn timeout_backlog_round_robins_known_sparse_targets() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let target_count = TIMEOUT_WORK_BUDGET + 2; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = vec![target("first", first.addr), target("second", second.addr)] + .into_iter() + .chain( + (2..target_count) + .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")), + ) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[2..] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + open_target_for_timeout_test(&runtime, &mut task, 1); + let now = Instant::now(); + for index in 0..2 { + for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 1).unwrap() { + replace_pending_for_timeout_test( + &mut task, + index, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + task.timeout_backlog.push_back(TimeoutBacklogEntry { + index, + generation: task.targets[index].instance.generation, + }); + } + task.timeout_cursor = 2; + + assert!(task.poll_timeout_pass(now)); + + assert_eq!( + timeout_losses_for(&observations, "first"), + TIMEOUT_WORK_BUDGET / 2 + ); + assert_eq!( + timeout_losses_for(&observations, "second"), + TIMEOUT_WORK_BUDGET / 2 + ); + assert_eq!(task.timeout_backlog.len(), 2); + close_timeout_test_target(&runtime, &mut task, 0); + close_timeout_test_target(&runtime, &mut task, 1); + drop(task); + first.finish(); + second.finish(); +} + +#[test] +fn timeout_backlog_does_not_starve_discovery() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let target_count = (2 * TIMEOUT_WORK_BUDGET) + 1; + let late_index = target_count - 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| match index { + 0 => target("backlog", first.addr), + index if index == late_index => target("late", second.addr), + _ => ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9"), + }) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[1..late_index] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + open_target_for_timeout_test(&runtime, &mut task, late_index); + let now = Instant::now(); + for seq in 0..u32::try_from(3 * TIMEOUT_WORK_BUDGET).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + replace_pending_for_timeout_test( + &mut task, + late_index, + timeout_probe( + 0, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + task.timeout_backlog.push_back(TimeoutBacklogEntry { + index: 0, + generation: task.targets[0].instance.generation, + }); + task.timeout_cursor = 1; + + assert!(task.poll_timeout_pass(now)); + assert_eq!(timeout_losses_for(&observations, "late"), 0); + assert_eq!(task.timeout_cursor, TIMEOUT_WORK_BUDGET + 1); + + assert!(task.poll_timeout_pass(now)); + assert_eq!(timeout_losses_for(&observations, "late"), 1); + assert_eq!(task.timeout_cursor, 0); + assert_eq!(task.timeout_backlog.len(), 1); + close_timeout_test_target(&runtime, &mut task, 0); + close_timeout_test_target(&runtime, &mut task, late_index); + drop(task); + first.finish(); + second.finish(); +} + +#[test] +fn stale_timeout_backlog_entry_is_dropped_after_pruning() { + let server = start_server(ServerBehavior::Echo, None); + let target_count = TIMEOUT_WORK_BUDGET + 2; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = vec![ + ManagedTargetConfig::new("retired", "127.0.0.1:9"), + target("survivor", server.addr), + ] + .into_iter() + .chain( + (2..target_count) + .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")), + ) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 1); + let now = Instant::now(); + replace_pending_for_timeout_test( + &mut task, + 1, + timeout_probe( + 0, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + task.timeout_backlog.push_back(TimeoutBacklogEntry { + index: 0, + generation: task.targets[0].instance.generation, + }); + task.targets[0].desired = false; + task.targets[0].state = TargetState::Terminal; + task.prune_undesired_terminal(); + task.timeout_cursor = 1; + + assert!(!task.poll_timeout_pass(now)); + assert!(task.timeout_backlog.is_empty()); + assert_eq!(timeout_losses_for(&observations, "survivor"), 0); + + task.timeout_cursor = 0; + assert!(!task.poll_timeout_pass(now)); + assert_eq!(timeout_losses_for(&observations, "survivor"), 1); + close_timeout_test_target(&runtime, &mut task, 0); + drop(task); + server.finish(); +} + +#[test] +fn immediate_timeout_backlog_skips_global_deadline_scan() { + let server = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, _) = ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap(); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let now = Instant::now(); + for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 1).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + task.timeout_backlog.push_back(TimeoutBacklogEntry { + index: 0, + generation: task.targets[0].instance.generation, + }); + task.deadline_inspections = 0; + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!(task.deadline_inspections, 0); + assert_eq!(task.timeout_backlog.len(), 1); + + assert!(!task.poll_timeout_pass(now)); + assert!(task.timeout_backlog.is_empty()); + task.deadline_inspections = 0; + poll_task_once(&runtime, &mut task); + assert_eq!(task.deadline_inspections, 1); + close_timeout_test_target(&runtime, &mut task, 0); + drop(task); + server.finish(); +} + #[test] fn timeout_discovery_reaches_later_target_after_budget_slice() { let server = start_server(ServerBehavior::Echo, None); -- 2.54.0 From 04ea4af079258bd051011478da1687ecb62df220 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 11:00:12 +0700 Subject: [PATCH 09/10] test(cli): synchronize multi-target peer close --- crates/irtt-cli/tests/client_multi_target.rs | 59 +++++++++++++++++++- 1 file changed, 56 insertions(+), 3 deletions(-) diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index b85bc62..987592a 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -4,7 +4,7 @@ use std::{ io::Read, net::{SocketAddr, UdpSocket}, process::{Command, Output, Stdio}, - sync::mpsc, + sync::{mpsc, Arc, Barrier}, thread, thread::JoinHandle, time::{Duration, Instant}, @@ -425,8 +425,9 @@ fn finite_multi_target_peer_close_is_accepted_as_completion() { #[test] fn continuous_all_peer_closed_targets_exit_nonzero() { let params = test_params(None, Duration::from_millis(10)); - let a = start_peer_close_server(params.clone()); - let b = start_peer_close_server(params); + let first_probe_barrier = Arc::new(Barrier::new(2)); + let a = start_synchronized_peer_close_server(params.clone(), Arc::clone(&first_probe_barrier)); + let b = start_synchronized_peer_close_server(params, first_probe_barrier); let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); command.args([ "--duration", @@ -860,6 +861,49 @@ fn start_peer_close_server(params: Params) -> FakeServer { start_gated_peer_close_server(params, Vec::new()) } +fn start_synchronized_peer_close_server( + params: Params, + first_probe_barrier: Arc, +) -> FakeServer { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .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, ¶ms), peer) + .unwrap(); + let (packet, peer) = recv_request(&socket); + let flags = packet[3]; + assert_eq!( + flags, 0, + "target closed before synchronized first peer-close probe" + ); + assert_eq!( + packet.len(), + echo_packet_len(false, ¶ms).unwrap(), + "expected a complete echo probe before synchronized peer-close reply" + ); + let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); + first_probe_barrier.wait(); + socket + .send_to( + &echo_reply_packet( + TOKEN, + seq, + ¶ms, + &TimestampFields::default(), + FLAG_REPLY | FLAG_CLOSE, + ), + peer, + ) + .unwrap(); + }); + FakeServer { addr, done } +} + fn start_gated_peer_close_server( params: Params, release_gates: Vec>, @@ -875,10 +919,19 @@ fn start_gated_peer_close_server( .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) .unwrap(); let (packet, peer) = recv_request(&socket); + let flags = packet[3]; + if flags & FLAG_CLOSE != 0 { + return; + } for gate in release_gates { gate.recv_timeout(Duration::from_secs(2)) .expect("timed out waiting to release peer-close reply"); } + assert_eq!( + packet.len(), + echo_packet_len(false, ¶ms).unwrap(), + "expected a complete echo probe before peer-close reply" + ); let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); socket .send_to( -- 2.54.0 From 301542186a28a55edb36564652beb176b8959f64 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 10 Aug 2026 11:09:00 +0700 Subject: [PATCH 10/10] test(cli): bound peer-close synchronization waits --- crates/irtt-cli/tests/client_multi_target.rs | 36 ++++++++++++++++---- 1 file changed, 30 insertions(+), 6 deletions(-) diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index 987592a..d6f39af 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -4,7 +4,7 @@ use std::{ io::Read, net::{SocketAddr, UdpSocket}, process::{Command, Output, Stdio}, - sync::{mpsc, Arc, Barrier}, + sync::mpsc, thread, thread::JoinHandle, time::{Duration, Instant}, @@ -425,9 +425,26 @@ fn finite_multi_target_peer_close_is_accepted_as_completion() { #[test] fn continuous_all_peer_closed_targets_exit_nonzero() { let params = test_params(None, Duration::from_millis(10)); - let first_probe_barrier = Arc::new(Barrier::new(2)); - let a = start_synchronized_peer_close_server(params.clone(), Arc::clone(&first_probe_barrier)); - let b = start_synchronized_peer_close_server(params, first_probe_barrier); + let (a_seen_tx, a_seen_rx) = mpsc::channel(); + let (a_release_tx, a_release_rx) = mpsc::channel(); + let (b_seen_tx, b_seen_rx) = mpsc::channel(); + let (b_release_tx, b_release_rx) = mpsc::channel(); + let a = start_synchronized_peer_close_server(params.clone(), a_seen_tx, a_release_rx); + let b = start_synchronized_peer_close_server(params, b_seen_tx, b_release_rx); + let coordinator = thread::spawn(move || { + a_seen_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting for target A's first probe"); + b_seen_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting for target B's first probe"); + a_release_tx + .send(()) + .expect("target A stopped waiting for peer-close release"); + b_release_tx + .send(()) + .expect("target B stopped waiting for peer-close release"); + }); let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); command.args([ "--duration", @@ -447,6 +464,7 @@ fn continuous_all_peer_closed_targets_exit_nonzero() { ]); let output = run_with_timeout(command, Duration::from_secs(3)); + coordinator.join().unwrap(); a.join(); b.join(); @@ -863,7 +881,8 @@ fn start_peer_close_server(params: Params) -> FakeServer { fn start_synchronized_peer_close_server( params: Params, - first_probe_barrier: Arc, + first_probe_seen_tx: mpsc::Sender<()>, + release_peer_close_rx: mpsc::Receiver<()>, ) -> FakeServer { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket @@ -887,7 +906,12 @@ fn start_synchronized_peer_close_server( "expected a complete echo probe before synchronized peer-close reply" ); let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); - first_probe_barrier.wait(); + first_probe_seen_tx + .send(()) + .expect("peer-close coordinator stopped waiting for first probe"); + release_peer_close_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting to release synchronized peer-close reply"); socket .send_to( &echo_reply_packet( -- 2.54.0