ronantakizawa/codereview-bench huggingface dataset:
Repository: tokio-rs/tokio
File: tokio/tests/io_async_fd.rs (Rust)

Code before:
    let (fd_a, fd_b) = socket::socketpair(
        AddressFamily::Unix,
        SockType::Stream,
        None,
        SockFlag::empty(),
    )
    .expect("socketpair");
    let fds = (FileDescriptor { fd: fd_a }, FileDescriptor { fd: fd_b });

    set_nonblocking(fds.0.fd.as_raw_fd());
    set_nonblocking(fds.1.fd.as_raw_fd());

    fds
}

fn drain(mut fd: &FileDescriptor, mut amt: usize) {
    let mut buf = [0u8; 512];
    while amt > 0 {
        match fd.read(&mut buf[..]) {
            Err(e) if e.kind() == ErrorKind::WouldBlock => {}
            Ok(0) => panic!("unexpected EOF"),
            Err(e) => panic!("unexpected error: {:?}", e),
            Ok(x) => {
                amt -= x;
                continue;
            }
        }
    }
}

#[tokio::test]
async fn initially_writable() {
    let (a, b) = socketpair();

    let afd_a = AsyncFd::new(a).unwrap();
    let afd_b = AsyncFd::new(b).unwrap();

    afd_a.writable().await.unwrap().clear_ready();
    afd_b.writable().await.unwrap().clear_ready();

    tokio::select! {
        biased;
        _ = tokio::time::sleep(Duration::from_millis(10)) => {},
        _ = afd_a.readable() => panic!("Unexpected readable state"),
        _ = afd_b.readable() => panic!("Unexpected readable state"),
    }
}

#[tokio::test]
async fn reset_readable() {
    let (a, mut b) = socketpair();

Review comment (suggestion):
Could this just be?
```suggestion
            Ok(x) => amt -= x,
```

Code after:
    let (fd_a, fd_b) = socket::socketpair(
        AddressFamily::Unix,
        SockType::Stream,
        None,
        SockFlag::empty(),
    )
    .expect("socketpair");
    let fds = (FileDescriptor { fd: fd_a }, FileDescriptor { fd: fd_b });

    set_nonblocking(fds.0.fd.as_raw_fd());
    set_nonblocking(fds.1.fd.as_raw_fd());

    fds
}

fn drain(mut fd: &FileDescriptor, mut amt: usize) {
    let mut buf = [0u8; 512];
    while amt > 0 {
        match fd.read(&mut buf[..]) {
            Err(e) if e.kind() == ErrorKind::WouldBlock => {}
            Ok(0) => panic!("unexpected EOF"),
            Err(e) => panic!("unexpected error: {:?}", e),
            Ok(x) => amt -= x,
        }
    }
}

#[tokio::test]
async fn initially_writable() {
    let (a, b) = socketpair();

    let afd_a = AsyncFd::new(a).unwrap();
    let afd_b = AsyncFd::new(b).unwrap();

    afd_a.writable().await.unwrap().clear_ready();
    afd_b.writable().await.unwrap().clear_ready();

    tokio::select! {
        biased;
        _ = tokio::time::sleep(Duration::from_millis(10)) => {},
        _ = afd_a.readable() => panic!("Unexpected readable state"),
        _ = afd_b.readable() => panic!("Unexpected readable state"),
    }
}

#[tokio::test]
async fn reset_readable() {
    let (a, mut b) = socketpair();

    let afd_a = AsyncFd::new(a).unwrap();
Repository: tokio-rs/tokio
File: tokio/tests/task_join_set.rs (Rust)

Code before:

    assert!(rt()
        .block_on(set.join_next())
        .unwrap()
        .unwrap_err()
        .is_cancelled());
}

#[tokio::test]
async fn join_all() {
    let mut set: JoinSet<i32> = JoinSet::new();

    for _ in 0..5 {
        set.spawn(async { 1 });
    }
    let res: Vec<i32> = set.join_all().await;

    assert_eq!(res.len(), 5);
    for itm in res.into_iter() {
        assert_eq!(itm, 1)
    }
}

#[tokio::test]
async fn task_panics() {
    let mut set: JoinSet<()> = JoinSet::new();

    let (tx, mut rx) = oneshot::channel();
    assert_eq!(set.len(), 0);

    set.spawn(async move {
        tokio::time::sleep(Duration::from_secs(2)).await;
        tx.send(()).unwrap();
    });
    assert_eq!(set.len(), 1);

    set.spawn(async {
        tokio::time::sleep(Duration::from_secs(1)).await;
        panic!();
    });
    assert_eq!(set.len(), 2);

    let panic = tokio::spawn(set.join_all()).await.unwrap_err();
    assert!(rx.try_recv().is_err());
    assert!(panic.is_panic());
}

