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

Code before:
#[cfg(feature = "io")]

use std::io::{Read, Write};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

/// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or
/// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
#[derive(Debug)]
pub struct SyncIoBridge<T> {
    src: T,
    rt: tokio::runtime::Handle,
}

impl<T: AsyncRead + Unpin> Read for SyncIoBridge<T> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let src = &mut self.src;
        self.rt.block_on(AsyncReadExt::read(src, buf))
    }

    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> std::io::Result<usize> {
        let src = &mut self.src;
        self.rt.block_on(src.read_to_end(buf))
    }

    fn read_to_string(&mut self, buf: &mut String) -> std::io::Result<usize> {
        let src = &mut self.src;

Review comment (suggestion):
This is unnecessary because the entire `io` module is not compiled without the feature.

Code after:
use std::io::{Read, Write};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

/// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or
/// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
#[derive(Debug)]
pub struct SyncIoBridge<T> {
    src: T,
    rt: tokio::runtime::Handle,
}

impl<T: AsyncRead + Unpin> Read for SyncIoBridge<T> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let src = &mut self.src;
        self.rt.block_on(AsyncReadExt::read(src, buf))
    }

    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> std::io::Result<usize> {
        let src = &mut self.src;
        self.rt.block_on(src.read_to_end(buf))
    }

    fn read_to_string(&mut self, buf: &mut String) -> std::io::Result<usize> {
        let src = &mut self.src;
        self.rt.block_on(src.read_to_string(buf))
    }
Repository: tokio-rs/tokio
File: tokio/src/sync/oneshot.rs (Rust)

Code before:
    impl std::error::Error for RecvError {}

    // ===== impl TryRecvError =====

    impl fmt::Display for TryRecvError {
        fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
            match self {
                TryRecvError::Empty => write!(fmt, "channel empty"),
                TryRecvError::Closed => write!(fmt, "channel closed"),
            }
        }
    }

    impl std::error::Error for TryRecvError {}
}

use self::error::*;

struct Inner<T> {
    /// Manages the state of the inner cell
    state: AtomicUsize,

    /// The value. This is set by `Sender` and read by `Receiver`. The state of
    /// the cell is tracked by `state`.
    value: UnsafeCell<Option<T>>,

    /// The task to notify when the receiver drops without consuming the value.
    tx_task: Task,

    /// The task to notify when the value is sent.
    rx_task: Task,
}

struct Task(UnsafeCell<MaybeUninit<Waker>>);

impl Task {
    unsafe fn will_wake(&self, cx: &mut Context<'_>) -> bool {
        self.with_task(|w| w.will_wake(cx.waker()))
    }

