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

Code before:
    target_pointer_width = "64",
    target_pointer_width = "128"
)))]
compile_error! {
    "Tokio requires the platform pointer width to be 32, 64, or 128 bits"
}

#[cfg(all(
    not(tokio_unstable),
    target_family = "wasm",
    any(
        feature = "fs",
        feature = "io-std",
        feature = "net",
        feature = "process",
        feature = "signal"
    )
))]
compile_error!("Only features sync,macros,io-util,rt,time are supported on wasm.");

#[cfg(all(
    not(tokio_unstable),
    all(target_family = "wasm", not(target_os = "wasi")),
    feature = "rt-multi-thread"
))]
compile_error!("rt-multi-thread feature only available on wasi targets.");

#[cfg(all(not(tokio_unstable), tokio_taskdump))]
compile_error!("The `tokio_taskdump` feature requires `--cfg tokio_unstable`.");

#[cfg(all(
    tokio_taskdump,
    not(all(
        target_os = "linux",
        any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")
    ))
))]
compile_error!(
    "The `tokio_taskdump` feature is only currently supported on \
linux, on `aarch64`, `x86` and `x86_64`."
);

// Includes re-exports used by macros.
//
// This module is not intended to be part of the public API. In general, any
// `doc(hidden)` code is not part of Tokio's public and stable API.
#[macro_use]
#[doc(hidden)]
pub mod macros;

cfg_fs! {

Review comment (suggestion):
It's possible I'm misreading these conditions, but as far as I can tell, this has the effect of making rt-multi-thread work on wasi _without_ the `--cfg tokio_unstable`. I don't want to do that in the same PR as the one that adds the initial support for it.

Code after:
    target_pointer_width = "64",
    target_pointer_width = "128"
)))]
compile_error! {
    "Tokio requires the platform pointer width to be 32, 64, or 128 bits"
}

#[cfg(all(
    not(tokio_unstable),
    target_family = "wasm",
    any(
        feature = "fs",
        feature = "io-std",
        feature = "net",
        feature = "process",
        feature = "rt-multi-thread",
        feature = "signal"
    )
))]
compile_error!("Only features sync,macros,io-util,rt,time are supported on wasm.");

#[cfg(all(not(tokio_unstable), tokio_taskdump))]
compile_error!("The `tokio_taskdump` feature requires `--cfg tokio_unstable`.");

#[cfg(all(
    tokio_taskdump,
    not(all(
        target_os = "linux",
        any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")
    ))
))]
compile_error!(
    "The `tokio_taskdump` feature is only currently supported on \
linux, on `aarch64`, `x86` and `x86_64`."
);

// Includes re-exports used by macros.
//
// This module is not intended to be part of the public API. In general, any
// `doc(hidden)` code is not part of Tokio's public and stable API.
#[macro_use]
#[doc(hidden)]
pub mod macros;

cfg_fs! {
    pub mod fs;
}

mod future;

pub mod io;
Repository: tokio-rs/tokio
File: benches/time_timeout.rs (Rust)

Code before:
//! Benchmark spawning a task onto the basic and threaded Tokio executors.
//! This essentially measure the time to enqueue a task in the local and remote
//! case.

use std::time::{Duration, Instant};

use criterion::{black_box, criterion_group, criterion_main, Criterion};
use tokio::time::timeout;

// a vevry quick async task, but might timeout
async fn quick_job() -> usize {
    1
}

fn single_thread_scheduler_timeout(c: &mut Criterion) {
    do_test(c, 1, "single_thread_timeout");
}

fn multi_thread_scheduler_timeout(c: &mut Criterion) {
    do_test(c, 8, "multi_thread_timeout-8");
}

fn do_test(c: &mut Criterion, workers: usize, name: &str) {
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .worker_threads(workers)
        .build()
        .unwrap();

    c.bench_function(name, |b| {
        b.iter_custom(|iters| {
            let start = Instant::now();
            runtime.block_on(async {
                black_box(spawn_job(iters as usize, workers).await);
            });
            start.elapsed()
        })
    });
}

