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

Code before:
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime and return a [`JoinHandle`].
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// When this function is called, if the thread pool is active, the provided future
    /// will start running in the background immediately.
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
    /// ```
    #[track_caller]
    pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>

Review comment (suggestion):
It is recommended to add two sentences explaining the behavior of `Runtime::spawn`: one for current thread runtime and another for multi-threaded runtime.

Please keep it concise—verbose documentation makes it harder for downstream users to understand.

Code after:
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// The provided future will start running in the background immediately
    /// when `spawn` is called, even if you don't await the returned
    /// `JoinHandle` (assuming that the runtime [is running][running-runtime]).
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    /// [running-runtime]: index.html#driving-the-runtime
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
    /// ```
Repository: tokio-rs/tokio
File: tokio/src/runtime/runtime.rs (Rust)

Code before:
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// Executes the given future. Depending on the current [`Runtime`],
    /// it may start immediately in the background or require blocking on the [`JoinHandle`].
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
    /// ```
    #[track_caller]

Review comment (suggestion):
```suggestion
    /// Executes the given future. Depending on the current [`Runtime`] configuration,
```

Code after:
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// The provided future will start running in the background immediately
    /// when `spawn` is called, even if you don't await the returned
    /// `JoinHandle` (assuming that the runtime [is running][running-runtime]).
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    /// [running-runtime]: index.html#driving-the-runtime
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
Repository: tokio-rs/tokio
File: tokio/src/runtime/runtime.rs (Rust)

Code before:
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// Executes the given future. Depending on the current [`Runtime`] configuration,
    /// it may start immediately in the background or require blocking on the [`JoinHandle`].
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
    /// ```
    #[track_caller]

Review comment (suggestion):
```suggestion
    /// Executes the given future. Depending on the current [Runtime] configuration,
```

Code after:
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// The provided future will start running in the background immediately
    /// when `spawn` is called, even if you don't await the returned
    /// `JoinHandle` (assuming that the runtime [is running][running-runtime]).
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    /// [running-runtime]: index.html#driving-the-runtime
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
Repository: tokio-rs/tokio
File: tokio/src/runtime/runtime.rs (Rust)

Code before:
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// Executes the given future. Depending on the current [`Runtime`] configuration,
    /// it may start immediately in the background or require blocking on the [`JoinHandle`].
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
    /// ```
    #[track_caller]
    pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>

Review comment (suggestion):
```suggestion
    /// it may start immediately in the background or require blocking on the [JoinHandle].
```

Code after:
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// The provided future will start running in the background immediately
    /// when `spawn` is called, even if you don't await the returned
    /// `JoinHandle` (assuming that the runtime [is running][running-runtime]).
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    /// [running-runtime]: index.html#driving-the-runtime
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
    /// ```
Repository: tokio-rs/tokio
File: tokio/src/runtime/runtime.rs (Rust)

Code before:
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// Executes the given future. Depending on the current [Runtime] configuration,
    /// it may start immediately in the background or require blocking on the [JoinHandle].
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
    /// ```
    #[track_caller]
    pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>

Review comment (bug):
This is wrong!

It never requires blocking on the `JoinHandle`.

As long as `Runtime::block_on()` is called *with any future*, spawned tasks will execute. The `JoinHandle` does not matter. You could even call `rt.block_on(pending())` in a background thread.

Code after:
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// let rt = Runtime::new()
    ///     .unwrap();
    ///
    /// let handle = rt.handle();
    ///
    /// // Use the handle...
    /// # }
    /// ```
    pub fn handle(&self) -> &Handle {
        &self.handle
    }

    /// Spawns a future onto the Tokio runtime.
    ///
    /// This spawns the given future onto the runtime's executor, usually a
    /// thread pool. The thread pool is then responsible for polling the future
    /// until it completes.
    ///
    /// The provided future will start running in the background immediately
    /// when `spawn` is called, even if you don't await the returned
    /// `JoinHandle` (assuming that the runtime [is running][running-runtime]).
    ///
    /// See [module level][mod] documentation for more details.
    ///
    /// [mod]: index.html
    /// [running-runtime]: index.html#driving-the-runtime
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(not(target_family = "wasm"))]
    /// # {
    /// use tokio::runtime::Runtime;
    ///
    /// # fn dox() {
    /// // Create the runtime
    /// let rt = Runtime::new().unwrap();
    ///
    /// // Spawn a future onto the runtime
    /// rt.spawn(async {
    ///     println!("now running on a worker thread");
    /// });
    /// # }
    /// # }
    /// ```
