ronantakizawa/codereview-bench huggingface dataset:
Repository: tokio-rs/tokio
File: tokio/src/runtime/scheduler/multi_thread/handle/metrics.rs (Rust)

Code before:
use super::Handle;

use crate::runtime::{SchedulerMetrics, WorkerMetrics};

impl Handle {
    pub(crate) fn num_workers(&self) -> usize {
        self.shared.worker_metrics.len()
    }

    pub(crate) fn num_blocking_threads(&self) -> usize {
        // workers are currently spawned within the blocking_spawned
        self.blocking_spawner.num_threads() - self.num_workers()
    }

    pub(crate) fn num_idle_blocking_threads(&self) -> usize {
        self.blocking_spawner.num_idle_threads()
    }

    pub(crate) fn active_tasks_count(&self) -> usize {
        self.shared.owned.active_tasks_count()
    }

    pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
        &self.shared.scheduler_metrics
    }

    pub(crate) fn worker_metrics(&self, worker: usize) -> &WorkerMetrics {
        &self.shared.worker_metrics[worker]
    }

    pub(crate) fn injection_queue_depth(&self) -> usize {
        self.shared.injection_queue_depth()
    }

    pub(crate) fn worker_local_queue_depth(&self, worker: usize) -> usize {
        self.shared.worker_local_queue_depth(worker)
    }

Review comment (bug):
It could underflow during runtime startup.
```suggestion
        self.blocking_spawner.num_threads().saturating_sub(self.num_workers())
```

Code after:
use super::Handle;

use crate::runtime::{SchedulerMetrics, WorkerMetrics};

impl Handle {
    pub(crate) fn num_workers(&self) -> usize {
        self.shared.worker_metrics.len()
    }

    pub(crate) fn num_blocking_threads(&self) -> usize {
        // workers are currently spawned using spawn_blocking
        self.blocking_spawner
            .num_threads()
            .saturating_sub(self.num_workers())
    }

    pub(crate) fn num_idle_blocking_threads(&self) -> usize {
        self.blocking_spawner.num_idle_threads()
    }

    pub(crate) fn active_tasks_count(&self) -> usize {
        self.shared.owned.active_tasks_count()
    }

    pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
        &self.shared.scheduler_metrics
    }

    pub(crate) fn worker_metrics(&self, worker: usize) -> &WorkerMetrics {
        &self.shared.worker_metrics[worker]
    }

    pub(crate) fn injection_queue_depth(&self) -> usize {
        self.shared.injection_queue_depth()
    }

    pub(crate) fn worker_local_queue_depth(&self, worker: usize) -> usize {
Repository: tokio-rs/tokio
File: tokio/src/runtime/scheduler/multi_thread/handle/metrics.rs (Rust)

Code before:
use super::Handle;

use crate::runtime::{SchedulerMetrics, WorkerMetrics};

impl Handle {
    pub(crate) fn num_workers(&self) -> usize {
        self.shared.worker_metrics.len()
    }

    pub(crate) fn num_blocking_threads(&self) -> usize {
        // workers are currently spawned within the blocking_spawned
        self.blocking_spawner.num_threads().saturating_sub(self.num_workers())
    }

    pub(crate) fn num_idle_blocking_threads(&self) -> usize {
        self.blocking_spawner.num_idle_threads()
    }

    pub(crate) fn active_tasks_count(&self) -> usize {
        self.shared.owned.active_tasks_count()
    }

    pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
        &self.shared.scheduler_metrics
    }

    pub(crate) fn worker_metrics(&self, worker: usize) -> &WorkerMetrics {
        &self.shared.worker_metrics[worker]
    }

    pub(crate) fn injection_queue_depth(&self) -> usize {
        self.shared.injection_queue_depth()
    }

    pub(crate) fn worker_local_queue_depth(&self, worker: usize) -> usize {
        self.shared.worker_local_queue_depth(worker)

Review comment (suggestion):
```suggestion
        // workers are currently spawned using spawn_blocking
```

Code after:
use super::Handle;

use crate::runtime::{SchedulerMetrics, WorkerMetrics};

impl Handle {
    pub(crate) fn num_workers(&self) -> usize {
        self.shared.worker_metrics.len()
    }

    pub(crate) fn num_blocking_threads(&self) -> usize {
        // workers are currently spawned using spawn_blocking
        self.blocking_spawner
            .num_threads()
            .saturating_sub(self.num_workers())
    }

    pub(crate) fn num_idle_blocking_threads(&self) -> usize {
        self.blocking_spawner.num_idle_threads()
    }

    pub(crate) fn active_tasks_count(&self) -> usize {
        self.shared.owned.active_tasks_count()
    }

    pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
        &self.shared.scheduler_metrics
    }

    pub(crate) fn worker_metrics(&self, worker: usize) -> &WorkerMetrics {
        &self.shared.worker_metrics[worker]
    }

    pub(crate) fn injection_queue_depth(&self) -> usize {
        self.shared.injection_queue_depth()
    }
Repository: tokio-rs/tokio
File: tokio/src/runtime/scheduler/multi_thread_alt/handle/metrics.rs (Rust)

Code before:
use super::Handle;

use crate::runtime::{SchedulerMetrics, WorkerMetrics};

impl Handle {
    pub(crate) fn num_workers(&self) -> usize {
        self.shared.worker_metrics.len()
    }

    pub(crate) fn num_blocking_threads(&self) -> usize {
        // workers are currently spawned within the blocking_spawned
        self.blocking_spawner.num_threads() - self.num_workers()
    }

    pub(crate) fn num_idle_blocking_threads(&self) -> usize {
        self.blocking_spawner.num_idle_threads()
    }

    pub(crate) fn active_tasks_count(&self) -> usize {
        self.shared.owned.active_tasks_count()
    }

    pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
        &self.shared.scheduler_metrics
    }

    pub(crate) fn worker_metrics(&self, worker: usize) -> &WorkerMetrics {
        &self.shared.worker_metrics[worker]
    }

    pub(crate) fn injection_queue_depth(&self) -> usize {
        self.shared.injection_queue_depth()
    }

    pub(crate) fn worker_local_queue_depth(&self, worker: usize) -> usize {
        self.shared.worker_local_queue_depth(worker)
    }

Review comment (suggestion):
```suggestion
        self.blocking_spawner.num_threads().saturating_sub(self.num_workers())
```

Code after:
use super::Handle;

use crate::runtime::{SchedulerMetrics, WorkerMetrics};

impl Handle {
    pub(crate) fn num_workers(&self) -> usize {
        self.shared.worker_metrics.len()
    }

    pub(crate) fn num_blocking_threads(&self) -> usize {
        // workers are currently spawned using spawn_blocking
        self.blocking_spawner
            .num_threads()
            .saturating_sub(self.num_workers())
    }

    pub(crate) fn num_idle_blocking_threads(&self) -> usize {
        self.blocking_spawner.num_idle_threads()
    }

    pub(crate) fn active_tasks_count(&self) -> usize {
        self.shared.owned.active_tasks_count()
    }

    pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
        &self.shared.scheduler_metrics
    }

    pub(crate) fn worker_metrics(&self, worker: usize) -> &WorkerMetrics {
        &self.shared.worker_metrics[worker]
    }

    pub(crate) fn injection_queue_depth(&self) -> usize {
        self.shared.injection_queue_depth()
    }

    pub(crate) fn worker_local_queue_depth(&self, worker: usize) -> usize {
Repository: tokio-rs/tokio
File: tokio/tests/io_poll_aio.rs (Rust)

Code before:
#![warn(rust_2018_idioms)]
#![cfg(all(target_os = "freebsd", feature = "net"))]

use mio_aio::{AioFsyncMode, SourceApi};
use std::{
    future::Future,
    io, mem,
    os::{
        fd::AsFd,
        unix::io::{AsRawFd, RawFd}
    },
    pin::{pin, Pin},
    task::{Context, Poll},
};
use tempfile::tempfile;
use tokio::io::bsd::{Aio, AioSource};
use tokio_test::assert_pending;

mod aio {
    use super::*;

    #[derive(Debug)]
    struct TokioSource<'fd>(mio_aio::Source<nix::sys::aio::AioFsync<'fd>>);

    impl<'fd> AioSource for TokioSource<'fd> {
        fn register(&mut self, kq: RawFd, token: usize) {
            self.0.register_raw(kq, token)
        }
        fn deregister(&mut self) {
            self.0.deregister_raw()
        }
    }

    /// A very crude implementation of an AIO-based future
    struct FsyncFut<'fd>(Aio<TokioSource<'fd>>);


Review comment (suggestion):
Tokio uses this convention for imports:
```suggestion
    os::fd::AsFd,
    os::unix::io::{AsRawFd, RawFd},
```

Code after:
#![warn(rust_2018_idioms)]
#![cfg(all(target_os = "freebsd", feature = "net"))]

use mio_aio::{AioFsyncMode, SourceApi};
use std::{
    future::Future,
    io, mem,
    os::fd::AsFd,
    os::unix::io::{AsRawFd, RawFd},
    pin::{pin, Pin},
    task::{Context, Poll},
};
use tempfile::tempfile;
use tokio::io::bsd::{Aio, AioSource};
use tokio_test::assert_pending;

mod aio {
    use super::*;

    #[derive(Debug)]
    struct TokioSource<'fd>(mio_aio::Source<nix::sys::aio::AioFsync<'fd>>);

    impl<'fd> AioSource for TokioSource<'fd> {
        fn register(&mut self, kq: RawFd, token: usize) {
            self.0.register_raw(kq, token)
        }
        fn deregister(&mut self) {
            self.0.deregister_raw()
        }
    }

    /// A very crude implementation of an AIO-based future
    struct FsyncFut<'fd>(Aio<TokioSource<'fd>>);

    impl<'fd> FsyncFut<'fd> {
        pub fn submit(self: Pin<&mut Self>) -> io::Result<()> {
Repository: tokio-rs/tokio
File: tokio/src/sync/tests/loom_oneshot.rs (Rust)

Code before:
        .unwrap();

        if let Some(mut tx) = tx {
            // Previous task parked, use a new task...
            block_on(OnClose::new(&mut tx));
        }
    });
}