    unsafe fn with_task<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&Waker) -> R,
    {
        self.0.with(|ptr| {
            let waker: *const Waker = (&*ptr).as_ptr();
            f(&*waker)
        })
    }

    unsafe fn drop_task(&self) {

Review comment (suggestion):
Can I ask why it's recommended to wrap `Sender` in an `Option` here? I can't seem to find a rationale!

Code after:
/// }
/// ```
///
/// If the sender is dropped without sending, the receiver will fail with
/// [`error::RecvError`]:
///
/// ```
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
///     let (tx, rx) = oneshot::channel::<u32>();
///
///     tokio::spawn(async move {
///         drop(tx);
///     });
///
///     match rx.await {
///         Ok(_) => panic!("This doesn't happen"),
///         Err(_) => println!("the sender dropped"),
///     }
/// }
/// ```
///
/// To use a `Sender` from a destructor, put it in an [`Option`] and call
/// [`Option::take`].
///
/// ```
/// use tokio::sync::oneshot;
///
/// struct SendOnDrop {
///     sender: Option<oneshot::Sender<&'static str>>,
/// }
/// impl Drop for SendOnDrop {
///     fn drop(&mut self) {
///         if let Some(sender) = self.sender.take() {
///             // Using `let _ =` to ignore send errors.
///             let _ = sender.send("I got dropped!");
///         }
///     }
/// }
///
/// #[tokio::main]
/// # async fn _doc() {}
/// # #[tokio::main(flavor = "current_thread")]
/// async fn main() {
///     let (send, recv) = oneshot::channel();
///
///     let send_on_drop = SendOnDrop { sender: Some(send) };
///     drop(send_on_drop);
///
Repository: tokio-rs/tokio
File: tokio-util/tests/framed_stream.rs (Rust)

Code before:
macro_rules! assert_read {
    ($e:expr, $n:expr) => {{
        let val = assert_ready!($e);
        assert_eq!(val.unwrap().unwrap(), $n);
    }};
}

#[tokio::test]
async fn return_none_after_error() {
    let mut io = FramedRead::new(
        Builder::new()
            .read(b"abcdef")
            .read_error(io::Error::new(io::ErrorKind::Other, "Resource errored out"))
            .read(b"more data")
            .build(),
        BytesCodec::new(),
    );

    let mut task = task::spawn(());

    task.enter(|cx, _| {
        assert_read!(pin!(io).poll_next(cx), b"abcdef".to_vec());
        let val = assert_ready!(pin!(io).poll_next(cx));
        assert!(val.unwrap().is_err());
        let val = assert_ready!(pin!(io).poll_next(cx));
        assert!(val.is_none());
        assert_read!(pin!(io).poll_next(cx), b"more data".to_vec());
    })
}


Review comment (suggestion):
I think the test reads a bit easier this way.
```suggestion
        assert!(assert_ready!(pin!(io).poll_next(cx)).unwrap().is_err());
        assert!(assert_ready!(pin!(io).poll_next(cx)).is_none());
```

Code after:
macro_rules! assert_read {
    ($e:expr, $n:expr) => {{
        let val = assert_ready!($e);
        assert_eq!(val.unwrap().unwrap(), $n);
    }};
}

#[tokio::test]
async fn return_none_after_error() {
    let mut io = FramedRead::new(
        Builder::new()
            .read(b"abcdef")
            .read_error(io::Error::new(io::ErrorKind::Other, "Resource errored out"))
            .read(b"more data")
            .build(),
        BytesCodec::new(),
    );

    let mut task = task::spawn(());

    task.enter(|cx, _| {
        assert_read!(pin!(io).poll_next(cx), b"abcdef".to_vec());
        assert!(assert_ready!(pin!(io).poll_next(cx)).unwrap().is_err());
        assert!(assert_ready!(pin!(io).poll_next(cx)).is_none());
        assert_read!(pin!(io).poll_next(cx), b"more data".to_vec());
    })
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/builder.rs (Rust)

Code before:
    }

    #[cfg(any(not(feature = "rt-multi-thread"), tokio_wasi))]
    fn default_worker_threads(_: Kind) -> Option<usize> {
        None
    }

    #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
    fn default_worker_threads(kind: Kind) -> Option<usize> {
        match kind {
            // Always return None if using current thread
            Kind::CurrentThread => return None,
            Kind::MultiThread => {}
        };
        match std::env::var(ENV_WORKER_THREADS) {
            Ok(s) => {
                let n: usize = s.parse().unwrap_or_else(|e| {
                    panic!(
                        "{} must be usize, error: {}, value: {}",
                        ENV_WORKER_THREADS, e, s
                    )
                });
                assert!(n > 0, "{} cannot be set to 0", ENV_WORKER_THREADS);
                Some(n)
            }
            Err(_) => None,
        }
    }
}

