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

Code before:

    let guard = fd.ready(Interest::ERROR).await.unwrap();
    assert_eq!(guard.ready(), Ready::ERROR);
}

#[derive(Debug, PartialEq, Eq)]
struct InvalidSource;

impl AsRawFd for InvalidSource {
    fn as_raw_fd(&self) -> RawFd {
        -1
    }
}

#[tokio::test]
async fn try_new() {
    let original = Arc::new(InvalidSource);

    let error = AsyncFd::try_new(original.clone()).unwrap_err();
    let (returned, _cause) = error.into_parts();

    assert!(Arc::ptr_eq(&original, &returned));
}

/// Regression test for issue #7563
///
/// Reproduces the bug where closing fd before dropping AsyncFd causes
/// OS deregister to fail, preventing cleanup and leaking ScheduledIo objects.
#[tokio::test]
async fn memory_leak_when_fd_closed_before_drop() {
    use std::os::unix::io::{AsRawFd, RawFd};
    use std::sync::Arc;
    use tokio::io::unix::AsyncFd;
    use tokio::runtime::Handle;

    use nix::sys::socket::{self, AddressFamily, SockFlag, SockType};

    struct RawFdWrapper {
        fd: RawFd,
    }

    impl AsRawFd for RawFdWrapper {
        fn as_raw_fd(&self) -> RawFd {
            self.fd
        }
    }

    let rt_handle = Handle::current();
    tokio::task::yield_now().await;
    let initial_count = rt_handle.io_total_registration_count();


Review comment (suggestion):
I think the old test should be restored.
It was testing `AsyncFd::try_with_interest()`

Code after:

    let guard = fd.ready(Interest::ERROR).await.unwrap();
    assert_eq!(guard.ready(), Ready::ERROR);
}

#[derive(Debug, PartialEq, Eq)]
struct InvalidSource;

impl AsRawFd for InvalidSource {
    fn as_raw_fd(&self) -> RawFd {
        -1
    }
}

#[tokio::test]
async fn try_new() {
    let original = Arc::new(InvalidSource);

    let error = AsyncFd::try_new(original.clone()).unwrap_err();
    let (returned, _cause) = error.into_parts();

    assert!(Arc::ptr_eq(&original, &returned));
}

#[tokio::test]
async fn try_with_interest() {
    let original = Arc::new(InvalidSource);

    let error = AsyncFd::try_with_interest(original.clone(), Interest::READABLE).unwrap_err();
    let (returned, _cause) = error.into_parts();

    assert!(Arc::ptr_eq(&original, &returned));
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/io/driver.rs (Rust)

Code before:
        }

        self.metrics.dec_fd_count();

        os_result?; // Return error after cleanup

        Ok(())
    }

    fn release_pending_registrations(&self) {
        if self.registrations.needs_release() {
            self.registrations.release(&mut self.synced.lock());
        }
    }
}

/// TEST PURPOSE RELATED TO PR #7773
#[cfg(feature = "full")]
impl Handle {
    /// Returns the number of pending registrations (test-only, not part of public API)
    #[doc(hidden)]
    #[allow(unreachable_pub)]
    pub fn pending_registration_count(&self) -> usize {
        self.registrations.pending_release_count()
    }
    /// Returns the total number of registrations in the main list (test-only)
    #[doc(hidden)]
    #[allow(unreachable_pub)]
    pub fn total_registration_count(&self) -> usize {
        self.registrations
            .total_registration_count(&mut self.synced.lock())
    }
}

impl fmt::Debug for Handle {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Handle")
    }
}

impl Direction {
    pub(super) fn mask(self) -> Ready {
        match self {
            Direction::Read => Ready::READABLE | Ready::READ_CLOSED,
            Direction::Write => Ready::WRITABLE | Ready::WRITE_CLOSED,
        }
    }
}


Review comment (suggestion):
```suggestion
    /// Returns the total number of registrations in the main list (test-only, not part of public API)
```

Code after:
        }

        self.metrics.dec_fd_count();

        os_result // Return error after cleanup
    }

    fn release_pending_registrations(&self) {
        if self.registrations.needs_release() {
            self.registrations.release(&mut self.synced.lock());
        }
    }
}

impl fmt::Debug for Handle {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Handle")
    }
}

