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

Code before:
    ///
    ///     assert!(tx.closed().now_or_never().is_some());
    /// }
    /// ```
    pub async fn closed(&self) {
        std::future::poll_fn(|_| {
            let tail = self.shared.tail.lock();

            if tail.closed || tail.rx_cnt == 0 {
                Poll::Ready(())
            } else {
                Poll::Pending
            }
        })
        .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");

    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):
I might be wrong here butI think this future will only get polled once because you are not waking up the task again.

Consider this example:
```rust
#[tokio::main]
async fn main() {
    let (tx, mut rx1) = tokio::sync::broadcast::channel(16);

    tokio::spawn(async move {
        let r = rx1.recv().await.unwrap();
        println!("Rx1 received: {}", r);
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    });

    tx.send(20).unwrap();

    println!("Waiting for subscribers to drop");
    tx.closed().await;
    println!("Subscribers dropped");
}
```
Here the call to `closed` will block because it will only be polled once right when you call it. But the subscriber is not dropped at that point, so the `poll_fn` will return `Pending`. But because the future is never be polled again, this will not resolve after that.

Code after:
    ///
    ///     let _ = tx.send(10);
    ///     assert!(tx.closed().now_or_never().is_none());
    ///
    ///     let _ = tokio::spawn(async move {
    ///         assert_eq!(rx2.recv().await.unwrap(), 10);
    ///     }).await;
    ///
    ///     assert!(tx.closed().now_or_never().is_some());
    /// }
    /// ```
    pub async fn closed(&self) {
        loop {
            let notified = self.shared.notify_last_rx_drop.notified();

            {
                // 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");

    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;