sync: fix cloning value when receiving from broadcast channel

The broadcast channel does not require values to implement `Sync` yet it calls
the `.clone()` method without synchronizing. This is unsound logic. This patch
adds per-value synchronization on receive to handle this case. It is unlikely
any usage of the broadcast channel is currently at risk of the unsoundeness
issue as it requires accessing a `!Sync` type during `.clone()`, which would be
very unusual when using the broadcast channel.

Author: carllerche

Diff for: tokio/src/sync/broadcast.rs

@@ -118,7 +118,7 @@
 
 use crate::loom::cell::UnsafeCell;
 use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
-use crate::loom::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard};
+use crate::loom::sync::{Arc, Mutex, MutexGuard};
 use crate::util::linked_list::{self, GuardedLinkedList, LinkedList};
 use crate::util::WakeList;
 
@@ -303,7 +303,7 @@ use self::error::{RecvError, SendError, TryRecvError};
 /// Data shared between senders and receivers.
 struct Shared<T> {
     /// slots in the channel.
-    buffer: Box<[RwLock<Slot<T>>]>,
+    buffer: Box<[Mutex<Slot<T>>]>,
 
     /// Mask a position -> index.
     mask: usize,
@@ -347,7 +347,7 @@ struct Slot<T> {
     ///
     /// The value is set by `send` when the write lock is held. When a reader
     /// drops, `rem` is decremented. When it hits zero, the value is dropped.
-    val: UnsafeCell<Option<T>>,
+    val: Option<T>,
 }
 
 /// An entry in the wait queue.
@@ -385,7 +385,7 @@ generate_addr_of_methods! {
 }
 
 struct RecvGuard<'a, T> {
-    slot: RwLockReadGuard<'a, Slot<T>>,
+    slot: MutexGuard<'a, Slot<T>>,
 }
 
 /// Receive a value future.
@@ -394,11 +394,15 @@ struct Recv<'a, T> {
     receiver: &'a mut Receiver<T>,
 
     /// Entry in the waiter `LinkedList`.
-    waiter: UnsafeCell<Waiter>,
+    waiter: WaiterCell,
 }
 
-unsafe impl<'a, T: Send> Send for Recv<'a, T> {}
-unsafe impl<'a, T: Send> Sync for Recv<'a, T> {}
+// The wrapper around `UnsafeCell` isolates the unsafe impl `Send` and `Sync`
+// from `Recv`.
+struct WaiterCell(UnsafeCell<Waiter>);
+
+unsafe impl Send for WaiterCell {}
+unsafe impl Sync for WaiterCell {}
 
 /// Max number of receivers. Reserve space to lock.
 const MAX_RECEIVERS: usize = usize::MAX >> 2;
@@ -466,12 +470,6 @@ pub fn channel<T: Clone>(capacity: usize) -> (Sender<T>, Receiver<T>) {
     (tx, rx)
 }
 
