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

Code before:
            {
                // Ensure the lock drops if the channel isn't closed
                let tail = self.shared.tail.lock();
                if tail.closed {
                    return;
                }
            }

            notified.await;
        }
    }

    fn close_channel(&self) {
        let mut tail = self.shared.tail.lock();
        tail.closed = true;

        self.shared.notify_rx(tail);
    }
}

/// Create a new `Receiver` which reads starting from the tail.
fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
    let mut tail = shared.tail.lock();

    assert!(tail.rx_cnt != MAX_RECEIVERS, "max receivers");

    tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");
    if tail.closed {
        // Potentially need to re-open the channel, if a new receiver has been added between calls
        // to poll()
        tail.closed = false;
    }

    let next = tail.pos;

    drop(tail);

    Receiver { shared, next }
}

/// List used in `Shared::notify_rx`. It wraps a guarded linked list
/// and gates the access to it on the `Shared.tail` mutex. It also empties
/// the list on drop.
struct WaitersList<'a, T> {
    list: GuardedLinkedList<Waiter, <Waiter as linked_list::Link>::Target>,
    is_empty: bool,
    shared: &'a Shared<T>,
}

impl<'a, T> Drop for WaitersList<'a, T> {
    fn drop(&mut self) {

Review comment (bug):
This causes the test `tests/sync_broadcast::resubscribe_to_closed_channel` to fail because it also reopens the channel when the sender got dropped before.
Maybe instead of checking for `tail.closed` to reopen the channel we could check for `tail.rx_cnt == 0` before the call to `tail.rx_cnt.checked_add(1)`? In that case the channel would only be reopened if it got closed due to no more receivers being "alive". 
I only ran the sync tests locally real quick and got no failures, so maybe its something to work with?

Code after:
            }

            notified.await;
        }
    }

    fn close_channel(&self) {
        let mut tail = self.shared.tail.lock();
        tail.closed = true;

        self.shared.notify_rx(tail);
    }
}

/// Create a new `Receiver` which reads starting from the tail.
fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
    let mut tail = shared.tail.lock();

    assert!(tail.rx_cnt != MAX_RECEIVERS, "max receivers");

    if tail.rx_cnt == 0 {
        // Potentially need to re-open the channel, if a new receiver has been added between calls
        // to poll(). Note that we use rx_cnt == 0 instead of is_closed since is_closed also
        // applies if the sender has been dropped
        tail.closed = false;
    }

    tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");
    let next = tail.pos;

    drop(tail);

    Receiver { shared, next }
}

/// List used in `Shared::notify_rx`. It wraps a guarded linked list
/// and gates the access to it on the `Shared.tail` mutex. It also empties
/// the list on drop.
struct WaitersList<'a, T> {
    list: GuardedLinkedList<Waiter, <Waiter as linked_list::Link>::Target>,
    is_empty: bool,
    shared: &'a Shared<T>,
}

impl<'a, T> Drop for WaitersList<'a, T> {
    fn drop(&mut self) {
        // If the list is not empty, we unlink all waiters from it.
        // We do not wake the waiters to avoid double panics.
        if !self.is_empty {
            let _lock_guard = self.shared.tail.lock();
            while self.list.pop_back().is_some() {}