diff --git a/crates/irtt-client/src/managed/schedule.rs b/crates/irtt-client/src/managed/schedule.rs index 4401abc..ec8d334 100644 --- a/crates/irtt-client/src/managed/schedule.rs +++ b/crates/irtt-client/src/managed/schedule.rs @@ -48,6 +48,11 @@ impl ProbeSchedule { self.next_send_at } + /// Wake for finite expiry independently of send readiness until finished. + pub(crate) fn end_deadline(&self) -> Option { + self.end_at.filter(|_| !self.is_finished()) + } + pub(crate) fn interval(&self) -> Duration { self.interval } @@ -152,3 +157,42 @@ pub(crate) fn instant_abs_diff(left: Instant, right: Instant) -> Duration { .or_else(|| right.checked_duration_since(left)) .unwrap_or(Duration::ZERO) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn lifetime_deadline_exists_only_for_an_unfinished_finite_schedule() { + let start = Instant::now(); + let end = start + Duration::from_millis(500); + let mut negotiated = NegotiatedParams { + params: irtt_proto::Params { + duration_ns: 500_000_000, + interval_ns: 1_000_000, + ..irtt_proto::Params::default() + }, + restrictions: Vec::new(), + }; + let mut finite = ProbeSchedule::new(start, &negotiated).unwrap(); + assert_eq!(finite.end_deadline(), Some(end)); + assert!(finite.permit_probe_at(end - Duration::from_nanos(1))); + assert_eq!(finite.end_deadline(), Some(end)); + assert!(!finite.permit_probe_at(end)); + assert!(finite.is_finished()); + assert_eq!(finite.end_deadline(), None); + + let mut finite = ProbeSchedule::new(start, &negotiated).unwrap(); + let commit = finite + .preflight_managed_commit(start, end - Duration::from_nanos(1)) + .unwrap(); + finite.commit(commit); + assert!(finite.is_finished()); + assert_eq!(finite.end_deadline(), None); + + negotiated.params.duration_ns = 0; + let mut continuous = ProbeSchedule::new(start, &negotiated).unwrap(); + assert!(continuous.permit_probe_at(end)); + assert_eq!(continuous.end_deadline(), None); + } +} diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 0c8a3dd..e99ab11 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -1936,6 +1936,12 @@ impl ManagedClientTask { non_send_deadline = non_send_deadline .into_iter() .chain(client.next_probe_timeout_deadline()) + .chain( + target + .schedule + .as_ref() + .and_then(ProbeSchedule::end_deadline), + ) .min(); if target.desired && target.retirement.is_none() && !target.send_waiting { send_deadline = send_deadline @@ -2399,6 +2405,71 @@ fn close_timeout_failure() -> ManagedTargetFailure { mod tests { use super::*; + // Real EAGAIN needs privileged queue shaping. Inspect only the owning + // deadline selection here, using a production socket without sending. + #[test] + fn finite_lifetime_deadline_survives_send_waiting_and_stagger_gating() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let start = Instant::now(); + let end = start + Duration::from_millis(500); + let negotiated = crate::NegotiatedParams { + params: irtt_proto::Params { + duration_ns: 500_000_000, + interval_ns: 1_000_000, + ..irtt_proto::Params::default() + }, + restrictions: Vec::new(), + }; + for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] { + let (mut task, _handle) = ManagedClient::task( + ManagedClientConfig { + pacing, + ..ManagedClientConfig::default() + }, + vec![ManagedTargetConfig::new("target", "127.0.0.1:9")], + ) + .unwrap(); + let client = AsyncClient::connect(ClientConfig { + server_addr: "127.0.0.1:9".into(), + ..ClientConfig::default() + }) + .await + .unwrap(); + task.state = DriverState::Running; + task.install_target_state(0, TargetState::Active { client }); + task.send_gate = Some(end + Duration::from_secs(1)); + + for send_waiting in [false, true] { + task.targets[0].send_waiting = send_waiting; + task.targets[0].schedule = + Some(ProbeSchedule::new(start, &negotiated).unwrap()); + let expected = if pacing == ManagedPacing::Burst && !send_waiting { + start + } else { + end + }; + assert_eq!(task.next_deadline(), Some(expected)); + + task.targets[0] + .schedule + .as_mut() + .unwrap() + .permit_probe_at(end); + assert_eq!(task.next_deadline(), None); + } + + let mut continuous = negotiated.clone(); + continuous.params.duration_ns = 0; + task.targets[0].schedule = Some(ProbeSchedule::new(start, &continuous).unwrap()); + assert_eq!(task.next_deadline(), None); + } + }); + } + #[test] fn release_gate_only_delays_an_existing_send_deadline() { let earlier = Instant::now();