#[tokio::test(start_paused = true)]
async fn abort_all() {
    let mut set: JoinSet<()> = JoinSet::new();


Review comment (bug):
This test is failing under the `panic=abort` condition, so you could use `#[cfg(panic = "unwind")]` gate to avoid this. (It seems that [wasm is `panic=abort` by default](https://users.rust-lang.org/t/catch-panic-in-wasm/57569/2).)

```suggestion
#[cfg(panic = "unwind")]
#[tokio::test]
async fn task_panics() {
```


Code after:

    assert!(rt()
        .block_on(set.join_next())
        .unwrap()
        .unwrap_err()
        .is_cancelled());
}

#[tokio::test]
async fn join_all() {
    let mut set: JoinSet<i32> = JoinSet::new();

    for _ in 0..5 {
        set.spawn(async { 1 });
    }
    let res: Vec<i32> = set.join_all().await;

    assert_eq!(res.len(), 5);
    for itm in res.into_iter() {
        assert_eq!(itm, 1)
    }
}

#[cfg(panic = "unwind")]
#[tokio::test(start_paused = true)]
async fn task_panics() {
    let mut set: JoinSet<()> = JoinSet::new();

    let (tx, mut rx) = oneshot::channel();
    assert_eq!(set.len(), 0);

    set.spawn(async move {
        tokio::time::sleep(Duration::from_secs(2)).await;
        tx.send(()).unwrap();
    });
    assert_eq!(set.len(), 1);

    set.spawn(async {
        tokio::time::sleep(Duration::from_secs(1)).await;
        panic!();
    });
    assert_eq!(set.len(), 2);

    let panic = tokio::spawn(set.join_all()).await.unwrap_err();
    assert!(rx.try_recv().is_err());
    assert!(panic.is_panic());
}

#[tokio::test(start_paused = true)]
async fn abort_all() {
    let mut set: JoinSet<()> = JoinSet::new();
Repository: tokio-rs/tokio
File: tokio/tests/task_join_set.rs (Rust)

Code before:
    }

    assert!(rt()
        .block_on(set.join_next())
        .unwrap()
        .unwrap_err()
        .is_cancelled());
}

#[tokio::test]
async fn join_all() {
    let mut set: JoinSet<i32> = JoinSet::new();

    for _ in 0..5 {
        set.spawn(async { 1 });
    }
    let res: Vec<i32> = set.join_all().await;

    assert_eq!(res.len(), 5);
    for itm in res.into_iter() {
        assert_eq!(itm, 1)
    }
}

#[cfg(panic = "unwind")]
#[tokio::test]
async fn task_panics() {
    let mut set: JoinSet<()> = JoinSet::new();

    let (tx, mut rx) = oneshot::channel();
    assert_eq!(set.len(), 0);

    set.spawn(async move {
        tokio::time::sleep(Duration::from_secs(2)).await;
        tx.send(()).unwrap();
    });
    assert_eq!(set.len(), 1);

    set.spawn(async {
        tokio::time::sleep(Duration::from_secs(1)).await;
        panic!();
    });
    assert_eq!(set.len(), 2);

    let panic = tokio::spawn(set.join_all()).await.unwrap_err();
    assert!(rx.try_recv().is_err());
    assert!(panic.is_panic());
}

#[tokio::test(start_paused = true)]
async fn abort_all() {

Review comment (suggestion):
This will make the test run much much faster by using [simulated time](https://docs.rs/tokio/latest/tokio/time/fn.pause.html).
```suggestion
#[tokio::test(start_paused = true)]
```

Code after:
    }

    assert!(rt()
        .block_on(set.join_next())
        .unwrap()
        .unwrap_err()
        .is_cancelled());
}

#[tokio::test]
async fn join_all() {
    let mut set: JoinSet<i32> = JoinSet::new();

    for _ in 0..5 {
        set.spawn(async { 1 });
    }
    let res: Vec<i32> = set.join_all().await;

    assert_eq!(res.len(), 5);
    for itm in res.into_iter() {
        assert_eq!(itm, 1)
    }
}

#[cfg(panic = "unwind")]
#[tokio::test(start_paused = true)]
async fn task_panics() {
    let mut set: JoinSet<()> = JoinSet::new();

    let (tx, mut rx) = oneshot::channel();
    assert_eq!(set.len(), 0);

    set.spawn(async move {
        tokio::time::sleep(Duration::from_secs(2)).await;
        tx.send(()).unwrap();
    });
    assert_eq!(set.len(), 1);

    set.spawn(async {
        tokio::time::sleep(Duration::from_secs(1)).await;
        panic!();
    });
    assert_eq!(set.len(), 2);

    let panic = tokio::spawn(set.join_all()).await.unwrap_err();
    assert!(rx.try_recv().is_err());
    assert!(panic.is_panic());
}

#[tokio::test(start_paused = true)]
async fn abort_all() {
Repository: tokio-rs/tokio
File: tokio-util/src/task/mod.rs (Rust)

Code before:
//! Extra utilities for spawning tasks

#[cfg(tokio_unstable)]
mod join_map;
mod spawn_pinned;
pub use spawn_pinned::LocalPoolHandle;

#[cfg(tokio_unstable)]
#[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "rt"))))]
pub use join_map::{JoinMap, JoinMapKeys};

pub mod task_tracker;
pub use task_tracker::TaskTracker;

pub mod spawn_abortable;


Review comment (suggestion):
I'm not sure this module name makes much sense considering the type is called `AbortOnDropHandle`.

Code after:
//! Extra utilities for spawning tasks

#[cfg(tokio_unstable)]
mod join_map;
mod spawn_pinned;
pub use spawn_pinned::LocalPoolHandle;

#[cfg(tokio_unstable)]
#[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "rt"))))]
pub use join_map::{JoinMap, JoinMapKeys};

pub mod task_tracker;
pub use task_tracker::TaskTracker;

mod abort_on_drop;
pub use abort_on_drop::AbortOnDropHandle;
Repository: tokio-rs/tokio
File: tokio-util/src/task/abort_on_drop.rs (Rust)

Code before:
    pin::Pin,
    task::{Context, Poll},
};

/// A wrapper around a [`tokio::task::JoinHandle`],
/// which [aborts] the task when it is dropped.
///
/// [aborts]: tokio::task::JoinHandle::abort
#[must_use = "Dropping the handle aborts the task immediately"]
#[derive(Debug)]
pub struct AbortOnDropHandle<T>(JoinHandle<T>);

impl<T> Drop for AbortOnDropHandle<T> {
    fn drop(&mut self) {
        self.0.abort()
    }
}

impl<T> AbortOnDropHandle<T> {
    /// Create an [`AbortOnDropHandle`] from a [`JoinHandle`].
    pub fn new(handle: JoinHandle<T>) -> Self {
        Self(handle)
    }

    /// Abort the task associated with this handle,
    /// equivalent of [`tokio::task::JoinHandle::abort`].
    pub fn abort(&self) {
        self.0.abort()
    }

    /// Checks if the task associated with this handle is finished,
    /// equivalent of [`tokio::task::JoinHandle::is_finished`].
    pub fn is_finished(&self) -> bool {
        self.0.is_finished()
    }

    /// Returns a new [`AbortHandle`] that can be used to remotely abort this task,
    /// equivalent of [`tokio::task::JoinHandle::abort_handle`].
    pub fn abort_handle(&self) -> AbortHandle {
        self.0.abort_handle()
    }
}

impl<T> Future for AbortOnDropHandle<T> {
    type Output = Result<T, JoinError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        Pin::new(&mut self.0).poll(cx)
    }
}


