Repository: tokio-rs/tokio
File: tokio/src/sync/broadcast.rs (Rust)

Code before:
}

impl<T> Shared<T> {
    fn notify_rx<'a, 'b: 'a>(&'b self, mut tail: MutexGuard<'a, Tail>) {
        // It is critical for `GuardedLinkedList` safety that the guard node is
        // pinned in memory and is not dropped until the guarded list is dropped.
        let guard = Waiter::new();
        pin!(guard);

        // We move all waiters to a secondary list. It uses a `GuardedLinkedList`
        // underneath to allow every waiter to safely remove itself from it.
        //
        // * This list will be still guarded by the `waiters` lock.
        //   `NotifyWaitersList` wrapper makes sure we hold the lock to modify it.
        // * This wrapper will empty the list on drop. It is critical for safety
        //   that we will not leave any list entry with a pointer to the local
        //   guard node after this function returns / panics.
        let mut list = WaitersList::new(std::mem::take(&mut tail.waiters), guard.as_ref(), self);

        let mut wakers = WakeList::new();
        'outer: loop {
            while wakers.can_push() {
                match list.pop_back_locked(&mut tail) {
                    Some(mut waiter) => {
                        // Safety: `tail` lock is still held.
                        let waiter = unsafe { waiter.as_mut() };

                        if let Some(waker) = waiter.waker.take() {
                            wakers.push(waker);
                        }

                        // `Release` is needed to synchronize with `Recv::drop`.
                        // It is critical to set this variable **after** waker
                        // is extracted, otherwise we may data race with `Recv::drop`.
                        assert!(waiter.queued.swap(false, Release));
                    }
                    None => {
                        break 'outer;
                    }
                }
            }

            // Release the lock before waking.
            drop(tail);

            // Before we acquire the lock again all sorts of things can happen:
            // some waiters may remove themselves from the list and new waiters
            // may be added. This is fine since at worst we will unnecessarily
            // wake up waiters which will then queue themselves again.

            wakers.wake_all();

Review comment (suggestion):
You can no longer take a mutable reference to the entire waiter. You don't have mutable access to `queued` anymore.

Code after:

impl<T> Shared<T> {
    fn notify_rx<'a, 'b: 'a>(&'b self, mut tail: MutexGuard<'a, Tail>) {
        // It is critical for `GuardedLinkedList` safety that the guard node is
        // pinned in memory and is not dropped until the guarded list is dropped.
        let guard = Waiter::new();
        pin!(guard);

        // We move all waiters to a secondary list. It uses a `GuardedLinkedList`
        // underneath to allow every waiter to safely remove itself from it.
        //
        // * This list will be still guarded by the `waiters` lock.
        //   `NotifyWaitersList` wrapper makes sure we hold the lock to modify it.
        // * This wrapper will empty the list on drop. It is critical for safety
        //   that we will not leave any list entry with a pointer to the local
        //   guard node after this function returns / panics.
        let mut list = WaitersList::new(std::mem::take(&mut tail.waiters), guard.as_ref(), self);

        let mut wakers = WakeList::new();
        'outer: loop {
            while wakers.can_push() {
                match list.pop_back_locked(&mut tail) {
                    Some(waiter) => {
                        unsafe {
                            // Safety: accessing `waker` is safe because
                            // the tail lock is held.
                            if let Some(waker) = (*waiter.as_ptr()).waker.take() {
                                wakers.push(waker);
                            }

                            // Safety: `queued` is atomic.
                            let queued = &(*waiter.as_ptr()).queued;
                            // `Relaxed` suffices because the tail lock is held.
                            assert!(queued.load(Relaxed));
                            // `Release` is needed to synchronize with `Recv::drop`.
                            // It is critical to set this variable **after** waker
                            // is extracted, otherwise we may data race with `Recv::drop`.
                            queued.store(false, Release);
                        }
                    }
                    None => {
                        break 'outer;
                    }
                }
            }

            // Release the lock before waking.
            drop(tail);

            // Before we acquire the lock again all sorts of things can happen:
            // some waiters may remove themselves from the list and new waiters