diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 3cb1504..cdb8a6e 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -18,6 +18,10 @@ use crate::{ timing::ClientTimestamp, }; +pub(crate) mod schedule; + +use schedule::ProbeSchedule; + #[cfg(test)] use crate::{ session::machine::{ @@ -37,6 +41,7 @@ use crate::{ #[derive(Debug)] pub struct Client { runtime: SessionMachine, + schedule: Option, socket: UdpSocket, remote: SocketAddr, recv_buffer: Vec, @@ -56,6 +61,7 @@ impl Client { Ok(Self { runtime, + schedule: None, socket, remote, recv_buffer: vec![0_u8; recv_buffer_size(false, None)?], @@ -79,26 +85,30 @@ impl Client { match self.socket.recv(&mut buf) { Ok(size) => { let reply = self.runtime.decode_open_reply(&buf[..size])?; + let opened_at = ClientTimestamp::now(); let remote = self.remote; let has_hmac = self.runtime.has_hmac(); let socket = &self.socket; let recv_buffer = &mut self.recv_buffer; + let mut schedule = None; - return self.runtime.accept_open_reply( - reply, - ClientTimestamp::now(), - |negotiated| { - let size = recv_buffer_size(has_hmac, Some(negotiated))?; - let negotiated_dscp = u8::try_from(negotiated.params.dscp) - .map_err(|_| ClientError::InvalidConfig { - reason: "negotiated dscp must be in range 0..=63" - .to_owned(), - })?; - apply_dscp_to_socket(socket, remote, negotiated_dscp)?; - recv_buffer.resize(size, 0); - Ok(()) - }, - ); + let outcome = + self.runtime + .accept_open_reply(reply, opened_at, |negotiated| { + let size = recv_buffer_size(has_hmac, Some(negotiated))?; + let negotiated_dscp = u8::try_from(negotiated.params.dscp) + .map_err(|_| ClientError::InvalidConfig { + reason: "negotiated dscp must be in range 0..=63" + .to_owned(), + })?; + apply_dscp_to_socket(socket, remote, negotiated_dscp)?; + recv_buffer.resize(size, 0); + schedule = + Some(ProbeSchedule::new(opened_at.mono, negotiated)?); + Ok(()) + })?; + self.schedule = schedule; + return Ok(outcome); } Err(err) if matches!( @@ -131,17 +141,22 @@ impl Client { pub fn close(&mut self) -> Result, ClientError> { let socket = &self.socket; let remote = self.remote; - self.runtime.close_with(|packet| { + let events = self.runtime.close_with(|packet| { clear_dscp_on_socket(socket, remote)?; socket.send(packet)?; Ok(()) - }) + })?; + self.schedule = None; + Ok(events) } /// Return the monotonic deadline for the next probe send, if another probe /// is scheduled. pub fn next_send_deadline(&self) -> Option { - self.runtime.next_send_deadline() + if !self.runtime.is_open() { + return None; + } + self.schedule.as_ref()?.next_send_deadline() } /// Return the local timeout used to classify pending probes as lost. @@ -199,11 +214,15 @@ impl Client { Err(err) => return Err(ClientError::Socket(err)), }; - self.runtime.process_received_echo_packet( + let events = self.runtime.process_received_echo_packet( &self.recv_buffer[..datagram.len], datagram.received_at, datagram.meta, - ) + )?; + if self.runtime.is_terminal() { + self.schedule = None; + } + Ok(events) } /// Receive and classify datagrams until a receive produces no events or the @@ -247,7 +266,11 @@ impl Client { /// pending probes have either replied or timed out. No-test and closed /// sessions are also considered complete. pub fn is_run_complete(&self) -> bool { - self.runtime.is_run_complete() + self.runtime.is_terminal() + || self + .schedule + .as_ref() + .is_some_and(|schedule| schedule.is_finished() && self.runtime.pending_is_empty()) } /// Return whether the session was closed by a peer close-flagged reply. @@ -280,17 +303,23 @@ impl Client { override_ts: Option, ) -> Result, ClientError> { let socket = &self.socket; - self.runtime.send_probe_with(override_ts, |packet| { - let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); - let send_call_start = Instant::now(); - let bytes = socket.send(packet)?; - let send_call = send_call_start.elapsed(); - Ok(SendProbeResult { - sent_at, - bytes, - send_call, + self.runtime.ensure_open()?; + let schedule = self + .schedule + .as_mut() + .expect("open sessions always have a probe schedule"); + self.runtime + .send_probe_with(schedule, override_ts, |packet| { + let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); + let send_call_start = Instant::now(); + let bytes = socket.send(packet)?; + let send_call = send_call_start.elapsed(); + Ok(SendProbeResult { + sent_at, + bytes, + send_call, + }) }) - }) } fn send_managed_probe_inner( @@ -299,8 +328,13 @@ impl Client { override_ts: Option, ) -> Result, ClientError> { let socket = &self.socket; + self.runtime.ensure_open()?; + let schedule = self + .schedule + .as_mut() + .expect("open sessions always have a probe schedule"); self.runtime - .send_probe_for_deadline(override_ts, scheduled_at, |packet| { + .send_probe_for_deadline(schedule, override_ts, scheduled_at, |packet| { let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); let send_call_start = Instant::now(); let bytes = socket.send(packet)?; diff --git a/crates/irtt-client/src/client/schedule.rs b/crates/irtt-client/src/client/schedule.rs new file mode 100644 index 0000000..f300437 --- /dev/null +++ b/crates/irtt-client/src/client/schedule.rs @@ -0,0 +1,312 @@ +use std::time::{Duration, Instant}; + +use crate::{error::ClientError, session::NegotiatedParams}; + +#[derive(Debug)] +pub(crate) struct ProbeSchedule { + start_at: Instant, + end_at: Option, + interval: Duration, + next_send_at: Option, +} + +#[derive(Debug)] +pub(crate) struct ScheduleCommit { + pub(crate) scheduled_at: Instant, + pub(crate) next_send_at: Option, + pub(crate) timer_error: Duration, +} + +impl ProbeSchedule { + pub(crate) fn new( + start_at: Instant, + negotiated: &NegotiatedParams, + ) -> Result { + let interval_ns = u64::try_from(negotiated.params.interval_ns) + .expect("validated positive negotiated interval"); + let interval = Duration::from_nanos(interval_ns); + let end_at = if negotiated.params.duration_ns > 0 { + let duration_ns = u64::try_from(negotiated.params.duration_ns) + .expect("validated positive negotiated duration"); + Some( + start_at + .checked_add(Duration::from_nanos(duration_ns)) + .ok_or_else(|| ClientError::NegotiationRejected { + reason: "duration is too large to schedule".to_owned(), + })?, + ) + } else { + None + }; + + Ok(Self { + start_at, + end_at, + interval, + next_send_at: Some(start_at), + }) + } + + pub(crate) fn next_send_deadline(&self) -> Option { + self.next_send_at + } + + pub(crate) fn permit_probe_at(&mut self, now: Instant) -> bool { + if self.next_send_at.is_none() { + return false; + } + if self.end_at.is_some_and(|end| now >= end) { + self.next_send_at = None; + return false; + } + true + } + + pub(crate) fn preflight_caller_commit( + &self, + sent_at: Instant, + next_packets_sent: u64, + ) -> Result { + let scheduled_at = self + .next_send_at + .expect("a caller schedule commit requires a permitted probe"); + let interval_ns = + u64::try_from(self.interval.as_nanos()).map_err(|_| ClientError::DurationOverflow)?; + let next_send_at = next_probe_deadline(self.start_at, interval_ns, next_packets_sent)?; + Ok(self.finish_commit(scheduled_at, next_send_at, sent_at)) + } + + pub(crate) fn preflight_managed_commit( + &self, + scheduled_at: Instant, + sent_at: Instant, + ) -> Result { + let (scheduled_at, next_send_at, _) = + advance_cadence(scheduled_at, self.interval, sent_at)?; + Ok(self.finish_commit(scheduled_at, next_send_at, sent_at)) + } + + pub(crate) fn commit(&mut self, commit: ScheduleCommit) { + self.next_send_at = commit.next_send_at; + } + + pub(crate) fn is_finished(&self) -> bool { + self.next_send_at.is_none() + } + + fn finish_commit( + &self, + scheduled_at: Instant, + next_send_at: Instant, + sent_at: Instant, + ) -> ScheduleCommit { + ScheduleCommit { + scheduled_at, + next_send_at: if self.end_at.is_some_and(|end| next_send_at >= end) { + None + } else { + Some(next_send_at) + }, + timer_error: instant_abs_diff(sent_at, scheduled_at), + } + } +} + +fn next_probe_deadline( + start: Instant, + interval_ns: u64, + packets_sent: u64, +) -> Result { + let offset_ns = interval_ns + .checked_mul(packets_sent) + .ok_or(ClientError::DurationOverflow)?; + start + .checked_add(Duration::from_nanos(offset_ns)) + .ok_or(ClientError::DurationOverflow) +} + +pub(crate) fn advance_cadence( + deadline: Instant, + interval: Duration, + now: Instant, +) -> Result<(Instant, Instant, u128), ClientError> { + if interval.is_zero() { + return Err(ClientError::InvalidConfig { + reason: "probe interval must be greater than zero".to_owned(), + }); + } + + let elapsed_slots = now + .checked_duration_since(deadline) + .map_or(0, |elapsed| elapsed.as_nanos() / interval.as_nanos()); + let scheduled_offset = duration_from_nanos( + interval + .as_nanos() + .checked_mul(elapsed_slots) + .ok_or(ClientError::DurationOverflow)?, + )?; + let next_offset = duration_from_nanos( + interval + .as_nanos() + .checked_mul( + elapsed_slots + .checked_add(1) + .ok_or(ClientError::DurationOverflow)?, + ) + .ok_or(ClientError::DurationOverflow)?, + )?; + let scheduled_at = deadline + .checked_add(scheduled_offset) + .ok_or(ClientError::DurationOverflow)?; + let next_at = deadline + .checked_add(next_offset) + .ok_or(ClientError::DurationOverflow)?; + Ok((scheduled_at, next_at, elapsed_slots)) +} + +fn duration_from_nanos(nanos: u128) -> Result { + const NANOS_PER_SECOND: u128 = 1_000_000_000; + let seconds = + u64::try_from(nanos / NANOS_PER_SECOND).map_err(|_| ClientError::DurationOverflow)?; + let subsec_nanos = u32::try_from(nanos % NANOS_PER_SECOND) + .expect("nanosecond remainder is always less than one second"); + Ok(Duration::new(seconds, subsec_nanos)) +} + +fn instant_abs_diff(left: Instant, right: Instant) -> Duration { + left.checked_duration_since(right) + .or_else(|| right.checked_duration_since(left)) + .unwrap_or(Duration::ZERO) +} + +#[cfg(test)] +mod tests { + use super::*; + use irtt_proto::Params; + + fn negotiated(interval_ns: i64, duration_ns: i64) -> NegotiatedParams { + NegotiatedParams { + params: Params { + interval_ns, + duration_ns, + ..Params::default() + }, + restrictions: Vec::new(), + } + } + + #[test] + fn initial_deadline_is_the_opening_time() { + let opened_at = Instant::now(); + + let schedule = ProbeSchedule::new(opened_at, &negotiated(10_000_000, 0)).unwrap(); + + assert_eq!(schedule.next_send_deadline(), Some(opened_at)); + assert!(!schedule.is_finished()); + } + + #[test] + fn caller_preflight_keeps_absolute_cadence_without_mutation() { + let start = Instant::now(); + let schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap(); + let sent_at = start + Duration::from_millis(4); + + let commit = schedule.preflight_caller_commit(sent_at, 3).unwrap(); + + assert_eq!(commit.scheduled_at, start); + assert_eq!(commit.next_send_at, Some(start + Duration::from_millis(30))); + assert_eq!(commit.timer_error, Duration::from_millis(4)); + assert_eq!(schedule.next_send_deadline(), Some(start)); + } + + #[test] + fn caller_commit_advances_exactly_one_cadence_slot() { + let start = Instant::now(); + let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap(); + + let commit = schedule.preflight_caller_commit(start, 1).unwrap(); + assert_eq!(schedule.next_send_deadline(), Some(start)); + + schedule.commit(commit); + assert_eq!( + schedule.next_send_deadline(), + Some(start + Duration::from_millis(10)) + ); + } + + #[test] + fn finite_duration_uses_an_exclusive_end_boundary() { + let start = Instant::now(); + let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 20_000_000)).unwrap(); + + let first = schedule.preflight_caller_commit(start, 1).unwrap(); + schedule.commit(first); + assert_eq!( + schedule.next_send_deadline(), + Some(start + Duration::from_millis(10)) + ); + + let second = schedule + .preflight_caller_commit(start + Duration::from_millis(10), 2) + .unwrap(); + schedule.commit(second); + + assert!(schedule.is_finished()); + assert_eq!(schedule.next_send_deadline(), None); + assert!(!schedule.permit_probe_at(start + Duration::from_millis(20))); + } + + #[test] + fn permit_at_exclusive_end_finishes_schedule_without_a_send() { + let start = Instant::now(); + let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 20_000_000)).unwrap(); + + assert!(!schedule.permit_probe_at(start + Duration::from_millis(20))); + assert!(schedule.is_finished()); + assert_eq!(schedule.next_send_deadline(), None); + } + + #[test] + fn managed_preflight_skips_missed_slots_without_mutation() { + let start = Instant::now(); + let schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap(); + let sent_at = start + Duration::from_millis(45); + + let commit = schedule.preflight_managed_commit(start, sent_at).unwrap(); + + assert_eq!(commit.scheduled_at, start + Duration::from_millis(40)); + assert_eq!(commit.next_send_at, Some(start + Duration::from_millis(50))); + assert_eq!(commit.timer_error, Duration::from_millis(5)); + assert_eq!(schedule.next_send_deadline(), Some(start)); + } + + #[test] + fn delayed_managed_commit_does_not_replay_catch_up_slots() { + let start = Instant::now(); + let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap(); + let delayed_send = start + Duration::from_millis(45); + + let commit = schedule + .preflight_managed_commit(start, delayed_send) + .unwrap(); + schedule.commit(commit); + + assert_eq!( + schedule.next_send_deadline(), + Some(start + Duration::from_millis(50)) + ); + assert!(schedule.next_send_deadline().unwrap() > delayed_send); + } + + #[test] + fn deadline_arithmetic_overflow_is_detected_before_commit() { + let start = Instant::now(); + let schedule = ProbeSchedule::new(start, &negotiated(2, 0)).unwrap(); + + assert!(matches!( + schedule.preflight_caller_commit(start, u64::MAX), + Err(ClientError::DurationOverflow) + )); + assert_eq!(schedule.next_send_deadline(), Some(start)); + } +} diff --git a/crates/irtt-client/src/client/tests/close.rs b/crates/irtt-client/src/client/tests/close.rs index b7152f0..f5c7f23 100644 --- a/crates/irtt-client/src/client/tests/close.rs +++ b/crates/irtt-client/src/client/tests/close.rs @@ -118,6 +118,7 @@ fn close_flagged_echo_reply_emits_reply_then_closes_without_sending_close() { )); assert_eq!(events.len(), 2); assert!(client.next_send_deadline().is_none()); + assert!(client.schedule.is_none()); assert!(matches!( client.send_probe(), Err(ClientError::AlreadyClosed) diff --git a/crates/irtt-client/src/client/tests/no_test.rs b/crates/irtt-client/src/client/tests/no_test.rs index 02b7b07..d1fc893 100644 --- a/crates/irtt-client/src/client/tests/no_test.rs +++ b/crates/irtt-client/src/client/tests/no_test.rs @@ -8,8 +8,11 @@ fn no_test_success_validates_params() { let server = no_test_server(params.clone(), 0); config.server_addr = server.addr.to_string(); let mut client = Client::connect(config).unwrap(); + assert!(client.next_send_deadline().is_none()); let negotiated = assert_no_test_completed(client.open().unwrap()); assert_eq!(negotiated.params, params); + assert!(client.next_send_deadline().is_none()); + assert!(client.schedule.is_none()); server.join(); } diff --git a/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs b/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs index a02c26c..4b6d7cc 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs @@ -20,6 +20,7 @@ fn send_probe_sends_valid_echo_request() { ..default_test_config(server.addr) }; let mut client = Client::connect(config).unwrap(); + assert!(client.next_send_deadline().is_none()); assert_open_started(client.open().unwrap()); let events = client.send_probe().unwrap(); assert_eq!(events.len(), 1); diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index f621e53..d554fa5 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -12,12 +12,13 @@ use std::{ }; use crate::{ + client::schedule::{advance_cadence, ProbeSchedule}, config::{ClientAuthConfig, ClientConfig}, error::ClientError, event::{ClientEvent, OpenOutcome}, metadata::ReceiveMeta, receive::recv_datagram_from, - session::machine::{advance_cadence, params_from_config, SendProbeResult, SessionMachine}, + session::machine::{params_from_config, SendProbeResult, SessionMachine}, socket::{bind_unconnected_udp_socket, validate_open_timeouts}, socket_options::apply_dscp_to_socket, timing::ClientTimestamp, @@ -626,6 +627,7 @@ struct TargetState { remote: SocketAddr, configured_auth: Option, runtime: SessionMachine, + schedule: Option, status: TargetStatus, open_packet: Vec, counters: TargetCounters, @@ -716,6 +718,7 @@ impl TargetState { remote: target.remote, configured_auth: target.auth, runtime, + schedule: None, status: TargetStatus::Opening { attempt: 0, next_send_at: now, @@ -749,6 +752,14 @@ impl TargetState { self.final_reason = Some(reason); } + fn is_run_complete(&self) -> bool { + self.runtime.is_terminal() + || self + .schedule + .as_ref() + .is_some_and(|schedule| schedule.is_finished() && self.runtime.pending_is_empty()) + } + fn outcome(&self, reason: ManagedTargetEndReason) -> ManagedTargetOutcome { ManagedTargetOutcome { id: self.id.clone(), @@ -872,16 +883,21 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc) -> Result<(), let mut target = target.lock().expect("target mutex poisoned"); match target.status { TargetStatus::Opening { .. } => { + let opened_at = datagram.received_at; + let mut schedule = None; let result = target.runtime.decode_open_reply(packet).and_then(|reply| { - target.runtime.accept_open_reply( - reply, - datagram.received_at, - |negotiated| validate_group_negotiation(&shared, negotiated), - ) + target + .runtime + .accept_open_reply(reply, opened_at, |negotiated| { + validate_group_negotiation(&shared, negotiated)?; + schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?); + Ok(()) + }) }); match result { Ok(outcome) => { + target.schedule = schedule; publish_open_outcome(&shared.hub, &mut target, outcome); wake_scheduler = true; } @@ -985,6 +1001,7 @@ fn process_active_target_packet( Ok(events) => { publish_events(hub, target, events); if target.runtime.is_peer_closed() { + target.schedule = None; target.mark_finished(ManagedTargetEndReason::PeerClosed); true } else { @@ -1251,7 +1268,7 @@ fn finish_completed_targets(socket: &UdpSocket, shared: &GroupShared, now: Insta for target in all_targets(shared) { let mut target = target.lock().expect("target mutex poisoned"); match target.status { - TargetStatus::Active if target.runtime.is_run_complete() => { + TargetStatus::Active if target.is_run_complete() => { if target.runtime.is_peer_closed() { target.mark_finished(ManagedTargetEndReason::PeerClosed); } else if target.runtime.has_timed_out_metadata() { @@ -1295,19 +1312,27 @@ fn send_echo_to_target( return; } let remote = target.remote; - let result = target + target .runtime - .send_probe_for_deadline(None, scheduled_at, |packet| { - let sent_at = ClientTimestamp::now(); - let send_call_start = Instant::now(); - let bytes = socket.send_to(packet, remote)?; - let send_call = send_call_start.elapsed(); - Ok(SendProbeResult { - sent_at, - bytes, - send_call, - }) - }); + .ensure_open() + .expect("active targets are open"); + let TargetState { + runtime, schedule, .. + } = &mut *target; + let schedule = schedule + .as_mut() + .expect("active targets always have a probe schedule"); + let result = runtime.send_probe_for_deadline(schedule, None, scheduled_at, |packet| { + let sent_at = ClientTimestamp::now(); + let send_call_start = Instant::now(); + let bytes = socket.send_to(packet, remote)?; + let send_call = send_call_start.elapsed(); + Ok(SendProbeResult { + sent_at, + bytes, + send_call, + }) + }); match result { Ok(events) => { @@ -1437,7 +1462,10 @@ fn close_locked_target( socket.send_to(packet, remote)?; Ok(()) }) { - Ok(events) => publish_events(hub, target, events), + Ok(events) => { + publish_events(hub, target, events); + target.schedule = None; + } Err(err) => { target.mark_finished(ManagedTargetEndReason::Failed( ManagedTargetFailure::runtime(&err), @@ -1619,7 +1647,7 @@ fn target_is_due_for_slot(target: &Arc>, scheduled_at: Instan fn target_next_send_deadline(target: &Arc>) -> Option { let target = target.lock().expect("target mutex poisoned"); if matches!(target.status, TargetStatus::Active) { - target.runtime.next_send_deadline() + target.schedule.as_ref()?.next_send_deadline() } else { None } @@ -1810,24 +1838,38 @@ mod tests { let open_packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms); let open_reply = target.runtime.decode_open_reply(&open_packet).unwrap(); + let opened_at = ClientTimestamp::now(); + let mut opened_schedule = None; let open_outcome = target .runtime - .accept_open_reply(open_reply, ClientTimestamp::now(), |_| Ok(())) + .accept_open_reply(open_reply, opened_at, |negotiated| { + opened_schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?); + Ok(()) + }) .unwrap(); + target.schedule = opened_schedule; publish_open_outcome(&hub, &mut target, open_outcome); let scheduled_at = Instant::now(); let sent_at = ClientTimestamp::now(); - let sent_events = target - .runtime - .send_probe_for_deadline(Some(sent_at), scheduled_at, |packet| { - Ok(SendProbeResult { - sent_at, - bytes: packet.len(), - send_call: Duration::ZERO, - }) - }) - .unwrap(); + let sent_events = { + let TargetState { + runtime, schedule, .. + } = &mut target; + runtime.send_probe_for_deadline( + schedule.as_mut().unwrap(), + Some(sent_at), + scheduled_at, + |packet| { + Ok(SendProbeResult { + sent_at, + bytes: packet.len(), + send_call: Duration::ZERO, + }) + }, + ) + } + .unwrap(); publish_events(&hub, &mut target, sent_events); ActiveTargetFixture { diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 1c01a7d..6ac80aa 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -7,6 +7,7 @@ use irtt_proto::{ }; use crate::{ + client::schedule::ProbeSchedule, config::{ ClientConfig, RunMode, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_UDP_PAYLOAD_LENGTH, }, @@ -56,13 +57,9 @@ struct ActiveSession { next_wire_seq: u32, highest_received_seq: Option, packets_sent: u64, - start_mono: Instant, - end_mono: Option, - next_send_at: Instant, pending: PendingMap, timed_out: TimedOutMap, completed: CompletedSet, - sending_done: bool, } #[derive(Debug, Clone, Copy)] @@ -159,33 +156,25 @@ impl SessionMachine { } } - pub(crate) fn next_send_deadline(&self) -> Option { - let MachineState::Open(session) = &self.state else { - return None; - }; - if session.sending_done { - return None; - } - Some(session.next_send_at) - } - pub(crate) fn probe_timeout(&self) -> Duration { self.config.probe_timeout } pub(crate) fn send_probe_with( &mut self, + schedule: &mut ProbeSchedule, override_ts: Option, send: F, ) -> Result, ClientError> where F: FnOnce(&[u8]) -> Result, { - self.send_probe_inner(override_ts, None, false, send) + self.send_probe_inner(schedule, override_ts, None, false, send) } pub(crate) fn send_probe_for_deadline( &mut self, + schedule: &mut ProbeSchedule, override_ts: Option, scheduled_at: Instant, send: F, @@ -193,11 +182,12 @@ impl SessionMachine { where F: FnOnce(&[u8]) -> Result, { - self.send_probe_inner(override_ts, Some(scheduled_at), true, send) + self.send_probe_inner(schedule, override_ts, Some(scheduled_at), true, send) } fn send_probe_inner( &mut self, + schedule: &mut ProbeSchedule, override_ts: Option, scheduled_at_override: Option, skip_missed_slots: bool, @@ -206,6 +196,11 @@ impl SessionMachine { where F: FnOnce(&[u8]) -> Result, { + let now = override_ts.unwrap_or_else(ClientTimestamp::now); + if !schedule.permit_probe_at(now.mono) { + return Ok(vec![]); + } + let (config, state) = (&self.config, &mut self.state); let probe_timeout = config.probe_timeout; let hmac_key = config.hmac_key.as_deref(); @@ -216,23 +211,9 @@ impl SessionMachine { MachineState::NoTestCompleted => return Err(ClientError::AlreadyCompleted), }; - if session.sending_done { - return Ok(vec![]); - } - - let now = override_ts.unwrap_or_else(ClientTimestamp::now); - - if let Some(end) = session.end_mono { - if now.mono >= end { - session.sending_done = true; - return Ok(vec![]); - } - } - session.pending.check_capacity()?; let wire_seq = session.next_wire_seq; - let scheduled_at = scheduled_at_override.unwrap_or(session.next_send_at); let request = EchoRequest { token: session.token, @@ -263,27 +244,17 @@ impl SessionMachine { counter: "packets_sent", })?; - let interval_ns = u64::try_from(session.negotiated.params.interval_ns) - .expect("validated positive negotiated interval"); - let interval = Duration::from_nanos(interval_ns); - let (scheduled_at, next_send_at) = if skip_missed_slots { - let (scheduled_at, next_send_at, _) = - advance_cadence(scheduled_at, interval, send_result.sent_at.mono)?; - (scheduled_at, next_send_at) + let schedule_commit = if skip_missed_slots { + schedule.preflight_managed_commit( + scheduled_at_override.expect("managed sends include a scheduled deadline"), + send_result.sent_at.mono, + )? } else { - ( - scheduled_at, - next_probe_deadline(session.start_mono, interval_ns, session.packets_sent)?, - ) + schedule.preflight_caller_commit(send_result.sent_at.mono, session.packets_sent)? }; - let timer_error = instant_abs_diff(send_result.sent_at.mono, scheduled_at); - session.next_send_at = next_send_at; - - if let Some(end) = session.end_mono { - if session.next_send_at >= end { - session.sending_done = true; - } - } + let scheduled_at = schedule_commit.scheduled_at; + let timer_error = schedule_commit.timer_error; + schedule.commit(schedule_commit); Ok(vec![ClientEvent::EchoSent { seq: wire_seq, @@ -358,14 +329,21 @@ impl SessionMachine { }]) } - pub(crate) fn is_run_complete(&self) -> bool { + pub(crate) fn pending_is_empty(&self) -> bool { match &self.state { - MachineState::Open(session) => session.sending_done && session.pending.len() == 0, + MachineState::Open(session) => session.pending.len() == 0, MachineState::NoTestCompleted | MachineState::Closed { .. } => true, MachineState::Connected => false, } } + pub(crate) fn is_terminal(&self) -> bool { + matches!( + self.state, + MachineState::NoTestCompleted | MachineState::Closed { .. } + ) + } + pub(crate) fn is_peer_closed(&self) -> bool { matches!( self.state, @@ -395,6 +373,10 @@ impl SessionMachine { matches!(self.state, MachineState::Open(_)) } + pub(crate) fn ensure_open(&self) -> Result<(), ClientError> { + self.open_session().map(|_| ()) + } + fn accept_normal_open( &mut self, reply: OpenReply, @@ -412,28 +394,15 @@ impl SessionMachine { )?; before_normal_open(&negotiated)?; - let end_mono = if negotiated.params.duration_ns > 0 { - Some(negotiated_end_mono( - now.mono, - negotiated.params.duration_ns, - )?) - } else { - None - }; - self.state = MachineState::Open(Box::new(ActiveSession { token, negotiated: negotiated.clone(), next_wire_seq: 0, highest_received_seq: None, packets_sent: 0, - start_mono: now.mono, - end_mono, - next_send_at: now.mono, pending: PendingMap::new(self.config.max_pending_probes), timed_out: TimedOutMap::new(self.config.max_pending_probes), completed: CompletedSet::new(self.config.max_pending_probes), - sending_done: false, })); let event = ClientEvent::SessionStarted { @@ -704,67 +673,6 @@ pub(crate) fn update_highest_received(highest_received_seq: &mut Option, wi })); } -fn next_probe_deadline( - start: Instant, - interval_ns: u64, - packets_sent: u64, -) -> Result { - let offset_ns = interval_ns - .checked_mul(packets_sent) - .ok_or(ClientError::DurationOverflow)?; - start - .checked_add(Duration::from_nanos(offset_ns)) - .ok_or(ClientError::DurationOverflow) -} - -pub(crate) fn advance_cadence( - deadline: Instant, - interval: Duration, - now: Instant, -) -> Result<(Instant, Instant, u128), ClientError> { - if interval.is_zero() { - return Err(ClientError::InvalidConfig { - reason: "probe interval must be greater than zero".to_owned(), - }); - } - - let elapsed_slots = now - .checked_duration_since(deadline) - .map_or(0, |elapsed| elapsed.as_nanos() / interval.as_nanos()); - let scheduled_offset = duration_from_nanos( - interval - .as_nanos() - .checked_mul(elapsed_slots) - .ok_or(ClientError::DurationOverflow)?, - )?; - let next_offset = duration_from_nanos( - interval - .as_nanos() - .checked_mul( - elapsed_slots - .checked_add(1) - .ok_or(ClientError::DurationOverflow)?, - ) - .ok_or(ClientError::DurationOverflow)?, - )?; - let scheduled_at = deadline - .checked_add(scheduled_offset) - .ok_or(ClientError::DurationOverflow)?; - let next_at = deadline - .checked_add(next_offset) - .ok_or(ClientError::DurationOverflow)?; - Ok((scheduled_at, next_at, elapsed_slots)) -} - -fn duration_from_nanos(nanos: u128) -> Result { - const NANOS_PER_SECOND: u128 = 1_000_000_000; - let seconds = - u64::try_from(nanos / NANOS_PER_SECOND).map_err(|_| ClientError::DurationOverflow)?; - let subsec_nanos = u32::try_from(nanos % NANOS_PER_SECOND) - .expect("nanosecond remainder is always less than one second"); - Ok(Duration::new(seconds, subsec_nanos)) -} - pub(crate) fn sequence_is_after(candidate: u32, current: u32) -> bool { candidate != current && candidate.wrapping_sub(current) < (1 << 31) } @@ -773,12 +681,6 @@ pub(crate) fn sequence_is_before(candidate: u32, current: u32) -> bool { current != candidate && current.wrapping_sub(candidate) < (1 << 31) } -fn instant_abs_diff(left: Instant, right: Instant) -> Duration { - left.checked_duration_since(right) - .or_else(|| right.checked_duration_since(left)) - .unwrap_or(Duration::ZERO) -} - pub(crate) fn compute_rtt( sent_at: &ClientTimestamp, received_at: &ClientTimestamp, @@ -929,13 +831,3 @@ fn config_duration_to_ns(field: &str, duration: Duration) -> Result Result { - debug_assert!(duration_ns > 0); - let duration_ns = u64::try_from(duration_ns).expect("validated positive negotiated duration"); - start - .checked_add(Duration::from_nanos(duration_ns)) - .ok_or_else(|| ClientError::NegotiationRejected { - reason: "duration is too large to schedule".to_owned(), - }) -}