vector_calls/
jitter.rs

1//! The adaptive jitter buffer: holds a few frames so the network's unevenness is
2//! smoothed out before the decoder, grows on late arrivals, and takes the latency
3//! back when the path settles. Time comes in as milliseconds from the caller.
4
5use super::stats::MediaStats;
6use super::FRAME_MS;
7use std::collections::BTreeMap;
8use std::sync::atomic::Ordering;
9
10/// Frames buffered before playout starts.
11pub const PREFILL: usize = 2;
12/// Deeper than this and playout jumps forward: latency costs more than the gap.
13pub const MAX_DEPTH: usize = 25;
14/// The deepest target the jitter estimate may ask for.
15pub const MAX_TARGET: usize = 20;
16/// How fast a remembered delay swing fades, in milliseconds per frame received:
17/// a 200 ms burst still shapes the target ten seconds later.
18const PEAK_DECAY_MS: f32 = 0.4;
19/// How long the buffer must sit above its target before it trims a frame. A burst
20/// leaves it deep; trimming at once would meet the next burst empty again.
21const SHED_AFTER_MS: u64 = 3000;
22/// Concealed frames in a row before playout stops guessing and waits. Each one
23/// stretches the last sound a little further; past three that reads as a slur,
24/// and a clean gap is easier on the ear.
25pub const MAX_PLC_RUN: u32 = 3;
26
27pub enum Pop {
28    Frame(Vec<u8>),
29    /// The frame is missing; rebuild it from the FEC data in its successor.
30    Fec(Vec<u8>),
31    Muted,
32    Conceal,
33    Wait,
34}
35
36pub struct Buffered {
37    payload: Vec<u8>,
38    muted: bool,
39}
40
41#[derive(Default)]
42pub struct Jitter {
43    frames: BTreeMap<u64, Buffered>,
44    /// Next sequence to play; None until prefilled.
45    next: Option<u64>,
46    /// Unwrapped sequence of the newest arrival, for the 16-bit wrap.
47    newest: Option<u64>,
48    /// When the newest frame arrived, in the caller's milliseconds, and its sequence.
49    last_arrival: Option<(u64, u64)>,
50    /// RFC 3550 style smoothed inter-arrival jitter, in milliseconds.
51    jitter_ms: f32,
52    /// The largest recent inter-arrival swing, decaying: bursts, not the average.
53    peak_ms: f32,
54    /// Since when the buffer has been deeper than its target.
55    over_since: Option<u64>,
56    shed_after: Option<u64>,
57}
58
59impl Jitter {
60    /// Frames waiting, for the depth readout.
61    pub fn buffered(&self) -> usize {
62        self.frames.len()
63    }
64
65    /// Frames to hold: enough for three times the smoothed jitter, or for the
66    /// biggest swing seen lately with a little to spare, whichever is more.
67    fn target_frames(&self) -> usize {
68        let need_ms = (self.jitter_ms * 3.0).max(self.peak_ms * 1.2);
69        let frames = (need_ms / FRAME_MS as f32).ceil() as usize;
70        (PREFILL + frames).clamp(PREFILL, MAX_TARGET)
71    }
72
73    fn unwrap_seq(&mut self, seq: u16) -> u64 {
74        let s = match self.newest {
75            None => seq as u64,
76            Some(newest) => {
77                let delta = seq.wrapping_sub(newest as u16) as i16 as i64;
78                (newest as i64 + delta).max(0) as u64
79            }
80        };
81        if self.newest.map_or(true, |n| s > n) {
82            self.newest = Some(s);
83        }
84        s
85    }
86
87    /// A frame arrived at `now_ms`.
88    pub fn push(&mut self, seq: u16, muted: bool, payload: &[u8], now_ms: u64, stats: &MediaStats) {
89        let s = self.unwrap_seq(seq);
90        let now = now_ms;
91        if let Some((at, prev_seq)) = self.last_arrival {
92            let expected = (s as i64 - prev_seq as i64) as f32 * FRAME_MS as f32;
93            let actual = now.saturating_sub(at) as f32;
94            let d = (actual - expected).abs();
95            self.jitter_ms += (d - self.jitter_ms) / 16.0;
96            self.peak_ms = d.max(self.peak_ms - PEAK_DECAY_MS);
97            stats.jitter_ms.store(self.jitter_ms as u32, Ordering::Relaxed);
98        }
99        self.last_arrival = Some((now, s));
100
101        if let Some(next) = self.next {
102            if s < next {
103                stats.late.fetch_add(1, Ordering::Relaxed);
104                return;
105            }
106        }
107        self.frames.insert(s, Buffered { payload: payload.to_vec(), muted });
108
109        if self.frames.len() > MAX_DEPTH {
110            // Jump to the newest PREFILL frames; what came before is late already.
111            let keep_from = *self.frames.keys().nth(self.frames.len() - PREFILL).unwrap();
112            let dropped = self.frames.range(..keep_from).count() as u64;
113            self.frames = self.frames.split_off(&keep_from);
114            stats.lost.fetch_add(dropped, Ordering::Relaxed);
115            self.next = Some(keep_from);
116        }
117    }
118
119    /// The next thing to play, decided at `now_ms`.
120    pub fn pop(&mut self, plc_run: u32, now_ms: u64, stats: &MediaStats) -> Pop {
121        let next = match self.next {
122            Some(n) => n,
123            None => {
124                if self.frames.len() < PREFILL {
125                    return Pop::Wait;
126                }
127                let first = *self.frames.keys().next().unwrap();
128                self.next = Some(first);
129                first
130            }
131        };
132        if self.frames.len() > self.target_frames() {
133            // Underruns stretched the buffer; once that has held for a while, take the
134            // latency back a frame at a time.
135            let now = now_ms;
136            let since = *self.over_since.get_or_insert(now);
137            if now.saturating_sub(since) >= self.shed_after.unwrap_or(SHED_AFTER_MS) {
138                if let Some(oldest) = self.frames.keys().next().copied() {
139                    self.frames.remove(&oldest);
140                    stats.shed.fetch_add(1, Ordering::Relaxed);
141                    self.next = Some(oldest + 1);
142                    self.over_since = Some(now);
143                    return self.pop(plc_run, now_ms, stats);
144                }
145            }
146        } else {
147            self.over_since = None;
148        }
149        if let Some(f) = self.frames.remove(&next) {
150            self.next = Some(next + 1);
151            return if f.muted { Pop::Muted } else { Pop::Frame(f.payload) };
152        }
153        match self.frames.keys().next().copied() {
154            Some(min) if min == next + 1 => {
155                // One slot of grace: on a jittery path the frame is usually just behind
156                // its successor. Only when the buffer is already full enough is it lost.
157                if plc_run == 0 && self.frames.len() < self.target_frames() {
158                    return Pop::Conceal;
159                }
160                self.next = Some(min);
161                Pop::Fec(self.frames[&min].payload.clone())
162            }
163            Some(min) => {
164                // A hole wider than one frame: guess twice, then skip to what we have.
165                if plc_run < 2 {
166                    self.next = Some(next + 1);
167                    Pop::Conceal
168                } else {
169                    stats.lost.fetch_add(min - next, Ordering::Relaxed);
170                    self.next = Some(min);
171                    self.pop(0, now_ms, stats)
172                }
173            }
174            None => {
175                // Nothing has arrived: fill the time without giving up on this frame, so
176                // its late arrival still plays and the buffer grows by the underrun.
177                if plc_run < MAX_PLC_RUN {
178                    Pop::Conceal
179                } else {
180                    Pop::Wait
181                }
182            }
183        }
184    }
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190
191    fn frame(n: u8) -> Vec<u8> {
192        vec![n]
193    }
194
195    #[test]
196    fn frames_play_in_order_after_prefill_and_a_gap_uses_fec() {
197        let stats = MediaStats::default();
198        let mut j = Jitter::default();
199        assert!(matches!(j.pop(0, 0, &stats), Pop::Wait));
200        j.push(0, false, &frame(0), 0, &stats);
201        assert!(matches!(j.pop(0, 0, &stats), Pop::Wait));
202        j.push(1, false, &frame(1), 0, &stats);
203        assert!(matches!(j.pop(0, 0, &stats), Pop::Frame(p) if p == frame(0)));
204        assert!(matches!(j.pop(0, 0, &stats), Pop::Frame(p) if p == frame(1)));
205        // Frame 2 lost, 3 arrives: one frame of grace, then 2 is rebuilt from 3's FEC,
206        // then 3 plays.
207        j.push(3, false, &frame(3), 0, &stats);
208        assert!(matches!(j.pop(0, 0, &stats), Pop::Conceal));
209        assert!(matches!(j.pop(1, 0, &stats), Pop::Fec(p) if p == frame(3)));
210        assert!(matches!(j.pop(0, 0, &stats), Pop::Frame(p) if p == frame(3)));
211        // Late arrival is dropped.
212        j.push(2, false, &frame(2), 0, &stats);
213        assert_eq!(stats.late.load(Ordering::Relaxed), 1);
214        // An underrun conceals but keeps waiting for frame 4, which then plays.
215        assert!(matches!(j.pop(0, 0, &stats), Pop::Conceal));
216        assert_eq!(j.next, Some(4));
217        j.push(4, false, &frame(4), 0, &stats);
218        assert!(matches!(j.pop(1, 0, &stats), Pop::Frame(p) if p == frame(4)));
219    }
220
221    #[test]
222    fn a_stretched_buffer_keeps_its_depth_at_first_then_sheds_to_target() {
223        let stats = MediaStats::default();
224        let mut j = Jitter::default();
225        for s in 0..8u16 {
226            j.push(s, false, &frame(s as u8), 0, &stats);
227        }
228        // Instant pushes read as jitter to the estimator; pin it so the target is PREFILL.
229        j.jitter_ms = 0.0;
230        j.peak_ms = 0.0;
231        // A fresh overshoot is kept.
232        assert_eq!(j.target_frames(), PREFILL);
233        assert!(matches!(j.pop(0, 0, &stats), Pop::Frame(p) if p == frame(0)));
234        assert_eq!(stats.shed.load(Ordering::Relaxed), 0);
235        // Once the overshoot has lasted long enough, frames go one per pop until in range.
236        j.shed_after = Some(0);
237        assert!(matches!(j.pop(0, 0, &stats), Pop::Frame(p) if p == frame(6)));
238        assert_eq!(stats.shed.load(Ordering::Relaxed), 5);
239        assert_eq!(stats.lost.load(Ordering::Relaxed), 0);
240        assert_eq!(j.frames.len(), 1);
241    }
242
243    #[test]
244    fn the_target_grows_with_jitter_and_remembers_bursts() {
245        let mut j = Jitter::default();
246        j.jitter_ms = 43.0;
247        assert_eq!(j.target_frames(), PREFILL + 7);
248        j.peak_ms = 300.0;
249        assert_eq!(j.target_frames(), PREFILL + 18);
250        j.jitter_ms = 1000.0;
251        assert_eq!(j.target_frames(), MAX_TARGET);
252    }
253
254    #[test]
255    fn sequence_wraps_and_deep_buffers_jump_forward() {
256        let stats = MediaStats::default();
257        let mut j = Jitter::default();
258        j.push(65534, false, &frame(1), 0, &stats);
259        j.push(65535, false, &frame(2), 0, &stats);
260        j.push(0, false, &frame(3), 0, &stats);
261        assert!(matches!(j.pop(0, 0, &stats), Pop::Frame(p) if p == frame(1)));
262        assert_eq!(j.next, Some(65535));
263        for s in 1..=(MAX_DEPTH as u16 + 2) {
264            j.push(s, false, &frame(9), 0, &stats);
265        }
266        assert!(j.frames.len() <= MAX_DEPTH, "{}", j.frames.len());
267        assert!(stats.lost.load(Ordering::Relaxed) > 0);
268        // Playout resumed at the jump target, not at the frame it was on.
269        assert!(j.next.unwrap() > 65537);
270    }
271}
vector_calls/jitter.rs

vector_calls/
jitter.rs