#[test]
fn checking_tx_send_ok_not_drop() {
    use std::cell::RefCell;
    loom::thread_local! {
        static IS_RX: RefCell<bool> = RefCell::new(true);
    }

    struct Msg;

    impl Drop for Msg {
        fn drop(&mut self) {
            IS_RX.with(|is_rx: &RefCell<_>| {
                // On `tx.send(msg)` returning `Err(msg)`,
                // we call `std::mem::forget(msg)`, so that
                // `drop` is not expected to be called in the
                // tx thread.
                assert!(*is_rx.borrow());
            });
        }
    }

    loom::model(|| {
        let (tx, rx) = oneshot::channel();

        // tx thread
        let tx_thread_join_handle = thread::spawn(move || {
            // Ensure that `Msg::drop` in this thread will see is_rx == false
            IS_RX.with(|is_rx: &RefCell<_>| {
                *is_rx.borrow_mut() = false;
            });
            if let Err(msg) = tx.send(Msg) {
                std::mem::forget(msg);
            }
        });

        // rx thread
        let rx_thread_join_handle = thread::spawn(move || {
            drop(rx);
        });

        tx_thread_join_handle.join().unwrap();
        rx_thread_join_handle.join().unwrap();

Review comment (suggestion):
If the newly added code in `oneshot.rs` is removed, we will hit this assertion.

Code after:
        if let Some(mut tx) = tx {
            // Previous task parked, use a new task...
            block_on(OnClose::new(&mut tx));
        }
    });
}

