add PendingQueue::append (#345)

which appends chunks to the back of the pending queue.

Consider the following two simultaneous calls to `send_payload_data`:

```
  1. [beginning_fragment1, data2, data3, ending_fragment4]
  2. [beginning_fragment5, data6, data7, ending_fragment8]
```

If `push` is used, due to individual locks, it's possible to end up with:

```
  [beginning_fragment1, data2, beginning_fragment5, data3, data6, ending_fragment4,
     data6, data7, ending_fragment8]
```

This will result in wrong data received by the remote.

Author: melekes

sctp/CHANGELOG.md

@@ -5,6 +5,7 @@
 * Increased minimum support rust version to `1.60.0`.
 * Do not loose data in `PollStream::poll_write` [#341](https://github.com/webrtc-rs/webrtc/pull/341).
 * `PollStream::poll_shutdown`: make sure to flush any writes before shutting down [#340](https://github.com/webrtc-rs/webrtc/pull/340)
+* Fixed a possible bug when adding chunks to pending queue [#345](https://github.com/webrtc-rs/webrtc/pull/345)
 
 ## v0.6.1
 

sctp/src/association/association_internal.rs

@@ -1996,22 +1996,6 @@ impl AssociationInternal {
         packets
     }
 
-    /// send_payload_data sends the data chunks.
-    async fn send_payload_data(&mut self, chunks: Vec<ChunkPayloadData>) -> Result<()> {
-        let state = self.get_state();
-        if state != AssociationState::Established {
-            return Err(Error::ErrPayloadDataStateNotExist);
-        }
-
-        // Push the chunks into the pending queue first.
-        for c in chunks {
-            self.pending_queue.push(c).await;
-        }
-
-        self.awake_write_loop();
-        Ok(())
-    }
-
     fn check_partial_reliability_status(&self, c: &ChunkPayloadData) {
         if !self.use_forward_tsn {
             return;

sctp/src/queue/pending_queue.rs

@@ -25,18 +25,58 @@ impl PendingQueue {
         PendingQueue::default()
     }
 
+    /// Appends a chunk to the back of the pending queue.
     pub(crate) async fn push(&self, c: ChunkPayloadData) {
-        self.n_bytes.fetch_add(c.user_data.len(), Ordering::SeqCst);
+        let user_data_len = c.user_data.len();
+
         if c.unordered {
             let mut unordered_queue = self.unordered_queue.lock().await;
             unordered_queue.push_back(c);
         } else {
             let mut ordered_queue = self.ordered_queue.lock().await;
             ordered_queue.push_back(c);
         }
+
+        self.n_bytes.fetch_add(user_data_len, Ordering::SeqCst);
         self.queue_len.fetch_add(1, Ordering::SeqCst);
     }
 
+    /// Appends chunks to the back of the pending queue.
+    ///
+    /// # Panics
+    ///
+    /// If it's a mix of unordered and ordered chunks.
+    pub(crate) async fn append(&self, chunks: Vec<ChunkPayloadData>) {
+        if chunks.is_empty() {
+            return;
+        }
+
+        let total_user_data_len = chunks.iter().fold(0, |acc, c| acc + c.user_data.len());
+        let chunks_len = chunks.len();
+
+        let unordered = chunks
+            .first()
+            .expect("chunks to not be empty because of the above check")
+            .unordered;
+        if unordered {
+            let mut unordered_queue = self.unordered_queue.lock().await;
+            for c in chunks {
+                assert!(c.unordered, "expected all chunks to be unordered");
+                unordered_queue.push_back(c);
+            }
+        } else {
+            let mut ordered_queue = self.ordered_queue.lock().await;
+            for c in chunks {
+                assert!(!c.unordered, "expected all chunks to be ordered");
+                ordered_queue.push_back(c);
+            }
+        }
+
+        self.n_bytes
+            .fetch_add(total_user_data_len, Ordering::SeqCst);
+        self.queue_len.fetch_add(chunks_len, Ordering::SeqCst);
+    }
+
     pub(crate) async fn peek(&self) -> Option<ChunkPayloadData> {
         if self.selected.load(Ordering::SeqCst) {
             if self.unordered_is_selected.load(Ordering::SeqCst) {

sctp/src/queue/queue_test.rs

@@ -425,6 +425,21 @@ async fn test_pending_queue_selection_persistence() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_pending_queue_append() -> Result<()> {
+    let pq = PendingQueue::new();
+    pq.append(vec![
+        make_data_chunk(0, false, NO_FRAGMENT),
+        make_data_chunk(1, false, NO_FRAGMENT),
+        make_data_chunk(3, false, NO_FRAGMENT),
+    ])
+    .await;
+    assert_eq!(30, pq.get_num_bytes(), "total bytes mismatch");
+    assert_eq!(3, pq.len(), "len mismatch");
+
+    Ok(())
+}
+
 ///////////////////////////////////////////////////////////////////
 //reassembly_queue_test
 ///////////////////////////////////////////////////////////////////

sctp/src/stream/mod.rs

@@ -488,10 +488,8 @@ impl Stream {
             return Err(Error::ErrPayloadDataStateNotExist);
         }
 
-        // Push the chunks into the pending queue first.
-        for c in chunks {
-            self.pending_queue.push(c).await;
-        }
+        // NOTE: append is used here instead of push in order to prevent chunks interlacing.
+        self.pending_queue.append(chunks).await;
 
         self.awake_write_loop();
         Ok(())