[RTCP] remove unnecessary async function/trait

Author: yngrtc

interceptor/src/nack/responder/mod.rs

@@ -69,8 +69,10 @@ impl ResponderInternal {
         };
 
         for n in &nack.nacks {
+            // can't use n.range() since this callback is async fn,
+            // instead, use NackPair into_iter()
             let stream2 = Arc::clone(&stream);
-            n.range(Box::new(
+            let f = Box::new(
                 move |seq: u16| -> Pin<Box<dyn Future<Output = bool> + Send + 'static>> {
                     let stream3 = Arc::clone(&stream2);
                     Box::pin(async move {
@@ -80,12 +82,15 @@ impl ResponderInternal {
                                 log::warn!("failed resending nacked packet: {}", err);
                             }
                         }
-
                         true
                     })
                 },
-            ))
-            .await;
+            );
+            for packet_id in n.into_iter() {
+                if !f(packet_id).await {
+                    return;
+                }
+            }
         }
     }
 }

rtcp/Cargo.toml

@@ -15,7 +15,3 @@ util = { version = "0.7.0", path = "../util", package = "webrtc-util", default-f
 
 bytes = "1"
 thiserror = "1.0"
-
-[dev-dependencies]
-tokio = { version = "1.19", features = ["sync"] }
-tokio-test = "0.4.0" # must match the min version of the `tokio` crate above

rtcp/src/transport_feedbacks/transport_layer_nack/mod.rs

@@ -7,8 +7,6 @@ use util::marshal::{Marshal, MarshalSize, Unmarshal};
 use bytes::{Buf, BufMut};
 use std::any::Any;
 use std::fmt;
-use std::future::Future;
-use std::pin::Pin;
 
 /// PacketBitmap shouldn't be used like a normal integral,
 /// so it's type is masked here. Access it with PacketList().
@@ -24,9 +22,6 @@ pub struct NackPair {
     pub lost_packets: PacketBitmap,
 }
 
-pub type RangeFn =
-    Box<dyn (Fn(u16) -> Pin<Box<dyn Future<Output = bool> + Send + 'static>>) + Send + Sync>;
-
 pub struct NackIterator {
     packet_id: u16,
     bitfield: PacketBitmap,
@@ -72,9 +67,12 @@ impl NackPair {
         self.into_iter().collect()
     }
 
-    pub async fn range(&self, f: RangeFn) {
+    pub fn range<F>(&self, f: F)
+    where
+        F: Fn(u16) -> bool,
+    {
         for packet_id in self.into_iter() {
-            if !f(packet_id).await {
+            if !f(packet_id) {
                 return;
             }
         }

rtcp/src/transport_feedbacks/transport_layer_nack/transport_layer_nack_test.rs

@@ -1,7 +1,6 @@
 use super::*;
 use bytes::Bytes;
-use std::sync::Arc;
-use tokio::sync::Mutex;
+use std::sync::{Arc, Mutex};
 
 #[test]
 fn test_transport_layer_nack_unmarshal() {
@@ -200,48 +199,38 @@ fn test_nack_pair() {
     );
 }
 
-#[tokio::test]
-async fn test_nack_pair_range() {
+#[test]
+fn test_nack_pair_range() {
     let n = NackPair {
         packet_id: 42,
         lost_packets: 2,
     };
 
     let out = Arc::new(Mutex::new(vec![]));
     let out1 = Arc::clone(&out);
-    n.range(Box::new(
-        move |s: u16| -> Pin<Box<dyn Future<Output = bool> + Send + 'static>> {
-            let out2 = Arc::clone(&out1);
-            Box::pin(async move {
-                let mut o = out2.lock().await;
-                o.push(s);
-                true
-            })
-        },
-    ))
-    .await;
+    n.range(move |s: u16| -> bool {
+        let out2 = Arc::clone(&out1);
+        let mut o = out2.lock().unwrap();
+        o.push(s);
+        true
+    });
 
     {
-        let o = out.lock().await;
+        let o = out.lock().unwrap();
         assert_eq!(*o, &[42, 44]);
     }
 
     let out = Arc::new(Mutex::new(vec![]));
     let out1 = Arc::clone(&out);
-    n.range(Box::new(
-        move |s: u16| -> Pin<Box<dyn Future<Output = bool> + Send + 'static>> {
-            let out2 = Arc::clone(&out1);
-            Box::pin(async move {
-                let mut o = out2.lock().await;
-                o.push(s);
-                false
-            })
-        },
-    ))
-    .await;
+    n.range(move |s: u16| -> bool {
+        let out2 = Arc::clone(&out1);
+        let mut o = out2.lock().unwrap();
+        o.push(s);
+        false
+    });
 
     {
-        let o = out.lock().await;
+        let o = out.lock().unwrap();
         assert_eq!(*o, &[42]);
     }
 }

rtp/Cargo.toml

@@ -11,7 +11,7 @@ repository = "https://github.com/webrtc-rs/rtp"
 rust-version = "1.63.0"
 
 [dependencies]
-util = { version = "0.7.0", path = "../util", package = "webrtc-util", default-features = false, features = ["marshal", "sync"] }
+util = { version = "0.7.0", path = "../util", package = "webrtc-util", default-features = false, features = ["marshal"] }
 
 bytes = "1"
 rand = "0.8.5"

rtp/src/sequence.rs

@@ -1,8 +1,7 @@
 use std::fmt;
+use std::sync::atomic::{AtomicU16, AtomicU64, Ordering};
 use std::sync::Arc;
 
-use util::sync::Mutex;
-
 /// Sequencer generates sequential sequence numbers for building RTP packets
 pub trait Sequencer: fmt::Debug {
     fn next_sequence_number(&self) -> u16;
@@ -20,10 +19,10 @@ impl Clone for Box<dyn Sequencer + Send + Sync> {
 /// number
 pub fn new_random_sequencer() -> impl Sequencer {
     let c = Counters {
-        sequence_number: rand::random::<u16>(),
-        roll_over_count: 0,
+        sequence_number: Arc::new(AtomicU16::new(rand::random::<u16>())),
+        roll_over_count: Arc::new(AtomicU64::new(0)),
     };
-    SequencerImpl(Arc::new(Mutex::new(c)))
+    SequencerImpl(c)
 }
 
 /// NewFixedSequencer returns a new sequencer starting from a specific
@@ -32,42 +31,39 @@ pub fn new_fixed_sequencer(s: u16) -> impl Sequencer {
     let sequence_number = if s == 0 { u16::MAX } else { s - 1 };
 
     let c = Counters {
-        sequence_number,
-        roll_over_count: 0,
+        sequence_number: Arc::new(AtomicU16::new(sequence_number)),
+        roll_over_count: Arc::new(AtomicU64::new(0)),
     };
 
-    SequencerImpl(Arc::new(Mutex::new(c)))
+    SequencerImpl(c)
 }
 
 #[derive(Debug, Clone)]
-struct SequencerImpl(Arc<Mutex<Counters>>);
+struct SequencerImpl(Counters);
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 struct Counters {
-    sequence_number: u16,
-    roll_over_count: u64,
+    sequence_number: Arc<AtomicU16>,
+    roll_over_count: Arc<AtomicU64>,
 }
 
 impl Sequencer for SequencerImpl {
     /// NextSequenceNumber increment and returns a new sequence number for
     /// building RTP packets
     fn next_sequence_number(&self) -> u16 {
-        let mut lock = self.0.lock();
-
-        if lock.sequence_number == u16::MAX {
-            lock.roll_over_count += 1;
-            lock.sequence_number = 0;
+        if self.0.sequence_number.load(Ordering::SeqCst) == u16::MAX {
+            self.0.roll_over_count.fetch_add(1, Ordering::SeqCst);
+            self.0.sequence_number.store(0, Ordering::SeqCst);
+            0
         } else {
-            lock.sequence_number += 1;
+            self.0.sequence_number.fetch_add(1, Ordering::SeqCst) + 1
         }
-
-        lock.sequence_number
     }
 
     /// RollOverCount returns the amount of times the 16bit sequence number
     /// has wrapped
     fn roll_over_count(&self) -> u64 {
-        self.0.lock().roll_over_count
+        self.0.roll_over_count.load(Ordering::SeqCst)
     }
 
     fn clone_to(&self) -> Box<dyn Sequencer + Send + Sync> {