Skip to content
rachel/esp32-radio
File

Blob: firmware/vendor/str0m/src/streams/register_nack.rs

rust650 lines
1use std::ops::Range;
2 
3use crate::rtp_::{Nack, NackEntry, ReportList, SeqNo};
4 
5/// Number of out of order packets we keep track of for reports
6const MAX_MISORDER: u64 = 100;
7 
8/// The max number of NACKs we perform for a single packet
9const MAX_NACKS: u8 = 5;
10 
11/// Circular buffer size
12const BUFFER_SIZE: u64 = MAX_MISORDER + 1;
13 
14#[derive(Debug)]
15pub struct NackRegister {
16    /// Status of packets indexed by wrapping SeqNo.
17    packets: Vec<PacketStatus>,
18 
19    /// Range of seq numbers considered NACK reporting.
20    active: Option<Range<SeqNo>>,
21}
22 
23#[derive(Debug, Default, Clone, Copy)]
24struct PacketStatus {
25    received: bool,
26    nack_count: u8,
27}
28 
29impl PacketStatus {
30    fn needs_nack(&self) -> bool {
31        !self.received && self.nack_count < MAX_NACKS
32    }
33 
34    fn mark_received(&mut self) -> bool {
35        let new = !self.received;
36        self.received = true;
37        new
38    }
39 
40    fn reset(&mut self) {
41        self.received = false;
42        self.nack_count = 0;
43    }
44}
45 
46struct NackIterator<'a> {
47    reg: &'a mut NackRegister,
48    next: u64,
49    end: u64,
50}
51 
52impl<'a> Iterator for NackIterator<'a> {
53    type Item = NackEntry;
54 
55    fn next(&mut self) -> Option<Self::Item> {
56        self.next =
57            (self.next..=self.end).find(|s| self.reg.packet_mut((*s).into()).needs_nack())?;
58 
59        let mut entry = NackEntry {
60            pid: self.next as u16,
61            blp: 0,
62        };
63 
64        self.reg.packet_mut(self.next.into()).nack_count += 1;
65        self.next += 1;
66 
67        for (i, s) in (self.next..self.end).take(16).enumerate() {
68            let packet = self.reg.packet_mut(s.into());
69            if packet.needs_nack() {
70                self.reg.packet_mut(self.next.into()).nack_count += 1;
71                entry.blp |= 1 << i
72            }
73            self.next += 1;
74        }
75 
76        Some(entry)
77    }
78}
79 
80impl NackRegister {
81    /// Creates a new register.
82    ///
83    /// The max_seq_no is to provide a starting point for ROC calculations.
84    pub fn new(max_seq_no: Option<SeqNo>) -> Self {
85        let mut n = NackRegister {
86            packets: vec![PacketStatus::default(); BUFFER_SIZE as usize],
87            active: None,
88        };
89 
90        if let Some(seq) = max_seq_no {
91            n.init_with_seq(seq);
92        }
93 
94        n
95    }
96 
97    pub fn accepts(&self, seq: SeqNo) -> bool {
98        let Some(active) = self.active.clone() else {
99            // if we don't have initialized, we do want the first packet.
100            return true;
101        };
102 
103        // behind the window
104        if seq < active.start {
105            return false;
106        }
107 
108        !self.packet(seq).received || seq > active.end
109    }
110 
111    pub fn update(&mut self, seq: SeqNo) -> bool {
112        let Some(active) = self.active.clone() else {
113            // automatically pick up the first seq number
114            self.init_with_seq(seq);
115            return true;
116        };
117 
118        if seq < active.start {
119            // skip old seq numbers, report as not new
120            return false;
121        }
122 
123        let new = !self.packet_mut(seq).received || seq > active.end;
124 
125        let end = active.end.max(seq);
126 
127        let start: SeqNo = {
128            let min = end.saturating_sub(MAX_MISORDER);
129            let mut start = (*active.start).max(min);
130            while start < *end {
131                if !self.packet_mut(start.into()).received && start != *seq {
132                    break;
133                }
134                start += 1;
135            }
136            start.into()
137        };
138 
139        // reset packets that are rolling our of the nack window
140        for (i, s) in (*active.start..*start).enumerate() {
141            let p = self.packet_mut(s.into());
142            if !p.received && s != *seq {
143                trace!("Seq no {} missing after {} attempts", s, p.nack_count);
144            }
145            self.packet_mut(s.into()).reset();
146 
147            if i > self.packets.len() {
148                // we have reset all entries already
149                break;
150            }
151        }
152 
153        if (start..=end).contains(&seq) {
154            self.packet_mut(seq).mark_received();
155        }
156 
157        self.active = Some(start..end);
158 
159        new
160    }
161 
162    fn init_with_seq(&mut self, seq: SeqNo) {
163        self.active = Some(seq..seq);
164        self.packet_mut(seq).mark_received();
165    }
166 
167    pub fn max_seq(&self) -> Option<SeqNo> {
168        self.active.as_ref().map(|a| a.end)
169    }
170 
171    /// Create a new nack report
172    ///
173    /// This modifies the state as it counts how many times packets have been nacked
174    pub fn nack_reports(&mut self) -> Option<impl Iterator<Item = Nack>> {
175        let Range { start, end } = self.active.clone()?;
176        let start = (*start..=*end).find(|s| self.packet_mut((*s).into()).needs_nack())?;
177 
178        Some(
179            ReportList::lists_from_iter(NackIterator {
180                reg: self,
181                next: start,
182                end: *end,
183            })
184            .into_iter()
185            .map(|reports| {
186                Nack {
187                    sender_ssrc: 0.into(),
188                    ssrc: 0.into(), // changed when sending
189                    reports,
190                }
191            }),
192        )
193    }
194 
195    fn as_index(&self, seq: SeqNo) -> usize {
196        (*seq % self.packets.len() as u64) as usize
197    }
198 
199    fn packet(&self, seq: SeqNo) -> &PacketStatus {
200        let index = self.as_index(seq);
201        &self.packets[index]
202    }
203 
204    fn packet_mut(&mut self, seq: SeqNo) -> &mut PacketStatus {
205        let index = self.as_index(seq);
206        &mut self.packets[index]
207    }
208}
209 
210#[cfg(test)]
211mod test {
212    use std::ops::Range;
213 
214    use crate::streams::register_nack::MAX_MISORDER;
215 
216    use super::NackRegister;
217 
218    fn assert_update(
219        reg: &mut NackRegister,
220        seq: u64,
221        expect_new: bool,
222        expect_received: bool,
223        expect_active: Range<u64>,
224    ) {
225        assert_eq!(
226            reg.update(seq.into()),
227            expect_new,
228            "seq {} was expected to{} be new",
229            seq,
230            if expect_new { "" } else { " NOT" }
231        );
232        let active = reg.active.clone().expect("nack range");
233        assert_eq!(
234            reg.packet_mut(seq.into()).received,
235            expect_received,
236            "seq {} expected to{} be received in {:?}",
237            seq,
238            if expect_received { "" } else { " NOT" },
239            active
240        );
241        assert_eq!(active, expect_active.start.into()..expect_active.end.into());
242        assert_not_dirty(reg);
243    }
244 
245    fn assert_not_dirty(reg: &NackRegister) {
246        // we should leave no dirty state outside of the nack window
247        let active = reg.active.clone().expect("nack range");
248        let active = (*active.start..=*active.end)
249            .map(|seq| reg.as_index(seq.into()))
250            .collect::<Vec<_>>();
251 
252        for i in 0..reg.packets.len() {
253            if active.contains(&i) {
254                continue;
255            }
256            assert!(
257                !reg.packets[i].received && reg.packets[i].nack_count == 0,
258                "dirty state at index {} outside of nack window {:?}",
259                i,
260                active,
261            );
262        }
263    }
264 
265    #[test]
266    fn active_window_sliding() {
267        let mut reg = NackRegister::new(None);
268 
269        assert!(reg.accepts(10.into()));
270        assert_update(&mut reg, 10, true, true, 10..10);
271 
272        // packet before window start is ignored
273        assert!(!reg.accepts(9.into()));
274        assert_update(&mut reg, 9, false, false, 10..10);
275 
276        // duped packet
277        assert!(!reg.accepts(10.into()));
278        assert_update(&mut reg, 10, false, true, 10..10);
279 
280        // future packets accepted, window not sliding
281        let next = 10 + MAX_MISORDER;
282        assert!(reg.accepts(next.into()));
283        assert_update(&mut reg, next, true, true, 11..next);
284        let next = 11 + MAX_MISORDER;
285        assert!(reg.accepts(next.into()));
286        assert_update(&mut reg, next, true, true, 11..next);
287 
288        // future packet accepted, sliding window
289        let next = 12 + MAX_MISORDER;
290        assert!(reg.accepts(next.into()));
291        assert_update(&mut reg, next, true, true, 12..next);
292 
293        // older packet received within window
294        let next = 13;
295        assert!(reg.accepts(next.into()));
296        assert_update(&mut reg, next, true, true, 12..(12 + MAX_MISORDER));
297 
298        // do not want the same packet again
299        assert!(!reg.accepts(next.into()));
300 
301        // future packet accepted, sliding window start skips over received
302        let next = 13 + MAX_MISORDER;
303        assert!(reg.accepts(next.into()));
304        assert_update(&mut reg, next, true, true, 14..next);
305 
306        // do not want the same packet again
307        assert!(!reg.accepts(next.into()));
308 
309        // older packet accepted, window star moves ahead
310        let next = 14;
311        assert!(reg.accepts(next.into()));
312        assert_update(&mut reg, next, true, false, 15..(13 + MAX_MISORDER));
313    }
314 
315    #[test]
316    fn nack_report_none() {
317        let mut reg = NackRegister::new(None);
318        assert!(reg.nack_reports().is_none());
319 
320        reg.update(110.into());
321        assert!(reg.nack_reports().is_none());
322 
323        reg.update(111.into());
324        assert!(reg.nack_reports().is_none());
325    }
326 
327    #[test]
328    fn nack_test_huge_seq_gap_no_hang() {
329        let mut reg = NackRegister::new(None);
330 
331        reg.update(0.into());
332        reg.update(18446744073709551515.into());
333    }
334 
335    #[test]
336    fn nack_report_one() {
337        let mut reg = NackRegister::new(None);
338        assert!(reg.nack_reports().is_none());
339 
340        reg.update(110.into());
341        assert!(reg.nack_reports().is_none());
342 
343        reg.update(112.into());
344        let report = reg.nack_reports().map(Vec::from_iter).expect("some report");
345        assert!(report.len() == 1);
346        assert_eq!(report[0].reports.len(), 1);
347        assert_eq!(report[0].reports[0].pid, 111);
348        assert_eq!(report[0].reports[0].blp, 0);
349    }
350 
351    #[test]
352    fn nack_report_two() {
353        let mut reg = NackRegister::new(None);
354        assert!(reg.nack_reports().is_none());
355 
356        reg.update(110.into());
357        assert!(reg.nack_reports().is_none());
358 
359        reg.update(113.into());
360        let report = reg.nack_reports().map(Vec::from_iter).expect("some report");
361        assert!(report.len() == 1);
362        assert_eq!(report[0].reports.len(), 1);
363        assert_eq!(report[0].reports[0].pid, 111);
364        assert_eq!(report[0].reports[0].blp, 0b1);
365    }
366 
367    #[test]
368    fn nack_report_with_hole() {
369        let mut reg = NackRegister::new(None);
370 
371        for i in &[100, 101, 103, 105, 106, 107, 108, 109, 110] {
372            reg.update((*i).into());
373        }
374 
375        let report = reg.nack_reports().map(Vec::from_iter).expect("some report");
376        assert!(report.len() == 1);
377        assert_eq!(report[0].reports.len(), 1);
378        assert_eq!(report[0].reports[0].pid, 102);
379        assert_eq!(report[0].reports[0].blp, 0b10);
380    }
381 
382    #[test]
383    fn nack_report_stop_at_17() {
384        let mut reg = NackRegister::new(None);
385 
386        let seq = &[
387            100, 101, 103, 104, 105, 106, 107, 108, 109, 110, //
388            111, 112, 113, 114, 115, 116, 117, 118, 119, 120, //
389            121, 122, 123, 125,
390        ];
391 
392        for i in seq {
393            reg.update((*i).into());
394        }
395 
396        let report = reg.nack_reports().map(Vec::from_iter).expect("some report");
397        assert_eq!(report.len(), 1);
398        assert_eq!(report[0].reports.len(), 2);
399        assert_eq!(report[0].reports[0].pid, 102);
400        assert_eq!(report[0].reports[0].blp, 0);
401    }
402 
403    #[test]
404    fn nack_report_hole_at_17() {
405        let mut reg = NackRegister::new(None);
406 
407        let seq = &[
408            100, 101, 103, 104, 105, 106, 107, 108, 109, 110, //
409            111, 112, 113, 114, 115, 116, 117, 119, 120, 121, //
410            122, 123, 124, 125, 126, 127, 128, 129,
411        ];
412 
413        for i in seq {
414            reg.update((*i).into());
415        }
416 
417        let report = reg.nack_reports().map(Vec::from_iter).expect("some report");
418        assert_eq!(report.len(), 1);
419        assert_eq!(report[0].reports.len(), 1);
420        assert_eq!(report[0].reports[0].pid, 102);
421        assert_eq!(report[0].reports[0].blp, 0b1000_0000_0000_0000);
422    }
423 
424    #[test]
425    fn nack_report_no_stop_all_there() {
426        let mut reg = NackRegister::new(None);
427 
428        let seq = &[
429            100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, //
430            111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, //
431            122, 123, 124, 125, 126, 127, 128, 129,
432        ];
433 
434        for i in seq {
435            reg.update((*i).into());
436        }
437 
438        assert!(reg.nack_reports().is_none());
439    }
440 
441    #[test]
442    fn nack_report_rtx() {
443        let mut reg = NackRegister::new(None);
444        for i in &[
445            100, 101, 102, 103, 104, 105, //
446        ] {
447            reg.update((*i).into());
448        }
449        assert!(reg.nack_reports().is_none());
450        let active = reg.active.clone().expect("nack range");
451        assert_eq!(*active.start, 105);
452 
453        for i in &[
454            106, 108, 109, 110, 111, 112, 113, 114, 115, //
455        ] {
456            reg.update((*i).into());
457        }
458        assert!(reg.nack_reports().is_some());
459        let active = reg.active.clone().expect("nack range");
460        assert_eq!(*active.start, 107);
461 
462        reg.update(107.into()); // Got 107 via RTX
463 
464        let nacks = reg.nack_reports().map(Vec::from_iter);
465        assert!(
466            nacks.is_none(),
467            "Expected no NACKs to be generated after repairing the stream, got {nacks:?}"
468        );
469        let active = reg.active.clone().expect("nack range");
470        assert_eq!(*active.start, 115);
471    }
472 
473    #[test]
474    fn nack_report_rollover_rtx() {
475        // This test is checking that after rollover nacks are not skipped because of
476        // packet position that would remain marked as received from before the rollover
477        let mut reg = NackRegister::new(None);
478        for i in &[
479            100, 101, 102, 103, 104, 105, 106, 108, 109, 110, 111, 112, 113, 114, 115,
480        ] {
481            reg.update((*i).into());
482        }
483 
484        reg.update(107.into()); // Got 107 via RTX
485        let active = reg.active.clone().expect("nack range");
486        assert_eq!(*active.start, 115);
487 
488        for i in 116..3106 {
489            reg.update(i.into());
490        }
491        let active = reg.active.clone().expect("nack range");
492        assert_eq!(*active.start, 3105);
493 
494        for i in &[3106, 3108, 3109, 3110, 3111, 3112, 3113, 3114, 3115] {
495            reg.update((*i).into()); // Missing at postion 107 again
496        }
497 
498        let active = reg.active.clone().expect("nack range");
499        assert_eq!(*active.start, 3107);
500    }
501 
502    #[test]
503    fn nack_report_rollover_rtx_with_seq_jump() {
504        let mut reg = NackRegister::new(None);
505 
506        // 2999 is missing
507        for i in 0..2999 {
508            reg.update(i.into());
509        }
510 
511        // 3002 is missing
512        reg.update(3003.into());
513        reg.update(3004.into());
514        reg.update(3000.into());
515        reg.update(3001.into());
516 
517        let reports = reg.nack_reports().map(Vec::from_iter).expect("some report");
518        assert_eq!(reports.len(), 1);
519        assert_eq!(reports[0].reports[0].pid, 2999);
520        assert_eq!(reports[0].reports[0].blp, 4);
521    }
522 
523    #[test]
524    fn out_of_order_and_rollover() {
525        let mut reg = NackRegister::new(None);
526 
527        reg.update(2998.into());
528        reg.update(2999.into());
529 
530        // receive older packet
531        reg.update(2995.into());
532 
533        // wrap
534        for i in 3000..5995 {
535            reg.update(i.into());
536        }
537 
538        // 5995 is missing
539 
540        reg.update(5996.into());
541        reg.update(5997.into());
542 
543        let reports = reg.nack_reports().map(Vec::from_iter).expect("some report");
544        assert_eq!(reports.len(), 1);
545        assert_eq!(reports[0].reports[0].pid, 5995);
546    }
547 
548    #[test]
549    fn nack_check_on_seq_rollover() {
550        let range = 65530..65541;
551        let missing = [65535_u64, 65536_u64, 65537_u64];
552        let expected = [65535_u16, 0_u16, 1_u16];
553 
554        for (missing, expected) in missing.iter().zip(expected.iter()) {
555            let mut seqs: Vec<_> = range.clone().collect();
556            let mut reg = NackRegister::new(None);
557 
558            seqs.retain(|x| *x != *missing);
559            for i in seqs.as_slice() {
560                reg.update((*i).into());
561            }
562 
563            let reports = reg.nack_reports().map(Vec::from_iter).expect("some report");
564            let pid = reports[0].reports[0].pid;
565            assert_eq!(pid, *expected);
566        }
567    }
568 
569    #[test]
570    fn nack_check_forward_at_boundary() {
571        let mut reg = NackRegister::new(None);
572        for i in 2996..=3003 {
573            reg.update(i.into());
574        }
575 
576        assert!(reg.nack_reports().is_none());
577        let active = reg.active.clone().expect("nack range");
578        assert_eq!(*active.start, 3003);
579 
580        for i in 3004..=3008 {
581            reg.update(i.into());
582        }
583 
584        let report = reg.nack_reports().map(Vec::from_iter);
585        assert!(report.is_none(), "Expected empty NACKs got {:?}", report);
586        let active = reg.active.clone().expect("nack range");
587        assert_eq!(*active.start, 3008);
588    }
589 
590    #[test]
591    fn nack_check_forward_at_u16_boundary() {
592        let mut reg = NackRegister::new(None);
593        for i in 65500..=65534 {
594            reg.update(i.into());
595        }
596        assert!(reg.nack_reports().is_none());
597        let active = reg.active.clone().expect("nack range");
598        assert_eq!(*active.start, 65534);
599 
600        for i in 65536..=65566 {
601            reg.update(i.into());
602        }
603 
604        assert!(reg.nack_reports().is_some());
605        let active = reg.active.clone().expect("nack range");
606        assert_eq!(*active.start, 65535);
607 
608        for i in 65567..=65666 {
609            reg.update(i.into());
610        }
611 
612        reg.update(65535.into());
613 
614        assert!(reg.nack_reports().is_none());
615        let active = reg.active.clone().expect("nack range");
616        assert_eq!(*active.start, 65666);
617    }
618 
619    #[test]
620    fn nack_reports_on_boundaries() {
621        let mut reg = NackRegister::new(None);
622 
623        for i in 0..=20 {
624            // gap must be at least 17 packets to separate the reports.
625            if i == 2 || i == 19 {
626                continue;
627            }
628            reg.update(i.into());
629        }
630 
631        let reports: Vec<_> = reg
632            .nack_reports()
633            .expect("should generate reports")
634            .flat_map(|nack| nack.reports)
635            .collect();
636 
637        assert_eq!(reports.len(), 2, "Should have found two NACK entries");
638        assert_eq!(reports[0].pid, 2, "First missing packet should be 2");
639        assert_eq!(
640            reports[0].blp, 0,
641            "No missing packets in blp bits following 2"
642        );
643        assert_eq!(reports[1].pid, 19, "Second missing packet should be 19");
644        assert_eq!(
645            reports[1].blp, 0,
646            "No missing packets in blp bits following 19"
647        );
648    }
649}