cfg_io_driver! {
    impl Builder {
        /// Enables the I/O driver.
        ///
        /// Doing this enables using net, process, signal, and some I/O types on
        /// the runtime.
        ///
        /// # Examples
        ///
        /// ```
        /// use tokio::runtime;
        ///
        /// let rt = runtime::Builder::new_multi_thread()
        ///     .enable_io()
        ///     .build()
        ///     .unwrap();
        /// ```
        pub fn enable_io(&mut self) -> &mut Self {
            self.enable_io = true;
            self
        }

Review comment (suggestion):
It seems natural to move the call to `num_cpus` here.

Code after:
}

cfg_io_driver! {
    impl Builder {
        /// Enables the I/O driver.
        ///
        /// Doing this enables using net, process, signal, and some I/O types on
        /// the runtime.
        ///
        /// # Examples
        ///
        /// ```
        /// use tokio::runtime;
        ///
        /// let rt = runtime::Builder::new_multi_thread()
        ///     .enable_io()
        ///     .build()
        ///     .unwrap();
        /// ```
        pub fn enable_io(&mut self) -> &mut Self {
            self.enable_io = true;
            self
        }

        /// Enables the I/O driver and configures the max number of events to be
        /// processed per tick.
        ///
        /// # Examples
        ///
        /// ```
        /// use tokio::runtime;
        ///
        /// let rt = runtime::Builder::new_current_thread()
        ///     .enable_io()
        ///     .max_io_events_per_tick(1024)
        ///     .build()
        ///     .unwrap();
        /// ```
        pub fn max_io_events_per_tick(&mut self, capacity: usize) -> &mut Self {
            self.nevents = capacity;
            self
        }
    }
}

cfg_time! {
    impl Builder {
        /// Enables the time driver.
        ///
        /// Doing this enables using `tokio::time` on the runtime.
        ///
Repository: tokio-rs/tokio
File: tokio/src/runtime/builder.rs (Rust)

Code before:
        Ok(Runtime::from_parts(
            Scheduler::CurrentThread(scheduler),
            handle,
            blocking_pool,
        ))
    }

    #[cfg(any(not(feature = "rt-multi-thread"), tokio_wasi))]
    fn default_worker_threads(_: Kind) -> Option<usize> {
        None
    }

    #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
    fn default_worker_threads(kind: Kind) -> Option<usize> {
        match kind {
            // Always return None if using current thread
            Kind::CurrentThread => return None,
            Kind::MultiThread => {}
        };
        match std::env::var(ENV_WORKER_THREADS) {
            Ok(s) => {
                let n: usize = s.parse().unwrap_or_else(|e| {
                    panic!(
                        "{} must be usize, error: {}, value: {}",
                        ENV_WORKER_THREADS, e, s
                    )
                });
                assert!(n > 0, "{} cannot be set to 0", ENV_WORKER_THREADS);
                Some(n)
            }
            Err(_) => None,
        }
    }
}

cfg_io_driver! {
    impl Builder {
        /// Enables the I/O driver.
        ///
        /// Doing this enables using net, process, signal, and some I/O types on
        /// the runtime.
        ///
        /// # Examples
        ///
        /// ```
        /// use tokio::runtime;
        ///
        /// let rt = runtime::Builder::new_multi_thread()
        ///     .enable_io()
        ///     .build()
        ///     .unwrap();

Review comment (suggestion):
Currently we panic only if the string is valid utf-8 but not an integer, since invalid unicode hits the `Err(_)` path.

Code after:
            Scheduler::CurrentThread(scheduler),
            handle,
            blocking_pool,
        ))
    }
}

cfg_io_driver! {
    impl Builder {
        /// Enables the I/O driver.
        ///
        /// Doing this enables using net, process, signal, and some I/O types on
        /// the runtime.
        ///
        /// # Examples
        ///
        /// ```
        /// use tokio::runtime;
        ///
        /// let rt = runtime::Builder::new_multi_thread()
        ///     .enable_io()
        ///     .build()
        ///     .unwrap();
        /// ```
        pub fn enable_io(&mut self) -> &mut Self {
            self.enable_io = true;
            self
        }

        /// Enables the I/O driver and configures the max number of events to be
        /// processed per tick.
        ///
        /// # Examples
        ///
        /// ```
        /// use tokio::runtime;
        ///
        /// let rt = runtime::Builder::new_current_thread()
        ///     .enable_io()
        ///     .max_io_events_per_tick(1024)
        ///     .build()
        ///     .unwrap();
        /// ```
        pub fn max_io_events_per_tick(&mut self, capacity: usize) -> &mut Self {
            self.nevents = capacity;
            self
        }
    }
}

