ronantakizawa/codereview-bench huggingface dataset:
Repository: tokio-rs/tokio
File: tokio/src/io/uring/write.rs (Rust)

Code before:
}

impl Cancellable for Write {
    fn cancel(self) -> CancelData {
        CancelData::Write(self)
    }
}

impl Op<Write> {
    /// Issue a write that starts at `buf_offset` within `buf` and writes `len` bytes
    /// into `file` at `file_offset`.
    pub(crate) fn write_at(
        fd: OwnedFd,
        buf: OwnedBuf,
        buf_offset: usize,
        len: u32,
        file_offset: u64,
    ) -> io::Result<Self> {
        // Check if `buf_offset` stays in bounds of the allocation
        debug_assert!(buf_offset + len as usize <= buf.len());

        // SAFETY:
        // - `buf_offset` stays in bounds of the allocation.
        // - `buf` is derived from an actual allocation, and the entire memory
        //    range is in bounds of that allocation.
        let ptr = unsafe { buf.as_ptr().add(buf_offset) };

        let sqe = opcode::Write::new(types::Fd(fd.as_raw_fd()), ptr, len)
            .offset(file_offset)
            .build();

        // SAFETY: parameters of the entry, such as `fd` and `buf`, are valid
        // until this operation completes.
        let op = unsafe { Op::new(sqe, Write { buf, fd }) };
        Ok(op)
    }
}


Review comment (suggestion):
```suggestion
        let ptr = buf[buf_offset .. buf_offset + len].as_ptr();
```

Code after:
}

impl Cancellable for Write {
    fn cancel(self) -> CancelData {
        CancelData::Write(self)
    }
}

impl Op<Write> {
    /// Issue a write that starts at `buf_offset` within `buf` and writes some bytes
    /// into `file` at `file_offset`.
    pub(crate) fn write_at(
        fd: OwnedFd,
        buf: OwnedBuf,
        buf_offset: usize,
        file_offset: u64,
    ) -> io::Result<Self> {
        // There is a cap on how many bytes we can write in a single uring write operation.
        // ref: https://github.com/axboe/liburing/discussions/497
        let len = u32::try_from(buf.as_ref().len() - buf_offset).unwrap_or(u32::MAX);

        let ptr = buf.as_ref()[buf_offset..buf_offset + len as usize].as_ptr();

        let sqe = opcode::Write::new(types::Fd(fd.as_raw_fd()), ptr, len)
            .offset(file_offset)
            .build();

        // SAFETY: parameters of the entry, such as `fd` and `buf`, are valid
        // until this operation completes.
        let op = unsafe { Op::new(sqe, Write { buf, fd }) };
        Ok(op)
    }
}
Repository: tokio-rs/tokio
File: tokio/src/sync/watch.rs (Rust)

Code before:
    ///
    /// Returns a [`RecvError`](error::RecvError) if and only if the channel has been closed.
    ///
    /// # Examples
    ///
    /// ## Basic usage
    /// ```
    /// use tokio::sync::watch;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, mut rx) = watch::channel("hello");
    ///
    ///     tx.send("goodbye").unwrap();
    ///
    ///     assert!(rx.has_changed().unwrap());
    ///     assert_eq!(*rx.borrow_and_update(), "goodbye");
    ///
    ///     // The value has been marked as seen
    ///     assert!(!rx.has_changed().unwrap());
    /// }
    /// ```
    ///
    /// ## Closed channel example
    /// ```
    /// use tokio::sync::watch;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, rx) = watch::channel("hello");
    ///     tx.send("goodbye").unwrap();
    ///
    ///     drop(tx);
    ///
    ///     // The channel is closed
    ///     assert!(rx.has_changed().is_err());
    /// }
    /// ```
    pub fn has_changed(&self) -> Result<bool, error::RecvError> {
        // Load the version from the state
        let state = self.shared.state.load();
        if state.is_closed() {
            // All senders have dropped.
            return Err(error::RecvError(()));
        }
        let new_version = state.version();

        Ok(self.version != new_version)
    }

    /// Marks the state as changed.

