diff --git a/crates/irtt-cli/src/lib.rs b/crates/irtt-cli/src/lib.rs index 70186d6..8b5308a 100644 --- a/crates/irtt-cli/src/lib.rs +++ b/crates/irtt-cli/src/lib.rs @@ -318,10 +318,17 @@ pub fn format_event(event: &ClientEvent, mode: OutputMode) -> Option { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +#[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct HumanEventStats { pub contributed_sample: bool, - pub rtt_ipdv: Option, + pub ipdv_pairs: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct HumanIpdvPair { + pub previous_logical_seq: u64, + pub current_logical_seq: u64, + pub rtt_ipdv: Duration, pub send_ipdv: Option, pub receive_ipdv: Option, } @@ -336,6 +343,17 @@ impl From for HumanEventStats { fn from(value: irtt_stats::EventStatsUpdate) -> Self { Self { contributed_sample: value.contributed_sample, + ipdv_pairs: value.ipdv_pairs.into_iter().map(Into::into).collect(), + } + } +} + +#[cfg(feature = "stats")] +impl From for HumanIpdvPair { + fn from(value: irtt_stats::IpdvPairUpdate) -> Self { + Self { + previous_logical_seq: value.previous_logical_seq, + current_logical_seq: value.current_logical_seq, rtt_ipdv: value.rtt_ipdv, send_ipdv: value.send_ipdv, receive_ipdv: value.receive_ipdv, @@ -352,6 +370,8 @@ pub fn format_human_event_with_options( stats: Option, options: HumanOutputOptions, ) -> String { + let stats = stats.as_ref(); + match event { ClientEvent::SessionStarted { remote, token, .. } => { format!("session started remote={remote} token={token:#x}") @@ -378,7 +398,12 @@ pub fn format_human_event_with_options( } write!(out, " rtt={}", format_duration(rtt.effective)).unwrap(); write_human_one_way(&mut out, *one_way); - write!(out, " ipdv={}", format_human_ipdv(stats)).unwrap(); + write!( + out, + " ipdv={}", + format_human_ipdv(stats, Some(*logical_seq)) + ) + .unwrap(); if let Some(processing) = server_timing.and_then(|timing| timing.processing) { write!(out, " proc={}", format_duration(processing)).unwrap(); } @@ -412,7 +437,7 @@ pub fn format_human_event_with_options( if let Some(rtt) = rtt { write!(out, " rtt={}", format_duration(rtt.effective)).unwrap(); write_human_one_way(&mut out, *one_way); - write!(out, " ipdv={}", format_human_ipdv(stats)).unwrap(); + write!(out, " ipdv={}", format_human_ipdv(stats, *logical_seq)).unwrap(); } write_human_received_stats(&mut out, *received_stats); out @@ -737,11 +762,22 @@ fn write_human_received_stats(out: &mut String, stats: Option) -> String { - match stats.and_then(|stats| stats.rtt_ipdv) { - Some(ipdv) => format_duration(ipdv), - None => "n/a".to_owned(), - } +fn format_human_ipdv(stats: Option<&HumanEventStats>, logical_seq: Option) -> String { + let Some(stats) = stats else { + return "n/a".to_owned(); + }; + + let pair = logical_seq + .and_then(|seq| { + stats + .ipdv_pairs + .iter() + .find(|pair| pair.current_logical_seq == seq) + }) + .or_else(|| stats.ipdv_pairs.last()); + + pair.map(|pair| format_duration(pair.rtt_ipdv)) + .unwrap_or_else(|| "n/a".to_owned()) } fn write_optional_u8(out: &mut String, key: &str, value: Option) { @@ -1335,9 +1371,13 @@ mod tests { &reply_event(), Some(HumanEventStats { contributed_sample: true, - rtt_ipdv: Some(Duration::from_micros(47)), - send_ipdv: None, - receive_ipdv: None, + ipdv_pairs: vec![HumanIpdvPair { + previous_logical_seq: 7, + current_logical_seq: 8, + rtt_ipdv: Duration::from_micros(47), + send_ipdv: None, + receive_ipdv: None, + }], }), ); diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index f3868ab..971a240 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -138,20 +138,19 @@ impl StatsCollector { pub type RollingSnapshot = CumulativeSnapshot; pub type FiniteSummary = CumulativeSnapshot; -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +#[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct EventStatsUpdate { pub contributed_sample: bool, - pub rtt_ipdv: Option, - pub send_ipdv: Option, - pub receive_ipdv: Option, + pub ipdv_pairs: Vec, } -impl EventStatsUpdate { - fn merge_ipdv(&mut self, other: Self) { - self.rtt_ipdv = self.rtt_ipdv.or(other.rtt_ipdv); - self.send_ipdv = self.send_ipdv.or(other.send_ipdv); - self.receive_ipdv = self.receive_ipdv.or(other.receive_ipdv); - } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct IpdvPairUpdate { + pub previous_logical_seq: u64, + pub current_logical_seq: u64, + pub rtt_ipdv: Duration, + pub send_ipdv: Option, + pub receive_ipdv: Option, } #[derive(Debug, Clone, PartialEq)] @@ -390,9 +389,13 @@ impl CoreStats { if self.samples.insert(seq, sample).is_none() { self.sample_order.push_back(seq); self.enforce_sequence_limit(); - update.merge_ipdv(self.try_ipdv_pair(seq)); + if let Some(pair) = self.try_ipdv_pair(seq) { + update.ipdv_pairs.push(pair); + } if let Some(next) = seq.checked_add(1) { - update.merge_ipdv(self.try_ipdv_pair(next)); + if let Some(pair) = self.try_ipdv_pair(next) { + update.ipdv_pairs.push(pair); + } } } } @@ -431,39 +434,46 @@ impl CoreStats { } } - fn try_ipdv_pair(&mut self, current_seq: u64) -> EventStatsUpdate { - let Some(previous_seq) = current_seq.checked_sub(1) else { - return EventStatsUpdate::default(); - }; + fn try_ipdv_pair(&mut self, current_seq: u64) -> Option { + let previous_seq = current_seq.checked_sub(1)?; + if !self.ipdv_pairs.insert(current_seq) { - return EventStatsUpdate::default(); + return None; } + let Some(previous) = self.samples.get(&previous_seq) else { self.ipdv_pairs.remove(¤t_seq); - return EventStatsUpdate::default(); - }; - let Some(current) = self.samples.get(¤t_seq) else { - self.ipdv_pairs.remove(¤t_seq); - return EventStatsUpdate::default(); + return None; }; - let rtt_ipdv = abs_i128_to_u64(current.rtt_primary_ns - previous.rtt_primary_ns); - self.ipdv_round_trip.push(rtt_ipdv); - let mut update = EventStatsUpdate { - rtt_ipdv: Some(Duration::from_nanos(rtt_ipdv)), - ..EventStatsUpdate::default() + let Some(current) = self.samples.get(¤t_seq) else { + self.ipdv_pairs.remove(¤t_seq); + return None; }; - if let Some(send_ipdv) = send_ipdv_ns(previous, current) { - let send_ipdv = abs_i128_to_u64(send_ipdv); - self.ipdv_send.push(send_ipdv); - update.send_ipdv = Some(Duration::from_nanos(send_ipdv)); + + // Compute everything before mutating metric fields, otherwise the borrow + // checker may quite reasonably start throwing furniture. + let rtt_ipdv = abs_i128_to_u64(current.rtt_primary_ns - previous.rtt_primary_ns); + let send_ipdv = send_ipdv_ns(previous, current).map(abs_i128_to_u64); + let receive_ipdv = receive_ipdv_ns(previous, current).map(abs_i128_to_u64); + + self.ipdv_round_trip.push(rtt_ipdv); + + if let Some(value) = send_ipdv { + self.ipdv_send.push(value); } - if let Some(receive_ipdv) = receive_ipdv_ns(previous, current) { - let receive_ipdv = abs_i128_to_u64(receive_ipdv); - self.ipdv_receive.push(receive_ipdv); - update.receive_ipdv = Some(Duration::from_nanos(receive_ipdv)); + + if let Some(value) = receive_ipdv { + self.ipdv_receive.push(value); } - update + + Some(IpdvPairUpdate { + previous_logical_seq: previous_seq, + current_logical_seq: current_seq, + rtt_ipdv: Duration::from_nanos(rtt_ipdv), + send_ipdv: send_ipdv.map(Duration::from_nanos), + receive_ipdv: receive_ipdv.map(Duration::from_nanos), + }) } fn snapshot(&self) -> CumulativeSnapshot { @@ -1251,6 +1261,24 @@ mod tests { assert_eq!(snapshot.loss.packet_loss_percent, 50.0); } + fn assert_no_ipdv_pairs(update: &EventStatsUpdate) { + assert!(update.ipdv_pairs.is_empty(), "{update:?}"); + } + + fn assert_one_ipdv_pair( + update: &EventStatsUpdate, + previous_logical_seq: u64, + current_logical_seq: u64, + rtt_ipdv: Duration, + ) -> &IpdvPairUpdate { + assert_eq!(update.ipdv_pairs.len(), 1, "{update:?}"); + let pair = &update.ipdv_pairs[0]; + assert_eq!(pair.previous_logical_seq, previous_logical_seq); + assert_eq!(pair.current_logical_seq, current_logical_seq); + assert_eq!(pair.rtt_ipdv, rtt_ipdv); + pair + } + #[test] fn ipdv_is_sequence_adjacent_and_gap_preserving() { let mut collector = StatsCollector::new(StatsConfig::finite()); @@ -1260,10 +1288,13 @@ mod tests { let snapshot = collector.cumulative(); assert!(first.contributed_sample); - assert_eq!(first.rtt_ipdv, None); + assert_no_ipdv_pairs(&first); + assert!(gap.contributed_sample); - assert_eq!(gap.rtt_ipdv, None); - assert_eq!(adjacent.rtt_ipdv, Some(Duration::from_millis(3))); + assert_no_ipdv_pairs(&gap); + + assert!(adjacent.contributed_sample); + assert_one_ipdv_pair(&adjacent, 2, 3, Duration::from_millis(3)); assert_eq!(snapshot.ipdv.round_trip.stats.count, 1); assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 3_000_000); } @@ -1275,8 +1306,10 @@ mod tests { let update = collector.process(&late_reply(0, 10, 10)); let snapshot = collector.cumulative(); + assert!(update.contributed_sample); - assert_eq!(update.rtt_ipdv, Some(Duration::from_millis(10))); + assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(10)); + assert_eq!(snapshot.ipdv.round_trip.stats.count, 1); assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 10_000_000); } @@ -1287,9 +1320,41 @@ mod tests { collector.process(&reply(0, 10, 10)); let update = collector.process(&reply(1, 13, 13)); - assert_eq!(update.rtt_ipdv, Some(Duration::from_millis(3))); - assert!(update.send_ipdv.is_some()); - assert!(update.receive_ipdv.is_some()); + assert!(update.contributed_sample); + + let pair = assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(3)); + assert!(pair.send_ipdv.is_some()); + assert!(pair.receive_ipdv.is_some()); + } + + #[test] + fn gap_fill_update_exposes_both_completed_ipdv_pairs() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + + let first = collector.process(&reply(0, 10, 10)); + let gap = collector.process(&reply(2, 20, 20)); + let fill = collector.process(&reply(1, 13, 13)); + + assert!(first.contributed_sample); + assert!(first.ipdv_pairs.is_empty()); + + assert!(gap.contributed_sample); + assert!(gap.ipdv_pairs.is_empty()); + + assert!(fill.contributed_sample); + assert_eq!(fill.ipdv_pairs.len(), 2); + + assert_eq!(fill.ipdv_pairs[0].previous_logical_seq, 0); + assert_eq!(fill.ipdv_pairs[0].current_logical_seq, 1); + assert_eq!(fill.ipdv_pairs[0].rtt_ipdv, Duration::from_millis(3)); + + assert_eq!(fill.ipdv_pairs[1].previous_logical_seq, 1); + assert_eq!(fill.ipdv_pairs[1].current_logical_seq, 2); + assert_eq!(fill.ipdv_pairs[1].rtt_ipdv, Duration::from_millis(7)); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.ipdv.round_trip.stats.count, 2); + assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 10_000_000); } #[test]