diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index 6433fb5..213c3cd 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -2,10 +2,7 @@ #![forbid(unsafe_code)] -use std::{ - collections::VecDeque, - time::{Duration, Instant, SystemTime, UNIX_EPOCH}, -}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use irtt_client::{ ClientEvent, ClientTimestamp, OneWayDelaySample, ReceivedStatsSample, RttSample, ServerTiming, @@ -14,11 +11,13 @@ use irtt_client::{ mod ipdv; mod loss; +mod rolling; mod time_stats; use ipdv::{IpdvSample, IpdvTracker}; use loss::loss_stats; pub use loss::LossStats; +use rolling::RollingEvents; use time_stats::TimeMetric; pub use time_stats::TimeStats; @@ -102,52 +101,6 @@ impl StatsCollector { } } -#[derive(Debug, Clone, PartialEq)] -struct RollingEvents { - count_limit: Option, - time_limit: Option, - count_events: Option>, - time_events: Option>, -} - -impl RollingEvents { - fn new(config: StatsConfig) -> Self { - Self { - count_limit: config.rolling_count, - time_limit: config.rolling_time, - count_events: config.rolling_count.map(|_| VecDeque::new()), - time_events: config.rolling_time.map(|_| VecDeque::new()), - } - } - - 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(duration), Some(window)) = (self.time_limit, self.time_events.as_mut()) { - let cutoff = event.at().checked_sub(duration); - window.push_back(event); - if let Some(cutoff) = cutoff { - while window.front().is_some_and(|event| event.at() < cutoff) { - window.pop_front(); - } - } - } - } - - fn count_snapshot(&self) -> Option { - self.count_events.as_ref().map(snapshot_window) - } - - fn time_snapshot(&self) -> Option { - self.time_events.as_ref().map(snapshot_window) - } -} - #[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct EventStatsUpdate { pub contributed_sample: bool, @@ -631,14 +584,6 @@ impl ReplySample { } } -fn snapshot_window(events: &VecDeque) -> Snapshot { - let mut core = CoreStats::new(SampleMode::RunningOnly); - for event in events { - core.apply(event.clone()); - } - core.snapshot() -} - fn system_time_ns(time: SystemTime) -> Option { if let Ok(duration) = time.duration_since(UNIX_EPOCH) { return Some(duration_ns_i128(duration)); diff --git a/crates/irtt-stats/src/rolling.rs b/crates/irtt-stats/src/rolling.rs new file mode 100644 index 0000000..fc5e3f7 --- /dev/null +++ b/crates/irtt-stats/src/rolling.rs @@ -0,0 +1,57 @@ +use std::{collections::VecDeque, time::Duration}; + +use crate::{CoreStats, SampleMode, Snapshot, StatsConfig, StatsEvent}; + +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct RollingEvents { + count_limit: Option, + time_limit: Option, + count_events: Option>, + time_events: Option>, +} + +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()), + time_events: config.rolling_time.map(|_| VecDeque::new()), + } + } + + 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(duration), Some(window)) = (self.time_limit, self.time_events.as_mut()) { + let cutoff = event.at().checked_sub(duration); + window.push_back(event); + if let Some(cutoff) = cutoff { + while window.front().is_some_and(|event| event.at() < cutoff) { + window.pop_front(); + } + } + } + } + + pub(crate) fn count_snapshot(&self) -> Option { + self.count_events.as_ref().map(snapshot_window) + } + + pub(crate) fn time_snapshot(&self) -> Option { + self.time_events.as_ref().map(snapshot_window) + } +} + +fn snapshot_window(events: &VecDeque) -> Snapshot { + let mut core = CoreStats::new(SampleMode::RunningOnly); + for event in events { + core.apply(event.clone()); + } + core.snapshot() +}