Handle empty RTP packets in Sample Builder (#303)

libWebRTC, among others, will sometime send padding RTP packets to 
keep send rates steady. When they do the packets are sent after a 
complete sequence of RTP packets that form a Sample/GOP/Frame. Example:

* P0, Timestamp=1, Start
* P1, T=1
* P2, T=1, End(RTP Marker Bit set)
* P3, T=1, Padding
* P4, T=1, Padding
* P5, T=2, Start
* P6, T=2 End

Here, when building a sample using the packets P5 and P6 we'll have 
`prev_dropped_packets = 2`, however these dropped packets are 
inconsequential as they don't carry media. The stream of samples is 
unbroken, despite having dropped some packets. With the introduction 
of `prev_padding_packets` it's possible to distinguish between dropped 
packets that are likely to break the sample stream and these 
unimportant padding packets.

Author: lookback-hugotunius

media/src/io/sample_builder/mod.rs

@@ -22,6 +22,7 @@ pub struct SampleBuilder<T: Depacketizer> {
     max_late_timestamp: u32,
     buffer: Vec<Option<Packet>>,
     prepared_samples: Vec<Option<Sample>>,
+    last_sample_timestamp: Option<u32>,
 
     /// Interface that allows us to take RTP packets to samples
     depacketizer: T,
@@ -40,6 +41,10 @@ pub struct SampleBuilder<T: Depacketizer> {
 
     /// number of packets forced to be dropped
     dropped_packets: u16,
+
+    /// number of padding packets detected and dropped. This number will be a subset of
+    /// `droppped_packets`
+    padding_packets: u16,
 }
 
 impl<T: Depacketizer> SampleBuilder<T> {
@@ -55,12 +60,14 @@ impl<T: Depacketizer> SampleBuilder<T> {
             max_late_timestamp: 0,
             buffer: vec![None; u16::MAX as usize + 1],
             prepared_samples: (0..=u16::MAX as usize).map(|_| None).collect(),
+            last_sample_timestamp: None,
             depacketizer,
             sample_rate,
             filled: SampleSequenceLocation::new(),
             active: SampleSequenceLocation::new(),
             prepared: SampleSequenceLocation::new(),
             dropped_packets: 0,
+            padding_packets: 0,
         }
     }
 
@@ -276,7 +283,19 @@ impl<T: Depacketizer> SampleBuilder<T> {
             .depacketizer
             .is_partition_head(&self.buffer[consume.head as usize].as_ref()?.payload)
         {
+            // libWebRTC will sometimes send several empty padding packets to smooth out send
+            // rate. These packets don't carry any media payloads.
+            let is_padding = consume.range(&self.buffer).all(|p| {
+                p.map(|p| {
+                    self.last_sample_timestamp == Some(p.header.timestamp) && p.payload.is_empty()
+                })
+                .unwrap_or(false)
+            });
+
             self.dropped_packets += consume.count();
+            if is_padding {
+                self.padding_packets += consume.count();
+            }
             self.purge_consumed_location(&consume, true);
             self.purge_consumed_buffers();
             return None;
@@ -301,9 +320,12 @@ impl<T: Depacketizer> SampleBuilder<T> {
             duration: Duration::from_secs_f64((samples as f64) / (self.sample_rate as f64)),
             packet_timestamp: sample_timestamp,
             prev_dropped_packets: self.dropped_packets,
+            prev_padding_packets: self.padding_packets,
         };
 
         self.dropped_packets = 0;
+        self.padding_packets = 0;
+        self.last_sample_timestamp = Some(sample_timestamp);
 
         self.prepared_samples[self.prepared.tail as usize] = Some(sample);
         self.prepared.tail = self.prepared.tail.wrapping_add(1);

media/src/io/sample_builder/sample_builder_test.rs

@@ -18,6 +18,7 @@ pub struct SampleBuilderTest {
     samples: Vec<Sample>,
     max_late: u16,
     max_late_timestamp: Duration,
+    extra_pop_attempts: usize,
 }
 
 pub struct FakeDepacketizer {
@@ -821,6 +822,119 @@ pub fn test_sample_builder() {
             max_late_timestamp: Duration::from_secs(2000),
             ..Default::default()
         },
+        SampleBuilderTest {
+            #[rustfmt::skip]
+            message: "Sample builder should recognise padding packets".into(),
+            packets: vec![
+                Packet {
+                    // First packet
+                    header: Header {
+                        sequence_number: 5000,
+                        timestamp: 1,
+                        ..Default::default()
+                    },
+                    payload: bytes!(1),
+                    ..Default::default()
+                },
+                Packet {
+                    // Second packet
+                    header: Header {
+                        sequence_number: 5001,
+                        timestamp: 1,
+                        ..Default::default()
+                    },
+                    payload: bytes!(2),
+                    ..Default::default()
+                },
+                Packet {
+                    // Third packet
+                    header: Header {
+                        sequence_number: 5002,
+                        timestamp: 1,
+                        marker: true,
+                        ..Default::default()
+                    },
+                    payload: bytes!(3),
+                    ..Default::default()
+                },
+                Packet {
+                    // Padding packet 1
+                    header: Header {
+                        sequence_number: 5003,
+                        timestamp: 1,
+                        ..Default::default()
+                    },
+                    payload: Bytes::from_static(&[]),
+                    ..Default::default()
+                },
+                Packet {
+                    // Padding packet 2
+                    header: Header {
+                        sequence_number: 5004,
+                        timestamp: 1,
+                        ..Default::default()
+                    },
+                    payload: Bytes::from_static(&[]),
+                    ..Default::default()
+                },
+                Packet {
+                    // Sixth packet
+                    header: Header {
+                        sequence_number: 5005,
+                        timestamp: 2,
+                        ..Default::default()
+                    },
+                    payload: bytes!(1),
+                    ..Default::default()
+                },
+                Packet {
+                    // Seventh packet
+                    header: Header {
+                        sequence_number: 5006,
+                        timestamp: 2,
+                        marker: true,
+                        ..Default::default()
+                    },
+                    payload: bytes!(7),
+                    ..Default::default()
+                },
+                Packet {
+                    // Seventh packet
+                    header: Header {
+                        sequence_number: 5007,
+                        timestamp: 3,
+                        ..Default::default()
+                    },
+                    payload: bytes!(1),
+                    ..Default::default()
+                },
+            ],
+            samples: vec![
+                Sample {
+                    // First sample
+                    data: bytes!(1, 2, 3),
+                    duration: Duration::from_secs(0),
+                    packet_timestamp: 1,
+                    prev_dropped_packets: 0,
+                    ..Default::default()
+                },
+                Sample {
+                    // Second sample
+                    data: bytes!(1, 7),
+                    duration: Duration::from_secs(1),
+                    packet_timestamp: 2,
+                    prev_dropped_packets: 2,
+                    prev_padding_packets: 2,
+                    ..Default::default()
+                },
+            ],
+            with_head_checker: true,
+            head_bytes: vec![bytes!(1)],
+            max_late: 50,
+            max_late_timestamp: Duration::from_secs(2000),
+            extra_pop_attempts: 1,
+            ..Default::default()
+        },
     ];
 
     for t in test_data {
@@ -843,12 +957,17 @@ pub fn test_sample_builder() {
             s.push(p)
         }
 
-        // Here we need some fancy loop that pops from s until empty. This propbably exists somewhere already.
-        // HAH, found it.
         while let Some(sample) = s.pop() {
             samples.push(sample)
         }
 
+        for _ in 0..t.extra_pop_attempts {
+            // Pop some more
+            while let Some(sample) = s.pop() {
+                samples.push(sample)
+            }
+        }
+
         // Current problem: Sample does not implement Eq. Either implement myself or find another way of comparison. (Derive does not work)
         assert_eq!(t.samples, samples, "{}", t.message);
     }

media/src/io/sample_builder/sample_sequence_location.rs

@@ -8,6 +8,26 @@ pub(crate) enum Comparison {
     After,
 }
 
+pub(crate) struct Iterator<'a, T> {
+    data: &'a [Option<T>],
+    sample: SampleSequenceLocation,
+    i: u16,
+}
+
+impl<'a, T> std::iter::Iterator for Iterator<'a, T> {
+    type Item = Option<&'a T>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.sample.compare(self.i) == Comparison::Inside {
+            let old_i = self.i as usize;
+            self.i = self.i.wrapping_add(1);
+            return Some(self.data[old_i].as_ref());
+        }
+
+        None
+    }
+}
+
 #[derive(Clone, Copy)]
 pub(crate) struct SampleSequenceLocation {
     /// head is the first packet in a sequence
@@ -50,4 +70,12 @@ impl SampleSequenceLocation {
         }
         Comparison::After
     }
+
+    pub(crate) fn range<'s, 'a, T>(&'s self, data: &'a [Option<T>]) -> Iterator<'a, T> {
+        Iterator {
+            data,
+            sample: *self,
+            i: self.head,
+        }
+    }
 }