Review comment (suggestion):
I think this looks slightly nicer like this:
```suggestion
    /// equivalent to [`JoinHandle::abort`].
    ///
    /// [`JoinHandle::abort`]: tokio::task::JoinHandle::abort
```
(3 instances of this)

Code after:
    pin::Pin,
    task::{Context, Poll},
};

/// A wrapper around a [`tokio::task::JoinHandle`],
/// which [aborts] the task when it is dropped.
///
/// [aborts]: tokio::task::JoinHandle::abort
#[must_use = "Dropping the handle aborts the task immediately"]
#[derive(Debug)]
pub struct AbortOnDropHandle<T>(JoinHandle<T>);

impl<T> Drop for AbortOnDropHandle<T> {
    fn drop(&mut self) {
        self.0.abort()
    }
}

impl<T> AbortOnDropHandle<T> {
    /// Create an [`AbortOnDropHandle`] from a [`JoinHandle`].
    pub fn new(handle: JoinHandle<T>) -> Self {
        Self(handle)
    }

    /// Abort the task associated with this handle,
    /// equivalent to [`JoinHandle::abort`].
    pub fn abort(&self) {
        self.0.abort()
    }

    /// Checks if the task associated with this handle is finished,
    /// equivalent to [`JoinHandle::is_finished`].
    pub fn is_finished(&self) -> bool {
        self.0.is_finished()
    }

