sctp: limit the bytes in the PendingQueue by using a semaphore (#367)

As discussed in #360 the pending queue can grow indefinitely if the sender writes packets faster than the association is able to transmit them.

This PR solves this by enforcing a limit on the pending queue. This blocks the sender until enough space is free.

Author: KillingSpark

data/src/data_channel/mod.rs

@@ -118,7 +118,9 @@ impl DataChannel {
             })
             .marshal()?;
 
-            stream.write_sctp(&msg, PayloadProtocolIdentifier::Dcep)?;
+            stream
+                .write_sctp(&msg, PayloadProtocolIdentifier::Dcep)
+                .await?;
         }
         Ok(DataChannel::new(stream, config))
     }
@@ -284,10 +286,13 @@ impl DataChannel {
         };
 
         let n = if data_len == 0 {
-            let _ = self.stream.write_sctp(&Bytes::from_static(&[0]), ppi)?;
+            let _ = self
+                .stream
+                .write_sctp(&Bytes::from_static(&[0]), ppi)
+                .await?;
             0
         } else {
-            let n = self.stream.write_sctp(data, ppi)?;
+            let n = self.stream.write_sctp(data, ppi).await?;
             self.bytes_sent.fetch_add(n, Ordering::SeqCst);
             n
         };
@@ -300,7 +305,8 @@ impl DataChannel {
         let ack = Message::DataChannelAck(DataChannelAck {}).marshal()?;
         Ok(self
             .stream
-            .write_sctp(&ack, PayloadProtocolIdentifier::Dcep)?)
+            .write_sctp(&ack, PayloadProtocolIdentifier::Dcep)
+            .await?)
     }
 
     /// Close closes the DataChannel and the underlying SCTP stream.

sctp/CHANGELOG.md

@@ -2,10 +2,14 @@
 
 ## Unreleased
 
