fix(stats): support sequence number wraparound in IPDV tracking
Use wrapping arithmetic instead of checked arithmetic to ensure inter-packet delay variation (IPDV) is correctly calculated when sequence numbers wrap from u32::MAX to 0.
This commit is contained in:
1 parent
64da60d480
commit
0b16ee802d
2 files changed
+113
-10
No files matched your search
@@ -48,10 +48,8 @@ impl IpdvTracker {
|
|||||||
if let Some(pair) = self.try_pair(seq) {
|
if let Some(pair) = self.try_pair(seq) {
|
||||||
pairs.push(pair);
|
pairs.push(pair);
|
||||||
}
|
}
|
||||||
if let Some(next) = seq.checked_add(1) {
|
if let Some(pair) = self.try_pair(seq.wrapping_add(1)) {
|
||||||
if let Some(pair) = self.try_pair(next) {
|
pairs.push(pair);
|
||||||
pairs.push(pair);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
pairs
|
pairs
|
||||||
}
|
}
|
||||||
@@ -66,15 +64,13 @@ impl IpdvTracker {
|
|||||||
};
|
};
|
||||||
if self.samples.remove(&seq).is_some() {
|
if self.samples.remove(&seq).is_some() {
|
||||||
self.completed_pairs.remove(&seq);
|
self.completed_pairs.remove(&seq);
|
||||||
if let Some(next) = seq.checked_add(1) {
|
self.completed_pairs.remove(&seq.wrapping_add(1));
|
||||||
self.completed_pairs.remove(&next);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn try_pair(&mut self, current_seq: u32) -> Option<CompletedIpdvPair> {
|
fn try_pair(&mut self, current_seq: u32) -> Option<CompletedIpdvPair> {
|
||||||
let previous_seq = current_seq.checked_sub(1)?;
|
let previous_seq = current_seq.wrapping_sub(1);
|
||||||
|
|
||||||
if !self.completed_pairs.insert(current_seq) {
|
if !self.completed_pairs.insert(current_seq) {
|
||||||
return None;
|
return None;
|
||||||
@@ -237,6 +233,52 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ipdv_tracker_completes_wrapped_adjacent_pair() {
|
||||||
|
let mut tracker = IpdvTracker::new(None);
|
||||||
|
assert!(tracker.insert(ipdv_sample(u32::MAX, 10)).is_empty());
|
||||||
|
|
||||||
|
let pairs = tracker.insert(ipdv_sample(0, 14));
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
pairs,
|
||||||
|
vec![CompletedIpdvPair {
|
||||||
|
previous_seq: u32::MAX,
|
||||||
|
current_seq: 0,
|
||||||
|
rtt_ipdv_ns: 4,
|
||||||
|
send_ipdv_ns: None,
|
||||||
|
receive_ipdv_ns: None,
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ipdv_tracker_wrap_gap_does_not_complete_pair() {
|
||||||
|
let mut tracker = IpdvTracker::new(None);
|
||||||
|
assert!(tracker.insert(ipdv_sample(u32::MAX - 1, 10)).is_empty());
|
||||||
|
|
||||||
|
assert!(tracker.insert(ipdv_sample(0, 14)).is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ipdv_tracker_late_wrapped_previous_completes_pair() {
|
||||||
|
let mut tracker = IpdvTracker::new(None);
|
||||||
|
assert!(tracker.insert(ipdv_sample(0, 14)).is_empty());
|
||||||
|
|
||||||
|
let pairs = tracker.insert(ipdv_sample(u32::MAX, 10));
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
pairs,
|
||||||
|
vec![CompletedIpdvPair {
|
||||||
|
previous_seq: u32::MAX,
|
||||||
|
current_seq: 0,
|
||||||
|
rtt_ipdv_ns: 4,
|
||||||
|
send_ipdv_ns: None,
|
||||||
|
receive_ipdv_ns: None,
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn ipdv_tracker_duplicate_sequence_does_not_emit_pair_again() {
|
fn ipdv_tracker_duplicate_sequence_does_not_emit_pair_again() {
|
||||||
let mut tracker = IpdvTracker::new(None);
|
let mut tracker = IpdvTracker::new(None);
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
mod common;
|
mod common;
|
||||||
|
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant, UNIX_EPOCH};
|
||||||
|
|
||||||
use common::{adjusted_reply, sent, ts, unadjusted_late_reply, unadjusted_reply};
|
use common::{adjusted_reply, sent, ts, unadjusted_late_reply, unadjusted_reply};
|
||||||
use irtt_client::{ClientEvent, PacketMeta};
|
use irtt_client::{ClientEvent, ClientTimestamp, PacketMeta, RttSample, SignedDuration};
|
||||||
use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig};
|
use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig};
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -113,6 +113,34 @@ fn assert_one_ipdv_pair(
|
|||||||
pair
|
pair
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn reply_at(seq: u32, raw_ms: u64, sent_ms: u64) -> ClientEvent {
|
||||||
|
let sent_at = ClientTimestamp {
|
||||||
|
mono: Instant::now() + Duration::from_millis(sent_ms),
|
||||||
|
wall: UNIX_EPOCH + Duration::from_millis(sent_ms),
|
||||||
|
};
|
||||||
|
let received_at = ClientTimestamp {
|
||||||
|
mono: sent_at.mono + Duration::from_millis(raw_ms),
|
||||||
|
wall: sent_at.wall + Duration::from_millis(raw_ms),
|
||||||
|
};
|
||||||
|
let raw = Duration::from_millis(raw_ms);
|
||||||
|
ClientEvent::EchoReply {
|
||||||
|
seq,
|
||||||
|
remote: "127.0.0.1:2112".parse().unwrap(),
|
||||||
|
sent_at,
|
||||||
|
received_at,
|
||||||
|
rtt: RttSample {
|
||||||
|
raw,
|
||||||
|
adjusted: None,
|
||||||
|
effective: SignedDuration::from_duration(raw),
|
||||||
|
},
|
||||||
|
server_timing: None,
|
||||||
|
one_way: None,
|
||||||
|
received_stats: None,
|
||||||
|
bytes: 64,
|
||||||
|
packet_meta: PacketMeta::default(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn ipdv_is_sequence_adjacent_and_gap_preserving() {
|
fn ipdv_is_sequence_adjacent_and_gap_preserving() {
|
||||||
let mut collector = StatsCollector::new(StatsConfig::finite());
|
let mut collector = StatsCollector::new(StatsConfig::finite());
|
||||||
@@ -133,6 +161,39 @@ fn ipdv_is_sequence_adjacent_and_gap_preserving() {
|
|||||||
assert_eq!(snapshot.ipdv.round_trip.total_ns, 3_000_000);
|
assert_eq!(snapshot.ipdv.round_trip.total_ns, 3_000_000);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ipdv_wraparound_sequence_is_adjacent() {
|
||||||
|
let mut collector = StatsCollector::new(StatsConfig::finite());
|
||||||
|
let first = collector.process(&reply_at(u32::MAX, 10, 0));
|
||||||
|
let wrapped = collector.process(&reply_at(0, 14, 10));
|
||||||
|
|
||||||
|
assert!(first.contributed_sample);
|
||||||
|
assert_no_ipdv_pairs(&first);
|
||||||
|
|
||||||
|
assert!(wrapped.contributed_sample);
|
||||||
|
assert_one_ipdv_pair(&wrapped, u32::MAX, 0, Duration::from_millis(4));
|
||||||
|
|
||||||
|
let snapshot = collector.snapshot();
|
||||||
|
assert_eq!(snapshot.ipdv.round_trip.count, 1);
|
||||||
|
assert_eq!(snapshot.ipdv.round_trip.total_ns, 4_000_000);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ipdv_wraparound_gap_is_preserved() {
|
||||||
|
let mut collector = StatsCollector::new(StatsConfig::finite());
|
||||||
|
let before_wrap = collector.process(&reply_at(u32::MAX - 1, 10, 0));
|
||||||
|
let wrapped = collector.process(&reply_at(0, 14, 20));
|
||||||
|
|
||||||
|
assert!(before_wrap.contributed_sample);
|
||||||
|
assert_no_ipdv_pairs(&before_wrap);
|
||||||
|
|
||||||
|
assert!(wrapped.contributed_sample);
|
||||||
|
assert_no_ipdv_pairs(&wrapped);
|
||||||
|
|
||||||
|
let snapshot = collector.snapshot();
|
||||||
|
assert_eq!(snapshot.ipdv.round_trip.count, 0);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn late_reply_can_complete_ipdv_pair() {
|
fn late_reply_can_complete_ipdv_pair() {
|
||||||
let mut collector = StatsCollector::new(StatsConfig::finite());
|
let mut collector = StatsCollector::new(StatsConfig::finite());
|
||||||
|
|||||||
Reference in new issue
Block a user