impl Direction {
    pub(super) fn mask(self) -> Ready {
        match self {
            Direction::Read => Ready::READABLE | Ready::READ_CLOSED,
            Direction::Write => Ready::WRITABLE | Ready::WRITE_CLOSED,
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/io/driver.rs (Rust)

Code before:

        Ok(scheduled_io)
    }

    /// Deregisters an I/O resource from the reactor.
    pub(super) fn deregister_source(
        &self,
        registration: &Arc<ScheduledIo>,
        source: &mut impl Source,
    ) -> io::Result<()> {
        // Deregister the source with the OS poller **first**
        // Cleanup ALWAYS happens
        let os_result = self.registry.deregister(source);

        if self
            .registrations
            .deregister(&mut self.synced.lock(), registration)
        {
            self.unpark();
        }

        self.metrics.dec_fd_count();

        os_result?; // Return error after cleanup

        Ok(())
    }

    fn release_pending_registrations(&self) {
        if self.registrations.needs_release() {
            self.registrations.release(&mut self.synced.lock());
        }
    }
}

impl fmt::Debug for Handle {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Handle")
    }
}

impl Direction {
    pub(super) fn mask(self) -> Ready {
        match self {
            Direction::Read => Ready::READABLE | Ready::READ_CLOSED,
            Direction::Write => Ready::WRITABLE | Ready::WRITE_CLOSED,
        }
    }
}


Review comment (suggestion):
```suggestion
        os_result // Return error after cleanup
```

Code after:

        Ok(scheduled_io)
    }

    /// Deregisters an I/O resource from the reactor.
    pub(super) fn deregister_source(
        &self,
        registration: &Arc<ScheduledIo>,
        source: &mut impl Source,
    ) -> io::Result<()> {
        // Deregister the source with the OS poller **first**
        // Cleanup ALWAYS happens
        let os_result = self.registry.deregister(source);

        if self
            .registrations
            .deregister(&mut self.synced.lock(), registration)
        {
            self.unpark();
        }

        self.metrics.dec_fd_count();

        os_result // Return error after cleanup
    }

    fn release_pending_registrations(&self) {
        if self.registrations.needs_release() {
            self.registrations.release(&mut self.synced.lock());
        }
    }
}

impl fmt::Debug for Handle {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Handle")
    }
}

impl Direction {
    pub(super) fn mask(self) -> Ready {
        match self {
            Direction::Read => Ready::READABLE | Ready::READ_CLOSED,
            Direction::Write => Ready::WRITABLE | Ready::WRITE_CLOSED,
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/tests/io_async_fd_memory_leak.rs (Rust)

Code before:
//! Regression test for issue #7563 - Memory leak when fd closed before AsyncFd drop
//!
//! This test uses a custom global allocator to track actual memory usage,
//! avoiding false positives from RSS measurements which include freed-but-retained memory.

#![cfg(all(unix, target_os = "linux"))]

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};

/// A tracking allocator that counts bytes currently allocated
struct TrackingAllocator;

static ALLOCATED: AtomicUsize = AtomicUsize::new(0);

unsafe impl GlobalAlloc for TrackingAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        let ptr = unsafe { System.alloc(layout) };
        if !ptr.is_null() {
            ALLOCATED.fetch_add(layout.size(), Ordering::Relaxed);
        }
        ptr
    }

    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        unsafe { System.dealloc(ptr, layout) };
        ALLOCATED.fetch_sub(layout.size(), Ordering::Relaxed);
    }

    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
        let new_ptr = unsafe { System.realloc(ptr, layout, new_size) };

Review comment (suggestion):
Please include a `#[cfg()]` for the feature flags required by this test. You may use "full" if you would like.

Code after:
//! Regression test for issue #7563 - Memory leak when fd closed before AsyncFd drop
//!
//! This test uses a custom global allocator to track actual memory usage,
//! avoiding false positives from RSS measurements which include freed-but-retained memory.

#![cfg(all(unix, target_os = "linux", feature = "full"))]

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};

/// A tracking allocator that counts bytes currently allocated
struct TrackingAllocator {
    allocated: AtomicUsize,
}

impl TrackingAllocator {
    const fn new() -> Self {
        Self {
            allocated: AtomicUsize::new(0),
        }
    }
}

unsafe impl GlobalAlloc for TrackingAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        let ptr = unsafe { System.alloc(layout) };
        if !ptr.is_null() {
            self.allocated.fetch_add(layout.size(), Ordering::Relaxed);
        }
        ptr
    }
Repository: tokio-rs/tokio
File: tokio/tests/io_async_fd_memory_leak.rs (Rust)

Code before:
//! Regression test for issue #7563 - Memory leak when fd closed before AsyncFd drop
//!
//! This test uses a custom global allocator to track actual memory usage,
//! avoiding false positives from RSS measurements which include freed-but-retained memory.

