Revert "sync: add broadcast::Sender::closed"

[Darksonn]

This reverts commit 5c8cd33.

Reason for revert: Failing tests:

---- tokio/src/sync/broadcast.rs - sync::broadcast::Sender<T>::closed (line 819) stdout ---- Test executable failed (exit status: 101).

stderr:
thread 'main' panicked at tokio/src/sync/broadcast.rs:22:5: assertion failed: tx.closed().now_or_never().is_some() stack backtrace:
   0: rust_begin_unwind
             at /rustc/9fc6b43126469e3858e2fe86cafb4f0fd5068869/library/std/src/panicking.rs:665:5
   1: core::panicking::panic_fmt
             at /rustc/9fc6b43126469e3858e2fe86cafb4f0fd5068869/library/core/src/panicking.rs:76:14
   2: core::panicking::panic
             at /rustc/9fc6b43126469e3858e2fe86cafb4f0fd5068869/library/core/src/panicking.rs:148:5
   3: rust_out::main::{{closure}}
   4: tokio::runtime::park::CachedParkThread::block_on::{{closure}}
   5: tokio::runtime::park::CachedParkThread::block_on
   6: tokio::runtime::context::blocking::BlockingRegionGuard::block_on
   7: tokio::runtime::scheduler::multi_thread::MultiThread::block_on::{{closure}}
   8: tokio::runtime::context::runtime::enter_runtime
   9: tokio::runtime::scheduler::multi_thread::MultiThread::block_on
  10: tokio::runtime::runtime::Runtime::block_on_inner
  11: tokio::runtime::runtime::Runtime::block_on
  12: rust_out::main
  13: core::ops::function::FnOnce::call_once
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.

cc @evanrittenhouse

[evanrittenhouse]

Sorry for the headache. One thing that immediately comes to mind - I wonder if the test is somewhat non-deterministic because we never await the task with rx1. Since futures won't yield to the runtime if they're instantly ready, I wonder if it's possible that we progress through the entire test with rx2's recv() (and the wrapping task) instantly ready, which I guess could sometimes happen before rx1's task is finished spawning/running?

Since tokio::main creates a multi-threaded runtime, I need to dig and see if that will always put rx1's task on a separate thread from the main thread. If so, then the above hypothesis could still occur since we'd have the overhead of spawning a new thread as well, but it feels a bit unlikely.

Something like this may be better for the test, since it'll force each receiver to drop before progressing:

    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, mut rx1) = broadcast::channel::<u32>(16);
    ///     let mut rx2 = tx.subscribe();
    ///
    ///     let _ = tx.send(10);
    ///
    ///     let _ = tokio::spawn(async move {
    ///         assert_eq!(rx1.recv().await.unwrap(), 10);
    ///     }).await;
    ///     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());
    /// }

[Darksonn]

Ah, yeah, it's probably just the test that's bad. How about just not having tasks at all?

#[tokio::main]
async fn main() {
    let (tx, mut rx1) = broadcast::channel::<u32>(16);
    let mut rx2 = tx.subscribe();

    assert_eq!(rx1.recv().await.unwrap(), 10);
    drop(rx1);

    let _ = tx.send(10);
    assert!(tx.closed().now_or_never().is_none());

    assert_eq!(rx2.recv().await.unwrap(), 10);
    drop(rx2);

    assert!(tx.closed().now_or_never().is_some());
}

[evanrittenhouse]

That works too! I originally had the tasks so we can move the rxs in, but drop() works just as well.

What's the process to move forward? Should I just raise a new PR that modifies the test?

[Darksonn]

Opening a PR would be great, thanks!

[evanrittenhouse]

Sure thing, here it is: https://github.com/tokio-rs/tokio/pull/7090/files