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::{ EventCounts, EventStatsUpdate, IpdvPairUpdate, IpdvStats, LateReplyMode, OneWayDelayStats, PacketCounts, RttStats, SampleMode, ServerProcessingStats, Snapshot, }; pub(crate) const CONTINUOUS_SEQUENCE_LIMIT: usize = 4096; #[derive(Debug, Clone, PartialEq)] pub(crate) struct CoreStats { events: EventCounts, packets: PacketCounts, 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, } impl CoreStats { pub(crate) fn new(sample_mode: SampleMode, late_replies: LateReplyMode) -> Self { // One decision for every public timing metric: in exact mode they all // retain their samples so they can all report an exact median. The // retention estimate in `retention.rs` counts the same set. let retain_exact = sample_mode == SampleMode::Exact; let sequence_limit = if retain_exact { None } else { 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: 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, } } pub(crate) fn apply(&mut self, event: StatsEvent) -> EventStatsUpdate { match event { StatsEvent::Sent { bytes, send_call_ns, timer_error_ns, .. } => { self.apply_sent(bytes, send_call_ns, timer_error_ns); EventStatsUpdate::default() } StatsEvent::UniqueReply { is_late, sample, .. } => self.apply_unique_reply(is_late, *sample), StatsEvent::DuplicateReply { bytes, .. } => { self.apply_duplicate_reply(bytes); EventStatsUpdate::default() } StatsEvent::Loss { .. } => { self.apply_loss(); EventStatsUpdate::default() } StatsEvent::Warning { .. } => { self.apply_warning(); EventStatsUpdate::default() } StatsEvent::UntrackedLate { bytes, .. } => { self.apply_untracked_late(bytes); EventStatsUpdate::default() } } } fn apply_sent(&mut self, bytes: usize, send_call_ns: i128, timer_error_ns: Option) { self.events.sent_events += 1; self.packets.packets_sent += 1; self.packets.bytes_sent = self.packets.bytes_sent.saturating_add(bytes as u64); self.send_call.push_ns(send_call_ns); if let Some(error) = timer_error_ns { self.timer_error.push_ns(error); } } fn apply_unique_reply(&mut self, is_late: bool, sample: ReplySample) -> EventStatsUpdate { self.account_unique_reply(is_late, &sample); // A matched late reply is always a late unique reply. Whether it is // also allowed to measure is the configured policy, not a property of // the event. if is_late && self.late_replies == LateReplyMode::CountOnly { return EventStatsUpdate::default(); } let mut update = EventStatsUpdate { contributed_sample: true, ..EventStatsUpdate::default() }; self.record_reply_metrics(&sample); update.ipdv_pairs = self.apply_ipdv_sample(sample.ipdv); update } fn account_unique_reply(&mut self, is_late: bool, sample: &ReplySample) { self.events.echo_replies += u64::from(!is_late); self.events.late_unique_replies += u64::from(is_late); self.packets.packets_received += 1; self.packets.unique_replies += 1; self.packets.late_packets += u64::from(is_late); self.packets.bytes_received = self .packets .bytes_received .saturating_add(sample.bytes as u64); if let Some(count) = sample.received_count { let count = u64::from(count); self.packets.server_packets_received = Some( self.packets .server_packets_received .map_or(count, |current| current.max(count)), ); } if let Some(window) = sample.received_window { self.packets.server_received_window = Some(window); } } fn record_reply_metrics(&mut self, sample: &ReplySample) { self.rtt_primary.push_ns(sample.ipdv.rtt_primary_ns); self.rtt_raw.push_ns(sample.rtt_raw_ns); if let Some(adjusted) = sample.rtt_adjusted_ns { self.rtt_adjusted.push_ns(adjusted); } if let Some(processing) = sample.server_processing_ns { self.server_processing.push_ns(processing); } if let Some(delay) = sample.send_delay_ns { self.send_delay.push_ns(delay); } if let Some(delay) = sample.receive_delay_ns { self.receive_delay.push_ns(delay); } } fn apply_ipdv_sample(&mut self, sample: IpdvSample) -> Vec { let mut updates = Vec::new(); for pair in self.ipdv_tracker.insert(sample) { let Some(rtt_ipdv) = duration_from_non_negative_i128_ns(pair.rtt_ipdv_ns) else { continue; }; let send_ipdv = pair .send_ipdv_ns .and_then(duration_from_non_negative_i128_ns); let receive_ipdv = pair .receive_ipdv_ns .and_then(duration_from_non_negative_i128_ns); self.ipdv_round_trip.push_ns(pair.rtt_ipdv_ns); if let Some(value) = pair.send_ipdv_ns { self.ipdv_send.push_ns(value); } if let Some(value) = pair.receive_ipdv_ns { self.ipdv_receive.push_ns(value); } updates.push(IpdvPairUpdate { previous_seq: pair.previous_seq, current_seq: pair.current_seq, rtt_ipdv, send_ipdv, receive_ipdv, }); } updates } fn apply_duplicate_reply(&mut self, bytes: usize) { self.events.duplicate_replies += 1; self.packets.packets_received += 1; self.packets.duplicates += 1; self.packets.bytes_received = self.packets.bytes_received.saturating_add(bytes as u64); } fn apply_loss(&mut self) { self.events.loss_events += 1; } fn apply_warning(&mut self) { self.events.warning_events += 1; } fn apply_untracked_late(&mut self, bytes: usize) { self.events.untracked_late_replies += 1; self.packets.packets_received += 1; self.packets.late_packets += 1; self.packets.bytes_received = self.packets.bytes_received.saturating_add(bytes as u64); } pub(crate) fn snapshot(&self) -> Snapshot { let packets = self.packets; Snapshot { events: self.events, packets, loss: loss_stats(packets, packets.server_packets_received), 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(), server_processing: self.server_processing_stats(), } } fn rtt_stats(&self) -> RttStats { RttStats { 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.snapshot(), send: self.ipdv_send.snapshot(), receive: self.ipdv_receive.snapshot(), } } fn one_way_delay_stats(&self) -> OneWayDelayStats { OneWayDelayStats { send_delay: self.send_delay.snapshot(), receive_delay: self.receive_delay.snapshot(), } } fn server_processing_stats(&self) -> ServerProcessingStats { ServerProcessingStats { processing: self.server_processing.snapshot(), } } } fn duration_from_non_negative_i128_ns(value: i128) -> Option { u64::try_from(value).ok().map(Duration::from_nanos) }