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

Code before:
use crate::io::sys;
use crate::io::{AsyncRead, AsyncWrite, ReadBuf};

use std::cmp;
use std::future::Future;
use std::io;
use std::io::prelude::*;
use std::mem::MaybeUninit;
use std::pin::Pin;
use std::task::{ready, Context, Poll};

/// `T` should not implement _both_ Read and Write.
#[derive(Debug)]
pub(crate) struct Blocking<T> {
    inner: Option<T>,
    state: State<T>,
    /// `true` if the lower IO layer needs flushing.
    need_flush: bool,
}

#[derive(Debug)]
pub(crate) struct Buf {
    buf: Vec<u8>,
    pos: usize,
    capacity: usize,
}

pub(crate) const DEFAULT_MAX_BUF_SIZE: usize = 2 * 1024 * 1024;

#[derive(Debug)]
enum State<T> {
    Idle(Option<Buf>),
    Busy(sys::Blocking<(io::Result<usize>, Buf, T)>),
}

cfg_io_blocking! {
    impl<T> Blocking<T> {
        #[cfg_attr(feature = "fs", allow(dead_code))]
        pub(crate) fn new(inner: T) -> Blocking<T> {
            Blocking {
                inner: Some(inner),
                state: State::Idle(Some(Buf::with_capacity(0))),
                need_flush: false,
            }
        }
    }
}

impl<T> AsyncRead for Blocking<T>
where

Review comment (suggestion):
This doesn't look too good but currently the implementation expects `max_buf_size` to always be followed, while `Vec` may sometimes allocate more space or future calls to `ensure_capacity_for` may request smaller capacities than what had already been previously allocated, so this maintains that property.

Code after:
use crate::io::sys;
use crate::io::{AsyncRead, AsyncWrite, ReadBuf};

use std::cmp;
use std::future::Future;
use std::io;
use std::io::prelude::*;
use std::mem::MaybeUninit;
use std::pin::Pin;
use std::task::{ready, Context, Poll};

/// `T` should not implement _both_ Read and Write.
#[derive(Debug)]
pub(crate) struct Blocking<T> {
    inner: Option<T>,
    state: State<T>,
    /// `true` if the lower IO layer needs flushing.
    need_flush: bool,
}

#[derive(Debug)]
pub(crate) struct Buf {
    buf: Vec<u8>,
    pos: usize,
}

pub(crate) const DEFAULT_MAX_BUF_SIZE: usize = 2 * 1024 * 1024;

#[derive(Debug)]
enum State<T> {
    Idle(Option<Buf>),
    Busy(sys::Blocking<(io::Result<usize>, Buf, T)>),
}

cfg_io_blocking! {
    impl<T> Blocking<T> {
        /// # Safety
        ///
        /// The `Read` implementation of `inner` must never read from the buffer
        /// it is borrowing and must correctly report the length of the data
        /// written into the buffer.
        #[cfg_attr(feature = "fs", allow(dead_code))]
        pub(crate) unsafe fn new(inner: T) -> Blocking<T> {
            Blocking {
                inner: Some(inner),
                state: State::Idle(Some(Buf::with_capacity(0))),
                need_flush: false,
            }
        }
    }
Repository: tokio-rs/tokio
File: tokio-util/src/lib.rs (Rust)

Code before:
    #[macro_use]
    mod tracing;

    pub mod codec;
}

cfg_net! {
    #[cfg(not(target_arch = "wasm32"))]
    pub mod udp;
    pub mod net;
}

cfg_compat! {
    pub mod compat;
}

cfg_io! {
    pub mod io;
}

cfg_rt! {
    pub mod context;
}

#[cfg(any(feature = "rt", feature = "join-map"))]
#[cfg_attr(docsrs,doc(cfg(any(feature = "rt", feature = "join-map"))))]
pub mod task;

cfg_time! {
    pub mod time;
}

pub mod sync;

pub mod either;

pub use bytes;

mod util;


