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

Code before:
            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;
    }
    assert!(coop_count == 0);
    assert_eq!(count, TASK_NUM);
}

#[tokio::test]
async fn test_join_queue_try_join_next_with_id() {
    // Note that this number is large enough to trigger coop as in
    // `test_join_queue_try_join_next_coop` test. Without using
    // `tokio::task::coop::unconstrained` inside `try_join_next_with_id`
    // this test fails on `assert_eq!(count, TASK_NUM)`.
    const TASK_NUM: u32 = 1000;

    let (send, recv) = tokio::sync::watch::channel(());

    let mut queue = JoinQueue::new();
    let mut spawned = Vec::with_capacity(TASK_NUM as usize);

    for _ in 0..TASK_NUM {
        let mut recv = recv.clone();
        let handle = queue.spawn(async move { recv.changed().await.unwrap() });

        spawned.push(handle.id());
    }
    drop(recv);

    assert!(queue.try_join_next_with_id().is_none());

    send.send_replace(());
    send.closed().await;

    let mut count = 0;

Review comment (suggestion):
```suggestion
async fn test_join_queue_try_join_next_with_id_disabled_coop() {
```

Code after:

    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;
            }
        }
    }
    assert_eq!(coop_count, 0);
    assert_eq!(count, TASK_NUM);
}

#[tokio::test]
async fn test_join_queue_try_join_next_with_id_disabled_coop() {
    // Note that this number is large enough to trigger coop as in
    // `test_join_queue_try_join_next_coop` test. Without using
    // `tokio::task::coop::unconstrained` inside `try_join_next_with_id`
    // this test fails on `assert_eq!(count, TASK_NUM)`.
    const TASK_NUM: u32 = 1000;

    let (send, recv) = tokio::sync::watch::channel(());

    let mut queue = JoinQueue::new();
    let mut spawned = Vec::with_capacity(TASK_NUM as usize);

    for _ in 0..TASK_NUM {
        let mut recv = recv.clone();
        let handle = queue.spawn(async move { recv.changed().await.unwrap() });

        spawned.push(handle.id());
    }
    drop(recv);

    assert!(queue.try_join_next_with_id().is_none());

    send.send_replace(());
    send.closed().await;
Repository: tokio-rs/tokio
File: tokio-util/tests/task_join_queue.rs (Rust)

Code before:
        }
    }

    assert_eq!(count, TASK_NUM);
    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;
    });

    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;


Review comment (suggestion):
Could you add a short comment on this function to explain that this is to test the idempotent of `try_join_next`.

Code after:
        }
    }

    assert_eq!(count, TASK_NUM);
    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();
Repository: tokio-rs/tokio
File: tokio-util/src/sync/tests/loom_cancellation_token.rs (Rust)