#[test]
fn checking_tx_send_ok_not_drop() {
    use std::borrow::Borrow;
    use std::cell::Cell;

    loom::thread_local! {
        static IS_RX: Cell<bool> = Cell::new(true);
    }

    struct Msg;

    impl Drop for Msg {
        fn drop(&mut self) {
            IS_RX.with(|is_rx: &Cell<_>| {
                // On `tx.send(msg)` returning `Err(msg)`,
                // we call `std::mem::forget(msg)`, so that
                // `drop` is not expected to be called in the
                // tx thread.
                assert!(is_rx.get());
            });
        }
    }

    let mut builder = loom::model::Builder::new();
    builder.preemption_bound = Some(2);

    builder.check(|| {
        let (tx, rx) = oneshot::channel();

        // tx thread
        let tx_thread_join_handle = thread::spawn(move || {
            // Ensure that `Msg::drop` in this thread will see is_rx == false
            IS_RX.with(|is_rx: &Cell<_>| {
                is_rx.set(false);
            });
            if let Err(msg) = tx.send(Msg) {
                std::mem::forget(msg);
            }
        });

        // main thread is the rx thread
        drop(rx);

        tx_thread_join_handle.join().unwrap();
Repository: tokio-rs/tokio
File: tokio/src/sync/tests/loom_oneshot.rs (Rust)

Code before:
                // we call `std::mem::forget(msg)`, so that
                // `drop` is not expected to be called in the
                // tx thread.
                assert!(*is_rx.borrow());
            });
        }
    }

    loom::model(|| {
        let (tx, rx) = oneshot::channel();

        // tx thread
        let tx_thread_join_handle = thread::spawn(move || {
            // Ensure that `Msg::drop` in this thread will see is_rx == false
            IS_RX.with(|is_rx: &RefCell<_>| {
                *is_rx.borrow_mut() = false;
            });
            if let Err(msg) = tx.send(Msg) {
                std::mem::forget(msg);
            }
        });

        // rx thread
        let rx_thread_join_handle = thread::spawn(move || {
            drop(rx);
        });

        tx_thread_join_handle.join().unwrap();
        rx_thread_join_handle.join().unwrap();
    });
}


Review comment (performance):
You can just call `drop(rx)` on the main thread. Spawning threads is very expensive in loom tests.

Code after:
                // we call `std::mem::forget(msg)`, so that
                // `drop` is not expected to be called in the
                // tx thread.
                assert!(is_rx.get());
            });
        }
    }

    let mut builder = loom::model::Builder::new();
    builder.preemption_bound = Some(2);

    builder.check(|| {
        let (tx, rx) = oneshot::channel();

        // tx thread
        let tx_thread_join_handle = thread::spawn(move || {
            // Ensure that `Msg::drop` in this thread will see is_rx == false
            IS_RX.with(|is_rx: &Cell<_>| {
                is_rx.set(false);
            });
            if let Err(msg) = tx.send(Msg) {
                std::mem::forget(msg);
            }
        });

        // main thread is the rx thread
        drop(rx);

        tx_thread_join_handle.join().unwrap();
    });
}
Repository: tokio-rs/tokio
File: tokio/tests/macros_test.rs (Rust)

Code before:
pub mod issue_5243 {
    macro_rules! mac {
        (async fn $name:ident() $b:block) => {
            #[::tokio::test]
            async fn $name() {
                $b
            }
        };
    }
    mac!(
        async fn foo() {}
    );
}

#[cfg(tokio_unstable)]
pub mod macro_rt_arg_unhandled_panic {
    use tokio_test::assert_err;