Review comment (suggestion):
```suggestion
#[cfg_attr(docsrs, doc(cfg(any(feature = "rt", feature = "join-map"))))]
```

Code after:

    pub mod codec;
}

cfg_net! {
    #[cfg(not(target_arch = "wasm32"))]
    pub mod udp;
    pub mod net;
}

cfg_compat! {
    pub mod compat;
}

cfg_io! {
    pub mod io;
}

cfg_rt! {
    pub mod context;
}

#[cfg(feature = "rt")]
pub mod task;

cfg_time! {
    pub mod time;
}

pub mod sync;

pub mod either;

pub use bytes;

mod util;

pub mod future;
Repository: tokio-rs/tokio
File: tokio-util/tests/sync_cancellation_token.rs (Rust)

Code before:
    {
        let (tx, rx) = oneshot::channel::<()>();

        let token = CancellationToken::new();
        let fut = token.run_until_cancelled(async move {
            rx.await.unwrap();
            42
        });
        pin!(fut);

        assert_eq!(
            Poll::Pending,
            fut.as_mut().poll(&mut Context::from_waker(&waker))
        );

        tx.send(()).unwrap();

        assert_eq!(
            Poll::Ready(Some(42)),
            fut.as_mut().poll(&mut Context::from_waker(&waker))
        );
    }
}

#[test]
fn run_until_cancelled_owned_test() {
    let (waker, _) = new_count_waker();

    {
        let token = CancellationToken::new();
        let to_cancel = token.clone();

        let fut = token.run_until_cancelled_owned(std::future::pending::<()>());
        pin!(fut);

        assert_eq!(
            Poll::Pending,
            fut.as_mut().poll(&mut Context::from_waker(&waker))
        );

        to_cancel.cancel();

        assert_eq!(
            Poll::Ready(None),
            fut.as_mut().poll(&mut Context::from_waker(&waker))
        );
    }

    {
        let (tx, rx) = oneshot::channel::<()>();


Review comment (suggestion):
I think it would be nice to add a case to this test that would only compile with the `run_until_cancelled_owned` and not with `run_until_cancelled` like you mentioned in the PR description as this was the reason behind this PR.

Code after:
    {
        let (tx, rx) = oneshot::channel::<()>();

        let token = CancellationToken::new();
        let fut = token.run_until_cancelled(async move {
            rx.await.unwrap();
            42
        });
        pin!(fut);

        assert_eq!(
            Poll::Pending,
            fut.as_mut().poll(&mut Context::from_waker(&waker))
        );

        tx.send(()).unwrap();

        assert_eq!(
            Poll::Ready(Some(42)),
            fut.as_mut().poll(&mut Context::from_waker(&waker))
        );
    }
}

#[test]
fn run_until_cancelled_owned_test() {
    let (waker, _) = new_count_waker();

    {
        let token = CancellationToken::new();
        let to_cancel = token.clone();

        let takes_ownership = move |token: CancellationToken| {
            token.run_until_cancelled_owned(std::future::pending::<()>())
        };

        let fut = takes_ownership(token);
        pin!(fut);

        assert_eq!(
            Poll::Pending,
            fut.as_mut().poll(&mut Context::from_waker(&waker))
        );

        to_cancel.cancel();

        assert_eq!(
            Poll::Ready(None),
            fut.as_mut().poll(&mut Context::from_waker(&waker))
        );
    }
Repository: tokio-rs/tokio
File: tokio/src/process/mod.rs (Rust)

Code before:
    ///
    /// Basic usage:
    ///
    /// ```no_run
    /// use tokio::process::Command;
    ///
    /// async fn run_ls() {
    ///     let output: std::process::Output = Command::new("ls")
    ///         .output()
    ///         .await
    ///         .expect("ls command failed to run");
    ///     println!("stderr of ls: {:?}", output.stderr);
    /// }
    /// ```
    pub fn output(&mut self) -> impl Future<Output = io::Result<Output>> {
        self.std.stdout(Stdio::piped());
        self.std.stderr(Stdio::piped());

        let child = self.spawn();

        async { child?.wait_with_output().await }
    }

    /// Returns the boolean value that was previously set by [`Command::kill_on_drop`].
    ///
    /// Note that if you have not previously called [`Command:: kill_on_drop`], the
    /// default value of `false` will be returned here.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::process::Command;
    ///
    /// let mut cmd = Command::new("echo");
    /// assert!(!cmd.get_kill_on_drop());
    ///
    /// cmd.kill_on_drop(true);
    /// assert!(cmd.get_kill_on_drop());
    /// ```
    pub fn get_kill_on_drop(&self) -> bool {
        self.kill_on_drop
    }
}

impl From<StdCommand> for Command {
    fn from(std: StdCommand) -> Command {
        Command {
            std,
            kill_on_drop: false,
        }
    }

Review comment (suggestion):
```suggestion
    /// Note that if you have not previously called [`Command::kill_on_drop`], the
```

Code after:
    ///
    /// Basic usage:
    ///
    /// ```no_run
    /// use tokio::process::Command;
    ///
    /// async fn run_ls() {
    ///     let output: std::process::Output = Command::new("ls")
    ///         .output()
    ///         .await
    ///         .expect("ls command failed to run");
    ///     println!("stderr of ls: {:?}", output.stderr);
    /// }
    /// ```
    pub fn output(&mut self) -> impl Future<Output = io::Result<Output>> {
        self.std.stdout(Stdio::piped());
        self.std.stderr(Stdio::piped());

        let child = self.spawn();

        async { child?.wait_with_output().await }
    }

    /// Returns the boolean value that was previously set by [`Command::kill_on_drop`].
    ///
    /// Note that if you have not previously called [`Command::kill_on_drop`], the
    /// default value of `false` will be returned here.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::process::Command;
    ///
    /// let mut cmd = Command::new("echo");
    /// assert!(!cmd.get_kill_on_drop());
    ///
    /// cmd.kill_on_drop(true);
    /// assert!(cmd.get_kill_on_drop());
    /// ```
    pub fn get_kill_on_drop(&self) -> bool {
        self.kill_on_drop
    }
}

impl From<StdCommand> for Command {
    fn from(std: StdCommand) -> Command {
        Command {
            std,
            kill_on_drop: false,
        }
    }
Repository: tokio-rs/tokio
File: tokio/src/sync/mpsc/chan.rs (Rust)

Code before:
    }

    pub(super) fn semaphore(&self) -> &S {
        &self.inner.semaphore
    }

    pub(super) fn sender_strong_count(&self) -> usize {
        self.inner.tx_count.load(Acquire)
    }

    pub(super) fn sender_weak_count(&self) -> usize {
        self.inner.tx_weak_count.load(Relaxed)
    }
}

