refactor(irtt-stats): extract rolling events module
This commit is contained in:
2 files changed
+60
-58
No files matched your search
@@ -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<usize>,
|
||||
time_limit: Option<Duration>,
|
||||
count_events: Option<VecDeque<StatsEvent>>,
|
||||
time_events: Option<VecDeque<StatsEvent>>,
|
||||
}
|
||||
|
||||
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<Snapshot> {
|
||||
self.count_events.as_ref().map(snapshot_window)
|
||||
}
|
||||
|
||||
fn time_snapshot(&self) -> Option<Snapshot> {
|
||||
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<StatsEvent>) -> 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<i128> {
|
||||
if let Ok(duration) = time.duration_since(UNIX_EPOCH) {
|
||||
return Some(duration_ns_i128(duration));
|
||||
|
||||
@@ -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<usize>,
|
||||
time_limit: Option<Duration>,
|
||||
count_events: Option<VecDeque<StatsEvent>>,
|
||||
time_events: Option<VecDeque<StatsEvent>>,
|
||||
}
|
||||
|
||||
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<Snapshot> {
|
||||
self.count_events.as_ref().map(snapshot_window)
|
||||
}
|
||||
|
||||
pub(crate) fn time_snapshot(&self) -> Option<Snapshot> {
|
||||
self.time_events.as_ref().map(snapshot_window)
|
||||
}
|
||||
}
|
||||
|
||||
fn snapshot_window(events: &VecDeque<StatsEvent>) -> Snapshot {
|
||||
let mut core = CoreStats::new(SampleMode::RunningOnly);
|
||||
for event in events {
|
||||
core.apply(event.clone());
|
||||
}
|
||||
core.snapshot()
|
||||
}
|
||||
Reference in new issue
Block a user