media/src/io/sample_builder/sample_sequence_location_test.rs

@@ -22,3 +22,24 @@ fn test_sample_sequence_location_compare() {
     assert_eq!(Comparison::After, s2.compare(32));
     assert_eq!(Comparison::After, s2.compare(128));
 }
+
+#[test]
+fn test_sample_sequence_location_range() {
+    let mut data: Vec<Option<u16>> = vec![None; u16::MAX as usize + 1];
+
+    data[65533] = Some(65533);
+    data[65535] = Some(65535);
+    data[0] = Some(0);
+    data[2] = Some(2);
+
+    let s = SampleSequenceLocation {
+        head: 65533,
+        tail: 3,
+    };
+    let reconstructed: Vec<_> = s.range(&data).map(|x| x.cloned()).collect();
+
+    assert_eq!(
+        reconstructed,
+        [Some(65533), None, Some(65535), Some(0), None, Some(2)]
+    );
+}

media/src/lib.rs

@@ -15,11 +15,58 @@ use std::time::{Duration, SystemTime};
 /// A Sample contains encoded media and timing information
 #[derive(Debug)]
 pub struct Sample {
+    /// The assembled data in the sample, as a bitstream.
+    ///
+    /// The format is Codec dependant, but is always a bitstream format
+    /// rather than the packetized format used when carried over RTP.
+    ///
+    /// See: [`rtp::packetizer::Depacketizer`] and implementations of it for more details.
     pub data: Bytes,
+
+    /// The wallclock time when this sample was generated.
     pub timestamp: SystemTime,
+
+    /// The duration of this sample
     pub duration: Duration,
+
+    /// The RTP packet timestamp of this sample.
+    ///
+    /// For all RTP packets that contributed to a single sample the timestamp is the same.
     pub packet_timestamp: u32,
+
+    /// The number of packets that were dropped prior to building this sample.
+    ///
+    /// Packets being dropped doesn't necessarily indicate something wrong, e.g., packets are sometimes
+    /// dropped because they aren't relevant for sample building.
     pub prev_dropped_packets: u16,
+
+    /// The number of packets that were identified as padding prior to building this sample.
+    ///
+    /// Some implementations, notably libWebRTC, send padding packets to keep the send rate steady.
+    /// These packets don't carry media and aren't useful for building samples.
+    ///
+    /// This field can be combined with [`Sample::prev_dropped_packets`] to determine if any
+    /// dropped packets are likely to have detrimental impact on the steadiness of the RTP stream.
+    ///
+    /// ## Example adjustment
+    ///
+    /// ```rust
+    /// # use bytes::Bytes;
+    /// # use std::time::{SystemTime, Duration};
+    /// # use webrtc_media::Sample;
+    /// # let sample = Sample {
+    /// #   data: Bytes::new(),
+    /// #   timestamp: SystemTime::now(),
+    /// #   duration: Duration::from_secs(0),
+    /// #   packet_timestamp: 0,
+    /// #   prev_dropped_packets: 10,
+    /// #   prev_padding_packets: 15
+    /// # };
+    /// #
+    /// let adjusted_dropped =
+    /// sample.prev_dropped_packets.saturating_sub(sample.prev_padding_packets);
+    /// ```
+    pub prev_padding_packets: u16,
 }
 
 impl Default for Sample {
@@ -30,6 +77,7 @@ impl Default for Sample {
             duration: Duration::from_secs(0),
             packet_timestamp: 0,
             prev_dropped_packets: 0,
+            prev_padding_packets: 0,
         }
     }
 }
@@ -54,6 +102,10 @@ impl PartialEq for Sample {
         if self.prev_dropped_packets != other.prev_dropped_packets {
             equal = false;
         }
+        if self.prev_padding_packets != other.prev_padding_packets {
+            equal = false;
+        }
+
         equal
     }
 }