diff --git a/crates/irtt-stats/src/ipdv.rs b/crates/irtt-stats/src/ipdv.rs new file mode 100644 index 0000000..1ebb716 --- /dev/null +++ b/crates/irtt-stats/src/ipdv.rs @@ -0,0 +1,263 @@ +use std::{ + collections::{HashMap, HashSet, VecDeque}, + time::{Duration, Instant}, +}; + +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct IpdvSample { + pub(crate) seq: u32, + pub(crate) rtt_primary_ns: i128, + pub(crate) client_send_mono: Instant, + pub(crate) client_receive_mono: Instant, + pub(crate) client_send_wall_ns: Option, + pub(crate) client_receive_wall_ns: Option, + pub(crate) server_receive_mono_ns: Option, + pub(crate) server_send_mono_ns: Option, + pub(crate) server_receive_wall_ns: Option, + pub(crate) server_send_wall_ns: Option, +} + +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct IpdvTracker { + samples: HashMap, + sample_order: VecDeque, + completed_pairs: HashSet, + sequence_limit: Option, +} + +impl IpdvTracker { + pub(crate) fn new(sequence_limit: Option) -> Self { + Self { + samples: HashMap::new(), + sample_order: VecDeque::new(), + completed_pairs: HashSet::new(), + sequence_limit, + } + } + + pub(crate) fn insert(&mut self, sample: IpdvSample) -> Vec { + let seq = sample.seq; + if self.samples.insert(seq, sample).is_some() { + return Vec::new(); + } + + self.sample_order.push_back(seq); + self.enforce_sequence_limit(); + + let mut pairs = Vec::with_capacity(2); + if let Some(pair) = self.try_pair(seq) { + pairs.push(pair); + } + if let Some(next) = seq.checked_add(1) { + if let Some(pair) = self.try_pair(next) { + pairs.push(pair); + } + } + pairs + } + + fn enforce_sequence_limit(&mut self) { + let Some(limit) = self.sequence_limit else { + return; + }; + while self.samples.len() > limit { + let Some(seq) = self.sample_order.pop_front() else { + break; + }; + if self.samples.remove(&seq).is_some() { + self.completed_pairs.remove(&seq); + if let Some(next) = seq.checked_add(1) { + self.completed_pairs.remove(&next); + } + } + } + } + + fn try_pair(&mut self, current_seq: u32) -> Option { + let previous_seq = current_seq.checked_sub(1)?; + + if !self.completed_pairs.insert(current_seq) { + return None; + } + + let Some(previous) = self.samples.get(&previous_seq) else { + self.completed_pairs.remove(¤t_seq); + return None; + }; + + let Some(current) = self.samples.get(¤t_seq) else { + self.completed_pairs.remove(¤t_seq); + return None; + }; + + Some(CompletedIpdvPair { + previous_seq, + current_seq, + rtt_ipdv_ns: abs_i128_ns(current.rtt_primary_ns - previous.rtt_primary_ns), + send_ipdv_ns: send_ipdv_ns(previous, current).map(abs_i128_ns), + receive_ipdv_ns: receive_ipdv_ns(previous, current).map(abs_i128_ns), + }) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct CompletedIpdvPair { + pub(crate) previous_seq: u32, + pub(crate) current_seq: u32, + pub(crate) rtt_ipdv_ns: i128, + pub(crate) send_ipdv_ns: Option, + pub(crate) receive_ipdv_ns: Option, +} + +fn send_ipdv_ns(previous: &IpdvSample, current: &IpdvSample) -> Option { + if let (Some(prev_server), Some(cur_server)) = ( + previous.server_receive_mono_ns, + current.server_receive_mono_ns, + ) { + return Some( + i128::from(cur_server) + - i128::from(prev_server) + - instant_diff_ns(current.client_send_mono, previous.client_send_mono), + ); + } + if let (Some(prev_server), Some(cur_server), Some(prev_client), Some(cur_client)) = ( + previous.server_receive_wall_ns, + current.server_receive_wall_ns, + previous.client_send_wall_ns, + current.client_send_wall_ns, + ) { + return Some(i128::from(cur_server) - i128::from(prev_server) - (cur_client - prev_client)); + } + None +} + +fn receive_ipdv_ns(previous: &IpdvSample, current: &IpdvSample) -> Option { + if let (Some(prev_server), Some(cur_server)) = + (previous.server_send_mono_ns, current.server_send_mono_ns) + { + return Some( + instant_diff_ns(current.client_receive_mono, previous.client_receive_mono) + - (i128::from(cur_server) - i128::from(prev_server)), + ); + } + if let (Some(prev_server), Some(cur_server), Some(prev_client), Some(cur_client)) = ( + previous.server_send_wall_ns, + current.server_send_wall_ns, + previous.client_receive_wall_ns, + current.client_receive_wall_ns, + ) { + return Some( + (cur_client - prev_client) - (i128::from(cur_server) - i128::from(prev_server)), + ); + } + None +} + +fn instant_diff_ns(current: Instant, previous: Instant) -> i128 { + if let Some(diff) = current.checked_duration_since(previous) { + duration_ns_i128(diff) + } else { + -duration_ns_i128(previous.duration_since(current)) + } +} + +fn duration_ns_i128(duration: Duration) -> i128 { + i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX) +} + +fn abs_i128_ns(value: i128) -> i128 { + value.saturating_abs() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ipdv_sample(seq: u32, rtt_primary_ns: i128) -> IpdvSample { + let now = Instant::now(); + IpdvSample { + seq, + rtt_primary_ns, + client_send_mono: now, + client_receive_mono: now, + client_send_wall_ns: None, + client_receive_wall_ns: None, + server_receive_mono_ns: None, + server_send_mono_ns: None, + server_receive_wall_ns: None, + server_send_wall_ns: None, + } + } + + #[test] + fn ipdv_tracker_completes_adjacent_pair() { + let mut tracker = IpdvTracker::new(None); + assert!(tracker.insert(ipdv_sample(0, 10)).is_empty()); + + let pairs = tracker.insert(ipdv_sample(1, 14)); + + assert_eq!( + pairs, + vec![CompletedIpdvPair { + previous_seq: 0, + current_seq: 1, + rtt_ipdv_ns: 4, + send_ipdv_ns: None, + receive_ipdv_ns: None, + }] + ); + } + + #[test] + fn ipdv_tracker_gap_fill_completes_both_adjacent_pairs() { + let mut tracker = IpdvTracker::new(None); + assert!(tracker.insert(ipdv_sample(0, 10)).is_empty()); + assert!(tracker.insert(ipdv_sample(2, 20)).is_empty()); + + let pairs = tracker.insert(ipdv_sample(1, 13)); + + assert_eq!( + pairs, + vec![ + CompletedIpdvPair { + previous_seq: 0, + current_seq: 1, + rtt_ipdv_ns: 3, + send_ipdv_ns: None, + receive_ipdv_ns: None, + }, + CompletedIpdvPair { + previous_seq: 1, + current_seq: 2, + rtt_ipdv_ns: 7, + send_ipdv_ns: None, + receive_ipdv_ns: None, + }, + ] + ); + } + + #[test] + fn ipdv_tracker_duplicate_sequence_does_not_emit_pair_again() { + let mut tracker = IpdvTracker::new(None); + assert!(tracker.insert(ipdv_sample(0, 10)).is_empty()); + assert_eq!(tracker.insert(ipdv_sample(1, 14)).len(), 1); + + assert!(tracker.insert(ipdv_sample(1, 18)).is_empty()); + } + + #[test] + fn ipdv_tracker_bounded_mode_evicts_old_sequence_state() { + let mut tracker = IpdvTracker::new(Some(2)); + assert!(tracker.insert(ipdv_sample(0, 10)).is_empty()); + assert_eq!(tracker.insert(ipdv_sample(1, 14)).len(), 1); + + let pairs = tracker.insert(ipdv_sample(2, 20)); + + assert_eq!(pairs.len(), 1); + assert!(!tracker.samples.contains_key(&0)); + assert!(!tracker.completed_pairs.contains(&0)); + assert!(!tracker.completed_pairs.contains(&1)); + assert!(tracker.completed_pairs.contains(&2)); + } +} diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index 77995c8..21fb8be 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -3,7 +3,7 @@ #![forbid(unsafe_code)] use std::{ - collections::{HashMap, HashSet, VecDeque}, + collections::VecDeque, time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; @@ -12,8 +12,10 @@ use irtt_client::{ SignedDuration, }; +mod ipdv; mod time_stats; +use ipdv::{IpdvSample, IpdvTracker}; use time_stats::TimeMetric; pub use time_stats::TimeStats; @@ -595,20 +597,6 @@ struct ReplySample { ipdv: IpdvSample, } -#[derive(Debug, Clone, PartialEq)] -struct IpdvSample { - seq: u32, - rtt_primary_ns: i128, - client_send_mono: Instant, - client_receive_mono: Instant, - client_send_wall_ns: Option, - client_receive_wall_ns: Option, - server_receive_mono_ns: Option, - server_send_mono_ns: Option, - server_receive_wall_ns: Option, - server_send_wall_ns: Option, -} - impl ReplySample { #[allow(clippy::too_many_arguments)] fn from_reply_parts( @@ -653,98 +641,6 @@ impl ReplySample { } } -#[derive(Debug, Clone, PartialEq)] -struct IpdvTracker { - samples: HashMap, - sample_order: VecDeque, - completed_pairs: HashSet, - sequence_limit: Option, -} - -impl IpdvTracker { - fn new(sequence_limit: Option) -> Self { - Self { - samples: HashMap::new(), - sample_order: VecDeque::new(), - completed_pairs: HashSet::new(), - sequence_limit, - } - } - - fn insert(&mut self, sample: IpdvSample) -> Vec { - let seq = sample.seq; - if self.samples.insert(seq, sample).is_some() { - return Vec::new(); - } - - self.sample_order.push_back(seq); - self.enforce_sequence_limit(); - - let mut pairs = Vec::with_capacity(2); - if let Some(pair) = self.try_pair(seq) { - pairs.push(pair); - } - if let Some(next) = seq.checked_add(1) { - if let Some(pair) = self.try_pair(next) { - pairs.push(pair); - } - } - pairs - } - - fn enforce_sequence_limit(&mut self) { - let Some(limit) = self.sequence_limit else { - return; - }; - while self.samples.len() > limit { - let Some(seq) = self.sample_order.pop_front() else { - break; - }; - if self.samples.remove(&seq).is_some() { - self.completed_pairs.remove(&seq); - if let Some(next) = seq.checked_add(1) { - self.completed_pairs.remove(&next); - } - } - } - } - - fn try_pair(&mut self, current_seq: u32) -> Option { - let previous_seq = current_seq.checked_sub(1)?; - - if !self.completed_pairs.insert(current_seq) { - return None; - } - - let Some(previous) = self.samples.get(&previous_seq) else { - self.completed_pairs.remove(¤t_seq); - return None; - }; - - let Some(current) = self.samples.get(¤t_seq) else { - self.completed_pairs.remove(¤t_seq); - return None; - }; - - Some(CompletedIpdvPair { - previous_seq, - current_seq, - rtt_ipdv_ns: abs_i128_ns(current.rtt_primary_ns - previous.rtt_primary_ns), - send_ipdv_ns: send_ipdv_ns(previous, current).map(abs_i128_ns), - receive_ipdv_ns: receive_ipdv_ns(previous, current).map(abs_i128_ns), - }) - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -struct CompletedIpdvPair { - previous_seq: u32, - current_seq: u32, - rtt_ipdv_ns: i128, - send_ipdv_ns: Option, - receive_ipdv_ns: Option, -} - fn snapshot_window(events: &VecDeque) -> Snapshot { let mut core = CoreStats::new(SampleMode::RunningOnly); for event in events { @@ -816,58 +712,6 @@ fn percent(numerator: f64, denominator: f64) -> f64 { 100.0 * numerator / denominator } -fn send_ipdv_ns(previous: &IpdvSample, current: &IpdvSample) -> Option { - if let (Some(prev_server), Some(cur_server)) = ( - previous.server_receive_mono_ns, - current.server_receive_mono_ns, - ) { - return Some( - i128::from(cur_server) - - i128::from(prev_server) - - instant_diff_ns(current.client_send_mono, previous.client_send_mono), - ); - } - if let (Some(prev_server), Some(cur_server), Some(prev_client), Some(cur_client)) = ( - previous.server_receive_wall_ns, - current.server_receive_wall_ns, - previous.client_send_wall_ns, - current.client_send_wall_ns, - ) { - return Some(i128::from(cur_server) - i128::from(prev_server) - (cur_client - prev_client)); - } - None -} - -fn receive_ipdv_ns(previous: &IpdvSample, current: &IpdvSample) -> Option { - if let (Some(prev_server), Some(cur_server)) = - (previous.server_send_mono_ns, current.server_send_mono_ns) - { - return Some( - instant_diff_ns(current.client_receive_mono, previous.client_receive_mono) - - (i128::from(cur_server) - i128::from(prev_server)), - ); - } - if let (Some(prev_server), Some(cur_server), Some(prev_client), Some(cur_client)) = ( - previous.server_send_wall_ns, - current.server_send_wall_ns, - previous.client_receive_wall_ns, - current.client_receive_wall_ns, - ) { - return Some( - (cur_client - prev_client) - (i128::from(cur_server) - i128::from(prev_server)), - ); - } - None -} - -fn instant_diff_ns(current: Instant, previous: Instant) -> i128 { - if let Some(diff) = current.checked_duration_since(previous) { - duration_ns_i128(diff) - } else { - -duration_ns_i128(previous.duration_since(current)) - } -} - fn system_time_ns(time: SystemTime) -> Option { if let Ok(duration) = time.duration_since(UNIX_EPOCH) { return Some(duration_ns_i128(duration)); @@ -886,10 +730,6 @@ fn signed_duration_ns(duration: SignedDuration) -> i128 { duration.ns } -fn abs_i128_ns(value: i128) -> i128 { - value.saturating_abs() -} - fn duration_from_non_negative_i128_ns(value: i128) -> Option { u64::try_from(value).ok().map(Duration::from_nanos) } @@ -899,22 +739,6 @@ mod tests { use super::*; use std::time::SystemTime; - fn ipdv_sample(seq: u32, rtt_primary_ns: i128) -> IpdvSample { - let now = Instant::now(); - IpdvSample { - seq, - rtt_primary_ns, - client_send_mono: now, - client_receive_mono: now, - client_send_wall_ns: None, - client_receive_wall_ns: None, - server_receive_mono_ns: None, - server_send_mono_ns: None, - server_receive_wall_ns: None, - server_send_wall_ns: None, - } - } - #[test] fn system_time_before_epoch_is_supported() { let before = UNIX_EPOCH - Duration::from_nanos(7); @@ -924,76 +748,4 @@ mod tests { let now = SystemTime::now(); assert!(system_time_ns(now).is_some()); } - - #[test] - fn ipdv_tracker_completes_adjacent_pair() { - let mut tracker = IpdvTracker::new(None); - assert!(tracker.insert(ipdv_sample(0, 10)).is_empty()); - - let pairs = tracker.insert(ipdv_sample(1, 14)); - - assert_eq!( - pairs, - vec![CompletedIpdvPair { - previous_seq: 0, - current_seq: 1, - rtt_ipdv_ns: 4, - send_ipdv_ns: None, - receive_ipdv_ns: None, - }] - ); - } - - #[test] - fn ipdv_tracker_gap_fill_completes_both_adjacent_pairs() { - let mut tracker = IpdvTracker::new(None); - assert!(tracker.insert(ipdv_sample(0, 10)).is_empty()); - assert!(tracker.insert(ipdv_sample(2, 20)).is_empty()); - - let pairs = tracker.insert(ipdv_sample(1, 13)); - - assert_eq!( - pairs, - vec![ - CompletedIpdvPair { - previous_seq: 0, - current_seq: 1, - rtt_ipdv_ns: 3, - send_ipdv_ns: None, - receive_ipdv_ns: None, - }, - CompletedIpdvPair { - previous_seq: 1, - current_seq: 2, - rtt_ipdv_ns: 7, - send_ipdv_ns: None, - receive_ipdv_ns: None, - }, - ] - ); - } - - #[test] - fn ipdv_tracker_duplicate_sequence_does_not_emit_pair_again() { - let mut tracker = IpdvTracker::new(None); - assert!(tracker.insert(ipdv_sample(0, 10)).is_empty()); - assert_eq!(tracker.insert(ipdv_sample(1, 14)).len(), 1); - - assert!(tracker.insert(ipdv_sample(1, 18)).is_empty()); - } - - #[test] - fn ipdv_tracker_bounded_mode_evicts_old_sequence_state() { - let mut tracker = IpdvTracker::new(Some(2)); - assert!(tracker.insert(ipdv_sample(0, 10)).is_empty()); - assert_eq!(tracker.insert(ipdv_sample(1, 14)).len(), 1); - - let pairs = tracker.insert(ipdv_sample(2, 20)); - - assert_eq!(pairs.len(), 1); - assert!(!tracker.samples.contains_key(&0)); - assert!(!tracker.completed_pairs.contains(&0)); - assert!(!tracker.completed_pairs.contains(&1)); - assert!(tracker.completed_pairs.contains(&2)); - } }