#![cfg(all(unix, target_os = "linux"))]

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};

/// A tracking allocator that counts bytes currently allocated
struct TrackingAllocator;

static ALLOCATED: AtomicUsize = AtomicUsize::new(0);

unsafe impl GlobalAlloc for TrackingAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        let ptr = unsafe { System.alloc(layout) };
        if !ptr.is_null() {
            ALLOCATED.fetch_add(layout.size(), Ordering::Relaxed);
        }
        ptr
    }

    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        unsafe { System.dealloc(ptr, layout) };
        ALLOCATED.fetch_sub(layout.size(), Ordering::Relaxed);
    }

    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
        let new_ptr = unsafe { System.realloc(ptr, layout, new_size) };
        if !new_ptr.is_null() {
            // Subtract old size, add new size
            if new_size > layout.size() {
                ALLOCATED.fetch_add(new_size - layout.size(), Ordering::Relaxed);
            } else {
                ALLOCATED.fetch_sub(layout.size() - new_size, Ordering::Relaxed);
            }
        }

Review comment (suggestion):
nit: this could be a field of the `TrackingAllocator` struct

Code after:
//! Regression test for issue #7563 - Memory leak when fd closed before AsyncFd drop
//!
//! This test uses a custom global allocator to track actual memory usage,
//! avoiding false positives from RSS measurements which include freed-but-retained memory.

#![cfg(all(unix, target_os = "linux", feature = "full"))]

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};

/// A tracking allocator that counts bytes currently allocated
struct TrackingAllocator {
    allocated: AtomicUsize,
}

impl TrackingAllocator {
    const fn new() -> Self {
        Self {
            allocated: AtomicUsize::new(0),
        }
    }
}

unsafe impl GlobalAlloc for TrackingAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        let ptr = unsafe { System.alloc(layout) };
        if !ptr.is_null() {
            self.allocated.fetch_add(layout.size(), Ordering::Relaxed);
        }
        ptr
    }

    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        unsafe { System.dealloc(ptr, layout) };
        self.allocated.fetch_sub(layout.size(), Ordering::Relaxed);
    }

    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
        let new_ptr = unsafe { System.realloc(ptr, layout, new_size) };
Repository: tokio-rs/tokio
File: tokio/tests/io_async_fd_memory_leak.rs (Rust)