-unsafe impl<T: Send> Send for Sender<T> {}
-unsafe impl<T: Send> Sync for Sender<T> {}
-
-unsafe impl<T: Send> Send for Receiver<T> {}
-unsafe impl<T: Send> Sync for Receiver<T> {}
-
 impl<T> Sender<T> {
     /// Creates the sending-half of the [`broadcast`] channel.
     ///
@@ -510,10 +508,10 @@ impl<T> Sender<T> {
         let mut buffer = Vec::with_capacity(capacity);
 
         for i in 0..capacity {
-            buffer.push(RwLock::new(Slot {
+            buffer.push(Mutex::new(Slot {
                 rem: AtomicUsize::new(0),
                 pos: (i as u64).wrapping_sub(capacity as u64),
-                val: UnsafeCell::new(None),
+                val: None,
             }));
         }
 
@@ -599,7 +597,7 @@ impl<T> Sender<T> {
         tail.pos = tail.pos.wrapping_add(1);
 
         // Get the slot
-        let mut slot = self.shared.buffer[idx].write().unwrap();
+        let mut slot = self.shared.buffer[idx].lock();
 
         // Track the position
         slot.pos = pos;
@@ -608,7 +606,7 @@ impl<T> Sender<T> {
         slot.rem.with_mut(|v| *v = rem);
 
         // Write the value
-        slot.val = UnsafeCell::new(Some(value));
+        slot.val = Some(value);
 
         // Release the slot lock before notifying the receivers.
         drop(slot);
@@ -695,7 +693,7 @@ impl<T> Sender<T> {
         while low < high {
             let mid = low + (high - low) / 2;
             let idx = base_idx.wrapping_add(mid) & self.shared.mask;
-            if self.shared.buffer[idx].read().unwrap().rem.load(SeqCst) == 0 {
+            if self.shared.buffer[idx].lock().rem.load(SeqCst) == 0 {
                 low = mid + 1;
             } else {
                 high = mid;
@@ -737,7 +735,7 @@ impl<T> Sender<T> {
         let tail = self.shared.tail.lock();
 
         let idx = (tail.pos.wrapping_sub(1) & self.shared.mask as u64) as usize;
-        self.shared.buffer[idx].read().unwrap().rem.load(SeqCst) == 0
+        self.shared.buffer[idx].lock().rem.load(SeqCst) == 0
     }
 
     /// Returns the number of active receivers.
@@ -1057,7 +1055,7 @@ impl<T> Receiver<T> {
         let idx = (self.next & self.shared.mask as u64) as usize;
 
         // The slot holding the next value to read
-        let mut slot = self.shared.buffer[idx].read().unwrap();
+        let mut slot = self.shared.buffer[idx].lock();
 
         if slot.pos != self.next {
             // Release the `slot` lock before attempting to acquire the `tail`
@@ -1074,7 +1072,7 @@ impl<T> Receiver<T> {
             let mut tail = self.shared.tail.lock();
 
             // Acquire slot lock again
-            slot = self.shared.buffer[idx].read().unwrap();
+            slot = self.shared.buffer[idx].lock();
 
             // Make sure the position did not change. This could happen in the
             // unlikely event that the buffer is wrapped between dropping the
@@ -1367,12 +1365,12 @@ impl<'a, T> Recv<'a, T> {
     fn new(receiver: &'a mut Receiver<T>) -> Recv<'a, T> {
         Recv {
             receiver,
-            waiter: UnsafeCell::new(Waiter {
+            waiter: WaiterCell(UnsafeCell::new(Waiter {
                 queued: AtomicBool::new(false),
                 waker: None,
                 pointers: linked_list::Pointers::new(),
                 _p: PhantomPinned,
-            }),
+            })),
         }
     }
 
@@ -1384,7 +1382,7 @@ impl<'a, T> Recv<'a, T> {
             is_unpin::<&mut Receiver<T>>();
 
             let me = self.get_unchecked_mut();
-            (me.receiver, &me.waiter)
+            (me.receiver, &me.waiter.0)
         }
     }
 }
@@ -1418,6 +1416,7 @@ impl<'a, T> Drop for Recv<'a, T> {
         // `Shared::notify_rx` before we drop the object.
         let queued = self
             .waiter
+            .0
             .with(|ptr| unsafe { (*ptr).queued.load(Acquire) });
 
         // If the waiter is queued, we need to unlink it from the waiters list.
@@ -1432,6 +1431,7 @@ impl<'a, T> Drop for Recv<'a, T> {
             // `Relaxed` order suffices because we hold the tail lock.
             let queued = self
                 .waiter
+                .0
                 .with_mut(|ptr| unsafe { (*ptr).queued.load(Relaxed) });
 
             if queued {
@@ -1440,7 +1440,7 @@ impl<'a, T> Drop for Recv<'a, T> {
                 // safety: tail lock is held and the wait node is verified to be in
                 // the list.
                 unsafe {
-                    self.waiter.with_mut(|ptr| {
+                    self.waiter.0.with_mut(|ptr| {
                         tail.waiters.remove((&mut *ptr).into());
                     });
                 }
@@ -1486,16 +1486,15 @@ impl<'a, T> RecvGuard<'a, T> {
     where
         T: Clone,
     {
-        self.slot.val.with(|ptr| unsafe { (*ptr).clone() })
+        self.slot.val.clone()
     }
 }
 
 impl<'a, T> Drop for RecvGuard<'a, T> {
     fn drop(&mut self) {
         // Decrement the remaining counter
         if 1 == self.slot.rem.fetch_sub(1, SeqCst) {
-            // Safety: Last receiver, drop the value
-            self.slot.val.with_mut(|ptr| unsafe { *ptr = None });
+            self.slot.val = None;
         }
     }
 }