-* Performance improvements
-  * improve algorithm used to push to pending queue from O(n*log(n)) to O(log(n)) [#365](https://github.com/webrtc-rs/webrtc/pull/365)
-  * reuse as many allocations as possible when marshaling [#364](https://github.com/webrtc-rs/webrtc/pull/364)
-  * The lock for the internal association was contended badly because marshaling was done while still in a critical section and also tokio was scheduling tasks badly [#363](https://github.com/webrtc-rs/webrtc/pull/363)
+* Limit the bytes in the PendingQueue to avoid packets accumulating there uncontrollably [367](https://github.com/webrtc-rs/webrtc/pull/367)
+* Improve algorithm used to push to pending queue from O(n*log(n)) to O(log(n)) [#365](https://github.com/webrtc-rs/webrtc/pull/365)
+* Reuse as many allocations as possible when marshaling [#364](https://github.com/webrtc-rs/webrtc/pull/364)
+* The lock for the internal association was contended badly because marshaling was done while still in a critical section and also tokio was scheduling tasks badly [#363](https://github.com/webrtc-rs/webrtc/pull/363)
+
+### Breaking
+
+* Make `sctp::Stream::write` & `sctp::Stream::write_sctp` async again [367](https://github.com/webrtc-rs/webrtc/pull/367)
 
 ## v0.7.0
 

sctp/examples/ping.rs

@@ -83,7 +83,7 @@ async fn main() -> Result<(), Error> {
         while ping_seq_num < 10 {
             let ping_msg = format!("ping {}", ping_seq_num);
             println!("sent: {}", ping_msg);
-            stream_tx.write(&Bytes::from(ping_msg))?;
+            stream_tx.write(&Bytes::from(ping_msg)).await?;
 
             ping_seq_num += 1;
         }

sctp/examples/pong.rs

@@ -86,7 +86,7 @@ async fn main() -> Result<(), Error> {
 
             let pong_msg = format!("pong [{}]", ping_msg);
             println!("sent: {}", pong_msg);
-            stream2.write(&Bytes::from(pong_msg))?;
+            stream2.write(&Bytes::from(pong_msg)).await?;
 
             tokio::time::sleep(Duration::from_secs(1)).await;
         }

sctp/examples/throughput.rs

@@ -136,7 +136,7 @@ fn main() -> Result<(), Error> {
 
                 let mut now = tokio::time::Instant::now();
                 let mut pkt_num = 0;
-                while stream.write(&buf.clone().into()).is_ok() {
+                while stream.write(&buf.clone().into()).await.is_ok() {
                     pkt_num += 1;
                     if now.elapsed().as_secs() == 1 {
                         println!("Send {} pkts", pkt_num);

sctp/src/association/association_internal.rs

@@ -347,7 +347,7 @@ impl AssociationInternal {
     ) -> Vec<Packet> {
         // Pop unsent data chunks from the pending queue to send as much as
         // cwnd and rwnd allow.
-        let (chunks, sis_to_reset) = self.pop_pending_data_chunks_to_send();
+        let (chunks, sis_to_reset) = self.pop_pending_data_chunks_to_send().await;
         if !chunks.is_empty() {
             // Start timer. (noop if already started)
             log::trace!("[{}] T3-rtx timer start (pt1)", self.name);
@@ -1717,7 +1717,7 @@ impl AssociationInternal {
         self.handle_peer_last_tsn_and_acknowledgement(false)
     }
 
-    fn send_reset_request(&mut self, stream_identifier: u16) -> Result<()> {
+    async fn send_reset_request(&mut self, stream_identifier: u16) -> Result<()> {
         let state = self.get_state();
         if state != AssociationState::Established {
             return Err(Error::ErrResetPacketInStateNotExist);
@@ -1733,7 +1733,7 @@ impl AssociationInternal {
             ..Default::default()
         };
 
-        self.pending_queue.push(c);
+        self.pending_queue.push(c).await;
         self.awake_write_loop();
 
         Ok(())
@@ -1798,7 +1798,7 @@ impl AssociationInternal {
     }
 
     /// Move the chunk peeked with self.pending_queue.peek() to the inflight_queue.
-    fn move_pending_data_chunk_to_inflight_queue(
+    async fn move_pending_data_chunk_to_inflight_queue(
         &mut self,
         beginning_fragment: bool,
         unordered: bool,
@@ -1840,7 +1840,7 @@ impl AssociationInternal {
 
     /// pop_pending_data_chunks_to_send pops chunks from the pending queues as many as
     /// the cwnd and rwnd allows to send.
-    fn pop_pending_data_chunks_to_send(&mut self) -> (Vec<ChunkPayloadData>, Vec<u16>) {
+    async fn pop_pending_data_chunks_to_send(&mut self) -> (Vec<ChunkPayloadData>, Vec<u16>) {
         let mut chunks = vec![];
         let mut sis_to_reset = vec![]; // stream identifiers to reset
 
@@ -1885,8 +1885,9 @@ impl AssociationInternal {
 
             self.rwnd -= data_len as u32;
 
-            if let Some(chunk) =
-                self.move_pending_data_chunk_to_inflight_queue(beginning_fragment, unordered)
+            if let Some(chunk) = self
+                .move_pending_data_chunk_to_inflight_queue(beginning_fragment, unordered)
+                .await
             {
                 chunks.push(chunk);
             }
@@ -1898,8 +1899,9 @@ impl AssociationInternal {
             if let Some(c) = self.pending_queue.peek() {
                 let (beginning_fragment, unordered) = (c.beginning_fragment, c.unordered);
 
-                if let Some(chunk) =
-                    self.move_pending_data_chunk_to_inflight_queue(beginning_fragment, unordered)
+                if let Some(chunk) = self
+                    .move_pending_data_chunk_to_inflight_queue(beginning_fragment, unordered)
+                    .await
                 {
                     chunks.push(chunk);
                 }

sctp/src/association/association_internal/association_internal_test.rs

@@ -437,8 +437,8 @@ async fn test_assoc_handle_init() -> Result<()> {
     Ok(())
 }
 
-#[test]
-fn test_assoc_max_message_size_default() -> Result<()> {
+#[tokio::test]
+async fn test_assoc_max_message_size_default() -> Result<()> {
     let mut a = create_association_internal(Config {
         net_conn: Arc::new(DumbConn {}),
         max_receive_buffer_size: 0,
@@ -458,7 +458,7 @@ fn test_assoc_max_message_size_default() -> Result<()> {
         let p = Bytes::from(vec![0u8; 65537]);
         let ppi = PayloadProtocolIdentifier::from(s.default_payload_type.load(Ordering::SeqCst));
 
-        if let Err(err) = s.write_sctp(&p.slice(..65536), ppi) {
+        if let Err(err) = s.write_sctp(&p.slice(..65536), ppi).await {
             assert_ne!(
                 Error::ErrOutboundPacketTooLarge,
                 err,
@@ -468,7 +468,7 @@ fn test_assoc_max_message_size_default() -> Result<()> {
             assert!(false, "should be error");
         }
 
-        if let Err(err) = s.write_sctp(&p.slice(..65537), ppi) {
+        if let Err(err) = s.write_sctp(&p.slice(..65537), ppi).await {
             assert_eq!(
                 Error::ErrOutboundPacketTooLarge,
                 err,
@@ -482,8 +482,8 @@ fn test_assoc_max_message_size_default() -> Result<()> {
     Ok(())
 }
 
-#[test]
-fn test_assoc_max_message_size_explicit() -> Result<()> {
+#[tokio::test]
+async fn test_assoc_max_message_size_explicit() -> Result<()> {
     let mut a = create_association_internal(Config {
         net_conn: Arc::new(DumbConn {}),
         max_receive_buffer_size: 0,
@@ -504,7 +504,7 @@ fn test_assoc_max_message_size_explicit() -> Result<()> {
         let p = Bytes::from(vec![0u8; 30001]);
         let ppi = PayloadProtocolIdentifier::from(s.default_payload_type.load(Ordering::SeqCst));
 
-        if let Err(err) = s.write_sctp(&p.slice(..30000), ppi) {
+        if let Err(err) = s.write_sctp(&p.slice(..30000), ppi).await {
             assert_ne!(
                 Error::ErrOutboundPacketTooLarge,
                 err,
@@ -514,7 +514,7 @@ fn test_assoc_max_message_size_explicit() -> Result<()> {
             assert!(false, "should be error");
         }
 
-        if let Err(err) = s.write_sctp(&p.slice(..30001), ppi) {
+        if let Err(err) = s.write_sctp(&p.slice(..30001), ppi).await {
             assert_eq!(
                 Error::ErrOutboundPacketTooLarge,
                 err,