replace encoding/decoding spin locks with mutex+condvar

Author: github-af

Cargo.toml

@@ -35,4 +35,3 @@ enum-glob-use = "allow"
 missing-errors-doc = "allow"
 nursery = { level = "deny", priority = -1 }
 or_fun_call = "allow"
-significant_drop_tightening = "allow"

src/receive/decode.rs

@@ -1,14 +1,16 @@
 //! Worker that decodes `RaptorQ` packets into protocol blocks
 
 use crate::{protocol, receive};
-use std::{sync, thread};
+use std::thread;
 
 pub fn start<ClientNew, ClientEnd>(
     receiver: &receive::Receiver<ClientNew, ClientEnd>,
 ) -> Result<(), receive::Error> {
     loop {
         match receiver.for_decode.recv()? {
             super::Reassembled::Block { id, packets } => {
+                log::debug!("received block {id} to decode");
+
                 match receiver.raptorq.decode(id, packets) {
                     None => {
                         log::error!("lost block {id} (failed to decode)");
@@ -17,23 +19,32 @@ pub fn start<ClientNew, ClientEnd>(
                     Some(block) => {
                         log::debug!("block {id} decoded with {} bytes!", block.len());
 
-                        'inner: loop {
-                            let block_to_dispatch = receiver
-                                .block_to_dispatch
-                                .load(sync::atomic::Ordering::SeqCst);
-
-                            if block_to_dispatch == id {
-                                receiver
-                                    .to_dispatch
-                                    .send(Some(protocol::Block::deserialize(block)))?;
-                                receiver
-                                    .block_to_dispatch
-                                    .fetch_add(1, sync::atomic::Ordering::SeqCst);
-                                break 'inner;
-                            }
-
-                            thread::yield_now();
-                        }
+                        let mut block_to_dispatch =
+                            receiver.block_to_dispatch.0.lock().map_err(|e| {
+                                receive::Error::Other(format!(
+                                    "failed to acquire block_to_dispatch mutex: {e}"
+                                ))
+                            })?;
+
+                        block_to_dispatch = receiver
+                            .block_to_dispatch
+                            .1
+                            .wait_while(block_to_dispatch, |block_to_dispatch| {
+                                *block_to_dispatch != id
+                            })
+                            .map_err(|e| {
+                                receive::Error::Other(format!(
+                                    "failed to wait_while block_to_dispatch mutex: {e}"
+                                ))
+                            })?;
+
+                        receiver
+                            .to_dispatch
+                            .send(Some(protocol::Block::deserialize(block)))?;
+
+                        *block_to_dispatch = block_to_dispatch.wrapping_add(1);
+                        drop(block_to_dispatch);
+                        receiver.block_to_dispatch.1.notify_all();
                     }
                 }
             }

src/receive/mod.rs

@@ -148,7 +148,7 @@ pub struct Receiver<ClientNew, ClientEnd> {
     config: Config,
     raptorq: protocol::RaptorQ,
     multiplex_control: semka::Sem,
-    block_to_dispatch: sync::atomic::AtomicU8,
+    block_to_dispatch: (sync::Mutex<u8>, sync::Condvar),
     to_reblock: crossbeam_channel::Sender<crate::udp::Datagrams>,
     for_reblock: crossbeam_channel::Receiver<crate::udp::Datagrams>,
     to_decode: crossbeam_channel::Sender<Reassembled>,
@@ -187,7 +187,7 @@ where
         let multiplex_control = semka::Sem::new(config.max_clients)
             .ok_or(Error::Other("failed to create semaphore".into()))?;
 
-        let block_to_dispatch = sync::atomic::AtomicU8::new(0);
+        let block_to_dispatch = (sync::Mutex::new(0), sync::Condvar::new());
 
         let (to_reblock, for_reblock) = crossbeam_channel::unbounded();
         let (to_decode, for_decode) = crossbeam_channel::unbounded();

src/receive/reblock.rs

@@ -2,7 +2,7 @@
 //! reordering
 
 use crate::{receive, udp};
-use std::{mem, sync, thread};
+use std::{mem, thread};
 
 pub const WINDOW_WIDTH: u8 = u8::MAX / 2;
 
@@ -63,9 +63,12 @@ pub fn start<ClientNew, ClientEnd>(
             let packet = raptorq::EncodingPacket::deserialize(first_datagram);
             cur_id = packet.payload_id().source_block_number();
 
-            receiver
-                .block_to_dispatch
-                .store(cur_id, sync::atomic::Ordering::SeqCst);
+            let mut block_to_dispatch = receiver.block_to_dispatch.0.lock().map_err(|e| {
+                receive::Error::Other(format!("failed to acquire block_to_dispatch mutex: {e}"))
+            })?;
+            *block_to_dispatch = cur_id;
+            drop(block_to_dispatch);
+            receiver.block_to_dispatch.1.notify_all();
 
             let mut id = cur_id;
             let last = id.wrapping_add(WINDOW_WIDTH);

src/send/encoding.rs

@@ -1,7 +1,6 @@
 //! Worker that encodes protocol blocks into `RaptorQ` packets
 
 use crate::send;
-use std::{sync::atomic, thread};
 
 pub fn start<C>(sender: &send::Sender<C>) -> Result<(), send::Error> {
     loop {
@@ -16,16 +15,22 @@ pub fn start<C>(sender: &send::Sender<C>) -> Result<(), send::Error> {
 
         let packets = sender.raptorq.encode(block_id, block.serialized());
 
-        'inner: loop {
-            let block_to_send = sender.block_to_send.load(atomic::Ordering::SeqCst);
+        let mut block_to_send = sender.block_to_send.0.lock().map_err(|e| {
+            send::Error::Other(format!("failed to acquire block_to_send mutex: {e}"))
+        })?;
 
-            if block_to_send == block_id {
-                sender.to_send.send(Some(packets))?;
-                sender.block_to_send.fetch_add(1, atomic::Ordering::SeqCst);
-                break 'inner;
-            }
+        block_to_send = sender
+            .block_to_send
+            .1
+            .wait_while(block_to_send, |block_to_send| *block_to_send != block_id)
+            .map_err(|e| {
+                send::Error::Other(format!("failed to wait_while block_to_send mutex: {e}"))
+            })?;
 
-            thread::yield_now();
-        }
+        sender.to_send.send(Some(packets))?;
+
+        *block_to_send = block_to_send.wrapping_add(1);
+        drop(block_to_send);
+        sender.block_to_send.1.notify_all();
     }
 }

src/send/mod.rs

@@ -117,7 +117,7 @@ pub struct Sender<C> {
     raptorq: protocol::RaptorQ,
     multiplex_control: semka::Sem,
     block_to_encode: sync::atomic::AtomicU8,
-    block_to_send: sync::atomic::AtomicU8,
+    block_to_send: (sync::Mutex<u8>, sync::Condvar),
     to_server: crossbeam_channel::Sender<Option<C>>,
     for_server: crossbeam_channel::Receiver<Option<C>>,
     to_encoding: crossbeam_channel::Sender<Option<(u8, protocol::Block)>>,
@@ -139,7 +139,7 @@ where
             .ok_or(Error::Other("failed to create semaphore".into()))?;
 
         let block_to_encode = sync::atomic::AtomicU8::new(0);
-        let block_to_send = sync::atomic::AtomicU8::new(0);
+        let block_to_send = (sync::Mutex::new(0), sync::Condvar::new());
 
         let (to_server, for_server) = crossbeam_channel::bounded(1);
         let (to_encoding, for_encoding) =