ronantakizawa/codereview-bench huggingface dataset:
Repository: tokio-rs/tokio
File: tokio/src/fs/read_uring.rs (Rust)

Code before:

            match res {
                Ok(size_read) => {
                    if size_read == 0 {
                        return Ok(r_buf);
                    }

                    let size_read_usize = size_read as usize;

                    let new_len = size_read_usize + r_buf.len();
                    // SAFETY: We didn't read more than what as reserved
                    // as capacity, the _size_read was initialized by the kernel
                    // via a mutable pointer
                    unsafe { r_buf.set_len(new_len) }

                    fd = r_fd;
                    buf = r_buf;
                    offset += size_read as u64;
                    read_len -= size_read;

                    // 1. In case of no size_hint and a large file, if we keep reading
                    // PROBE_SIZE, we want to increment number of short reads in order to gradually
                    // increase read size per Op submission
                    // 2. In case of small reads by the kernel, also gradually increase
                    // read size per Op submission to read files in lesser cycles
                    if size_read_usize <= buf.len() {
                        consecutive_short_reads += 1;
                    } else {
                        consecutive_short_reads = 0;
                    }

                    // keep reading if there's something left to be read
                    if read_len > 0 {
                        continue;
                    } else {
                        break;
                    }
                }
                Err(e) => {
                    if e.kind() == ErrorKind::Interrupted {
                        buf = r_buf;
                        fd = r_fd;

                        continue;
                    } else {
                        return Err(e);
                    }
                }
            }
        }
    }

Review comment (suggestion):
`buf.len` (`r_buf`) is updated at line 119 and it accumulates the old length + `size_read_usize`.
I think this `if` check will always pass.

Code after:
    Ok((r_fd, r_buf, is_eof))
}

// Takes a amount of length to read and returns a singluar read in the buffer
//
// Returns the file descriptor, buffer and EOF reached or not
async fn op_read(
    mut fd: OwnedFd,
    mut buf: Vec<u8>,
    offset: &mut u64,
    read_len: u32,
) -> io::Result<(OwnedFd, Vec<u8>, bool)> {
    loop {
        let (res, r_fd, r_buf) = Op::read(fd, buf, read_len, *offset).await;

        match res {
            Err(e) if e.kind() == ErrorKind::Interrupted => {
                buf = r_buf;
                fd = r_fd;
            }
            Err(e) => return Err(e),
            Ok(size_read) => {
                *offset += size_read as u64;

                return Ok((r_fd, r_buf, size_read == 0));
            }
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/io/uring/read.rs (Rust)

Code before:

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 uinitialized memory. The read happens on unitialized
    // buffer and no overwiting 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
        let written = buf.len();
        let slice: &mut [u8] = &mut buf[written..];
        let buf_mut_ptr = slice.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
        unsafe { Op::new(read_op, Read { fd, buf }) }
    }
}


Review comment (suggestion):
This function should be marked as `unsafe` if the caller needs to follow this.

Code after:

        (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)
            .offset(offset)
            .build();

        // SAFETY: Parameters are valid for the entire duration of the operation
        unsafe { Op::new(read_op, Read { fd, buf }) }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/io/uring/read.rs (Rust)

Code before:
    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 uinitialized memory. The read happens on unitialized
    // buffer and no overwiting 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
        let written = buf.len();
        let slice: &mut [u8] = &mut buf[written..];
        let buf_mut_ptr = slice.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
        unsafe { Op::new(read_op, Read { fd, buf }) }
    }
}


Review comment (suggestion):
```suggestion
        assert!(buf.spare_capacity_mut().len() <= len as usize);
        let buf_mut_ptr = buf.spare_capacity_mut().as_mut_ptr();
```

Code after:
            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)
            .offset(offset)
            .build();

        // SAFETY: Parameters are valid for the entire duration of the operation
        unsafe { Op::new(read_op, Read { fd, buf }) }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/fs/read_uring.rs (Rust)

