Pause in TrackBinding when direction is not send

When direction is such that we are not sending. The writing of RTP
packets is stopped already in TrackBinding to avoid allocating
sequence numbers when they are not needed.

Author: algesten

webrtc/src/rtp_transceiver/rtp_sender/mod.rs

@@ -332,6 +332,7 @@ impl RTCRtpSender {
                     .await,
                 ssrc: context.ssrc,
                 write_stream: context.write_stream.clone(),
+                paused: self.paused.clone(),
             };
 
             t.bind(&new_context).await
@@ -392,6 +393,7 @@ impl RTCRtpSender {
                 write_stream: Some(
                     Arc::clone(&write_stream) as Arc<dyn TrackLocalWriter + Send + Sync>
                 ),
+                paused: self.paused.clone(),
             };
 
             let codec = if let Some(t) = &*track {

webrtc/src/track/track_local/mod.rs

@@ -35,6 +35,7 @@ pub struct TrackLocalContext {
     pub(crate) params: RTCRtpParameters,
     pub(crate) ssrc: SSRC,
     pub(crate) write_stream: Option<Arc<dyn TrackLocalWriter + Send + Sync>>,
+    pub(crate) paused: Arc<AtomicBool>,
 }
 
 impl TrackLocalContext {
@@ -104,6 +105,13 @@ pub(crate) struct TrackBinding {
     ssrc: SSRC,
     payload_type: PayloadType,
     write_stream: Option<Arc<dyn TrackLocalWriter + Send + Sync>>,
+    paused: Arc<AtomicBool>,
+}
+
+impl TrackBinding {
+    pub fn is_paused(&self) -> bool {
+        self.paused.load(Ordering::SeqCst)
+    }
 }
 
 pub(crate) struct InterceptorToTrackLocalWriter {
@@ -118,6 +126,10 @@ impl InterceptorToTrackLocalWriter {
             paused,
         }
     }
+
+    fn is_paused(&self) -> bool {
+        self.paused.load(Ordering::SeqCst)
+    }
 }
 
 impl std::fmt::Debug for InterceptorToTrackLocalWriter {
@@ -129,7 +141,7 @@ impl std::fmt::Debug for InterceptorToTrackLocalWriter {
 #[async_trait]
 impl TrackLocalWriter for InterceptorToTrackLocalWriter {
     async fn write_rtp(&self, pkt: &rtp::packet::Packet) -> Result<usize> {
-        if self.paused.load(Ordering::SeqCst) {
+        if self.is_paused() {
             return Ok(0);
         }
 

webrtc/src/track/track_local/track_local_static_rtp.rs

@@ -28,6 +28,16 @@ impl TrackLocalStaticRTP {
     pub fn codec(&self) -> RTCRtpCodecCapability {
         self.codec.clone()
     }
+
+    pub async fn any_binding_paused(&self) -> bool {
+        let bindings = self.bindings.lock().await;
+        bindings.iter().any(|b| b.paused.load(Ordering::SeqCst))
+    }
+
+    pub async fn all_binding_paused(&self) -> bool {
+        let bindings = self.bindings.lock().await;
+        bindings.iter().all(|b| b.paused.load(Ordering::SeqCst))
+    }
 }
 
 #[async_trait]
@@ -50,6 +60,7 @@ impl TrackLocal for TrackLocalStaticRTP {
                     payload_type: codec.payload_type,
                     write_stream: t.write_stream(),
                     id: t.id(),
+                    paused: t.paused.clone(),
                 }));
             }
 
@@ -122,6 +133,9 @@ impl TrackLocalWriter for TrackLocalStaticRTP {
             bindings.clone()
         };
         for b in bindings {
+            if b.is_paused() {
+                continue;
+            }
             pkt.header.ssrc = b.ssrc;
             pkt.header.payload_type = b.payload_type;
             if let Some(write_stream) = &b.write_stream {

webrtc/src/track/track_local/track_local_static_sample.rs

@@ -3,6 +3,7 @@ use super::*;
 use crate::error::flatten_errs;
 
 use crate::track::RTP_OUTBOUND_MTU;
+use log::warn;
 use media::Sample;
 use tokio::sync::Mutex;
 
@@ -11,6 +12,7 @@ struct TrackLocalStaticSampleInternal {
     packetizer: Option<Box<dyn rtp::packetizer::Packetizer + Send + Sync>>,
     sequencer: Option<Box<dyn rtp::sequence::Sequencer + Send + Sync>>,
     clock_rate: f64,
+    did_warn_about_wonky_pause: bool,
 }
 
 /// TrackLocalStaticSample is a TrackLocal that has a pre-set codec and accepts Samples.
@@ -32,6 +34,7 @@ impl TrackLocalStaticSample {
                 packetizer: None,
                 sequencer: None,
                 clock_rate: 0.0f64,
+                did_warn_about_wonky_pause: false,
             }),
         }
     }
@@ -52,6 +55,40 @@ impl TrackLocalStaticSample {
             return Ok(());
         }
 
+        let (any_paused, all_paused) = (
+            self.rtp_track.any_binding_paused().await,
+            self.rtp_track.all_binding_paused().await,
+        );
+
+        if all_paused {
+            // Abort already here to not increment sequence numbers.
+            return Ok(());
+        }
+
+        if any_paused {
+            // This is a problem state due to how this impl is structured. The sequencer will allocate
+            // one sequence number per RTP packet regardless of how many TrackBinding that will send
+            // the packet. I.e. we get the same sequence number per multiple SSRC, which is not good
+            // for SRTP, but that's how it works.
+            //
+            // Chrome has further a problem with regards to jumps in sequence number. Consider this:
+            //
+            // 1. Create track local
+            // 2. Bind track local to track 1.
+            // 3. Bind track local to track 2.
+            // 4. Pause track 1.
+            // 5. Keep sending...
+            //
+            // At this point, the track local will keep incrementing the sequence number, because we have
+            // one binding that is still active. However Chrome can only accept a relatively small jump
+            // in SRTP key deriving, which means if this pause state of one binding persists for a longer
+            // time, the track can never be resumed (against Chrome).
+            if !internal.did_warn_about_wonky_pause {
+                internal.did_warn_about_wonky_pause = true;
+                warn!("Detected multiple track bindings where only one was paused");
+            }
+        }
+
         // skip packets by the number of previously dropped packets
         if let Some(sequencer) = &internal.sequencer {
             for _ in 0..sample.prev_dropped_packets {