Review comment (suggestion):
```suggestion
    /// ## Closed channel example

```

Code after:
    /// # Errors
    ///
    /// Returns a [`RecvError`](error::RecvError) if and only if the channel has been closed.
    ///
    /// # Examples
    ///
    /// ## Basic usage
    ///
    /// ```
    /// use tokio::sync::watch;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, mut rx) = watch::channel("hello");
    ///
    ///     tx.send("goodbye").unwrap();
    ///
    ///     assert!(rx.has_changed().unwrap());
    ///     assert_eq!(*rx.borrow_and_update(), "goodbye");
    ///
    ///     // The value has been marked as seen
    ///     assert!(!rx.has_changed().unwrap());
    /// }
    /// ```
    ///
    /// ## Closed channel example
    ///
    /// ```
    /// use tokio::sync::watch;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, rx) = watch::channel("hello");
    ///     tx.send("goodbye").unwrap();
    ///
    ///     drop(tx);
    ///
    ///     // The channel is closed
    ///     assert!(rx.has_changed().is_err());
    /// }
    /// ```
    pub fn has_changed(&self) -> Result<bool, error::RecvError> {
        // Load the version from the state
        let state = self.shared.state.load();
        if state.is_closed() {
            // All senders have dropped.
            return Err(error::RecvError(()));
        }
        let new_version = state.version();

        Ok(self.version != new_version)
Repository: tokio-rs/tokio
File: tokio/tests/sync_watch.rs (Rust)

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

    drop(rx);

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

#[tokio::test]
async fn changed_succeeds_on_closed_channel_with_unseen_value() {
    let (tx, mut rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.changed()
        .await
        .expect("`changed` call does not return an error if the last value is not seen.");
}

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

    rx.changed()
        .await
        .expect_err("`has_changed` iff channel is closed.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_unseen_value() {
    let (tx, rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` iff channel is closed.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_seen_value() {

Review comment (suggestion):
```suggestion
        .expect("should not return error as long as the current value is not seen");
```

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

    drop(rx);

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

#[tokio::test]
async fn changed_succeeds_on_closed_channel_with_unseen_value() {
    let (tx, mut rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.changed()
        .await
        .expect("should not return error as long as the current value is not seen");
}

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

    rx.changed()
        .await
        .expect_err("should return error if the tx is closed and the current value is seen");
}

#[test]
fn has_changed_errors_on_closed_channel_with_unseen_value() {
    let (tx, rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` returns an error if and only if channel is closed. Even if the current value is not seen.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_seen_value() {
Repository: tokio-rs/tokio
File: tokio/tests/sync_watch.rs (Rust)

Code before:
    drop(tx);

    rx.changed()
        .await
        .expect("`changed` call does not return an error if the last value is not seen.");
}

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

    rx.changed()
        .await
        .expect_err("`has_changed` iff channel is closed.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_unseen_value() {
    let (tx, rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` iff channel is closed.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_seen_value() {
    let (tx, rx) = watch::channel("A");
    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` call returns an error if the last value is seen.");
}

#[tokio::test]
async fn wait_for_errors_on_closed_channel_true_predicate() {
    let (tx, mut rx) = watch::channel("A");
    tx.send("B").unwrap();
    drop(tx);

    rx.wait_for(|_| true).await.expect(
        "`wait_for` call does not return error even if channel is closed when predicate is true for last value.",
    );
}


Review comment (suggestion):
```suggestion
        .expect_err("`has_changed` should return error as long as the channel is closed.");
```

Code after:
    drop(tx);

    rx.changed()
        .await
        .expect("should not return error as long as the current value is not seen");
}

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

    rx.changed()
        .await
        .expect_err("should return error if the tx is closed and the current value is seen");
}

#[test]
fn has_changed_errors_on_closed_channel_with_unseen_value() {
    let (tx, rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` returns an error if and only if channel is closed. Even if the current value is not seen.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_seen_value() {
    let (tx, rx) = watch::channel("A");
    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` returns an error if and only if channel is closed.");
}

#[tokio::test]
async fn wait_for_errors_on_closed_channel_true_predicate() {
    let (tx, mut rx) = watch::channel("A");
    tx.send("B").unwrap();
    drop(tx);

    rx.wait_for(|_| true).await.expect(
        "`wait_for` call does not return error even if channel is closed when predicate is true for last value.",
    );
}
Repository: tokio-rs/tokio
File: tokio/tests/sync_watch.rs (Rust)

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

    rx.changed()
        .await
        .expect_err("`has_changed` iff channel is closed.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_unseen_value() {
    let (tx, rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` iff channel is closed.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_seen_value() {
    let (tx, rx) = watch::channel("A");
    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` call returns an error if the last value is seen.");
}

#[tokio::test]
async fn wait_for_errors_on_closed_channel_true_predicate() {
    let (tx, mut rx) = watch::channel("A");
    tx.send("B").unwrap();
    drop(tx);

    rx.wait_for(|_| true).await.expect(
        "`wait_for` call does not return error even if channel is closed when predicate is true for last value.",
    );
}


Review comment (suggestion):
```suggestion
        .expect_err("`has_changed` should return an error if the last value is seen.");
```

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

    rx.changed()
        .await
        .expect_err("should return error if the tx is closed and the current value is seen");
}

#[test]
fn has_changed_errors_on_closed_channel_with_unseen_value() {
    let (tx, rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` returns an error if and only if channel is closed. Even if the current value is not seen.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_seen_value() {
    let (tx, rx) = watch::channel("A");
    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` returns an error if and only if channel is closed.");
}

#[tokio::test]
async fn wait_for_errors_on_closed_channel_true_predicate() {
    let (tx, mut rx) = watch::channel("A");
    tx.send("B").unwrap();
    drop(tx);

    rx.wait_for(|_| true).await.expect(
        "`wait_for` call does not return error even if channel is closed when predicate is true for last value.",
    );
}
Repository: tokio-rs/tokio
File: tokio/tests/sync_watch.rs (Rust)

Code before:
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn changed_succeeds_on_closed_channel_with_unseen_value() {
    let (tx, mut rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.changed()
        .await
        .expect("`changed` call does not return an error if the last value is not seen.");
}

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

    rx.changed()
        .await
        .expect_err("`has_changed` iff channel is closed.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_unseen_value() {
    let (tx, rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` iff channel is closed.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_seen_value() {
    let (tx, rx) = watch::channel("A");
    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` call returns an error if the last value is seen.");
}

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

Review comment (suggestion):
```suggestion
        .expect_err("should return error if the tx is closed and the current value is seen");
```

Code after:
            }
        } => {},
        _ = tokio::task::yield_now() => {},
    }
}

#[tokio::test]
async fn changed_succeeds_on_closed_channel_with_unseen_value() {
    let (tx, mut rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.changed()
        .await
        .expect("should not return error as long as the current value is not seen");
}

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

    rx.changed()
        .await
        .expect_err("should return error if the tx is closed and the current value is seen");
}

#[test]
fn has_changed_errors_on_closed_channel_with_unseen_value() {
    let (tx, rx) = watch::channel("A");
    tx.send("B").unwrap();

    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` returns an error if and only if channel is closed. Even if the current value is not seen.");
}

#[test]
fn has_changed_errors_on_closed_channel_with_seen_value() {
    let (tx, rx) = watch::channel("A");
    drop(tx);

    rx.has_changed()
        .expect_err("`has_changed` returns an error if and only if channel is closed.");
}

#[tokio::test]
async fn wait_for_errors_on_closed_channel_true_predicate() {
    let (tx, mut rx) = watch::channel("A");
Repository: tokio-rs/tokio
File: tokio-util/tests/panic.rs (Rust)

Code before:
    let panic_location_file = test_panic(|| {
        let rt = basic();
        rt.block_on(async {
            let mut queue = task::spawn(DelayQueue::<u32>::with_capacity(3));

            queue.reserve((1 << 30) as usize);
        });
    });

    // The panic location should be in this file
    assert_eq!(&panic_location_file.unwrap(), file!());

    Ok(())
}

#[test]
fn future_ext_to_panic_caller() -> Result<(), Box<dyn Error>> {
    use tokio::{sync::oneshot, time::Duration};
    use tokio_util::future::FutureExt;

    let panic_location_file = test_panic(|| {
        let rt = basic();
        rt.block_on(async {
            let (_tx, rx) = oneshot::channel::<()>();
            let res = rx.timeout(Duration::from_millis(10)).await;
            let _ = res.unwrap();
        });
    });

    // The panic location should be in this file
    assert_eq!(&panic_location_file.unwrap(), file!());

    Ok(())
}

fn basic() -> Runtime {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
}


Review comment (suggestion):
I believe the `#[track_caller]` is still useful for the no-runtime error you get when using `timeout` when not inside of `rt.block_on`.

Code after:
    let panic_location_file = test_panic(|| {
        let rt = basic();
        rt.block_on(async {
            let mut queue = task::spawn(DelayQueue::<u32>::with_capacity(3));

            queue.reserve((1 << 30) as usize);
        });
    });

    // The panic location should be in this file
    assert_eq!(&panic_location_file.unwrap(), file!());

    Ok(())
}

#[test]
fn future_ext_to_panic_caller() -> Result<(), Box<dyn Error>> {
    use tokio::{sync::oneshot, time::Duration};
    use tokio_util::future::FutureExt;

    let panic_location_file = test_panic(|| {
        let (_tx, rx) = oneshot::channel::<()>();
        // this panics because there is no runtime available
        let _res = rx.timeout(Duration::from_millis(10));
    });

    // The panic location should be in this file
    assert_eq!(&panic_location_file.unwrap(), file!());

    Ok(())
}

fn basic() -> Runtime {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
}
Repository: tokio-rs/tokio
File: tokio-util/tests/task_join_queue.rs (Rust)

Code before:
    assert_eq!(joined, spawned);
}

#[tokio::test]
async fn test_join_queue_try_join_next() {
    let mut queue = JoinQueue::new();
    let (tx1, rx1) = oneshot::channel::<()>();
    queue.spawn(async {
        let _ = rx1.await;
    });
    let (tx2, rx2) = oneshot::channel::<()>();
    queue.spawn(async {
        let _ = rx2.await;
    });
    let (tx3, rx3) = oneshot::channel::<()>();
    queue.spawn(async {
        let _ = rx3.await;
    });

    assert_eq!(queue.len(), 3);
    assert!(queue.try_join_next().is_none());
    assert_eq!(queue.len(), 3);
    assert!(queue.try_join_next().is_none());
    assert_eq!(queue.len(), 3);
    assert!(queue.try_join_next().is_none());
    assert_eq!(queue.len(), 3);

    tx1.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 3);
    assert!(queue.try_join_next().is_some());
    assert_eq!(queue.len(), 2);
    assert!(queue.try_join_next().is_none());
    assert_eq!(queue.len(), 2);
    assert!(queue.try_join_next().is_none());
    assert_eq!(queue.len(), 2);

    tx3.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 2);
    assert!(queue.try_join_next().is_none());
    assert_eq!(queue.len(), 2);
    assert!(queue.try_join_next().is_none());
    assert_eq!(queue.len(), 2);

    tx2.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 2);