Code before:
    use nix::sys::socket::{self, AddressFamily, SockFlag, SockType};
    use std::os::unix::io::{AsRawFd, RawFd};
    use std::sync::Arc;
    use tokio::io::unix::AsyncFd;

    struct RawFdWrapper {
        fd: RawFd,
    }

    impl AsRawFd for RawFdWrapper {
        fn as_raw_fd(&self) -> RawFd {
            self.fd
        }
    }

    struct ArcFd(Arc<RawFdWrapper>);

    impl AsRawFd for ArcFd {
        fn as_raw_fd(&self) -> RawFd {
            self.0.as_raw_fd()
        }
    }

    fn set_nonblocking(fd: RawFd) {
        unsafe {
            let flags = libc::fcntl(fd, libc::F_GETFL);
            libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK);
        }
    }

    // Warm up - let runtime and allocator stabilize
    for _ in 0..100 {
        tokio::task::yield_now().await;
    }

    const ITERATIONS: usize = 1000;

    // Phase 1: Warm up allocations
    for _ in 0..ITERATIONS {
        let (fd_a, _fd_b) = socket::socketpair(
            AddressFamily::Unix,
            SockType::Stream,
            None,
            SockFlag::empty(),
        )
        .unwrap();

        let raw_fd = fd_a.as_raw_fd();
        set_nonblocking(raw_fd);
        std::mem::forget(fd_a);


Review comment (suggestion):
Handle negative value as at https://github.com/F4RAN/tokio/blob/fdf771e34b10e9f5ce4acbb96427e2ae331d6fd0/tokio/tests/io_async_fd.rs#L107-L113

Code after:

#[global_allocator]
static GLOBAL: TrackingAllocator = TrackingAllocator::new();

fn allocated_bytes() -> usize {
    GLOBAL.allocated.load(Ordering::Relaxed)
}

#[tokio::test]
async fn memory_leak_when_fd_closed_before_drop() {
    use nix::sys::socket::{self, AddressFamily, SockFlag, SockType};
    use std::os::unix::io::{AsRawFd, RawFd};
    use std::sync::Arc;
    use tokio::io::unix::AsyncFd;

    struct RawFdWrapper {
        fd: RawFd,
    }

    impl AsRawFd for RawFdWrapper {
        fn as_raw_fd(&self) -> RawFd {
            self.fd
        }
    }

    struct ArcFd(Arc<RawFdWrapper>);

    impl AsRawFd for ArcFd {
        fn as_raw_fd(&self) -> RawFd {
            self.0.as_raw_fd()
        }
    }

    fn set_nonblocking(fd: RawFd) {
        use nix::fcntl::{OFlag, F_GETFL, F_SETFL};

        let flags = nix::fcntl::fcntl(fd, F_GETFL).expect("fcntl(F_GETFL)");

        if flags < 0 {
            panic!(
                "bad return value from fcntl(F_GETFL): {} ({:?})",
                flags,
                nix::Error::last()
            );
        }

        let flags = OFlag::from_bits_truncate(flags) | OFlag::O_NONBLOCK;

        nix::fcntl::fcntl(fd, F_SETFL(flags)).expect("fcntl(F_SETFL)");
    }
Repository: tokio-rs/tokio
File: tokio/src/fs/write.rs (Rust)

Code before:
    feature = "fs",
    target_os = "linux"
))]
async fn write_uring(path: &Path, mut buf: OwnedBuf) -> io::Result<()> {
    use crate::{fs::OpenOptions, runtime::driver::op::Op};
    use std::os::fd::OwnedFd;

    let file = OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .open(path)
        .await?;

    let mut fd: OwnedFd = file
        .try_into_std()
        .expect("unexpected in-flight operation detected")
        .into();

    let total: usize = buf.as_ref().len();
    let mut buf_offset: usize = 0;
    let mut file_offset: u64 = 0;
    while buf_offset < total {
        let (n, _buf, _fd) = Op::write_at(fd, buf, buf_offset, file_offset)?.await;

        let is_eint = |err: &io::Error| {
            err.raw_os_error() == Some(libc::EINTR) || err.kind() == io::ErrorKind::Interrupted
        };

        let n = match n {
            Ok(0) => return Err(io::ErrorKind::WriteZero.into()),
            Ok(n) => n,
            Err(e) if is_eint(&e) => 0,
            Err(e) => return Err(e),
        };

        buf = _buf;
        fd = _fd;
        buf_offset += n as usize;
        file_offset += n as u64;
    }

    Ok(())
}

async fn write_spawn_blocking(path: &Path, contents: OwnedBuf) -> io::Result<()> {
    let path = path.to_owned();
    asyncify(move || std::fs::write(path, contents)).await
}


Review comment (suggestion):
```suggestion
        let is_eintr = |err: &io::Error| {
```

Code after:
    feature = "fs",
    target_os = "linux"
))]
async fn write_uring(path: &Path, mut buf: OwnedBuf) -> io::Result<()> {
    use crate::{fs::OpenOptions, runtime::driver::op::Op};
    use std::os::fd::OwnedFd;

    let file = OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .open(path)
        .await?;

    let mut fd: OwnedFd = file
        .try_into_std()
        .expect("unexpected in-flight operation detected")
        .into();

    let total: usize = buf.as_ref().len();
    let mut buf_offset: usize = 0;
    let mut file_offset: u64 = 0;
    while buf_offset < total {
        let (res, _buf, _fd) = Op::write_at(fd, buf, buf_offset, file_offset)?.await;

        let n = match res {
            Ok(0) => return Err(io::ErrorKind::WriteZero.into()),
            Ok(n) => n,
            Err(e) if e.kind() == io::ErrorKind::Interrupted => 0,
            Err(e) => return Err(e),
        };

        buf = _buf;
        fd = _fd;
        buf_offset += n as usize;
        file_offset += n as u64;
    }

    Ok(())
}

async fn write_spawn_blocking(path: &Path, contents: OwnedBuf) -> io::Result<()> {
    let path = path.to_owned();
    asyncify(move || std::fs::write(path, contents)).await
}
Repository: tokio-rs/tokio
File: tokio/src/fs/write.rs (Rust)

Code before:
    let file = OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .open(path)
        .await?;

    let mut fd: OwnedFd = file
        .try_into_std()
        .expect("unexpected in-flight operation detected")
        .into();

    let total: usize = buf.as_ref().len();
    let mut buf_offset: usize = 0;
    let mut file_offset: u64 = 0;
    while buf_offset < total {
        let (n, _buf, _fd) = Op::write_at(fd, buf, buf_offset, file_offset)?.await;

        let is_eint = |err: &io::Error| {
            err.raw_os_error() == Some(libc::EINTR) || err.kind() == io::ErrorKind::Interrupted
        };

        let n = match n {
            Ok(0) => return Err(io::ErrorKind::WriteZero.into()),
            Ok(n) => n,
            Err(e) if is_eint(&e) => 0,
            Err(e) => return Err(e),
        };

        buf = _buf;
        fd = _fd;
        buf_offset += n as usize;
        file_offset += n as u64;
    }

    Ok(())
}