Repository: tokio-rs/tokio
File: tokio/src/signal/windows/sys.rs (Rust)

Code before:
            _ => None,
        }
    }

    fn for_each<'a, F>(&'a self, mut f: F)
    where
        F: FnMut(&'a EventInfo),
    {
        f(&self.ctrl_break);
        f(&self.ctrl_close);
        f(&self.ctrl_c);
        f(&self.ctrl_logoff);
        f(&self.ctrl_shutdown);
    }
}

#[derive(Debug, Default)]
pub(crate) struct OsExtraData {}

fn global_init() -> io::Result<()> {
    static INIT: OnceLock<Result<(), i32>> = OnceLock::new();

    INIT.get_or_init(|| unsafe {
        let rc = console::SetConsoleCtrlHandler(Some(handler), 1);
        if rc == 0 {
            Err(io::Error::last_os_error().raw_os_error().unwrap())
        } else {
            Ok(())
        }
    })
    .map_err(io::Error::from_raw_os_error)
}

unsafe extern "system" fn handler(ty: u32) -> BOOL {
    let globals = globals();
    globals.record_event(ty as EventId);

    // According to https://docs.microsoft.com/en-us/windows/console/handlerroutine
    // the handler routine is always invoked in a new thread, thus we don't
    // have the same restrictions as in Unix signal handlers, meaning we can
    // go ahead and perform the broadcast here.
    let event_was_handled = globals.broadcast();

    if event_was_handled && event_requires_infinite_sleep_in_handler(ty) {
        loop {
            std::thread::park();
        }
    }

    if event_was_handled {
        1

Review comment (suggestion):
Same comment as on the unix implementation, I'd rather avoid the `.unwrap` here if possible

Code after:
            _ => None,
        }
    }

    fn for_each<'a, F>(&'a self, mut f: F)
    where
        F: FnMut(&'a EventInfo),
    {
        f(&self.ctrl_break);
        f(&self.ctrl_close);
        f(&self.ctrl_c);
        f(&self.ctrl_logoff);
        f(&self.ctrl_shutdown);
    }
}

#[derive(Debug, Default)]
pub(crate) struct OsExtraData {}

fn global_init() -> io::Result<()> {
    static INIT: OnceLock<Result<(), Option<i32>>> = OnceLock::new();

    INIT.get_or_init(|| {
        let rc = unsafe { console::SetConsoleCtrlHandler(Some(handler), 1) };
        if rc == 0 {
            Err(io::Error::last_os_error().raw_os_error())
        } else {
            Ok(())
        }
    })
    .map_err(|e| {
        e.map_or_else(
            || io::Error::new(io::ErrorKind::Other, "registering signal handler failed"),
            io::Error::from_raw_os_error,
        )
    })
}