    #[tokio::test(unhandled_panic = "shutdown_runtime")]
    #[should_panic]
    async fn unhandled_panic_shutdown_runtime() {
        let rt = tokio::spawn(async {
            panic!("This panic should shutdown the runtime.");
        })
        .await;
        assert_err!(rt);
    }

    #[tokio::test(unhandled_panic = "ignore")]
    async fn unhandled_panic_ignore() {
        let rt = tokio::spawn(async {
            panic!("This panic should be forwarded to rt as an error.");
        })
        .await;
        assert_err!(rt);
    }
}


Review comment (suggestion):
```suggestion
        let _ = tokio::spawn(async {
            panic!("This panic should shutdown the runtime.");
        })
        .await;
```

Code after:
pub mod issue_5243 {
    macro_rules! mac {
        (async fn $name:ident() $b:block) => {
            #[::tokio::test]
            async fn $name() {
                $b
            }
        };
    }
    mac!(
        async fn foo() {}
    );
}

#[cfg(tokio_unstable)]
pub mod macro_rt_arg_unhandled_panic {
    use tokio_test::assert_err;

    #[tokio::test(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
    #[should_panic]
    async fn unhandled_panic_shutdown_runtime() {
        let _ = tokio::spawn(async {
            panic!("This panic should shutdown the runtime.");
        })
        .await;
    }

    #[tokio::test(flavor = "current_thread", unhandled_panic = "ignore")]
    async fn unhandled_panic_ignore() {
        let rt = tokio::spawn(async {
            panic!("This panic should be forwarded to rt as an error.");
        })
        .await;
        assert_err!(rt);
    }
}
Repository: tokio-rs/tokio
File: tokio-macros/src/lib.rs (Rust)

Code before:
/// }
/// ```
///
/// Equivalent code not using `#[tokio::main]`
///
/// ```rust
/// use tokio as tokio1;
///
/// fn main() {
///     tokio1::runtime::Builder::new_multi_thread()
///         .enable_all()
///         .build()
///         .unwrap()
///         .block_on(async {
///             println!("Hello world");
///         })
/// }
/// ```
///
/// ### Configure unhandled panic behavior
///
/// Available options are `shutdown_runtime` and `ignore`. For more details, see
/// [`Builder::unhandled_panic`].
///
/// ```ignore
/// #[tokio::main(unhandled_panic = "shutdown_runtime")]
/// async fn main() {
///     let _ = tokio::spawn(async {
///         panic!("This panic will shutdown the runtime.");
///     }).await;
/// }
/// ```
///
/// Equivalent code not using `#[tokio::test]`
///
/// ```ignore
/// fn main() {
///     tokio::runtime::Builder::new_current_thread()
///         .enable_all()
///         .unhandled_panic(UnhandledPanic::ShutdownRuntime)
///         .build()
///         .unwrap()
///         .block_on(async {
///             let _ = tokio::spawn(async {
///                 panic!("This panic will shutdown the runtime.");
///             }).await;
///         })
/// }
/// ```
///
/// **Note**: This option depends on Tokio's [unstable API][unstable]. See [the

Review comment (suggestion):
Since unhandled_panic on multi thread runtime is not implemented yet, so for now it would be better to specify the runtime flavor as well.

```suggestion
/// #[tokio::main(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
```