cfg_time! {
Repository: tokio-rs/tokio
File: tokio/src/sync/watch.rs (Rust)

Code before:
    /// This can be useful for reusing the buffers inside a watched value.
    /// Additionally, this method permits sending values even when there are no
    /// receivers.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::watch;
    ///
    /// let (tx, _rx) = watch::channel(1);
    /// assert_eq!(tx.send_replace(2), 1);
    /// assert_eq!(tx.send_replace(3), 2);
    /// ```
    pub fn send_replace(&self, mut value: T) -> T {
        // swap old watched value with the new one
        self.send_modify(|old| mem::swap(old, &mut value));

        value
    }

    /// Returns a reference to the most recently sent value
    ///
    /// Outstanding borrows hold a read lock. This means that long lived borrows
    /// could cause the send half to block. It is recommended to keep the borrow
    /// as short lived as possible.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::watch;
    ///
    /// let (tx, _) = watch::channel("hello");
    /// assert_eq!(*tx.borrow(), "hello");
    /// ```
    pub fn borrow(&self) -> Ref<'_, T> {
        let inner = self.shared.value.read().unwrap();
        Ref { inner }
    }

    /// Checks if the channel has been closed. This happens when all receivers
    /// have dropped.
    ///
    /// # Examples
    ///
    /// ```
    /// let (tx, rx) = tokio::sync::watch::channel(());
    /// assert!(!tx.is_closed());
    ///
    /// drop(rx);
    /// assert!(tx.is_closed());
    /// ```

Review comment (suggestion):
The `send` method currently uses `send_replace`. Let's change it to call `send_modify` directly.

Code after:

        self.shared.notify_rx.notify_waiters();
    }

    /// Sends a new value via the channel, notifying all receivers and returning
    /// the previous value in the channel.
    ///
    /// This can be useful for reusing the buffers inside a watched value.
    /// Additionally, this method permits sending values even when there are no
    /// receivers.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::watch;
    ///
    /// let (tx, _rx) = watch::channel(1);
    /// assert_eq!(tx.send_replace(2), 1);
    /// assert_eq!(tx.send_replace(3), 2);
    /// ```
    pub fn send_replace(&self, mut value: T) -> T {
        // swap old watched value with the new one
        self.send_modify(|old| mem::swap(old, &mut value));

        value
    }

    /// Returns a reference to the most recently sent value
    ///
    /// Outstanding borrows hold a read lock. This means that long lived borrows
    /// could cause the send half to block. It is recommended to keep the borrow
    /// as short lived as possible.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::watch;
    ///
    /// let (tx, _) = watch::channel("hello");
    /// assert_eq!(*tx.borrow(), "hello");
    /// ```
    pub fn borrow(&self) -> Ref<'_, T> {
        let inner = self.shared.value.read().unwrap();
        Ref { inner }
    }

    /// Checks if the channel has been closed. This happens when all receivers
    /// have dropped.
    ///
    /// # Examples
    ///
Repository: tokio-rs/tokio
File: tokio/src/blocking.rs (Rust)

Code before:
        pub(crate) use crate::runtime::spawn_mandatory_blocking;
    }

    pub(crate) use crate::task::JoinHandle;
}

cfg_not_rt! {
    use std::fmt;
    use std::future::Future;
    use std::pin::Pin;
    use std::task::{Context, Poll};

    pub(crate) fn spawn_blocking<F, R>(_f: F) -> JoinHandle<R>
    where
        F: FnOnce() -> R + Send + 'static,
        R: Send + 'static,
    {
        assert_send_sync::<JoinHandle<std::cell::Cell<()>>>();
        panic!("requires the `rt` Tokio feature flag")
    }

    cfg_fs! {
        pub(crate) fn spawn_mandatory_blocking<F, R>(_f: F) -> JoinHandle<R>
        where
            F: FnOnce() -> R + Send + 'static,
        R: Send + 'static,
        {
            panic!("requires the `rt` Tokio feature flag")
        }
    }

    pub(crate) struct JoinHandle<R> {
        _p: std::marker::PhantomData<R>,
    }

    unsafe impl<T: Send> Send for JoinHandle<T> {}
    unsafe impl<T: Send> Sync for JoinHandle<T> {}

    impl<R> Future for JoinHandle<R> {
        type Output = Result<R, std::io::Error>;

        fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
            unreachable!()
        }
    }

    impl<T> fmt::Debug for JoinHandle<T>
    where
        T: fmt::Debug,
    {
        fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {

Review comment (suggestion):
Rustfmt doesn't look inside macro invocations.
```suggestion
            F: FnOnce() -> R + Send + 'static,
            R: Send + 'static,
```

Code after:
        pub(crate) use crate::runtime::spawn_mandatory_blocking;
    }

    pub(crate) use crate::task::JoinHandle;
}