impl<T, S: Semaphore> Drop for Rx<T, S> {
    fn drop(&mut self) {
        use super::block::Read::Value;

        self.close();

        self.inner.rx_fields.with_mut(|rx_fields_ptr| {
            let rx_fields = unsafe { &mut *rx_fields_ptr };
            struct Guard<'a, T, S: Semaphore> {
                list: &'a mut list::Rx<T>,
                tx: &'a CachePadded<crate::sync::mpsc::list::Tx<T>>,
                sem: &'a S,
            }

            impl<'a, T, S: Semaphore> Guard<'a, T, S> {
                fn drain(&mut self) {
                    // call T's destructor.
                    while let Some(Value(_)) = self.list.pop(self.tx) {
                        self.sem.add_permit();
                    }
                }
            }

            impl<'a, T, S: Semaphore> Drop for Guard<'a, T, S> {
                fn drop(&mut self) {
                    self.drain();
                }
            }

            let mut guard = Guard {
                list: &mut rx_fields.list,
                tx: &self.inner.tx,
                sem: &self.inner.semaphore,
            };

            guard.drain();

Review comment (suggestion):
I think you can just do this?
```suggestion
                tx: &'a list::Tx<T>,
```

Code after:
    }

    pub(super) fn semaphore(&self) -> &S {
        &self.inner.semaphore
    }

    pub(super) fn sender_strong_count(&self) -> usize {
        self.inner.tx_count.load(Acquire)
    }

    pub(super) fn sender_weak_count(&self) -> usize {
        self.inner.tx_weak_count.load(Relaxed)
    }
}

impl<T, S: Semaphore> Drop for Rx<T, S> {
    fn drop(&mut self) {
        use super::block::Read::Value;

        self.close();

        self.inner.rx_fields.with_mut(|rx_fields_ptr| {
            let rx_fields = unsafe { &mut *rx_fields_ptr };
            struct Guard<'a, T, S: Semaphore> {
                list: &'a mut list::Rx<T>,
                tx: &'a list::Tx<T>,
                sem: &'a S,
            }

            impl<'a, T, S: Semaphore> Guard<'a, T, S> {
                fn drain(&mut self) {
                    // call T's destructor.
                    while let Some(Value(_)) = self.list.pop(self.tx) {
                        self.sem.add_permit();
                    }
                }
            }

            impl<'a, T, S: Semaphore> Drop for Guard<'a, T, S> {
                fn drop(&mut self) {
                    self.drain();
                }
            }

            let mut guard = Guard {
                list: &mut rx_fields.list,
                tx: &self.inner.tx,
                sem: &self.inner.semaphore,
            };

            guard.drain();
Repository: tokio-rs/tokio
File: tokio/src/task/coop/mod.rs (Rust)

Code before:
//! functions, which force tasks to return to the executor periodically.
//!
//!
//! #### unconstrained
//!
//! If necessary, [`task::unconstrained`] lets you opt a future out of Tokio's cooperative
//! scheduling. When a future is wrapped with `unconstrained`, it will never be forced to yield to
//! Tokio. For example:
//!
//! ```
//! # #[tokio::main]
//! # async fn main() {
//! use tokio::{task, sync::mpsc};
//!
//! let fut = async {
//!     let (tx, mut rx) = mpsc::unbounded_channel();
//!
//!     for i in 0..1000 {
//!         let _ = tx.send(());
//!         // This will always be ready. If coop was in effect, this code would be forced to yield
//!         // periodically. However, if left unconstrained, then this code will never yield.
//!         rx.recv().await;
//!     }
//! };
//!
//! task::unconstrained(fut).await;
//! # }
//! ```
//! [`poll`]: method@std::future::Future::poll
//! [`task::unconstrained`]: crate::task::unconstrained()

cfg_rt! {
    mod consume_budget;
    pub use consume_budget::consume_budget;
}

// ```ignore
// # use tokio_stream::{Stream, StreamExt};
// async fn drop_all<I: Stream + Unpin>(mut input: I) {
//     while let Some(_) = input.next().await {
//         tokio::coop::proceed().await;
//     }
// }
// ```
//
// The `proceed` future will coordinate with the executor to make sure that
// every so often control is yielded back to the executor so it can run other
// tasks.
//
// # Placing yield points
//

Review comment (suggestion):
If we are moving this, we should use the new path in examples.
```suggestion
//! task::coop::unconstrained(fut).await;
```

Code after:
//!
//! #### unconstrained
//!
//! If necessary, [`task::unconstrained`] lets you opt a future out of Tokio's cooperative
//! scheduling. When a future is wrapped with `unconstrained`, it will never be forced to yield to
//! Tokio. For example:
//!
//! ```
//! # #[tokio::main]
//! # async fn main() {
//! use tokio::{task, sync::mpsc};
//!
//! let fut = async {
//!     let (tx, mut rx) = mpsc::unbounded_channel();
//!
//!     for i in 0..1000 {
//!         let _ = tx.send(());
//!         // This will always be ready. If coop was in effect, this code would be forced to yield
//!         // periodically. However, if left unconstrained, then this code will never yield.
//!         rx.recv().await;
//!     }
//! };
//!
//! task::coop::unconstrained(fut).await;
//! # }
//! ```
//! [`poll`]: method@std::future::Future::poll
//! [`task::unconstrained`]: crate::task::unconstrained()

cfg_rt! {
    mod consume_budget;
    pub use consume_budget::consume_budget;

    mod unconstrained;
    pub use unconstrained::{unconstrained, Unconstrained};
}

// ```ignore
// # use tokio_stream::{Stream, StreamExt};
// async fn drop_all<I: Stream + Unpin>(mut input: I) {
//     while let Some(_) = input.next().await {
//         tokio::coop::proceed().await;
//     }
// }
// ```
//
// The `proceed` future will coordinate with the executor to make sure that
// every so often control is yielded back to the executor so it can run other
// tasks.
//
// # Placing yield points
Repository: tokio-rs/tokio
File: tokio/src/task/mod.rs (Rust)

Code before:
//!
//! [`task::spawn_blocking`]: crate::task::spawn_blocking
//! [`task::block_in_place`]: crate::task::block_in_place
//! [rt-multi-thread]: ../runtime/index.html#threaded-scheduler
//! [`task::yield_now`]: crate::task::yield_now()
//! [`thread::yield_now`]: std::thread::yield_now

cfg_rt! {
    pub use crate::runtime::task::{JoinError, JoinHandle};

    mod blocking;
    pub use blocking::spawn_blocking;

    mod spawn;
    pub use spawn::spawn;

    cfg_rt_multi_thread! {
        pub use blocking::block_in_place;
    }

    mod yield_now;
    pub use yield_now::yield_now;

    pub mod coop;
    #[doc(hidden)]
    #[deprecated]
    pub use coop::consume_budget;

    mod local;
    pub use local::{spawn_local, LocalSet, LocalEnterGuard};

    mod task_local;
    pub use task_local::LocalKey;

    mod unconstrained;
    pub use unconstrained::{unconstrained, Unconstrained};

    #[doc(inline)]
    pub use join_set::JoinSet;
    pub use crate::runtime::task::AbortHandle;

    // Uses #[cfg(...)] instead of macro since the macro adds docsrs annotations.
    #[cfg(not(tokio_unstable))]
    mod join_set;
    #[cfg(tokio_unstable)]
    pub mod join_set;

    pub use crate::runtime::task::{Id, id, try_id};

    cfg_trace! {
        mod builder;

Review comment (suggestion):
```suggestion
    #[deprecated = "Moved to tokio::task::coop::consume_budget"]
```

Code after:
//!
//! [`task::spawn_blocking`]: crate::task::spawn_blocking
//! [`task::block_in_place`]: crate::task::block_in_place
//! [rt-multi-thread]: ../runtime/index.html#threaded-scheduler
//! [`task::yield_now`]: crate::task::yield_now()
//! [`thread::yield_now`]: std::thread::yield_now

cfg_rt! {
    pub use crate::runtime::task::{JoinError, JoinHandle};

    mod blocking;
    pub use blocking::spawn_blocking;

    mod spawn;
    pub use spawn::spawn;

    cfg_rt_multi_thread! {
        pub use blocking::block_in_place;
    }

    mod yield_now;
    pub use yield_now::yield_now;

    pub mod coop;
    #[doc(hidden)]
    #[deprecated = "Moved to tokio::task::coop::consume_budget"]
    pub use coop::consume_budget;
    #[doc(hidden)]
    #[deprecated = "Moved to tokio::task::coop::unconstrained"]
    pub use coop::unconstrained;
    #[doc(hidden)]
    #[deprecated = "Moved to tokio::task::coop::Unconstrained"]
    pub use coop::Unconstrained;

    mod local;
    pub use local::{spawn_local, LocalSet, LocalEnterGuard};

    mod task_local;
    pub use task_local::LocalKey;

    #[doc(inline)]
    pub use join_set::JoinSet;
    pub use crate::runtime::task::AbortHandle;

    // Uses #[cfg(...)] instead of macro since the macro adds docsrs annotations.
    #[cfg(not(tokio_unstable))]
    mod join_set;
    #[cfg(tokio_unstable)]
    pub mod join_set;

    pub use crate::runtime::task::{Id, id, try_id};
Repository: tokio-rs/tokio
File: tokio/src/signal/windows/sys.rs (Rust)

Code before:
        let rc = console::SetConsoleCtrlHandler(Some(handler), 1);
        let ret = if rc == 0 {
            Err(io::Error::last_os_error())
        } else {
            Ok(())
        };

        init = Some(ret);
    });

    init.unwrap_or_else(|| Ok(()))
}

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_requires_infinite_sleep_in_handler(ty) {
        loop {
            std::thread::sleep(std::time::Duration::MAX);
        }
    }

    if event_was_handled {
        1
    } else {
        // No one is listening for this notification any more
        // let the OS fire the next (possibly the default) handler.
        0
    }
}

#[cfg(all(test, not(loom)))]
mod tests {
    use super::*;
    use crate::runtime::Runtime;

    use tokio_test::{assert_ok, assert_pending, assert_ready_ok, task};

    #[test]
    fn ctrl_c() {
        let rt = rt();
        let _enter = rt.enter();

        let mut ctrl_c = task::spawn(crate::signal::ctrl_c());

Review comment (suggestion):
```suggestion
            std::thread::park();
```

Code after:
        let rc = console::SetConsoleCtrlHandler(Some(handler), 1);
        let ret = if rc == 0 {
            Err(io::Error::last_os_error())
        } else {
            Ok(())
        };

        init = Some(ret);
    });

    init.unwrap_or_else(|| Ok(()))
}

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
    } else {
        // No one is listening for this notification any more
        // let the OS fire the next (possibly the default) handler.
        0
    }
}

