Don't generate empty TWCC feedback packets (#324)

lookback-hugotunius · web-flow · commit 7220f446a908 · 2022-11-01T16:20:37.000+01:00
* Don't generate empty TWCC feedback packets

The previous version of this code would very often generate a feedback
packet with no chunks. These packets would be ignored by libWebRTC which
logs an error message saying:

```
Buffer too small (16 bytes) to fit a FeedbackPacket. Minimum size = 18
```

When receiving such a packet.

It also felt weird to clear the received packets when no chunks were
generated for those packets.
diff --git a/interceptor/src/twcc/mod.rs b/interceptor/src/twcc/mod.rs
@@ -56,12 +56,11 @@ impl Recorder {
 
     /// build_feedback_packet creates a new RTCP packet containing a TWCC feedback report.
     pub fn build_feedback_packet(&mut self) -> Vec<Box<dyn rtcp::packet::Packet + Send + Sync>> {
-        let mut feedback = Feedback::new(self.sender_ssrc, self.media_ssrc, self.fb_pkt_cnt);
-        self.fb_pkt_cnt = self.fb_pkt_cnt.wrapping_add(1);
         if self.received_packets.len() < 2 {
-            self.received_packets.clear();
-            return vec![Box::new(feedback.get_rtcp())];
+            return vec![];
         }
+        let mut feedback = Feedback::new(self.sender_ssrc, self.media_ssrc, self.fb_pkt_cnt);
+        self.fb_pkt_cnt = self.fb_pkt_cnt.wrapping_add(1);
 
         self.received_packets
             .sort_by(|a: &PktInfo, b: &PktInfo| -> Ordering {
diff --git a/interceptor/src/twcc/receiver/mod.rs b/interceptor/src/twcc/receiver/mod.rs
@@ -8,7 +8,7 @@ use crate::*;
 use receiver_stream::ReceiverStream;
 
 use rtp::extension::transport_cc_extension::TransportCcExtension;
-use std::time::{Duration, SystemTime};
+use std::time::Duration;
 use tokio::sync::{mpsc, Mutex};
 use tokio::time::MissedTickBehavior;
 use util::Unmarshal;
@@ -44,7 +44,7 @@ impl InterceptorBuilder for ReceiverBuilder {
                 streams: Mutex::new(HashMap::new()),
                 close_rx: Mutex::new(Some(close_rx)),
             }),
-            start_time: SystemTime::now(),
+            start_time: tokio::time::Instant::now(),
             packet_chan_tx,
             wg: Mutex::new(Some(WaitGroup::new())),
             close_tx: Mutex::new(Some(close_tx)),
@@ -72,7 +72,8 @@ struct ReceiverInternal {
 pub struct Receiver {
     internal: Arc<ReceiverInternal>,
 
-    start_time: SystemTime,
+    // we use tokio's Instant because it makes testing easier via `tokio::time::advance`.
+    start_time: tokio::time::Instant,
     packet_chan_tx: mpsc::Sender<Packet>,
 
     wg: Mutex<Option<WaitGroup>>,
@@ -132,6 +133,10 @@ impl Receiver {
                         recorder.build_feedback_packet()
                     };
 
+                    if pkts.is_empty() {
+                        continue;
+                    }
+
                     if let Err(err) = rtcp_writer.write(&pkts, &a).await{
                         log::error!("rtcp_writer.write got err: {}", err);
                     }
diff --git a/interceptor/src/twcc/receiver/receiver_stream.rs b/interceptor/src/twcc/receiver/receiver_stream.rs
@@ -5,7 +5,8 @@ pub(super) struct ReceiverStream {
     hdr_ext_id: u8,
     ssrc: u32,
     packet_chan_tx: mpsc::Sender<Packet>,
-    start_time: SystemTime,
+    // we use tokio's Instant because it makes testing easier via `tokio::time::advance`.
+    start_time: tokio::time::Instant,
 }
 
 impl ReceiverStream {
@@ -14,7 +15,7 @@ impl ReceiverStream {
         hdr_ext_id: u8,
         ssrc: u32,
         packet_chan_tx: mpsc::Sender<Packet>,
-        start_time: SystemTime,
+        start_time: tokio::time::Instant,
     ) -> Self {
         ReceiverStream {
             parent_rtp_reader,
@@ -43,10 +44,8 @@ impl RTPReader for ReceiverStream {
                 .send(Packet {
                     hdr: p.header,
                     sequence_number: tcc_ext.transport_sequence,
-                    arrival_time: SystemTime::now()
-                        .duration_since(self.start_time)
-                        .unwrap_or_else(|_| Duration::from_secs(0))
-                        .as_micros() as i64,
+                    arrival_time: (tokio::time::Instant::now() - self.start_time).as_micros()
+                        as i64,
                     ssrc: self.ssrc,
                 })
                 .await;
diff --git a/interceptor/src/twcc/receiver/receiver_test.rs b/interceptor/src/twcc/receiver/receiver_test.rs
@@ -26,18 +26,13 @@ async fn test_twcc_receiver_interceptor_before_any_packets() -> Result<()> {
     )
     .await;
 
-    let pkts = stream.written_rtcp().await.unwrap();
-    assert_eq!(pkts.len(), 1);
-    if let Some(tlcc) = pkts[0].as_any().downcast_ref::<TransportLayerCc>() {
-        assert_eq!(0, tlcc.packet_status_count);
-        assert_eq!(0, tlcc.fb_pkt_count);
-        assert_eq!(0, tlcc.base_sequence_number);
-        assert_eq!(0, tlcc.media_ssrc);
-        assert_eq!(0, tlcc.reference_time);
-        assert_eq!(0, tlcc.recv_deltas.len());
-        assert_eq!(0, tlcc.packet_chunks.len());
-    } else {
-        assert!(false);
+    tokio::select! {
+        pkts = stream.written_rtcp() => {
+            assert!(pkts.map(|p| p.is_empty()).unwrap_or(true), "Should not have sent an RTCP packet before receiving the first RTP packets")
+        }
+        _ = tokio::time::sleep(Duration::from_millis(300)) => {
+            // All good
+        }
     }
 
     stream.close().await?;
@@ -101,9 +96,7 @@ async fn test_twcc_receiver_interceptor_after_rtp_packets() -> Result<()> {
     Ok(())
 }
 
-//TODO: remove this conditional test
-#[cfg(not(any(target_os = "macos", target_os = "windows")))]
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn test_twcc_receiver_interceptor_different_delays_between_rtp_packets() -> Result<()> {
     let builder = Receiver::builder().with_interval(Duration::from_millis(500));
     let icpr = builder.build("")?;
@@ -124,7 +117,7 @@ async fn test_twcc_receiver_interceptor_different_delays_between_rtp_packets() -
 
     let delays = vec![0, 10, 100, 200];
     for (i, d) in delays.iter().enumerate() {
-        tokio::time::sleep(Duration::from_millis(*d)).await;
+        tokio::time::advance(Duration::from_millis(*d)).await;
 
         let mut hdr = rtp::header::Header::default();
         let tcc = TransportCcExtension {
@@ -139,13 +132,17 @@ async fn test_twcc_receiver_interceptor_different_delays_between_rtp_packets() -
                 ..Default::default()
             })
             .await;
+
+        // Yield so this packet can be processed
+        tokio::task::yield_now().await;
     }
 
-    // tick immediately, let's ignore the first rtcp pkt
-    let _ = stream.written_rtcp().await.unwrap();
+    // Force a packet to be generated
+    tokio::time::advance(Duration::from_millis(2001)).await;
+    tokio::task::yield_now().await;
 
-    // the second 500ms tick will works
     let pkts = stream.written_rtcp().await.unwrap();
+
     assert_eq!(pkts.len(), 1);
     if let Some(cc) = pkts[0].as_any().downcast_ref::<TransportLayerCc>() {
         assert_eq!(0, cc.base_sequence_number);
@@ -171,9 +168,7 @@ async fn test_twcc_receiver_interceptor_different_delays_between_rtp_packets() -
     Ok(())
 }
 
-//TODO: remove this conditional test
-#[cfg(not(target_os = "macos"))]
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn test_twcc_receiver_interceptor_packet_loss() -> Result<()> {
     let builder = Receiver::builder().with_interval(Duration::from_secs(2));
     let icpr = builder.build("")?;
@@ -192,26 +187,21 @@ async fn test_twcc_receiver_interceptor_packet_loss() -> Result<()> {
     )
     .await;
 
-    let sequence_number_to_delay: HashMap<u16, u64> = [
+    let sequence_number_to_delay = &[
         (0, 0),
         (1, 10),
         (4, 100),
         (8, 200),
         (9, 20),
         (10, 20),
         (30, 300),
-    ]
-    .iter()
-    .cloned()
-    .collect();
-
-    for i in &[0, 1, 4, 8, 9, 10, 30] {
-        let d = sequence_number_to_delay.get(i).unwrap();
-        tokio::time::sleep(Duration::from_millis(*d)).await;
+    ];
 
+    for (i, d) in sequence_number_to_delay {
+        tokio::time::advance(Duration::from_millis(*d)).await;
         let mut hdr = rtp::header::Header::default();
         let tcc = TransportCcExtension {
-            transport_sequence: *i as u16,
+            transport_sequence: *i,
         }
         .marshal()?;
         hdr.set_extension(1, tcc)?;
@@ -221,13 +211,17 @@ async fn test_twcc_receiver_interceptor_packet_loss() -> Result<()> {
                 ..Default::default()
             })
             .await;
+
+        // Yield so this packet can be processed
+        tokio::task::yield_now().await;
     }
 
-    // tick immediately, let's ignore the first rtcp pkt
-    let _ = stream.written_rtcp().await.unwrap();
+    // Force a packet to be generated
+    tokio::time::advance(Duration::from_millis(2001)).await;
+    tokio::task::yield_now().await;
 
-    // the second 500ms tick will works
     let pkts = stream.written_rtcp().await.unwrap();
+
     assert_eq!(pkts.len(), 1);
     if let Some(cc) = pkts[0].as_any().downcast_ref::<TransportLayerCc>() {
         assert_eq!(0, cc.base_sequence_number);
diff --git a/interceptor/src/twcc/twcc_test.rs b/interceptor/src/twcc/twcc_test.rs
@@ -494,19 +494,18 @@ fn test_build_feedback_packet_rolling() -> Result<()> {
     let mut r = Recorder::new(5000);
 
     let mut arrival_time = SCALE_FACTOR_REFERENCE_TIME as i64;
-    add_run(&mut r, &[0], &[arrival_time]);
+    add_run(&mut r, &[3], &[arrival_time]);
 
     let rtcp_packets = r.build_feedback_packet();
-    assert_eq!(1, rtcp_packets.len()); // Empty TWCC
+    assert_eq!(0, rtcp_packets.len());
 
     add_run(
         &mut r,
-        &[4, 8, 9, 10],
+        &[4, 8, 9],
         &[
             increase_time(&mut arrival_time, TYPE_TCC_DELTA_SCALE_FACTOR),
             increase_time(&mut arrival_time, TYPE_TCC_DELTA_SCALE_FACTOR),
             increase_time(&mut arrival_time, TYPE_TCC_DELTA_SCALE_FACTOR),
-            increase_time(&mut arrival_time, TYPE_TCC_DELTA_SCALE_FACTOR),
         ],
     );
 
@@ -516,27 +515,27 @@ fn test_build_feedback_packet_rolling() -> Result<()> {
     let expected = TransportLayerCc {
         sender_ssrc: 5000,
         media_ssrc: 5000,
-        base_sequence_number: 4,
+        base_sequence_number: 3,
         reference_time: 1,
-        fb_pkt_count: 1,
+        fb_pkt_count: 0,
         packet_status_count: 7,
         packet_chunks: vec![PacketStatusChunk::StatusVectorChunk(StatusVectorChunk {
             type_tcc: StatusChunkTypeTcc::StatusVectorChunk,
             symbol_size: SymbolSizeTypeTcc::TwoBit,
             symbol_list: vec![
+                SymbolTypeTcc::PacketReceivedSmallDelta,
                 SymbolTypeTcc::PacketReceivedSmallDelta,
                 SymbolTypeTcc::PacketNotReceived,
                 SymbolTypeTcc::PacketNotReceived,
                 SymbolTypeTcc::PacketNotReceived,
                 SymbolTypeTcc::PacketReceivedSmallDelta,
                 SymbolTypeTcc::PacketReceivedSmallDelta,
-                SymbolTypeTcc::PacketReceivedSmallDelta,
             ],
         })],
         recv_deltas: vec![
             RecvDelta {
                 type_tcc_packet: SymbolTypeTcc::PacketReceivedSmallDelta,
-                delta: TYPE_TCC_DELTA_SCALE_FACTOR,
+                delta: 0,
             },
             RecvDelta {
                 type_tcc_packet: SymbolTypeTcc::PacketReceivedSmallDelta,