unsafe extern "system" fn handler(ty: u32) -> BOOL {
    let globals = globals();
    globals.record_event(ty as EventId);

    // According to https://docs.microsoft.com/en-us/windows/console/handlerroutine
    // the handler routine is always invoked in a new thread, thus we don't
    // have the same restrictions as in Unix signal handlers, meaning we can
    // go ahead and perform the broadcast here.
    let event_was_handled = globals.broadcast();

    if event_was_handled && event_requires_infinite_sleep_in_handler(ty) {
        loop {
            std::thread::park();
Repository: tokio-rs/tokio
File: tokio/tests/rt_basic.rs (Rust)

Code before:
fn before_park_yields() {
    use futures::task::ArcWake;
    use std::sync::Arc;
    use tokio::runtime::Builder;
    use tokio::sync::Notify;

    struct MyWaker(Notify);

    impl ArcWake for MyWaker {
        fn wake_by_ref(arc_self: &Arc<Self>) {
            arc_self.0.notify_one();
        }
    }

    let notify = Arc::new(MyWaker(Notify::new()));
    let notify2 = notify.clone();
    let waker = futures::task::waker(notify2);
    let woken = Arc::new(AtomicBool::new(false));
    let woken2 = woken.clone();

    let rt = Builder::new_current_thread()
        .enable_all()
        .on_thread_park(move || {
            if !woken2.swap(true, Ordering::SeqCst) {
                let mut cx = Context::from_waker(&waker);
                let mut fut = std::pin::pin!(tokio::task::yield_now());
                let _ = Pin::new(&mut fut).poll(&mut cx);
            }
        })
        .build()
        .unwrap();

    rt.block_on(async {
        notify.0.notified().await;
    });

    assert!(woken.load(Ordering::SeqCst));
}


Review comment (suggestion):
```suggestion
                // `yield_now` pushes the waker to the defer slot.
                let mut fut = std::pin::pin!(tokio::task::yield_now());
```

Code after:
fn before_park_yields() {
    use futures::task::ArcWake;
    use std::sync::Arc;
    use tokio::runtime::Builder;
    use tokio::sync::Notify;

    struct MyWaker(Notify);

    impl ArcWake for MyWaker {
        fn wake_by_ref(arc_self: &Arc<Self>) {
            arc_self.0.notify_one();
        }
    }

    let notify = Arc::new(MyWaker(Notify::new()));
    let notify2 = notify.clone();
    let waker = futures::task::waker(notify2);
    let woken = Arc::new(AtomicBool::new(false));
    let woken2 = woken.clone();

    let rt = Builder::new_current_thread()
        .enable_all()
        .on_thread_park(move || {
            if !woken2.swap(true, Ordering::SeqCst) {
                let mut cx = Context::from_waker(&waker);
                // `yield_now` pushes the waker to the defer slot.
                let fut = std::pin::pin!(tokio::task::yield_now());
                let _ = fut.poll(&mut cx);
            }
        })
        .build()
        .unwrap();

    rt.block_on(async {
        notify.0.notified().await;
    });

    assert!(woken.load(Ordering::SeqCst));
}
Repository: tokio-rs/tokio
File: tokio-stream/tests/stream_fuse.rs (Rust)

Code before:
#[tokio::test]
async fn basic_usage() {
    let mut stream = Alternate { state: 0 };

    // the stream goes back and forth
    assert_eq!(stream.next().await, Some(0));
    assert_eq!(stream.next().await, None);
    assert_eq!(stream.next().await, Some(2));
    assert_eq!(stream.next().await, None);

    // however, once it is fused
    let mut stream = stream.fuse();

    assert_eq!(stream.size_hint(), (0, None));
    assert_eq!(stream.next().await, Some(4));

    assert_eq!(stream.size_hint(), (0, None));
    assert_eq!(stream.next().await, None);

    // it will always return `None` after the first time.
    assert_eq!(stream.size_hint(), (0, Some(0)));
    assert_eq!(stream.next().await, None);
    assert_eq!(stream.size_hint(), (0, Some(0)));
}

#[tokio::test]
async fn interval_stream_is_never_terminated() {
    use futures_core::stream::FusedStream;
    use tokio_stream::wrappers::IntervalStream;

    let interval = tokio::time::interval(std::time::Duration::from_millis(1));
    let stream = IntervalStream::new(interval);

    assert!(!stream.is_terminated());
}


Review comment (suggestion):
```suggestion
#[tokio::test]
#[cfg(feature = "time")]
```

IntervalStream is available only if the `time` feature is enabled
https://github.com/robcholz/tokio/blob/7a2135f426ab560a9efb320620f80e1935c43d31/tokio-stream/src/wrappers.rs#L36-L39

Code after:
#[tokio::test]
async fn basic_usage() {
    let mut stream = Alternate { state: 0 };

    // the stream goes back and forth
    assert_eq!(stream.next().await, Some(0));
    assert_eq!(stream.next().await, None);
    assert_eq!(stream.next().await, Some(2));
    assert_eq!(stream.next().await, None);

    // however, once it is fused
    let mut stream = stream.fuse();

    assert_eq!(stream.size_hint(), (0, None));
    assert_eq!(stream.next().await, Some(4));

    assert_eq!(stream.size_hint(), (0, None));
    assert_eq!(stream.next().await, None);

    // it will always return `None` after the first time.
    assert_eq!(stream.size_hint(), (0, Some(0)));
    assert_eq!(stream.next().await, None);
    assert_eq!(stream.size_hint(), (0, Some(0)));
}

#[tokio::test]
#[cfg(feature = "time")]
async fn interval_stream_is_never_terminated() {
    use futures_core::stream::FusedStream;
    use tokio_stream::wrappers::IntervalStream;

    let interval = tokio::time::interval(std::time::Duration::from_millis(1));
    let stream = IntervalStream::new(interval);

    assert!(!stream.is_terminated());
}