Review comment (performance):
Emm, I guess we repeated it three times because of the length of `JoinQueue` is 3. I think only one time is enough, we don't want to introduce too much complexity into the test set.

Code after:
    assert_eq!(joined, spawned);
}

#[tokio::test]
async fn test_join_queue_try_join_next() {
    let mut queue = JoinQueue::new();
    let (tx1, rx1) = oneshot::channel::<()>();
    queue.spawn(async {
        let _ = rx1.await;
    });
    let (tx2, rx2) = oneshot::channel::<()>();
    queue.spawn(async {
        let _ = rx2.await;
    });
    let (tx3, rx3) = oneshot::channel::<()>();
    queue.spawn(async {
        let _ = rx3.await;
    });

    // This function also checks that calling `queue.try_join_next()` repeatedly when
    // no task is ready is idempotent, i.e. that it does not change the queue state.
    fn check_try_join_next_is_noop(queue: &mut JoinQueue<()>) {
        let len = queue.len();
        for _ in 0..5 {
            assert!(queue.try_join_next().is_none());
            assert_eq!(queue.len(), len);
        }
    }

    assert_eq!(queue.len(), 3);
    check_try_join_next_is_noop(&mut queue);

    tx1.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 3);
    assert!(queue.try_join_next().is_some());
    assert_eq!(queue.len(), 2);
    check_try_join_next_is_noop(&mut queue);

    tx3.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 2);
    check_try_join_next_is_noop(&mut queue);

    tx2.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 2);
    assert!(queue.try_join_next().is_some());
