rtc_interceptor/gcc/
slope.rs

1//! Turning grouped acknowledgements into a filtered delay trend.
2
3use super::arrival_group::{ArrivalGroupAccumulator, InterGroupDelay};
4use super::kalman::Kalman;
5use crate::rtpfb::acknowledgement::PacketReport;
6use std::time::{Duration, Instant};
7
8/// One filtered delay-gradient reading.
9#[derive(Debug, Clone, Copy, PartialEq)]
10pub struct DelayTrend {
11    /// The raw inter-group measurement, in milliseconds.
12    pub measurement_ms: f64,
13    /// The filtered trend, in milliseconds. This is what the overuse detector compares.
14    pub estimate_ms: f64,
15    /// When the measurement belongs to.
16    pub at: Instant,
17    /// Bytes in the group this reading closed, for the received-rate calculation.
18    pub size: usize,
19}
20
21/// Grouping plus filtering: acknowledgements in, a delay trend out.
22///
23/// The two halves are separate types because they are separately testable — the accumulator against
24/// hand-built arrival patterns, the filter against numeric sequences — and this joins them without
25/// adding behaviour of its own.
26#[derive(Debug, Clone, Default)]
27pub struct SlopeEstimator {
28    groups: ArrivalGroupAccumulator,
29    kalman: Kalman,
30}
31
32impl SlopeEstimator {
33    /// A slope estimator with the draft's default grouping and tuning.
34    pub fn new() -> Self {
35        Self::default()
36    }
37
38    /// A slope estimator grouping packets sent within `burst_interval` of each other.
39    pub fn with_burst_interval(burst_interval: Duration) -> Self {
40        Self {
41            groups: ArrivalGroupAccumulator::new(burst_interval),
42            kalman: Kalman::new(),
43        }
44    }
45
46    /// The current filtered trend, in milliseconds.
47    pub fn estimate_ms(&self) -> f64 {
48        self.kalman.estimate()
49    }
50
51    /// Feed one report; returns a reading when a group closed.
52    pub fn accumulate(&mut self, report: &PacketReport) -> Option<DelayTrend> {
53        let delay = self.groups.accumulate(report)?;
54        Some(self.filter(delay))
55    }
56
57    /// Close the stream, emitting the reading for the group still open.
58    pub fn flush(&mut self) -> Option<DelayTrend> {
59        let delay = self.groups.flush()?;
60        Some(self.filter(delay))
61    }
62
63    fn filter(&mut self, delay: InterGroupDelay) -> DelayTrend {
64        DelayTrend {
65            measurement_ms: delay.delta_ms,
66            estimate_ms: self.kalman.update(delay.delta_ms),
67            at: delay.at,
68            size: delay.size,
69        }
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76    use rtcp::transport_feedbacks::cc_feedback_report::Ecn;
77
78    fn report(departure: Instant, arrival_ms: u64) -> PacketReport {
79        PacketReport {
80            ssrc: 1,
81            id: 0,
82            rtp_sequence_number: 0,
83            is_twcc: true,
84            twcc_sequence_number: 0,
85            size: 1200,
86            arrived: true,
87            departure,
88            arrival: Some(Duration::from_millis(arrival_ms)),
89            ecn: Ecn::default(),
90        }
91    }
92
93    /// End to end over the two halves: a non-queueing path reads flat.
94    #[test]
95    fn a_steady_path_reads_flat() {
96        let epoch = Instant::now();
97        let mut slope = SlopeEstimator::new();
98
99        for burst in 0..40u64 {
100            slope.accumulate(&report(
101                epoch + Duration::from_millis(burst * 20),
102                100 + burst * 20,
103            ));
104        }
105
106        assert!(
107            slope.estimate_ms().abs() < 1.0,
108            "a steady path should read about zero, got {}",
109            slope.estimate_ms()
110        );
111    }
112
113    /// And a queue building reads positive and rising — the signal P7-05's detector triggers on.
114    #[test]
115    fn a_queueing_path_reads_positive() {
116        let epoch = Instant::now();
117        let mut slope = SlopeEstimator::new();
118
119        for burst in 0..40u64 {
120            slope.accumulate(&report(
121                epoch + Duration::from_millis(burst * 20),
122                100 + burst * 26,
123            ));
124        }
125
126        assert!(
127            slope.estimate_ms() > 3.0,
128            "6 ms of queue per group should read clearly positive, got {}",
129            slope.estimate_ms()
130        );
131    }
132
133    /// The reading carries the group's size, which is what the received-rate calculation consumes.
134    #[test]
135    fn a_reading_carries_the_group_size() {
136        let epoch = Instant::now();
137        let mut slope = SlopeEstimator::new();
138
139        slope.accumulate(&report(epoch, 100));
140        slope.accumulate(&report(epoch + Duration::from_millis(20), 120));
141        let reading = slope
142            .accumulate(&report(epoch + Duration::from_millis(40), 140))
143            .expect("the second group closes against the first");
144
145        assert_eq!(1200, reading.size);
146    }
147}
rtc_interceptor/gcc/slope.rs

rtc_interceptor/gcc/
slope.rs