Code before:
                    // SAFETY: We didn't read more than what as reserved
                    // as capacity, the _size_read was initialized by the kernel
                    // via a mutable pointer
                    unsafe { r_buf.set_len(new_len) }

                    let requested = read_len;

                    fd = r_fd;
                    buf = r_buf;
                    offset += size_read as u64;
                    read_len -= size_read;

                    // 1. In case of no size_hint and a large file, if we keep reading
                    // PROBE_SIZE, we want to increment number of short reads in order to gradually
                    // increase read size per Op submission
                    // 2. In case of small reads by the kernel, also gradually increase
                    // read size per Op submission to read files in lesser cycles
                    if size_read <= requested {
                        consecutive_short_reads += 1;
                    } else {
                        consecutive_short_reads = 0;
                    }

                    // keep reading if there's something left to be read
                    if read_len > 0 {
                        continue;
                    } else {
                        break;
                    }
                }
                Err(e) if e.kind() == ErrorKind::Interrupted => {
                    buf = r_buf;
                    fd = r_fd;

                    continue;
                }
                Err(e) => return Err(e),
            }
        }
    }
}

async fn small_probe_read(
    mut fd: OwnedFd,
    mut buf: Vec<u8>,
    offset: u64,
) -> io::Result<(u32, OwnedFd, Vec<u8>)> {
    let mut temp_arr = [0; PROBE_SIZE];
    let has_enough = buf.len() > PROBE_SIZE;

    if has_enough {

Review comment (refactor):
I think we can remove this consecutive short reads logic.
```suggestion
                    // keep reading if there's something left to be read
                    if read_len == size_read {
                        continue;
```

Code after:
    r_buf.splice(back_bytes_len..back_bytes_len, temp_arr);

    Ok((r_fd, r_buf, is_eof))
}

// Takes a amount of length to read and returns a singluar read in the buffer
//
// Returns the file descriptor, buffer and EOF reached or not
async fn op_read(
    mut fd: OwnedFd,
    mut buf: Vec<u8>,
    offset: &mut u64,
    read_len: u32,
) -> io::Result<(OwnedFd, Vec<u8>, bool)> {
    loop {
        let (res, r_fd, r_buf) = Op::read(fd, buf, read_len, *offset).await;

        match res {
            Err(e) if e.kind() == ErrorKind::Interrupted => {
                buf = r_buf;
                fd = r_fd;
            }
            Err(e) => return Err(e),
            Ok(size_read) => {
                *offset += size_read as u64;

                return Ok((r_fd, r_buf, size_read == 0));
            }
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/fs/read_uring.rs (Rust)

Code before:
                Err(e) if e.kind() == ErrorKind::Interrupted => {
                    buf = r_buf;
                    fd = r_fd;

                    continue;
                }
                Err(e) => return Err(e),
            }
        }
    }
}