Code before:
fn drop_token_no_child() {
    loom::model(|| {
        let token = CancellationToken::new();
        let token1 = token.clone();
        let token2 = token.clone();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(token2);
        });

        let th3 = thread::spawn(move || {
            drop(token);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

// Temporarily disabled due to a false positive in loom -
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344#[ignore]
#[test]
fn drop_token_with_children() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let child_token1 = token1.child_token();
        let child_token2 = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(child_token1);
        });

        let th3 = thread::spawn(move || {
            drop(child_token2);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}


Review comment (suggestion):
```suggestion
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
#[ignore]
```

Code after:
fn drop_token_no_child() {
    loom::model(|| {
        let token = CancellationToken::new();
        let token1 = token.clone();
        let token2 = token.clone();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(token2);
        });

        let th3 = thread::spawn(move || {
            drop(token);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

// Temporarily disabled due to a false positive in loom -
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
#[ignore]
#[test]
fn drop_token_with_children() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let child_token1 = token1.child_token();
        let child_token2 = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(child_token1);
        });

        let th3 = thread::spawn(move || {
            drop(child_token2);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}
Repository: tokio-rs/tokio
File: tokio-util/src/sync/tests/loom_cancellation_token.rs (Rust)

Code before:
fn drop_token_with_children() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let child_token1 = token1.child_token();
        let child_token2 = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(child_token1);
        });

        let th3 = thread::spawn(move || {
            drop(child_token2);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

// Temporarily disabled due to a false positive in loom -
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344#[ignore]
#[test]
fn drop_and_cancel_token() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let token2 = token1.clone();
        let child_token = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            token2.cancel();
        });

        let th3 = thread::spawn(move || {
            drop(child_token);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}


Review comment (suggestion):
```suggestion
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
#[ignore]
```

Code after:
#[test]
fn drop_token_with_children() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let child_token1 = token1.child_token();
        let child_token2 = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(child_token1);
        });

        let th3 = thread::spawn(move || {
            drop(child_token2);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

// Temporarily disabled due to a false positive in loom -
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
#[ignore]
#[test]
fn drop_and_cancel_token() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let token2 = token1.clone();
        let child_token = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            token2.cancel();
        });

        let th3 = thread::spawn(move || {
            drop(child_token);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
Repository: tokio-rs/tokio
File: tokio-util/src/sync/tests/loom_cancellation_token.rs (Rust)

Code before:
fn drop_token_no_child() {
    loom::model(|| {
        let token = CancellationToken::new();
        let token1 = token.clone();
        let token2 = token.clone();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(token2);
        });

        let th3 = thread::spawn(move || {
            drop(token);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

// Temporarily disabled due to a false positive in loom -
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344#[ignore]
#[ignore]
#[test]
fn drop_token_with_children() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let child_token1 = token1.child_token();
        let child_token2 = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(child_token1);
        });

        let th3 = thread::spawn(move || {
            drop(child_token2);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

Review comment (suggestion):
```suggestion
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
```

Code after:
fn drop_token_no_child() {
    loom::model(|| {
        let token = CancellationToken::new();
        let token1 = token.clone();
        let token2 = token.clone();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(token2);
        });

        let th3 = thread::spawn(move || {
            drop(token);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

// Temporarily disabled due to a false positive in loom -
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
#[ignore]
#[test]
fn drop_token_with_children() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let child_token1 = token1.child_token();
        let child_token2 = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(child_token1);
        });

        let th3 = thread::spawn(move || {
            drop(child_token2);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}
Repository: tokio-rs/tokio
File: tokio-util/src/sync/tests/loom_cancellation_token.rs (Rust)

Code before:
fn drop_token_with_children() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let child_token1 = token1.child_token();
        let child_token2 = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(child_token1);
        });

        let th3 = thread::spawn(move || {
            drop(child_token2);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

// Temporarily disabled due to a false positive in loom -
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344#[ignore]
#[ignore]
#[test]
fn drop_and_cancel_token() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let token2 = token1.clone();
        let child_token = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            token2.cancel();
        });

        let th3 = thread::spawn(move || {
            drop(child_token);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

Review comment (suggestion):
```suggestion
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
```

Code after:
fn drop_token_with_children() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let child_token1 = token1.child_token();
        let child_token2 = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            drop(child_token1);
        });

        let th3 = thread::spawn(move || {
            drop(child_token2);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}

// Temporarily disabled due to a false positive in loom -
// see https://github.com/tokio-rs/tokio/pull/7644#issuecomment-3328381344
#[ignore]
#[test]
fn drop_and_cancel_token() {
    loom::model(|| {
        let token1 = CancellationToken::new();
        let token2 = token1.clone();
        let child_token = token1.child_token();

        let th1 = thread::spawn(move || {
            drop(token1);
        });

        let th2 = thread::spawn(move || {
            token2.cancel();
        });

        let th3 = thread::spawn(move || {
            drop(child_token);
        });

        assert_ok!(th1.join());
        assert_ok!(th2.join());
        assert_ok!(th3.join());
    });
}
Repository: tokio-rs/tokio
File: tokio/src/io/uring/read.rs (Rust)

Code before:
use crate::runtime::driver::op::{CancelData, Cancellable, Completable, CqeResult, Op};
use io_uring::{opcode, types};
use std::{
    io,
    os::fd::{AsRawFd, OwnedFd},
};

#[derive(Debug)]
pub(crate) struct Read {
    fd: OwnedFd,
}

impl Completable for Read {
    type Output = (u32, OwnedFd);
    fn complete(self, cqe: CqeResult) -> io::Result<Self::Output> {
        let res = cqe.result?;

        Ok((res, self.fd))
    }
}

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

impl Op<Read> {
    /// Submit a request to open a file
    pub(crate) fn read(fd: OwnedFd, buf: &mut Vec<u8>, len: u32, offset: u64) -> io::Result<Self> {
        let buf_mut_ptr = buf.as_mut_ptr();

        let read_op = opcode::Read::new(types::Fd(fd.as_raw_fd()), buf_mut_ptr, len)
            .offset(offset)
            .build();


Review comment (suggestion):
You must transfer ownership of the buffer to the `Read` operation. Passing a `&mut Vec<u8>` is not enough.

Code after:
use crate::runtime::driver::op::{CancelData, Cancellable, Completable, CqeResult, Op};

use io_uring::{opcode, types};
use std::io::{self, Error};
use std::os::fd::{AsRawFd, OwnedFd};

#[derive(Debug)]
pub(crate) struct Read {
    fd: OwnedFd,
    buf: Vec<u8>,
}

impl Completable for Read {
    type Output = (io::Result<u32>, OwnedFd, Vec<u8>);

    fn complete(self, cqe: CqeResult) -> Self::Output {
        let mut buf = self.buf;

        if let Ok(len) = cqe.result {
            let new_len = buf.len() + len as usize;
            // SAFETY: Kernel read len bytes
            unsafe { buf.set_len(new_len) };
        }

        (cqe.result, self.fd, buf)
    }

    fn complete_with_error(self, err: Error) -> Self::Output {
        (Err(err), self.fd, self.buf)
    }
}

impl Cancellable for Read {
    fn cancel(self) -> CancelData {
        CancelData::Read(self)
    }
Repository: tokio-rs/tokio
File: tokio/src/io/uring/read.rs (Rust)

Code before:
    io,
    os::fd::{AsRawFd, OwnedFd},
};

#[derive(Debug)]
pub(crate) struct Read {
    fd: OwnedFd,
}

impl Completable for Read {
    type Output = (u32, OwnedFd);
    fn complete(self, cqe: CqeResult) -> io::Result<Self::Output> {
        let res = cqe.result?;

        Ok((res, self.fd))
    }
}

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

impl Op<Read> {
    /// Submit a request to open a file
    pub(crate) fn read(fd: OwnedFd, buf: &mut Vec<u8>, len: u32, offset: u64) -> io::Result<Self> {
        let buf_mut_ptr = buf.as_mut_ptr();

        let read_op = opcode::Read::new(types::Fd(fd.as_raw_fd()), buf_mut_ptr, len)
            .offset(offset)
            .build();

        // SAFETY: Parameters are valid for the entire duration of the operation
        let op = unsafe { Op::new(read_op, Read { fd }) };

        Ok(op)
    }
}


Review comment (suggestion):
```suggestion
    /// Submit a request to read from a file
```

Code after:
use std::io::{self, Error};
use std::os::fd::{AsRawFd, OwnedFd};

#[derive(Debug)]
pub(crate) struct Read {
    fd: OwnedFd,
    buf: Vec<u8>,
}

impl Completable for Read {
    type Output = (io::Result<u32>, OwnedFd, Vec<u8>);

    fn complete(self, cqe: CqeResult) -> Self::Output {
        let mut buf = self.buf;

        if let Ok(len) = cqe.result {
            let new_len = buf.len() + len as usize;
            // SAFETY: Kernel read len bytes
            unsafe { buf.set_len(new_len) };
        }

        (cqe.result, self.fd, buf)
    }

    fn complete_with_error(self, err: Error) -> Self::Output {
        (Err(err), self.fd, self.buf)
    }
}

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

impl Op<Read> {
    // Submit a request to read a FD at given length and offset into a
    // dynamic buffer with uninitialized memory. The read happens on uninitialized
    // buffer and no overwriting happens.

    // SAFETY: The `len` of the amount to be read and the buffer that is passed
    // should have capacity > len.
    //
    // If `len` read is higher than vector capacity then setting its length by
    // the caller in terms of size_read can be unsound.
    pub(crate) fn read(fd: OwnedFd, mut buf: Vec<u8>, len: u32, offset: u64) -> Self {
        // don't overwrite on already written part
        assert!(buf.spare_capacity_mut().len() >= len as usize);
        let buf_mut_ptr = buf.spare_capacity_mut().as_mut_ptr().cast();

        let read_op = opcode::Read::new(types::Fd(fd.as_raw_fd()), buf_mut_ptr, len)