async fn spawn_job(iters: usize, procs: usize) {
    let mut handles = Vec::with_capacity(procs);

Review comment (suggestion):
This is using a multi-threaded runtime with one thread. It would also be nice to include the current-thread runtime.

Code after:
use std::time::{Duration, Instant};

use criterion::{black_box, criterion_group, criterion_main, Criterion};
use tokio::{
    runtime::Runtime,
    time::{sleep, timeout},
};

// a very quick async task, but might timeout
async fn quick_job() -> usize {
    1
}

fn build_run_time(workers: usize) -> Runtime {
    if workers == 1 {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
    } else {
        tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .worker_threads(workers)
            .build()
            .unwrap()
    }
}

fn single_thread_scheduler_timeout(c: &mut Criterion) {
    do_timeout_test(c, 1, "single_thread_timeout");
}

fn multi_thread_scheduler_timeout(c: &mut Criterion) {
    do_timeout_test(c, 8, "multi_thread_timeout-8");
}

fn do_timeout_test(c: &mut Criterion, workers: usize, name: &str) {
    let runtime = build_run_time(workers);
    c.bench_function(name, |b| {
        b.iter_custom(|iters| {
            let start = Instant::now();
            runtime.block_on(async {
Repository: tokio-rs/tokio
File: benches/time_timeout.rs (Rust)

Code before:
//! Benchmark spawning a task onto the basic and threaded Tokio executors.
//! This essentially measure the time to enqueue a task in the local and remote
//! case.

use std::time::{Duration, Instant};

use criterion::{black_box, criterion_group, criterion_main, Criterion};
use tokio::{
    runtime::Runtime,
    time::{sleep, timeout},
};

// a vevry quick async task, but might timeout
async fn quick_job() -> usize {
    1
}

fn build_run_time(workers: usize) -> Runtime {
    if workers == 1 {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .worker_threads(workers)
            .build()
            .unwrap()
    } else {
        tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .worker_threads(workers)
            .build()
            .unwrap()
    }
}

fn single_thread_scheduler_timeout(c: &mut Criterion) {
    do_timeout_test(c, 1, "single_thread_timeout");
}

fn multi_thread_scheduler_timeout(c: &mut Criterion) {
    do_timeout_test(c, 8, "multi_thread_timeout-8");
}

fn do_timeout_test(c: &mut Criterion, workers: usize, name: &str) {
    let runtime = build_run_time(workers);
    c.bench_function(name, |b| {
        b.iter_custom(|iters| {
            let start = Instant::now();
            runtime.block_on(async {

Review comment (suggestion):
I think if we use the current_thread runtime, we wouldn't need to specify the number of worker threads.

Code after:
use std::time::{Duration, Instant};

use criterion::{black_box, criterion_group, criterion_main, Criterion};
use tokio::{
    runtime::Runtime,
    time::{sleep, timeout},
};

// a very quick async task, but might timeout
async fn quick_job() -> usize {
    1
}

fn build_run_time(workers: usize) -> Runtime {
    if workers == 1 {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
    } else {
        tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .worker_threads(workers)
            .build()
            .unwrap()
    }
}

fn single_thread_scheduler_timeout(c: &mut Criterion) {
    do_timeout_test(c, 1, "single_thread_timeout");
}

fn multi_thread_scheduler_timeout(c: &mut Criterion) {
    do_timeout_test(c, 8, "multi_thread_timeout-8");
}

fn do_timeout_test(c: &mut Criterion, workers: usize, name: &str) {
    let runtime = build_run_time(workers);
    c.bench_function(name, |b| {
        b.iter_custom(|iters| {
            let start = Instant::now();
            runtime.block_on(async {
                black_box(spawn_timeout_job(iters as usize, workers).await);
            });
            start.elapsed()
        })
    });
Repository: tokio-rs/tokio
File: tokio/src/sync/notify.rs (Rust)

Code before:
        if matches!(*state, State::Waiting) {
            let mut waiters = notify.waiters.lock();
            let mut notify_state = notify.state.load(SeqCst);

            // We hold the lock, so this field is not concurrently accessed by
            // `notify_*` functions and we can use the relaxed ordering.
            let (notification, strategy) = waiter.notification.load(Relaxed);

            // remove the entry from the list (if not already removed)
            //
            // Safety: we hold the lock, so we have an exclusive access to every list the
            // waiter may be contained in. If the node is not contained in the `waiters`
            // list, then it is contained by a guarded list used by `notify_waiters`.
            unsafe { waiters.remove(NonNull::from(waiter)) };

            if waiters.is_empty() && get_state(notify_state) == WAITING {
                notify_state = set_state(notify_state, EMPTY);
                notify.state.store(notify_state, SeqCst);
            }

            // See if the node was notified but not received. In this case, if
            // the notification was triggered via `notify_one`, it must be sent
            // to the next waiter.
            if notification == Some(Notification::One) {
                if let Some(waker) =
                    notify_locked(&mut waiters, &notify.state, notify_state, strategy.unwrap())
                {
                    drop(waiters);
                    waker.wake();
                }
            }
        }
    }
}

/// # Safety
///
/// `Waiter` is forced to be !Unpin.
unsafe impl linked_list::Link for Waiter {
    type Handle = NonNull<Waiter>;
    type Target = Waiter;

    fn as_raw(handle: &NonNull<Waiter>) -> NonNull<Waiter> {
        *handle
    }

    unsafe fn from_raw(ptr: NonNull<Waiter>) -> NonNull<Waiter> {
        ptr
    }

    unsafe fn pointers(target: NonNull<Waiter>) -> NonNull<linked_list::Pointers<Waiter>> {

Review comment (suggestion):
I think we can drop the `Option` around the strategy. Then we don't need an `unwrap` here. We can ignore the strategy when we are waking everything.

Code after:
            // the notification was triggered via `notify_one`, it must be sent
            // to the next waiter.
            if let Some(Notification::One(strategy)) = notification {
                if let Some(waker) =
                    notify_locked(&mut waiters, &notify.state, notify_state, strategy)
                {
                    drop(waiters);
                    waker.wake();
                }
            }
        }
    }
}

/// # Safety
///
/// `Waiter` is forced to be !Unpin.
unsafe impl linked_list::Link for Waiter {
    type Handle = NonNull<Waiter>;
    type Target = Waiter;

    fn as_raw(handle: &NonNull<Waiter>) -> NonNull<Waiter> {
        *handle
    }

    unsafe fn from_raw(ptr: NonNull<Waiter>) -> NonNull<Waiter> {
        ptr
    }

    unsafe fn pointers(target: NonNull<Waiter>) -> NonNull<linked_list::Pointers<Waiter>> {
        Waiter::addr_of_pointers(target)
    }
}

fn is_unpin<T: Unpin>() {}
Repository: tokio-rs/tokio
File: tokio/tests/sync_notify.rs (Rust)

Code before:

    // by default fifo
    notify.notify_one();

    drop(notified1);

    // next waiter should be the one to be to woken up
    assert_ready!(notified2.poll());
    assert_pending!(notified3.poll());
}

#[test]
fn notified_multi_notify_drop_one_last_in() {
    let notify = Notify::new();
    let mut notified1 = spawn(async { notify.notified().await });
    let mut notified2 = spawn(async { notify.notified().await });
    let mut notified3 = spawn(async { notify.notified().await });

    // add waiters by order of poll execution
    assert_pending!(notified1.poll());
    assert_pending!(notified2.poll());
    assert_pending!(notified3.poll());

    notify.notify_one_last_in();

    drop(notified1);

    // latest waiter added should be the one to woken up
    assert_ready!(notified3.poll());
    assert_pending!(notified2.poll());
}

#[test]
fn notify_in_drop_after_wake() {
    use futures::task::ArcWake;
    use std::future::Future;
    use std::sync::Arc;

    let notify = Arc::new(Notify::new());

    struct NotifyOnDrop(Arc<Notify>);

    impl ArcWake for NotifyOnDrop {
        fn wake_by_ref(_arc_self: &Arc<Self>) {}
    }

    impl Drop for NotifyOnDrop {
        fn drop(&mut self) {
            self.0.notify_waiters();
        }
    }

Review comment (suggestion):
Shouldn't you drop `notified3` to trigger the forwarding logic?
```suggestion
    drop(notified3);
```

Code after:

    // by default fifo
    notify.notify_one();

    drop(notified1);

    // next waiter should be the one to be to woken up
    assert_ready!(notified2.poll());
    assert_pending!(notified3.poll());
}

#[test]
fn notified_multi_notify_last_drop() {
    let notify = Notify::new();
    let mut notified1 = spawn(async { notify.notified().await });
    let mut notified2 = spawn(async { notify.notified().await });
    let mut notified3 = spawn(async { notify.notified().await });

    // add waiters by order of poll execution
    assert_pending!(notified1.poll());
    assert_pending!(notified2.poll());
    assert_pending!(notified3.poll());

    notify.notify_last();

    drop(notified3);

    // latest waiter added should be the one to woken up
    assert_ready!(notified2.poll());
    assert_pending!(notified1.poll());
}

#[test]
fn notify_in_drop_after_wake() {
    use futures::task::ArcWake;
    use std::future::Future;
    use std::sync::Arc;

    let notify = Arc::new(Notify::new());

    struct NotifyOnDrop(Arc<Notify>);

    impl ArcWake for NotifyOnDrop {
        fn wake_by_ref(_arc_self: &Arc<Self>) {}
    }

    impl Drop for NotifyOnDrop {
        fn drop(&mut self) {
            self.0.notify_waiters();
        }
    }
Repository: tokio-rs/tokio
File: tokio/src/runtime/time/mod.rs (Rust)

Code before:
    fn set_next_wake(&self, value: Option<NonZeroU64>) {
        let val = value.map(|v| v.get()).unwrap_or(0);
        self.next_wake.store(val, Ordering::Relaxed);
    }

    fn get_next_wake(&self) -> Option<NonZeroU64> {
        let val = self.next_wake.load(Ordering::Relaxed);
        NonZeroU64::new(val)
    }
}

impl fmt::Debug for Inner {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("Inner").finish()
    }
}

mod rand {
    use crate::util::rand::FastRand;
    use std::cell::Cell;

    // Used by `TimerEntry`.
    pub(crate) fn thread_rng_n(n: u32) -> u32 {
        thread_local! {
            static THREAD_RNG: Cell<FastRand> = Cell::new(FastRand::new());
        }

        THREAD_RNG.with(|cell_rng| {
            let mut rng = cell_rng.get();
            let ret = rng.fastrand_n(n);
            cell_rng.set(rng);
            ret
        })
    }
}

#[cfg(test)]
mod tests;


Review comment (suggestion):
Please do not create new thread locals. Some platforms have small limits on the number of thread locals that each process can have.

There should already be a random number generator in the runtime context somewhere.

Code after:
        }
    }

    cfg_test_util! {
        fn did_wake(&self) -> bool {
            self.inner.did_wake.swap(false, Ordering::SeqCst)
        }
    }
}

// ===== impl Inner =====

impl Inner {
    /// Locks the driver's sharded wheel structure.
    pub(super) fn lock_sharded_wheel(
        &self,
        shard_id: u32,
    ) -> crate::loom::sync::MutexGuard<'_, Wheel> {
        let index = shard_id % (self.wheels.len() as u32);
        // Safety: This modulo operation ensures that the index is not out of bounds.
        unsafe { self.wheels.get_unchecked(index as usize).lock() }
    }

    // Check whether the driver has been shutdown
    pub(super) fn is_shutdown(&self) -> bool {
        self.is_shutdown.load(Ordering::SeqCst)
    }

    // Gets the number of shards.
    fn get_shard_size(&self) -> u32 {
        self.wheels.len() as u32
    }
}

impl fmt::Debug for Inner {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("Inner").finish()
    }
}

#[cfg(test)]
mod tests;
Repository: tokio-rs/tokio
File: tokio/src/runtime/time/entry.rs (Rust)

Code before:
    /// Returns the last-registered waker, if any.
    ///
    /// SAFETY: The driver lock must be held while invoking this function, and
    /// the entry must not be in any wheel linked lists.
    pub(super) unsafe fn fire(self, completed_state: TimerResult) -> Option<Waker> {
        self.inner.as_ref().state.fire(completed_state)
    }
}

impl Drop for TimerEntry {
    fn drop(&mut self) {
        unsafe { Pin::new_unchecked(self) }.as_mut().cancel();
    }
}

// Generates a shard id. If current thread is a worker thread, we use its worker index as a shard id.
// Otherwise, we use a random number generator to obtain the shard id.
cfg_rt! {
    fn generate_shard_id(shard_size: u32) -> u32 {
        let id = context::with_scheduler(|ctx| match ctx {
            Some(scheduler::Context::CurrentThread(_ctx)) => 0,
            #[cfg(feature = "rt-multi-thread")]
            Some(scheduler::Context::MultiThread(ctx)) => ctx.get_worker_index() as u32,
            #[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
            Some(scheduler::Context::MultiThreadAlt(ctx)) => ctx.get_worker_index() as u32,
            _ => context::thread_rng_n(shard_size),
        });
        id % shard_size
    }
}

cfg_not_rt! {
    fn generate_shard_id(shard_size: u32) -> u32 {
        context::thread_rng_n(shard_size)
    }
}


Review comment (suggestion):
```suggestion
            None => context::thread_rng_n(shard_size),
```

Code after:
    /// Returns the last-registered waker, if any.
    ///
    /// SAFETY: The driver lock must be held while invoking this function, and
    /// the entry must not be in any wheel linked lists.
    pub(super) unsafe fn fire(self, completed_state: TimerResult) -> Option<Waker> {
        self.inner.as_ref().state.fire(completed_state)
    }
}

impl Drop for TimerEntry {
    fn drop(&mut self) {
        unsafe { Pin::new_unchecked(self) }.as_mut().cancel();
    }
}

// Generates a shard id. If current thread is a worker thread, we use its worker index as a shard id.
// Otherwise, we use a random number generator to obtain the shard id.
cfg_rt! {
    fn generate_shard_id(shard_size: u32) -> u32 {
        let id = context::with_scheduler(|ctx| match ctx {
            Some(scheduler::Context::CurrentThread(_ctx)) => 0,
            #[cfg(feature = "rt-multi-thread")]
            Some(scheduler::Context::MultiThread(ctx)) => ctx.get_worker_index() as u32,
            #[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
            Some(scheduler::Context::MultiThreadAlt(ctx)) => ctx.get_worker_index() as u32,
            None => context::thread_rng_n(shard_size),
        });
        id % shard_size
    }
}

cfg_not_rt! {
    fn generate_shard_id(shard_size: u32) -> u32 {
        context::thread_rng_n(shard_size)
    }
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/scheduler/multi_thread_alt/worker.rs (Rust)

Code before:
        // Smooth out jitter
        if abs_diff(self.global_queue_interval, next) > 2 {
            self.global_queue_interval = next;
        }
    }

    fn shutdown_clear_defer(&self, cx: &Context) {
        let mut defer = cx.defer.borrow_mut();

        for task in defer.drain(..) {
            drop(task);
        }
    }
}

impl Context {
    pub(crate) fn defer(&self, waker: &Waker) {
        // TODO: refactor defer across all runtimes
        waker.wake_by_ref();
    }

    fn shared(&self) -> &Shared {
        &self.handle.shared
    }

    #[allow(dead_code)]
    pub(crate) fn get_worker_index(&self) -> usize {
        self.index
    }
}

impl Core {
    fn next_local_task(&mut self) -> Option<Notified> {
        self.next_lifo_task().or_else(|| self.run_queue.pop())
    }

    fn next_lifo_task(&mut self) -> Option<Notified> {
        self.lifo_slot.take()
    }
}

impl Shared {
    fn next_remote_task_synced(&self, synced: &mut Synced) -> Option<Notified> {
        // safety: we only have access to a valid `Synced` in this file.
        unsafe { self.inject.pop(&mut synced.inject) }
    }

    pub(super) fn schedule_task(&self, task: Notified, is_yield: bool) {
        use std::ptr;

        with_current(|maybe_cx| {

Review comment (suggestion):
```suggestion
    #[cfg_attr(not(feature = "time"), allow(dead_code))]
```

Code after:
        // Smooth out jitter
        if abs_diff(self.global_queue_interval, next) > 2 {
            self.global_queue_interval = next;
        }
    }

    fn shutdown_clear_defer(&self, cx: &Context) {
        let mut defer = cx.defer.borrow_mut();

        for task in defer.drain(..) {
            drop(task);
        }
    }
}

impl Context {
    pub(crate) fn defer(&self, waker: &Waker) {
        // TODO: refactor defer across all runtimes
        waker.wake_by_ref();
    }

    fn shared(&self) -> &Shared {
        &self.handle.shared
    }

    #[cfg_attr(not(feature = "time"), allow(dead_code))]
    pub(crate) fn get_worker_index(&self) -> usize {
        self.index
    }
}

impl Core {
    fn next_local_task(&mut self) -> Option<Notified> {
        self.next_lifo_task().or_else(|| self.run_queue.pop())
    }

    fn next_lifo_task(&mut self) -> Option<Notified> {
        self.lifo_slot.take()
    }
}

impl Shared {
    fn next_remote_task_synced(&self, synced: &mut Synced) -> Option<Notified> {
        // safety: we only have access to a valid `Synced` in this file.
        unsafe { self.inject.pop(&mut synced.inject) }
    }

    pub(super) fn schedule_task(&self, task: Notified, is_yield: bool) {
        use std::ptr;

        with_current(|maybe_cx| {