cfg_not_rt! {
    use std::fmt;
    use std::future::Future;
    use std::pin::Pin;
    use std::task::{Context, Poll};

    pub(crate) fn spawn_blocking<F, R>(_f: F) -> JoinHandle<R>
    where
        F: FnOnce() -> R + Send + 'static,
        R: Send + 'static,
    {
        assert_send_sync::<JoinHandle<std::cell::Cell<()>>>();
        panic!("requires the `rt` Tokio feature flag")
    }

    cfg_fs! {
        pub(crate) fn spawn_mandatory_blocking<F, R>(_f: F) -> Option<JoinHandle<R>>
        where
            F: FnOnce() -> R + Send + 'static,
            R: Send + 'static,
        {
            panic!("requires the `rt` Tokio feature flag")
        }
    }

    pub(crate) struct JoinHandle<R> {
        _p: std::marker::PhantomData<R>,
    }

    unsafe impl<T: Send> Send for JoinHandle<T> {}
    unsafe impl<T: Send> Sync for JoinHandle<T> {}

    impl<R> Future for JoinHandle<R> {
        type Output = Result<R, std::io::Error>;

        fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
            unreachable!()
        }
    }

    impl<T> fmt::Debug for JoinHandle<T>
    where
        T: fmt::Debug,
    {
        fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
Repository: tokio-rs/tokio
File: tokio/src/sync/oneshot.rs (Rust)

Code before:
    /// # Panics
    ///
    /// This function panics if called within an asynchronous execution
    /// context.
    ///
    /// # Examples
    ///
    /// ```
    /// use std::thread;
    /// use tokio::sync::oneshot;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, rx) = oneshot::channel::<u8>();
    ///
    ///     let sync_code = thread::spawn(move || {
    ///         assert_eq!(Some(10), rx.blocking_recv());
    ///     });
    ///
    ///     let _ = tx.send(10);
    ///     sync_code.join().unwrap();
    /// }
    /// ```
    pub fn blocking_recv(self) -> Option<T> {
        crate::future::block_on(self).ok()
    }
}

impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
        if let Some(inner) = self.inner.as_ref() {
            inner.close();
            #[cfg(all(tokio_unstable, feature = "tracing"))]
            self.resource_span.in_scope(|| {
                tracing::trace!(
                target: "runtime::resource::state_update",
                rx_dropped = true,
                rx_dropped.op = "override",
                )
            });
        }
    }
}

impl<T> Future for Receiver<T> {
    type Output = Result<T, RecvError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // If `inner` is `None`, then `poll()` has already completed.
        #[cfg(all(tokio_unstable, feature = "tracing"))]
        let _res_span = self.resource_span.clone().entered();

Review comment (suggestion):
This should be using the `RecvError` like `.await` does for consistency.

Code after:
    /// # Panics
    ///
    /// This function panics if called within an asynchronous execution
    /// context.
    ///
    /// # Examples
    ///
    /// ```
    /// use std::thread;
    /// use tokio::sync::oneshot;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, rx) = oneshot::channel::<u8>();
    ///
    ///     let sync_code = thread::spawn(move || {
    ///         assert_eq!(Ok(10), rx.blocking_recv());
    ///     });
    ///
    ///     let _ = tx.send(10);
    ///     sync_code.join().unwrap();
    /// }
    /// ```
    #[cfg(feature = "sync")]
    pub fn blocking_recv(self) -> Result<T, RecvError> {
        crate::future::block_on(self)
    }
}

impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
        if let Some(inner) = self.inner.as_ref() {
            inner.close();
            #[cfg(all(tokio_unstable, feature = "tracing"))]
            self.resource_span.in_scope(|| {
                tracing::trace!(
                target: "runtime::resource::state_update",
                rx_dropped = true,
                rx_dropped.op = "override",
                )
            });
        }
    }
}

impl<T> Future for Receiver<T> {
    type Output = Result<T, RecvError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // If `inner` is `None`, then `poll()` has already completed.
        #[cfg(all(tokio_unstable, feature = "tracing"))]