diff --git a/Cargo.lock b/Cargo.lock index 12e15b5..7722fb2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -779,6 +779,7 @@ name = "irtt-stats" version = "0.5.1" dependencies = [ "irtt-client", + "measurement-stats", "proptest", ] @@ -914,6 +915,12 @@ dependencies = [ "digest 0.11.3", ] +[[package]] +name = "measurement-stats" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0edbfa45a060f44b15568637075ecebbe258c22f86ce6f2b5d58d7a64cbdb689" + [[package]] name = "memchr" version = "2.8.3" diff --git a/crates/irtt-stats/Cargo.toml b/crates/irtt-stats/Cargo.toml index e568eb8..c0cb606 100644 --- a/crates/irtt-stats/Cargo.toml +++ b/crates/irtt-stats/Cargo.toml @@ -12,6 +12,7 @@ keywords = ["irtt", "networking", "latency", "statistics"] categories = ["network-programming"] [dependencies] +measurement-stats = "0.1.0" irtt-client.workspace = true [[example]] diff --git a/crates/irtt-stats/README.md b/crates/irtt-stats/README.md index ee04402..ed46b97 100644 --- a/crates/irtt-stats/README.md +++ b/crates/irtt-stats/README.md @@ -10,6 +10,11 @@ This crate does not open sessions or drive sockets. Feed it the `ClientEvent` stream produced by an `irtt-client` `Client`, `AsyncClient`, or managed session via `StatsCollector::process`, and read back a `Snapshot`. +Generic timing accumulation and count-window storage come from +`measurement-stats`. `TimeStats` remains available from `irtt_stats` as a +re-export. Event interpretation, IPDV pairing, loss, and snapshot reconstruction +remain here. + A `StatsCollector` has one sequence/IPDV namespace and one packet count, both scoped to a single target. A managed session covering multiple targets needs one collector per target — as the `irtt-rs` CLI does — since a target's @@ -26,6 +31,15 @@ bounded running statistics instead (no exact median) plus a bounded unbounded sessions. Rolling-window snapshots are always running-only and report no medians, whichever `StatsConfig` produced them. +An enabled time-based rolling window has no event-count cap, including in +continuous mode. It expires only on new events, anchored at their timestamps; +reading a snapshot does not advance time. Use count-based rolling storage for a +hard bound on retained events. The memory estimate excludes time-window storage. + +Exact snapshots sort a temporary copy of each metric's retained samples, adding +O(n) scratch memory and O(n log n) computation for that metric. The retention +estimate is not peak snapshot or allocator accounting. + See `examples/` in the repository for a runnable comparison of both modes. ## Documentation diff --git a/crates/irtt-stats/src/core.rs b/crates/irtt-stats/src/core.rs index 4a4f8f2..f8802b6 100644 --- a/crates/irtt-stats/src/core.rs +++ b/crates/irtt-stats/src/core.rs @@ -1,9 +1,10 @@ use std::time::Duration; +use measurement_stats::TimeAccumulator; + use crate::ipdv::{IpdvSample, IpdvTracker}; use crate::loss::loss_stats; use crate::normalization::{ReplySample, StatsEvent}; -use crate::time_stats::TimeMetric; use crate::{ EventCounts, EventStatsUpdate, IpdvPairUpdate, IpdvStats, LateReplyMode, OneWayDelayStats, PacketCounts, RttStats, SampleMode, ServerProcessingStats, Snapshot, @@ -15,17 +16,17 @@ pub(crate) const CONTINUOUS_SEQUENCE_LIMIT: usize = 4096; pub(crate) struct CoreStats { events: EventCounts, packets: PacketCounts, - send_call: TimeMetric, - timer_error: TimeMetric, - rtt_primary: TimeMetric, - rtt_raw: TimeMetric, - rtt_adjusted: TimeMetric, - ipdv_round_trip: TimeMetric, - ipdv_send: TimeMetric, - ipdv_receive: TimeMetric, - send_delay: TimeMetric, - receive_delay: TimeMetric, - server_processing: TimeMetric, + send_call: TimeAccumulator, + timer_error: TimeAccumulator, + rtt_primary: TimeAccumulator, + rtt_raw: TimeAccumulator, + rtt_adjusted: TimeAccumulator, + ipdv_round_trip: TimeAccumulator, + ipdv_send: TimeAccumulator, + ipdv_receive: TimeAccumulator, + send_delay: TimeAccumulator, + receive_delay: TimeAccumulator, + server_processing: TimeAccumulator, ipdv_tracker: IpdvTracker, late_replies: LateReplyMode, } @@ -42,20 +43,25 @@ impl CoreStats { Some(CONTINUOUS_SEQUENCE_LIMIT) }; + let metric = || match sample_mode { + SampleMode::Exact => TimeAccumulator::exact_unbounded(), + SampleMode::RunningOnly => TimeAccumulator::running(), + }; + Self { events: EventCounts::default(), packets: PacketCounts::default(), - send_call: TimeMetric::new(retain_exact), - timer_error: TimeMetric::new(retain_exact), - rtt_primary: TimeMetric::new(retain_exact), - rtt_raw: TimeMetric::new(retain_exact), - rtt_adjusted: TimeMetric::new(retain_exact), - ipdv_round_trip: TimeMetric::new(retain_exact), - ipdv_send: TimeMetric::new(retain_exact), - ipdv_receive: TimeMetric::new(retain_exact), - send_delay: TimeMetric::new(retain_exact), - receive_delay: TimeMetric::new(retain_exact), - server_processing: TimeMetric::new(retain_exact), + send_call: metric(), + timer_error: metric(), + rtt_primary: metric(), + rtt_raw: metric(), + rtt_adjusted: metric(), + ipdv_round_trip: metric(), + ipdv_send: metric(), + ipdv_receive: metric(), + send_delay: metric(), + receive_delay: metric(), + server_processing: metric(), ipdv_tracker: IpdvTracker::new(sequence_limit), late_replies, } @@ -227,8 +233,8 @@ impl CoreStats { events: self.events, packets, loss: loss_stats(packets), - send_call: self.send_call.stats(), - timer_error: self.timer_error.stats(), + send_call: self.send_call.snapshot(), + timer_error: self.timer_error.snapshot(), rtt: self.rtt_stats(), ipdv: self.ipdv_stats(), one_way_delay: self.one_way_delay_stats(), @@ -238,30 +244,30 @@ impl CoreStats { fn rtt_stats(&self) -> RttStats { RttStats { - primary: self.rtt_primary.stats(), - raw: self.rtt_raw.stats(), - adjusted: self.rtt_adjusted.stats(), + primary: self.rtt_primary.snapshot(), + raw: self.rtt_raw.snapshot(), + adjusted: self.rtt_adjusted.snapshot(), } } fn ipdv_stats(&self) -> IpdvStats { IpdvStats { - round_trip: self.ipdv_round_trip.stats(), - send: self.ipdv_send.stats(), - receive: self.ipdv_receive.stats(), + round_trip: self.ipdv_round_trip.snapshot(), + send: self.ipdv_send.snapshot(), + receive: self.ipdv_receive.snapshot(), } } fn one_way_delay_stats(&self) -> OneWayDelayStats { OneWayDelayStats { - send_delay: self.send_delay.stats(), - receive_delay: self.receive_delay.stats(), + send_delay: self.send_delay.snapshot(), + receive_delay: self.receive_delay.snapshot(), } } fn server_processing_stats(&self) -> ServerProcessingStats { ServerProcessingStats { - processing: self.server_processing.stats(), + processing: self.server_processing.snapshot(), } } } diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index 0d604b0..6461898 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -15,8 +15,9 @@ //! [`StatsConfig::continuous`] ([`SampleMode::RunningOnly`]) keeps only //! running statistics — no exact median — and bounds its adjacent-sequence //! IPDV tracking, so memory stays bounded for a long-running or unbounded -//! session. Rolling snapshots always use running statistics regardless of -//! this setting. +//! session. An enabled time-based rolling window has no item cap and is not +//! hard-bounded; it expires only when new events arrive. Rolling snapshots +//! always use running statistics regardless of this setting. //! //! See `examples/` in the repository for a runnable comparison of both modes. @@ -34,13 +35,13 @@ mod loss; mod normalization; mod retention; mod rolling; -mod time_stats; use core::CoreStats; pub use loss::LossStats; +#[doc(inline)] +pub use measurement_stats::TimeStats; use normalization::normalize_event; use rolling::RollingEvents; -pub use time_stats::TimeStats; #[derive(Debug, Clone, Copy, PartialEq, Eq)] /// Configuration for statistics collection. @@ -54,6 +55,10 @@ pub struct StatsConfig { /// and one unique reply event. pub rolling_count: Option, /// Time span of recent normalized events retained for time-based rolling snapshots. + /// + /// This window has no item cap, even in continuous mode. Its memory depends + /// on event volume inside the interval. Expiry occurs on insertion, anchored + /// at the latest event timestamp, not when a snapshot is read. pub rolling_time: Option, /// Whether matched late replies contribute measurements. pub late_replies: LateReplyMode, @@ -117,7 +122,8 @@ impl StatsConfig { /// /// Continuous mode uses running statistics, retains no exact samples so no /// timing metric reports a median, and bounds adjacent-sequence IPDV - /// tracking for long-running sessions. + /// tracking for long-running sessions. An explicitly enabled `rolling_time` + /// window is not hard-bounded; use `rolling_count` for bounded event storage. /// /// # Example /// diff --git a/crates/irtt-stats/src/rolling.rs b/crates/irtt-stats/src/rolling.rs index 0f03361..c521de1 100644 --- a/crates/irtt-stats/src/rolling.rs +++ b/crates/irtt-stats/src/rolling.rs @@ -1,14 +1,15 @@ use std::{collections::VecDeque, time::Duration}; +use measurement_stats::CountWindow; + use crate::{ core::CoreStats, normalization::StatsEvent, LateReplyMode, SampleMode, Snapshot, StatsConfig, }; #[derive(Debug, Clone, PartialEq)] pub(crate) struct RollingEvents { - count_limit: Option, time_limit: Option, - count_events: Option>, + count_events: Option>, time_events: Option>, late_replies: LateReplyMode, } @@ -16,20 +17,16 @@ pub(crate) struct RollingEvents { impl RollingEvents { pub(crate) fn new(config: StatsConfig) -> Self { Self { - count_limit: config.rolling_count, time_limit: config.rolling_time, - count_events: config.rolling_count.map(|_| VecDeque::new()), + count_events: config.rolling_count.map(CountWindow::new), time_events: config.rolling_time.map(|_| VecDeque::new()), late_replies: config.late_replies, } } pub(crate) fn push(&mut self, event: StatsEvent) { - if let (Some(limit), Some(window)) = (self.count_limit, self.count_events.as_mut()) { - window.push_back(event.clone()); - while window.len() > limit { - window.pop_front(); - } + if let Some(window) = self.count_events.as_mut() { + window.push(event.clone()); } if let (Some(duration), Some(window)) = (self.time_limit, self.time_events.as_mut()) { @@ -46,19 +43,22 @@ impl RollingEvents { pub(crate) fn count_snapshot(&self) -> Option { self.count_events .as_ref() - .map(|events| snapshot_window(events, self.late_replies)) + .map(|events| snapshot_window(events.iter(), self.late_replies)) } pub(crate) fn time_snapshot(&self) -> Option { self.time_events .as_ref() - .map(|events| snapshot_window(events, self.late_replies)) + .map(|events| snapshot_window(events.iter(), self.late_replies)) } } /// Recompute a window snapshot under the same normalized semantics as the /// cumulative collector, including its late-reply measurement policy. -fn snapshot_window(events: &VecDeque, late_replies: LateReplyMode) -> Snapshot { +fn snapshot_window<'a>( + events: impl Iterator, + late_replies: LateReplyMode, +) -> Snapshot { let mut core = CoreStats::new(SampleMode::RunningOnly, late_replies); for event in events { core.apply(event.clone()); diff --git a/crates/irtt-stats/src/time_stats.rs b/crates/irtt-stats/src/time_stats.rs deleted file mode 100644 index c7f5feb..0000000 --- a/crates/irtt-stats/src/time_stats.rs +++ /dev/null @@ -1,160 +0,0 @@ -#[derive(Debug, Clone, Copy, PartialEq)] -/// Summary statistics for timing values. -/// -/// All fields use nanoseconds except `variance_ns2`, which is nanoseconds -/// squared. Values may represent signed timing quantities such as scheduling -/// error or one-way delay. -pub struct TimeStats { - /// Number of samples included in this summary. - pub count: u64, - /// Sum of all samples, in nanoseconds. - pub total_ns: i128, - /// Smallest sample, in nanoseconds. - pub min_ns: Option, - /// Largest sample, in nanoseconds. - pub max_ns: Option, - /// Arithmetic mean, in nanoseconds. - pub mean_ns: f64, - /// Median, in nanoseconds, when exact samples were retained. - /// - /// This is `None` when the collector was not configured to retain exact - /// samples, and also when it was but this metric received none. - pub median_ns: Option, - /// Sample variance, in nanoseconds squared. - pub variance_ns2: f64, -} - -impl TimeStats { - /// Returns the sample standard deviation, in nanoseconds. - pub fn stddev_ns(&self) -> f64 { - self.variance_ns2.sqrt() - } - - /// Returns whether this summary contains no samples. - pub fn is_empty(&self) -> bool { - self.count == 0 - } -} - -#[derive(Debug, Clone, PartialEq)] -pub(crate) struct TimeMetric { - running: RunningStats, - samples: Option>, -} - -impl TimeMetric { - pub(crate) fn new(retain_samples: bool) -> Self { - Self { - running: RunningStats::default(), - samples: retain_samples.then(Vec::new), - } - } - - pub(crate) fn push_ns(&mut self, value: i128) { - self.running.push(value); - if let Some(samples) = self.samples.as_mut() { - samples.push(value); - } - } - - pub(crate) fn stats(&self) -> TimeStats { - self.running - .stats(self.samples.as_ref().and_then(|samples| median_ns(samples))) - } -} - -#[derive(Debug, Clone, PartialEq, Default)] -struct RunningStats { - count: u64, - total_ns: i128, - min_ns: Option, - max_ns: Option, - mean_ns: f64, - m2_ns2: f64, -} - -impl RunningStats { - fn push(&mut self, value: i128) { - self.count += 1; - self.total_ns = self.total_ns.saturating_add(value); - self.min_ns = Some(self.min_ns.map_or(value, |min| min.min(value))); - self.max_ns = Some(self.max_ns.map_or(value, |max| max.max(value))); - let x = value as f64; - let delta = x - self.mean_ns; - self.mean_ns += delta / self.count as f64; - let delta2 = x - self.mean_ns; - self.m2_ns2 += delta * delta2; - } - - fn stats(&self, median_ns: Option) -> TimeStats { - TimeStats { - count: self.count, - total_ns: self.total_ns, - min_ns: self.min_ns, - max_ns: self.max_ns, - mean_ns: if self.count == 0 { 0.0 } else { self.mean_ns }, - median_ns, - variance_ns2: sample_variance(self.count, self.m2_ns2), - } - } -} - -fn sample_variance(count: u64, m2: f64) -> f64 { - if count < 2 { - 0.0 - } else { - m2 / (count - 1) as f64 - } -} - -fn median_ns(samples: &[i128]) -> Option { - if samples.is_empty() { - return None; - } - let mut sorted = samples.to_vec(); - sorted.sort_unstable(); - let mid = sorted.len() / 2; - Some(if sorted.len() % 2 == 1 { - sorted[mid] as f64 - } else { - (sorted[mid - 1] as f64 + sorted[mid] as f64) / 2.0 - }) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn running_duration_stats_use_sample_variance() { - let mut metric = TimeMetric::new(false); - metric.push_ns(1); - metric.push_ns(2); - metric.push_ns(3); - let stats = metric.stats(); - assert_eq!(stats.count, 3); - assert_eq!(stats.total_ns, 6); - assert_eq!(stats.min_ns, Some(1)); - assert_eq!(stats.max_ns, Some(3)); - assert_eq!(stats.mean_ns, 2.0); - assert_eq!(stats.variance_ns2, 1.0); - assert_eq!(stats.stddev_ns(), 1.0); - } - - #[test] - fn exact_median_handles_odd_and_even_samples() { - assert_eq!(median_ns(&[3, 1, 2]), Some(2.0)); - assert_eq!(median_ns(&[4, 1, 2, 3]), Some(2.5)); - assert_eq!(median_ns(&[-5, 1, 3]), Some(1.0)); - assert_eq!(median_ns(&[-5, 1, 3, 7]), Some(2.0)); - } - - #[test] - fn single_sample_stddev_is_zero() { - let mut metric = TimeMetric::new(false); - metric.push_ns(42); - let stats = metric.stats(); - assert_eq!(stats.variance_ns2, 0.0); - assert_eq!(stats.stddev_ns(), 0.0); - } -}