    /// Returns a new [`AbortHandle`] that can be used to remotely abort this task,
    /// equivalent to [`JoinHandle::abort_handle`].
    pub fn abort_handle(&self) -> AbortHandle {
        self.0.abort_handle()
    }
}

impl<T> Future for AbortOnDropHandle<T> {
    type Output = Result<T, JoinError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        Pin::new(&mut self.0).poll(cx)
    }
}
Repository: tokio-rs/tokio
File: tokio-util/src/task/mod.rs (Rust)

Code before:
//! Extra utilities for spawning tasks

#[cfg(tokio_unstable)]
mod join_map;
mod spawn_pinned;
pub use spawn_pinned::LocalPoolHandle;

#[cfg(tokio_unstable)]
#[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "rt"))))]
pub use join_map::{JoinMap, JoinMapKeys};

pub mod task_tracker;
pub use task_tracker::TaskTracker;

pub mod abort_on_drop;


Review comment (suggestion):
No reason to expose a module with one item.
```suggestion
mod abort_on_drop;
pub use abort_on_drop::AbortOnDropHandle;
```

Code after:
//! Extra utilities for spawning tasks

#[cfg(tokio_unstable)]
mod join_map;
mod spawn_pinned;
pub use spawn_pinned::LocalPoolHandle;

#[cfg(tokio_unstable)]
#[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "rt"))))]
pub use join_map::{JoinMap, JoinMapKeys};

pub mod task_tracker;
pub use task_tracker::TaskTracker;

mod abort_on_drop;
pub use abort_on_drop::AbortOnDropHandle;
Repository: tokio-rs/tokio
File: tokio/src/macros/join.rs (Rust)

Code before:
        /// async fn more_async_work() {
        ///     // more here
        /// }
        ///
        /// #[tokio::main]
        /// async fn main() {
        ///     let (first, second) = tokio::join!(
        ///         do_stuff_async(),
        ///         more_async_work());
        ///
        ///     // do something with the values
        /// }
        /// ```
        #[macro_export]
        #[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
        $join
    };
}

#[cfg(doc)]
doc! {macro_rules! join {
    (@ {
        ( $($count:tt)* )
        ( $($total:tt)* )
        $( ( $($skip:tt)* ) $e:expr, )*
    }) => {
        unimplemented!()
    };

    (@ { ( $($s:tt)* ) ( $($n:tt)* ) $($t:tt)* } $e:expr, $($r:tt)* ) => {
        unimplemented!()
    };

    ( $($e:expr),+ $(,)?) => {
        unimplemented!()
    };

    () => { async {}.await }
}}