async fn small_probe_read(
    mut fd: OwnedFd,
    mut buf: Vec<u8>,
    offset: u64,
) -> io::Result<(u32, OwnedFd, Vec<u8>)> {
    let mut temp_arr = [0; PROBE_SIZE];
    let has_enough = buf.len() > PROBE_SIZE;

    if has_enough {
        // if we have more than PROBE_SIZE bytes in the buffer already then
        // don't call reserve as we might potentially read 0 bytes
        let back_bytes_len = buf.len() - PROBE_SIZE;
        temp_arr.copy_from_slice(&buf[back_bytes_len..]);
        // We're decreasing the length of the buffer and len is greater
        // than PROBE_SIZE. So we can read into the discarded length
        buf.truncate(back_bytes_len);
    } else {
        // we don't even have PROBE_SIZE length in the buffer, we need this
        // reservation
        buf.reserve_exact(PROBE_SIZE);
    }

    loop {
        let (res, r_fd, mut r_buf) = Op::read(fd, buf, PROBE_SIZE_U32, offset).await;

        match res {
            // return early if we inserted into reserved PROBE_SIZE
            // bytes
            Ok(size_read) if !has_enough => return Ok((size_read, r_fd, r_buf)),
            Ok(size_read) => {
                let old_len = r_buf.len() - (size_read as usize);

                r_buf.splice(old_len..old_len, temp_arr);

                return Ok((size_read, r_fd, r_buf));
            }
            Err(e) if e.kind() == ErrorKind::Interrupted => {
                buf = r_buf;
                fd = r_fd;

Review comment (suggestion):
This is a bit hard to follow, we may need a dedicated `Read` type just `[u8; N]`, for example:

<img src="https://github.com/user-attachments/assets/2c6c2089-dd9f-4c22-a07b-bd3536f9c953" />


Code after:
// Takes a amount of length to read and returns a singluar read in the buffer
//
// Returns the file descriptor, buffer and EOF reached or not
async fn op_read(
    mut fd: OwnedFd,
    mut buf: Vec<u8>,
    offset: &mut u64,
    read_len: u32,
) -> io::Result<(OwnedFd, Vec<u8>, bool)> {
    loop {
        let (res, r_fd, r_buf) = Op::read(fd, buf, read_len, *offset).await;

        match res {
            Err(e) if e.kind() == ErrorKind::Interrupted => {
                buf = r_buf;
                fd = r_fd;
            }
            Err(e) => return Err(e),
            Ok(size_read) => {
                *offset += size_read as u64;

                return Ok((r_fd, r_buf, size_read == 0));
            }
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/fs/read_uring.rs (Rust)

Code before:
        buf.try_reserve(PROBE_SIZE)?;

        return op_read(fd, buf, offset, &mut read_len).await;
    }

    let mut temp_arr = [0; PROBE_SIZE];
    let back_bytes_len = buf.len() - PROBE_SIZE;

    temp_arr.copy_from_slice(&buf[back_bytes_len..]);

    // We're decreasing the length of the buffer and len is greater
    // than PROBE_SIZE. So we can read into the discarded length
    buf.truncate(back_bytes_len);

    let (size_read, r_fd, mut r_buf) = op_read(fd, buf, offset, &mut read_len).await?;

    r_buf.splice(back_bytes_len..back_bytes_len, temp_arr);

    Ok((size_read, r_fd, r_buf))
}

async fn op_read(
    mut fd: OwnedFd,
    mut buf: Vec<u8>,
    offset: &mut u64,
    read_len: &mut u32,
) -> io::Result<(u32, OwnedFd, Vec<u8>)> {
    loop {
        let (res, r_fd, r_buf) = Op::read(fd, buf, *read_len, *offset).await;

        match res {
            Err(e) if e.kind() == ErrorKind::Interrupted => {
                buf = r_buf;
                fd = r_fd;
            }
            Err(e) => return Err(e),
            Ok(size_read) => {
                *offset += size_read as u64;
                *read_len -= size_read;

                if *read_len == 0 || size_read == 0 {
                    return Ok((size_read, r_fd, r_buf));
                }

                buf = r_buf;
                fd = r_fd;
            }
        }
    }
}


Review comment (suggestion):
There are no cases where the caller reads the value of `read_len` after the call, so a ref is not needed.
```suggestion
    read_len: u32,
```

Code after:

    Ok((r_fd, r_buf, is_eof))
}

// Takes a amount of length to read and returns a singluar read in the buffer
//
// Returns the file descriptor, buffer and EOF reached or not
async fn op_read(
    mut fd: OwnedFd,
    mut buf: Vec<u8>,
    offset: &mut u64,
    read_len: u32,
) -> io::Result<(OwnedFd, Vec<u8>, bool)> {
    loop {
        let (res, r_fd, r_buf) = Op::read(fd, buf, read_len, *offset).await;

        match res {
            Err(e) if e.kind() == ErrorKind::Interrupted => {
                buf = r_buf;
                fd = r_fd;
            }
            Err(e) => return Err(e),
            Ok(size_read) => {
                *offset += size_read as u64;

                return Ok((r_fd, r_buf, size_read == 0));
            }
        }
    }
}
Repository: tokio-rs/tokio
File: tokio-util/src/io/write_all_vectored.rs (Rust)

Code before:
fn advance_slices<'a>(bufs: &mut &mut [IoSlice<'a>], n: usize) {
    // Number of buffers to remove.
    let mut remove = 0;
    // Remaining length before reaching n. This prevents overflow
    // that could happen if the length of slices in `bufs` were instead
    // accumulated. Those slice may be aliased and, if they are large
    // enough, their added length may overflow a `usize`.
    let mut left = n;
    for buf in bufs.iter() {
        if let Some(remainder) = left.checked_sub(buf.len()) {
            left = remainder;
            remove += 1;
        } else {
            break;
        }
    }

    *bufs = &mut std::mem::take(bufs)[remove..];
    if let Some(first) = bufs.first_mut() {
        // SAFETY: We are recreating an IoSlice from a sub-slice of itself.
        // use raw pointers to bypass the borrow checker's lifetime rules.
        unsafe {
            let new_ptr = first.as_ptr().add(left);
            let new_len = first.len() - left;
            *first = IoSlice::new(std::slice::from_raw_parts(new_ptr, new_len));
        }
    } else {
        assert!(left == 0, "advancing io slices beyond their length");
    }
}


Review comment (suggestion):
This doesn't perform any bounds checks. I would prefer to instead sub-slice the `IoSlice` and then extend the lifetime using [the same strategy as fill_buf](https://github.com/tokio-rs/tokio/blob/c23735d43be36d31911dd2c832f36e5e6a6078a1/tokio/src/io/util/fill_buf.rs#L39)

Code after:
    // Remaining length before reaching n. This prevents overflow
    // that could happen if the length of slices in `bufs` were instead
    // accumulated. Those slice may be aliased and, if they are large
    // enough, their added length may overflow a `usize`.
    let mut left = n;
    for buf in bufs.iter() {
        if let Some(remainder) = left.checked_sub(buf.len()) {
            left = remainder;
            remove += 1;
        } else {
            break;
        }
    }

    *bufs = &mut std::mem::take(bufs)[remove..];
    if let Some(first) = bufs.first_mut() {
        let buf = &first[..left];
        // necessary due to limitating in the borrow checker,
        // when tokio MSRV reaches 1.81.0 this entire function
        // can be replaced with `IoSlice::advance_slices`
        //
        // SAFETY: transmute a sub-slice of an IoSlice<'a> back to
        // the lifetime `'a`. This is safe because the underlying memory
        // is guaranteed to live for 'a, we have shared access, and no
        // underlying data is reinterpreted to a different type.
        unsafe {
            *first = IoSlice::new(std::mem::transmute::<&[u8], &'a [u8]>(buf));
        }
    } else {
        assert!(left == 0, "advancing io slices beyond their length");
    }
}
Repository: tokio-rs/tokio
File: tokio-util/src/io/write_all_vectored.rs (Rust)

Code before:

// copied from `std::IoSlice::advance_slices`
// replace with method when MSRV is 1.81.0
fn advance_slices<'a>(bufs: &mut &mut [IoSlice<'a>], n: usize) {
    // Number of buffers to remove.
    let mut remove = 0;
    // Remaining length before reaching n. This prevents overflow
    // that could happen if the length of slices in `bufs` were instead
    // accumulated. Those slice may be aliased and, if they are large
    // enough, their added length may overflow a `usize`.
    let mut left = n;
    for buf in bufs.iter() {
        if let Some(remainder) = left.checked_sub(buf.len()) {
            left = remainder;
            remove += 1;
        } else {
            break;
        }
    }

    *bufs = &mut std::mem::take(bufs)[remove..];
    if let Some(first) = bufs.first_mut() {
        let buf = &first[..left];
        // SAFETY: necessary due to limitating in the borrow checker,
        // when tokio MSRV reaches 1.81.0 this entire function
        // can be replaced with `IoSlice::advance_slices`
        unsafe {
            *first = IoSlice::new(std::mem::transmute::<&[u8], &'a [u8]>(buf));
        }
    } else {
        assert!(left == 0, "advancing io slices beyond their length");
    }
}


Review comment (suggestion):
This comment is good, but it's not a safety comment. Safety comments should explain why the unsafe code is correct, not why it's necessary.

Code after:
fn advance_slices<'a>(bufs: &mut &mut [IoSlice<'a>], n: usize) {
    // Number of buffers to remove.
    let mut remove = 0;
    // Remaining length before reaching n. This prevents overflow
    // that could happen if the length of slices in `bufs` were instead
    // accumulated. Those slice may be aliased and, if they are large
    // enough, their added length may overflow a `usize`.
    let mut left = n;
    for buf in bufs.iter() {
        if let Some(remainder) = left.checked_sub(buf.len()) {
            left = remainder;
            remove += 1;
        } else {
            break;
        }
    }

    *bufs = &mut std::mem::take(bufs)[remove..];
    if let Some(first) = bufs.first_mut() {
        let buf = &first[..left];
        // necessary due to limitating in the borrow checker,
        // when tokio MSRV reaches 1.81.0 this entire function
        // can be replaced with `IoSlice::advance_slices`
        //
        // SAFETY: transmute a sub-slice of an IoSlice<'a> back to
        // the lifetime `'a`. This is safe because the underlying memory
        // is guaranteed to live for 'a, we have shared access, and no
        // underlying data is reinterpreted to a different type.
        unsafe {
            *first = IoSlice::new(std::mem::transmute::<&[u8], &'a [u8]>(buf));
        }
    } else {
        assert!(left == 0, "advancing io slices beyond their length");
    }
}