Repository: tokio-rs/tokio
File: tokio-util/tests/task_join_queue.rs (Rust)

Code before:
    check_try_join_next_is_noop(&mut queue);

    tx3.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 2);
    check_try_join_next_is_noop(&mut queue);

    tx2.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 2);
    assert!(queue.try_join_next().is_some());
    assert_eq!(queue.len(), 1);
    assert!(queue.try_join_next().is_some());
    assert!(queue.is_empty());
    check_try_join_next_is_noop(&mut queue);
}

#[tokio::test]
async fn test_join_queue_try_join_next_disabled_coop() {
    // This number is large enough to trigger coop. Without using `tokio::task::coop::unconstrained`
    // inside `try_join_next` this test fails on `assert!(coop_count == 0)`.
    const TASK_NUM: u32 = 1000;

    static SEM: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(0);

    let mut queue = JoinQueue::new();

    for _ in 0..TASK_NUM {
        queue.spawn(async {
            SEM.add_permits(1);
        });
    }

    let _ = SEM.acquire_many(TASK_NUM).await.unwrap();

    let mut count = 0;
    let mut coop_count = 0;
    while !queue.is_empty() {
        match queue.try_join_next() {
            Some(Ok(())) => {}
            Some(Err(err)) => panic!("failed: {err}"),
            None => {
                coop_count += 1;
                tokio::task::yield_now().await;
                continue;
            }
        }
        count += 1;
    }

Review comment (suggestion):
Will `Arc` work? I'd like to avoid `static` in tests because a `static` var is a global state.

Code after:
    assert!(queue.try_join_next().is_some());
    assert_eq!(queue.len(), 2);
    check_try_join_next_is_noop(&mut queue);

    tx3.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 2);
    check_try_join_next_is_noop(&mut queue);

    tx2.send(()).unwrap();
    tokio::task::yield_now().await;

    assert_eq!(queue.len(), 2);
    assert!(queue.try_join_next().is_some());
    assert_eq!(queue.len(), 1);
    assert!(queue.try_join_next().is_some());
    assert!(queue.is_empty());
    check_try_join_next_is_noop(&mut queue);
}

#[tokio::test]
async fn test_join_queue_try_join_next_disabled_coop() {
    // This number is large enough to trigger coop. Without using `tokio::task::coop::unconstrained`
    // inside `try_join_next` this test fails on `assert!(coop_count == 0)`.
    const TASK_NUM: u32 = 1000;

    let sem: std::sync::Arc<tokio::sync::Semaphore> =
        std::sync::Arc::new(tokio::sync::Semaphore::new(0));

    let mut queue = JoinQueue::new();

    for _ in 0..TASK_NUM {
        let sem = sem.clone();
        queue.spawn(async move {
            sem.add_permits(1);
        });
    }

    let _ = sem.acquire_many(TASK_NUM).await.unwrap();

    let mut count = 0;
    let mut coop_count = 0;
    while !queue.is_empty() {
        match queue.try_join_next() {
            Some(Ok(())) => count += 1,
            Some(Err(err)) => panic!("failed: {err}"),
            None => {
                coop_count += 1;
                tokio::task::yield_now().await;
            }