#[cfg(not(doc))]
doc! {macro_rules! join {
    (@ {
        // One `_` for each branch in the `join!` macro. This is not used once
        // normalization is complete.
        ( $($count:tt)* )

        // The expression `0+1+1+ ... +1` equal to the number of branches.
        ( $($total:tt)* )

        // Normalized join! branches

Review comment (suggestion):
This isn't useful for the reader. They don't pass a count and a total and a skip. The macro just takes a list of futures. A list of futures looks like this:
```rs
($($future:expr),*) => { unimplemented!() }
```

Code after:
        /// async fn more_async_work() {
        ///     // more here
        /// }
        ///
        /// #[tokio::main]
        /// async fn main() {
        ///     let (first, second) = tokio::join!(
        ///         do_stuff_async(),
        ///         more_async_work());
        ///
        ///     // do something with the values
        /// }
        /// ```
        #[macro_export]
        #[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
        $join
    };
}

#[cfg(doc)]
doc! {macro_rules! join {
    ($($future:expr),*) => { unimplemented!() }
}}

#[cfg(not(doc))]
doc! {macro_rules! join {
    (@ {
        // One `_` for each branch in the `join!` macro. This is not used once
        // normalization is complete.
        ( $($count:tt)* )

        // The expression `0+1+1+ ... +1` equal to the number of branches.
        ( $($total:tt)* )

        // Normalized join! branches
        $( ( $($skip:tt)* ) $e:expr, )*

    }) => {{
        use $crate::macros::support::{maybe_done, poll_fn, Future, Pin};
        use $crate::macros::support::Poll::{Ready, Pending};

        // Safety: nothing must be moved out of `futures`. This is to satisfy
        // the requirement of `Pin::new_unchecked` called below.
        //
        // We can't use the `pin!` macro for this because `futures` is a tuple
        // and the standard library provides no way to pin-project to the fields
        // of a tuple.
        let mut futures = ( $( maybe_done($e), )* );

        // This assignment makes sure that the `poll_fn` closure only has a
        // reference to the futures, instead of taking ownership of them. This
Repository: tokio-rs/tokio
File: tokio/tests/sync_watch.rs (Rust)

Code before:
#[tokio::test]
async fn receiver_is_notified_when_last_sender_is_dropped() {
    let (tx1, mut rx) = watch::channel(0);
    let tx2 = tx1.clone();

    let mut t = spawn(rx.changed());
    assert_pending!(t.poll());

    drop(tx1);
    assert!(!t.is_woken());
    drop(tx2);

    assert!(t.is_woken());
}

#[tokio::test]
async fn receiver_changed_is_cooperative() {
    let (tx, mut rx) = watch::channel(());

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                let _ = rx.changed().await;
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_wait_for_is_cooperative() {
    let (tx, mut rx) = watch::channel(0);

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                let _ = rx.wait_for(|val| *val == 1).await;
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn sender_closed_is_cooperative() {

Review comment (bug):
You may need to send another message for this test to actually work? Did you verify that these tests fail without the change?

Code after:
#[tokio::test]
async fn receiver_is_notified_when_last_sender_is_dropped() {
    let (tx1, mut rx) = watch::channel(0);
    let tx2 = tx1.clone();

    let mut t = spawn(rx.changed());
    assert_pending!(t.poll());

    drop(tx1);
    assert!(!t.is_woken());
    drop(tx2);

    assert!(t.is_woken());
}

#[tokio::test]
async fn receiver_changed_is_cooperative() {
    let (tx, mut rx) = watch::channel(());

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                assert!(rx.changed().await.is_err());
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_changed_is_cooperative_ok() {
    let (tx, mut rx) = watch::channel(());

    tokio::select! {
        biased;
        _ = async {
            loop {
                assert!(tx.send(()).is_ok());
                assert!(rx.changed().await.is_ok());
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_wait_for_is_cooperative() {
    let (tx, mut rx) = watch::channel(0);
Repository: tokio-rs/tokio
File: tokio/tests/sync_watch.rs (Rust)

Code before:
    let (tx, mut rx) = watch::channel(());

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                let _ = rx.changed().await;
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_wait_for_is_cooperative() {
    let (tx, mut rx) = watch::channel(0);

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                let _ = rx.wait_for(|val| *val == 1).await;
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn sender_closed_is_cooperative() {
    let (tx, rx) = watch::channel(());

    drop(rx);
    tokio::select! {
        _ = async {
            loop {
                let _ = tx.closed().await;
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}


Review comment (suggestion):
```suggestion
                assert!(rx.wait_for(|val| *val == 1).await.is_err());
```

Code after:
    let (tx, mut rx) = watch::channel(());

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                assert!(rx.changed().await.is_err());
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_changed_is_cooperative_ok() {
    let (tx, mut rx) = watch::channel(());

    tokio::select! {
        biased;
        _ = async {
            loop {
                assert!(tx.send(()).is_ok());
                assert!(rx.changed().await.is_ok());
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_wait_for_is_cooperative() {
    let (tx, mut rx) = watch::channel(0);

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                assert!(rx.wait_for(|val| *val == 1).await.is_err());
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_wait_for_is_cooperative_ok() {
    let (tx, mut rx) = watch::channel(0);
Repository: tokio-rs/tokio
File: tokio/tests/sync_watch.rs (Rust)

Code before:
            loop {
                let _ = rx.changed().await;
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_wait_for_is_cooperative() {
    let (tx, mut rx) = watch::channel(0);

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                let _ = rx.wait_for(|val| *val == 1).await;
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn sender_closed_is_cooperative() {
    let (tx, rx) = watch::channel(());

    drop(rx);
    tokio::select! {
        _ = async {
            loop {
                let _ = tx.closed().await;
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}


Review comment (suggestion):
Please add another version of this test where `wait_for` returns `Ok` instead of an error.

Code after:
            loop {
                assert!(rx.changed().await.is_err());
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_changed_is_cooperative_ok() {
    let (tx, mut rx) = watch::channel(());

    tokio::select! {
        biased;
        _ = async {
            loop {
                assert!(tx.send(()).is_ok());
                assert!(rx.changed().await.is_ok());
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_wait_for_is_cooperative() {
    let (tx, mut rx) = watch::channel(0);

    drop(tx);

    tokio::select! {
        biased;
        _ = async {
            loop {
                assert!(rx.wait_for(|val| *val == 1).await.is_err());
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn receiver_wait_for_is_cooperative_ok() {
    let (tx, mut rx) = watch::channel(0);

    tokio::select! {
        biased;
        _ = async {
            loop {
                assert!(tx.send(1).is_ok());
                assert!(rx.wait_for(|val| *val == 1).await.is_ok());