From 4b174ce2c95fe1d1a217917db93fcc935e17e0da Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Mon, 31 Mar 2025 09:42:43 -0700 Subject: [PATCH] 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. --- tokio/src/sync/broadcast.rs | 55 ++++++++++++++++++------------------- 1 file changed, 27 insertions(+), 28 deletions(-) diff --git a/tokio/src/sync/broadcast.rs b/tokio/src/sync/broadcast.rs index ba0a44fb8..10dfd0efb 100644 --- a/tokio/src/sync/broadcast.rs +++ b/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 { /// slots in the channel. - buffer: Box<[RwLock>]>, + buffer: Box<[Mutex>]>, /// Mask a position -> index. mask: usize, @@ -347,7 +347,7 @@ struct Slot { /// /// 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>, + val: Option, } /// An entry in the wait queue. @@ -385,7 +385,7 @@ generate_addr_of_methods! { } struct RecvGuard<'a, T> { - slot: RwLockReadGuard<'a, Slot>, + slot: MutexGuard<'a, Slot>, } /// Receive a value future. @@ -394,11 +394,15 @@ struct Recv<'a, T> { receiver: &'a mut Receiver, /// Entry in the waiter `LinkedList`. - waiter: UnsafeCell, + 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); + +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(capacity: usize) -> (Sender, Receiver) { (tx, rx) } -unsafe impl Send for Sender {} -unsafe impl Sync for Sender {} - -unsafe impl Send for Receiver {} -unsafe impl Sync for Receiver {} - impl Sender { /// Creates the sending-half of the [`broadcast`] channel. /// @@ -510,10 +508,10 @@ impl Sender { 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 Sender { 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 Sender { 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 Sender { 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 Sender { 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 Receiver { 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 Receiver { 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) -> 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>(); 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,7 +1486,7 @@ impl<'a, T> RecvGuard<'a, T> { where T: Clone, { - self.slot.val.with(|ptr| unsafe { (*ptr).clone() }) + self.slot.val.clone() } } @@ -1494,8 +1494,7 @@ 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; } } }