async fn write_spawn_blocking(path: &Path, contents: OwnedBuf) -> io::Result<()> {
    let path = path.to_owned();
    asyncify(move || std::fs::write(path, contents)).await
}


Review comment (suggestion):
```suggestion
            Err(e) if is_eintr(&e) => 0,
```

Code after:
    let file = OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .open(path)
        .await?;

    let mut fd: OwnedFd = file
        .try_into_std()
        .expect("unexpected in-flight operation detected")
        .into();

    let total: usize = buf.as_ref().len();
    let mut buf_offset: usize = 0;
    let mut file_offset: u64 = 0;
    while buf_offset < total {
        let (res, _buf, _fd) = Op::write_at(fd, buf, buf_offset, file_offset)?.await;

        let n = match res {
            Ok(0) => return Err(io::ErrorKind::WriteZero.into()),
            Ok(n) => n,
            Err(e) if e.kind() == io::ErrorKind::Interrupted => 0,
            Err(e) => return Err(e),
        };

        buf = _buf;
        fd = _fd;
        buf_offset += n as usize;
        file_offset += n as u64;
    }

    Ok(())
}

async fn write_spawn_blocking(path: &Path, contents: OwnedBuf) -> io::Result<()> {
    let path = path.to_owned();
    asyncify(move || std::fs::write(path, contents)).await
}
Repository: tokio-rs/tokio
File: tokio/src/fs/write.rs (Rust)

Code before:
    feature = "fs",
    target_os = "linux"
))]
async fn write_uring(path: &Path, mut buf: OwnedBuf) -> io::Result<()> {
    use crate::{fs::OpenOptions, runtime::driver::op::Op};
    use std::os::fd::OwnedFd;

    let file = OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .open(path)
        .await?;

    let mut fd: OwnedFd = file
        .try_into_std()
        .expect("unexpected in-flight operation detected")
        .into();

    let total: usize = buf.as_ref().len();
    let mut buf_offset: usize = 0;
    let mut file_offset: u64 = 0;
    while buf_offset < total {
        let (n, _buf, _fd) = Op::write_at(fd, buf, buf_offset, file_offset)?.await;

        let n = match n {
            Ok(0) => return Err(io::ErrorKind::WriteZero.into()),
            Ok(n) => n,
            Err(e) if e.kind() == io::ErrorKind::Interrupted => 0,
            Err(e) => return Err(e),
        };

        buf = _buf;
        fd = _fd;
        buf_offset += n as usize;
        file_offset += n as u64;
    }

    Ok(())
}

async fn write_spawn_blocking(path: &Path, contents: OwnedBuf) -> io::Result<()> {
    let path = path.to_owned();
    asyncify(move || std::fs::write(path, contents)).await
}


Review comment (suggestion):
```suggestion
        let n = match res {
```

Since `n` is a `Result`, I prefer to rename it to `res`.

Code after:
    feature = "fs",
    target_os = "linux"
))]
async fn write_uring(path: &Path, mut buf: OwnedBuf) -> io::Result<()> {
    use crate::{fs::OpenOptions, runtime::driver::op::Op};
    use std::os::fd::OwnedFd;

    let file = OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .open(path)
        .await?;

    let mut fd: OwnedFd = file
        .try_into_std()
        .expect("unexpected in-flight operation detected")
        .into();

    let total: usize = buf.as_ref().len();
    let mut buf_offset: usize = 0;
    let mut file_offset: u64 = 0;
    while buf_offset < total {
        let (res, _buf, _fd) = Op::write_at(fd, buf, buf_offset, file_offset)?.await;

        let n = match res {
            Ok(0) => return Err(io::ErrorKind::WriteZero.into()),
            Ok(n) => n,
            Err(e) if e.kind() == io::ErrorKind::Interrupted => 0,
            Err(e) => return Err(e),
        };

        buf = _buf;
        fd = _fd;
        buf_offset += n as usize;
        file_offset += n as u64;
    }

    Ok(())
}

async fn write_spawn_blocking(path: &Path, contents: OwnedBuf) -> io::Result<()> {
    let path = path.to_owned();
    asyncify(move || std::fs::write(path, contents)).await
}