#[cfg(all(test, not(loom)))]
mod tests {
    use super::*;
    use crate::runtime::Runtime;

    use tokio_test::{assert_ok, assert_pending, assert_ready_ok, task};

    unsafe fn raise_event(signum: u32) {
        if event_requires_infinite_sleep_in_handler(signum) {
            // Those events will enter an infinite loop in `handler`, so
            // we need to run them on a separate thread
            std::thread::spawn(move || super::handler(signum));
        } else {
Repository: tokio-rs/tokio
File: tokio-util/src/io/sync_bridge.rs (Rust)

Code before:

    /// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or
    /// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
    ///
    /// This is the same as [`SyncIoBridge::new`], but allows passing an arbitrary handle and hence may
    /// be initially invoked outside of an asynchronous context.
    pub fn new_with_handle(src: T, rt: tokio::runtime::Handle) -> Self {
        Self { src, rt }
    }

    /// Consume this bridge, returning the underlying stream.
    pub fn into_inner(self) -> T {
        self.src
    }

    /// Returns a shared reference to the inner resource.
    ///
    /// It is inadvisable to directly read/write from the underlying resource.
    pub fn get_ref(&self) -> &T {
        &self.src
    }

    /// Returns a mutable reference to the inner resource.
    ///
    /// It is inadvisable to directly read/write from the underlying resource.
    pub fn get_mut(&mut self) -> &T {
        &mut self.src
    }
}


Review comment (suggestion):
```suggestion
    pub fn get_mut(&mut self) -> &mut T {
```

Code after:

    /// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or
    /// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
    ///
    /// This is the same as [`SyncIoBridge::new`], but allows passing an arbitrary handle and hence may
    /// be initially invoked outside of an asynchronous context.
    pub fn new_with_handle(src: T, rt: tokio::runtime::Handle) -> Self {
        Self { src, rt }
    }

    /// Consume this bridge, returning the underlying stream.
    pub fn into_inner(self) -> T {
        self.src
    }
}

impl<T> AsMut<T> for SyncIoBridge<T> {
    fn as_mut(&mut self) -> &mut T {
        &mut self.src
    }
}

impl<T> AsRef<T> for SyncIoBridge<T> {
    fn as_ref(&self) -> &T {
        &self.src
    }
}