refactor(stats): make late-reply sampling explicit #92
No files matched your search
@@ -258,7 +258,7 @@ impl TuiState {
|
||||
let Some(target) = self.targets.get_mut(target_idx) else {
|
||||
return;
|
||||
};
|
||||
process_tui_stats(event, &mut target.stats);
|
||||
target.stats.process(event);
|
||||
match event {
|
||||
ClientEvent::SessionStarted {
|
||||
remote,
|
||||
@@ -616,43 +616,21 @@ impl GroupPacingArg {
|
||||
}
|
||||
}
|
||||
|
||||
/// Statistics policy for the TUI.
|
||||
///
|
||||
/// The TUI treats late replies as diagnostics, so it counts a matched late
|
||||
/// reply without letting it contribute timing samples. That is an explicit
|
||||
/// statistics policy rather than something the TUI expresses by degrading the
|
||||
/// event it forwards.
|
||||
fn stats_config(continuous: bool) -> irtt_stats::StatsConfig {
|
||||
if continuous {
|
||||
let base = if continuous {
|
||||
irtt_stats::StatsConfig::continuous()
|
||||
} else {
|
||||
irtt_stats::StatsConfig::finite()
|
||||
}
|
||||
}
|
||||
|
||||
fn process_tui_stats(event: &ClientEvent, stats: &mut StatsCollector) {
|
||||
if let ClientEvent::LateReply {
|
||||
seq,
|
||||
highest_seen,
|
||||
remote,
|
||||
received_at,
|
||||
bytes,
|
||||
packet_meta,
|
||||
..
|
||||
} = event
|
||||
{
|
||||
// The TUI treats late replies as diagnostics, even when retained send
|
||||
// metadata lets the client attach RTT fields to the event.
|
||||
let late_counter_event = ClientEvent::LateReply {
|
||||
seq: *seq,
|
||||
highest_seen: *highest_seen,
|
||||
remote: *remote,
|
||||
sent_at: None,
|
||||
received_at: *received_at,
|
||||
rtt: None,
|
||||
server_timing: None,
|
||||
one_way: None,
|
||||
received_stats: None,
|
||||
bytes: *bytes,
|
||||
packet_meta: *packet_meta,
|
||||
};
|
||||
stats.process(&late_counter_event);
|
||||
} else {
|
||||
stats.process(event);
|
||||
};
|
||||
irtt_stats::StatsConfig {
|
||||
late_replies: irtt_stats::LateReplyMode::CountOnly,
|
||||
..base
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2267,12 +2245,15 @@ mod tests {
|
||||
assert!(target.graph_history.is_empty());
|
||||
let snapshot = target.stats.snapshot();
|
||||
assert_eq!(snapshot.events.duplicate_replies, 1);
|
||||
assert_eq!(snapshot.events.late_unique_replies, 0);
|
||||
assert_eq!(snapshot.events.untracked_late_replies, 1);
|
||||
assert_eq!(snapshot.packets.unique_replies, 0);
|
||||
// A matched late reply is a late unique reply. The TUI suppresses its
|
||||
// measurements through the late-reply policy, not by reclassifying it.
|
||||
assert_eq!(snapshot.events.late_unique_replies, 1);
|
||||
assert_eq!(snapshot.events.untracked_late_replies, 0);
|
||||
assert_eq!(snapshot.packets.unique_replies, 1);
|
||||
assert_eq!(snapshot.packets.duplicates, 1);
|
||||
assert_eq!(snapshot.packets.late_packets, 1);
|
||||
assert_eq!(snapshot.rtt.primary.count, 0);
|
||||
assert_eq!(snapshot.ipdv.round_trip.count, 0);
|
||||
assert!(state
|
||||
.recent_events
|
||||
.iter()
|
||||
@@ -2283,6 +2264,32 @@ mod tests {
|
||||
.any(|event| event.contains("late seq=8")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn untracked_late_replies_stay_untracked_in_the_tui() {
|
||||
let mut state = TuiState::default();
|
||||
let target = target_instance("target");
|
||||
let late = ClientEvent::LateReply {
|
||||
seq: 8,
|
||||
highest_seen: 9,
|
||||
remote: remote(),
|
||||
sent_at: None,
|
||||
received_at: ts(Duration::from_secs(2)),
|
||||
rtt: None,
|
||||
server_timing: None,
|
||||
one_way: None,
|
||||
received_stats: None,
|
||||
bytes: 64,
|
||||
packet_meta: PacketMeta::default(),
|
||||
};
|
||||
state.process_target_event(&target, &late);
|
||||
|
||||
let snapshot = primary_target(&state).stats.snapshot();
|
||||
assert_eq!(snapshot.events.untracked_late_replies, 1);
|
||||
assert_eq!(snapshot.events.late_unique_replies, 0);
|
||||
assert_eq!(snapshot.packets.unique_replies, 0);
|
||||
assert_eq!(snapshot.packets.late_packets, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn target_scoped_echo_reply_updates_only_that_target() {
|
||||
let mut state =
|
||||
|
||||
@@ -5,8 +5,8 @@ use crate::loss::loss_stats;
|
||||
use crate::normalization::{ReplySample, StatsEvent};
|
||||
use crate::time_stats::TimeMetric;
|
||||
use crate::{
|
||||
EventCounts, EventStatsUpdate, IpdvPairUpdate, IpdvStats, OneWayDelayStats, PacketCounts,
|
||||
RttStats, SampleMode, ServerProcessingStats, Snapshot,
|
||||
EventCounts, EventStatsUpdate, IpdvPairUpdate, IpdvStats, LateReplyMode, OneWayDelayStats,
|
||||
PacketCounts, RttStats, SampleMode, ServerProcessingStats, Snapshot,
|
||||
};
|
||||
|
||||
const CONTINUOUS_SEQUENCE_LIMIT: usize = 4096;
|
||||
@@ -27,10 +27,11 @@ pub(crate) struct CoreStats {
|
||||
receive_delay: TimeMetric,
|
||||
server_processing: TimeMetric,
|
||||
ipdv_tracker: IpdvTracker,
|
||||
late_replies: LateReplyMode,
|
||||
}
|
||||
|
||||
impl CoreStats {
|
||||
pub(crate) fn new(sample_mode: SampleMode) -> Self {
|
||||
pub(crate) fn new(sample_mode: SampleMode, late_replies: LateReplyMode) -> Self {
|
||||
let sequence_limit = if sample_mode == SampleMode::Exact {
|
||||
None
|
||||
} else {
|
||||
@@ -52,6 +53,7 @@ impl CoreStats {
|
||||
receive_delay: TimeMetric::new(sample_mode == SampleMode::Exact),
|
||||
server_processing: TimeMetric::new(false),
|
||||
ipdv_tracker: IpdvTracker::new(sequence_limit),
|
||||
late_replies,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -97,12 +99,19 @@ impl CoreStats {
|
||||
}
|
||||
|
||||
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.account_unique_reply(is_late, &sample);
|
||||
self.record_reply_metrics(&sample);
|
||||
update.ipdv_pairs = self.apply_ipdv_sample(sample.ipdv);
|
||||
|
||||
|
||||
@@ -36,6 +36,8 @@ pub struct StatsConfig {
|
||||
pub rolling_count: Option<usize>,
|
||||
/// Time span of recent normalized events retained for time-based rolling snapshots.
|
||||
pub rolling_time: Option<Duration>,
|
||||
/// Whether matched late replies contribute measurements.
|
||||
pub late_replies: LateReplyMode,
|
||||
}
|
||||
|
||||
impl StatsConfig {
|
||||
@@ -49,6 +51,7 @@ impl StatsConfig {
|
||||
samples: SampleMode::Exact,
|
||||
rolling_count: None,
|
||||
rolling_time: None,
|
||||
late_replies: LateReplyMode::Measure,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -62,10 +65,34 @@ impl StatsConfig {
|
||||
samples: SampleMode::RunningOnly,
|
||||
rolling_count: None,
|
||||
rolling_time: None,
|
||||
late_replies: LateReplyMode::Measure,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
|
||||
/// Whether a matched late reply contributes measurements.
|
||||
///
|
||||
/// This is a measurement policy, not a classification. A late reply that the
|
||||
/// client matched to retained send state is a late unique reply in both modes,
|
||||
/// and a late reply that could not be matched remains an untracked late reply
|
||||
/// in both modes. Only what a matched late reply is allowed to measure differs,
|
||||
/// so a caller that treats late replies as diagnostics does not have to degrade
|
||||
/// the event itself to suppress its timing contribution.
|
||||
pub enum LateReplyMode {
|
||||
/// Matched late replies contribute timing, one-way, server-processing, and
|
||||
/// IPDV measurements like any other unique reply.
|
||||
#[default]
|
||||
Measure,
|
||||
/// Matched late replies are counted but measure nothing.
|
||||
///
|
||||
/// Packet, byte, late, and server-reported receive counters still update,
|
||||
/// so the reply stays visible as a late unique reply. Timing metrics,
|
||||
/// one-way and server-processing measurements, and IPDV do not update, and
|
||||
/// [`EventStatsUpdate::contributed_sample`] is `false`.
|
||||
CountOnly,
|
||||
}
|
||||
|
||||
impl Default for StatsConfig {
|
||||
fn default() -> Self {
|
||||
Self::finite()
|
||||
@@ -95,7 +122,7 @@ impl StatsCollector {
|
||||
/// Creates a collector with the supplied configuration.
|
||||
pub fn new(config: StatsConfig) -> Self {
|
||||
Self {
|
||||
cumulative: CoreStats::new(config.samples),
|
||||
cumulative: CoreStats::new(config.samples, config.late_replies),
|
||||
rolling: RollingEvents::new(config),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
use std::{collections::VecDeque, time::Duration};
|
||||
|
||||
use crate::{core::CoreStats, normalization::StatsEvent, SampleMode, Snapshot, StatsConfig};
|
||||
use crate::{
|
||||
core::CoreStats, normalization::StatsEvent, LateReplyMode, SampleMode, Snapshot, StatsConfig,
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub(crate) struct RollingEvents {
|
||||
@@ -8,6 +10,7 @@ pub(crate) struct RollingEvents {
|
||||
time_limit: Option<Duration>,
|
||||
count_events: Option<VecDeque<StatsEvent>>,
|
||||
time_events: Option<VecDeque<StatsEvent>>,
|
||||
late_replies: LateReplyMode,
|
||||
}
|
||||
|
||||
impl RollingEvents {
|
||||
@@ -17,6 +20,7 @@ impl RollingEvents {
|
||||
time_limit: config.rolling_time,
|
||||
count_events: config.rolling_count.map(|_| VecDeque::new()),
|
||||
time_events: config.rolling_time.map(|_| VecDeque::new()),
|
||||
late_replies: config.late_replies,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40,16 +44,22 @@ impl RollingEvents {
|
||||
}
|
||||
|
||||
pub(crate) fn count_snapshot(&self) -> Option<Snapshot> {
|
||||
self.count_events.as_ref().map(snapshot_window)
|
||||
self.count_events
|
||||
.as_ref()
|
||||
.map(|events| snapshot_window(events, self.late_replies))
|
||||
}
|
||||
|
||||
pub(crate) fn time_snapshot(&self) -> Option<Snapshot> {
|
||||
self.time_events.as_ref().map(snapshot_window)
|
||||
self.time_events
|
||||
.as_ref()
|
||||
.map(|events| snapshot_window(events, self.late_replies))
|
||||
}
|
||||
}
|
||||
|
||||
fn snapshot_window(events: &VecDeque<StatsEvent>) -> Snapshot {
|
||||
let mut core = CoreStats::new(SampleMode::RunningOnly);
|
||||
/// Recompute a window snapshot under the same normalized semantics as the
|
||||
/// cumulative collector, including its late-reply measurement policy.
|
||||
fn snapshot_window(events: &VecDeque<StatsEvent>, late_replies: LateReplyMode) -> Snapshot {
|
||||
let mut core = CoreStats::new(SampleMode::RunningOnly, late_replies);
|
||||
for event in events {
|
||||
core.apply(event.clone());
|
||||
}
|
||||
|
||||
@@ -8,7 +8,9 @@ use common::{
|
||||
use irtt_client::{
|
||||
ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, SignedDuration,
|
||||
};
|
||||
use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig};
|
||||
use irtt_stats::{
|
||||
EventStatsUpdate, IpdvPairUpdate, LateReplyMode, SampleMode, StatsCollector, StatsConfig,
|
||||
};
|
||||
|
||||
#[test]
|
||||
fn running_only_samples_avoid_finite_retention() {
|
||||
@@ -342,6 +344,7 @@ fn rolling_count_eviction_recomputes_from_bounded_events() {
|
||||
samples: SampleMode::RunningOnly,
|
||||
rolling_count: Some(2),
|
||||
rolling_time: None,
|
||||
late_replies: LateReplyMode::Measure,
|
||||
});
|
||||
collector.process(&sent(0, ts(0)));
|
||||
collector.process(&unadjusted_reply(0, 10));
|
||||
@@ -359,6 +362,7 @@ fn rolling_time_eviction_uses_event_timestamps() {
|
||||
samples: SampleMode::RunningOnly,
|
||||
rolling_count: None,
|
||||
rolling_time: Some(Duration::from_millis(15)),
|
||||
late_replies: LateReplyMode::Measure,
|
||||
});
|
||||
collector.process(&sent(0, ts(0)));
|
||||
collector.process(&sent(1, ts(10)));
|
||||
@@ -420,3 +424,120 @@ fn directional_loss_uses_server_received_count_when_available() {
|
||||
assert_eq!(loss.packet_loss_percent, 50.0);
|
||||
assert_eq!(loss.upstream_loss_percent, 50.0);
|
||||
}
|
||||
|
||||
fn count_only(base: StatsConfig) -> StatsConfig {
|
||||
StatsConfig {
|
||||
late_replies: LateReplyMode::CountOnly,
|
||||
..base
|
||||
}
|
||||
}
|
||||
|
||||
fn untracked_late_reply(seq: u32, received_ms: u64) -> ClientEvent {
|
||||
ClientEvent::LateReply {
|
||||
seq,
|
||||
highest_seen: seq + 1,
|
||||
remote: "127.0.0.1:2112".parse().unwrap(),
|
||||
sent_at: None,
|
||||
received_at: ts(received_ms),
|
||||
rtt: None,
|
||||
server_timing: None,
|
||||
one_way: None,
|
||||
received_stats: None,
|
||||
bytes: 64,
|
||||
packet_meta: PacketMeta::default(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn matched_late_reply_measures_under_the_default_policy() {
|
||||
let mut collector = StatsCollector::new(StatsConfig::finite());
|
||||
collector.process(&sent(0, ts(0)));
|
||||
let update = collector.process(&unadjusted_late_reply(0, 10));
|
||||
|
||||
assert!(update.contributed_sample);
|
||||
|
||||
let snapshot = collector.snapshot();
|
||||
assert_eq!(snapshot.events.late_unique_replies, 1);
|
||||
assert_eq!(snapshot.events.untracked_late_replies, 0);
|
||||
assert_eq!(snapshot.packets.unique_replies, 1);
|
||||
assert_eq!(snapshot.packets.late_packets, 1);
|
||||
assert_eq!(snapshot.packets.packets_received, 1);
|
||||
assert_eq!(snapshot.packets.bytes_received, 64);
|
||||
assert_eq!(snapshot.rtt.primary.count, 1);
|
||||
assert_eq!(snapshot.server_processing.processing.count, 1);
|
||||
assert_eq!(snapshot.one_way_delay.send_delay.count, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn matched_late_reply_counts_without_measuring_under_count_only() {
|
||||
let mut collector = StatsCollector::new(count_only(StatsConfig::finite()));
|
||||
collector.process(&sent(0, ts(0)));
|
||||
let update = collector.process(&unadjusted_late_reply(0, 10));
|
||||
|
||||
assert!(!update.contributed_sample);
|
||||
assert!(update.ipdv_pairs.is_empty());
|
||||
|
||||
let snapshot = collector.snapshot();
|
||||
// Still a matched late unique reply, not a reclassified untracked one.
|
||||
assert_eq!(snapshot.events.late_unique_replies, 1);
|
||||
assert_eq!(snapshot.events.untracked_late_replies, 0);
|
||||
assert_eq!(snapshot.packets.unique_replies, 1);
|
||||
assert_eq!(snapshot.packets.late_packets, 1);
|
||||
assert_eq!(snapshot.packets.packets_received, 1);
|
||||
assert_eq!(snapshot.packets.bytes_received, 64);
|
||||
// ...but it measures nothing.
|
||||
assert_eq!(snapshot.rtt.primary.count, 0);
|
||||
assert_eq!(snapshot.rtt.raw.count, 0);
|
||||
assert_eq!(snapshot.server_processing.processing.count, 0);
|
||||
assert_eq!(snapshot.one_way_delay.send_delay.count, 0);
|
||||
assert_eq!(snapshot.one_way_delay.receive_delay.count, 0);
|
||||
assert_eq!(snapshot.ipdv.round_trip.count, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn count_only_late_replies_do_not_complete_ipdv_pairs() {
|
||||
let mut collector = StatsCollector::new(count_only(StatsConfig::finite()));
|
||||
collector.process(&unadjusted_reply(0, 10));
|
||||
let update = collector.process(&unadjusted_late_reply(1, 13));
|
||||
|
||||
assert!(update.ipdv_pairs.is_empty());
|
||||
assert_eq!(collector.snapshot().ipdv.round_trip.count, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn untracked_late_replies_stay_untracked_in_both_policies() {
|
||||
for config in [StatsConfig::finite(), count_only(StatsConfig::finite())] {
|
||||
let mut collector = StatsCollector::new(config);
|
||||
let update = collector.process(&untracked_late_reply(9, 100));
|
||||
|
||||
assert!(!update.contributed_sample);
|
||||
|
||||
let snapshot = collector.snapshot();
|
||||
assert_eq!(snapshot.events.untracked_late_replies, 1);
|
||||
assert_eq!(snapshot.events.late_unique_replies, 0);
|
||||
assert_eq!(snapshot.packets.unique_replies, 0);
|
||||
assert_eq!(snapshot.packets.late_packets, 1);
|
||||
assert_eq!(snapshot.packets.packets_received, 1);
|
||||
assert_eq!(snapshot.rtt.primary.count, 0);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rolling_snapshots_use_the_same_late_reply_policy() {
|
||||
let mut collector = StatsCollector::new(StatsConfig {
|
||||
rolling_count: Some(8),
|
||||
rolling_time: None,
|
||||
..count_only(StatsConfig::continuous())
|
||||
});
|
||||
collector.process(&sent(0, ts(0)));
|
||||
collector.process(&unadjusted_reply(0, 10));
|
||||
collector.process(&unadjusted_late_reply(1, 10));
|
||||
|
||||
let rolling = collector.rolling_count().unwrap();
|
||||
assert_eq!(rolling.events.late_unique_replies, 1);
|
||||
assert_eq!(rolling.events.untracked_late_replies, 0);
|
||||
assert_eq!(rolling.packets.unique_replies, 2);
|
||||
assert_eq!(rolling.packets.late_packets, 1);
|
||||
assert_eq!(rolling.rtt.primary.count, 1);
|
||||
assert_eq!(rolling.ipdv.round_trip.count, 0);
|
||||
}
|
||||
Reference in new issue
Block a user