Code after:
/// }
/// ```
///
/// Equivalent code not using `#[tokio::main]`
///
/// ```rust
/// use tokio as tokio1;
///
/// fn main() {
///     tokio1::runtime::Builder::new_multi_thread()
///         .enable_all()
///         .build()
///         .unwrap()
///         .block_on(async {
///             println!("Hello world");
///         })
/// }
/// ```
///
/// ### Configure unhandled panic behavior
///
/// Available options are `shutdown_runtime` and `ignore`. For more details, see
/// [`Builder::unhandled_panic`].
///
/// This option is only compatible with the `current_thread` runtime.
///
/// ```no_run
/// #[cfg(tokio_unstable)]
/// #[tokio::main(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
/// async fn main() {
///     let _ = tokio::spawn(async {
///         panic!("This panic will shutdown the runtime.");
///     }).await;
/// }
/// # #[cfg(not(tokio_unstable))]
/// # fn main() { }
/// ```
///
/// Equivalent code not using `#[tokio::main]`
///
/// ```no_run
/// #[cfg(tokio_unstable)]
/// fn main() {
///     tokio::runtime::Builder::new_current_thread()
///         .enable_all()
///         .unhandled_panic(UnhandledPanic::ShutdownRuntime)
///         .build()
///         .unwrap()
///         .block_on(async {
///             let _ = tokio::spawn(async {
///                 panic!("This panic will shutdown the runtime.");
Repository: tokio-rs/tokio
File: tokio-macros/src/lib.rs (Rust)

Code before:
///             assert!(true);
///         })
/// }
/// ```
///
/// Note that `start_paused` requires the `test-util` feature to be enabled.
///
/// ### Rename package
///
/// ```rust
/// use tokio as tokio1;
///
/// #[tokio1::test(crate = "tokio1")]
/// async fn my_test() {
///     println!("Hello world");
/// }
/// ```
///
/// ### Configure unhandled panic behavior
///
/// Available options are `shutdown_runtime` and `ignore`. For more details, see
/// [`Builder::unhandled_panic`].
///
/// This option is only compatible with the `current_thread` runtime.
///
/// ```ignore
/// #[tokio::test(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
/// async fn my_test() {
///     let _ = tokio::spawn(async {
///         panic!("This panic will shutdown the runtime.");
///     }).await;
/// }
/// ```
///
/// Equivalent code not using `#[tokio::test]`
///
/// ```ignore
/// #[test]
/// fn my_test() {
///     tokio::runtime::Builder::new_current_thread()
///         .enable_all()
///         .unhandled_panic(UnhandledPanic::ShutdownRuntime)
///         .build()
///         .unwrap()
///         .block_on(async {
///             let _ = tokio::spawn(async {
///                 panic!("This panic will shutdown the runtime.");
///             }).await;
///         })
/// }
/// ```

Review comment (suggestion):
For consistency with the other examples on this page, can you use `no_run` instead of `ignore`?
(The `ignore` annotation makes the doc a bit scary.)
<img width="262" alt="Screen Shot 2024-06-03 at 19 57 29" src="https://github.com/tokio-rs/tokio/assets/55653825/ebb130a5-7d7f-476a-9c69-4d29b93d137f">



Code after:
///     tokio::runtime::Builder::new_current_thread()
///         .enable_all()
///         .start_paused(true)
///         .build()
///         .unwrap()
///         .block_on(async {
///             assert!(true);
///         })
/// }
/// ```
///
/// Note that `start_paused` requires the `test-util` feature to be enabled.
///
/// ### Rename package
///
/// ```rust
/// use tokio as tokio1;
///
/// #[tokio1::test(crate = "tokio1")]
/// async fn my_test() {
///     println!("Hello world");
/// }
/// ```
///
/// ### Configure unhandled panic behavior
///
/// Available options are `shutdown_runtime` and `ignore`. For more details, see
/// [`Builder::unhandled_panic`].
///
/// This option is only compatible with the `current_thread` runtime.
///
/// ```no_run
/// #[cfg(tokio_unstable)]
/// #[tokio::test(flavor = "current_thread", unhandled_panic = "shutdown_runtime")]
/// async fn my_test() {
///     let _ = tokio::spawn(async {
///         panic!("This panic will shutdown the runtime.");
///     }).await;
/// }
/// # #[cfg(not(tokio_unstable))]
/// # fn main() { }
/// ```
///
/// Equivalent code not using `#[tokio::test]`
///
/// ```no_run
/// #[cfg(tokio_unstable)]
/// #[test]
/// fn my_test() {
///     tokio::runtime::Builder::new_current_thread()
///         .enable_all()