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

Code before:
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consumes this [`SinkWriter`], returning the underlying sink.
    pub fn into_inner(self) -> S {
        self.inner
    }
}
impl<S, E> AsyncWrite for SinkWriter<S>
where
    for<'a> S: Sink<&'a [u8], Error = E>,
    E: Into<io::Error>,
{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, io::Error>> {
        let mut this = self.project();

        ready!(this.inner.as_mut().poll_ready(cx).map_err(Into::into))?;
        Poll::Ready(match this.inner.as_mut().start_send(buf) {
            Ok(()) => Ok(buf.len()),
            Err(e) => Err(e.into()),
        })
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
        self.project().inner.poll_flush(cx).map_err(Into::into)
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
        self.project().inner.poll_close(cx).map_err(Into::into)
    }
}


Review comment (suggestion):
No reason to be fancy here.
```suggestion
        match this.inner.as_mut().start_send(buf) {
            Ok(()) => Poll::Ready(Ok(buf.len())),
            Err(e) => Poll::Ready(Err(e.into())),
        }
```

Code after:
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consumes this [`SinkWriter`], returning the underlying sink.
    pub fn into_inner(self) -> S {
        self.inner
    }
}
impl<S, E> AsyncWrite for SinkWriter<S>
where
    for<'a> S: Sink<&'a [u8], Error = E>,
    E: Into<io::Error>,
{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, io::Error>> {
        let mut this = self.project();

        ready!(this.inner.as_mut().poll_ready(cx).map_err(Into::into))?;
        match this.inner.as_mut().start_send(buf) {
            Ok(()) => Poll::Ready(Ok(buf.len())),
            Err(e) => Poll::Ready(Err(e.into())),
        }
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
        self.project().inner.poll_flush(cx).map_err(Into::into)
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
        self.project().inner.poll_close(cx).map_err(Into::into)
    }
}
Repository: tokio-rs/tokio
File: tokio/src/signal/windows.rs (Rust)

Code before:
//! Windows-specific types for signal handling.
//!
//! This module is only defined on Windows and allows receiving "ctrl-c",
//! "ctrl-break", "ctrl-logoff", "ctrl-shutdown", and "ctrl-close"
//! notifications. These events are listened for via the `SetConsoleCtrlHandler`
//! function which receives the corresponding windows_sys event type.

#![cfg(any(windows, docsrs))]
#![cfg_attr(docsrs, doc(cfg(all(windows, feature = "signal"))))]

use crate::signal::RxFuture;
use std::io;
use std::task::{Context, Poll};

#[cfg(not(docsrs))]
#[path = "windows/sys.rs"]
mod imp;
#[cfg(not(docsrs))]
pub(crate) use self::imp::{OsExtraData, OsStorage};

#[cfg(docsrs)]
#[path = "windows/stub.rs"]
mod imp;

/// Creates a new event which receives "ctrl-c" notifications sent to the
/// process.
///
/// # Examples
///
/// ```rust,no_run
/// use tokio::signal::windows::ctrl_c;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
///     // An infinite stream of CTRL-C events.
///     let mut stream = ctrl_c()?;
///
///     // Print whenever a CTRL-C event is received.
///     for countdown in (0..3).rev() {
///         stream.recv().await;
///         println!("got CTRL-C. {} more to exit", countdown);
///     }
///
///     Ok(())
/// }
/// ```
pub fn ctrl_c() -> io::Result<CtrlC> {
    Ok(CtrlC {
        inner: self::imp::ctrl_c()?,
    })

Review comment (suggestion):
```suggestion
/// Creates a new listener which receives "ctrl-c" notifications sent to the
```

Code after:
//! Windows-specific types for signal handling.
//!
//! This module is only defined on Windows and allows receiving "ctrl-c",
//! "ctrl-break", "ctrl-logoff", "ctrl-shutdown", and "ctrl-close"
//! notifications. These events are listened for via the `SetConsoleCtrlHandler`
//! function which receives the corresponding windows_sys event type.

#![cfg(any(windows, docsrs))]
#![cfg_attr(docsrs, doc(cfg(all(windows, feature = "signal"))))]

use crate::signal::RxFuture;
use std::io;
use std::task::{Context, Poll};

#[cfg(not(docsrs))]
#[path = "windows/sys.rs"]
mod imp;
#[cfg(not(docsrs))]
pub(crate) use self::imp::{OsExtraData, OsStorage};

#[cfg(docsrs)]
#[path = "windows/stub.rs"]
mod imp;

/// Creates a new listener which receives "ctrl-c" notifications sent to the
/// process.
///
/// # Examples
///
/// ```rust,no_run
/// use tokio::signal::windows::ctrl_c;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
///     // A listener of CTRL-C events.
///     let mut signal = ctrl_c()?;
///
///     // Print whenever a CTRL-C event is received.
///     for countdown in (0..3).rev() {
///         signal.recv().await;
///         println!("got CTRL-C. {} more to exit", countdown);
///     }
///
///     Ok(())
/// }
/// ```
pub fn ctrl_c() -> io::Result<CtrlC> {
    Ok(CtrlC {
        inner: self::imp::ctrl_c()?,
    })
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_ext/take.rs (Rust)

Code before:
                    Some(_) => {
                        *self.as_mut().project().remaining -= 1;
                    }
                    None => {
                        *self.as_mut().project().remaining = 0;
                    }
                }
                ready
            })
        } else {
            Poll::Ready(None)
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        if self.remaining == 0 {
            return (0, Some(0));
        }

        let (lower, upper) = self.stream.size_hint();

        let lower = cmp::min(lower, self.remaining);

        let upper = match upper {
            Some(x) if x < self.remaining => Some(x),
            _ => Some(self.remaining),
        };

        (lower, upper)
    }
}


Review comment (suggestion):
You are making a bunch of unrelated changes. If these are clippy warnings, then please submit those fixes in a separate PR.

Code after:
                    Some(_) => {
                        *self.as_mut().project().remaining -= 1;
                    }
                    None => {
                        *self.as_mut().project().remaining = 0;
                    }
                }
                ready
            })
        } else {
            Poll::Ready(None)
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        if self.remaining == 0 {
            return (0, Some(0));
        }

        let (lower, upper) = self.stream.size_hint();

        let lower = cmp::min(lower, self.remaining as usize);

        let upper = match upper {
            Some(x) if x < self.remaining as usize => Some(x),
            _ => Some(self.remaining as usize),
        };

        (lower, upper)
    }
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/task/join.rs (Rust)

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

impl<T> UnwindSafe for JoinHandle<T> {}
impl<T> RefUnwindSafe for JoinHandle<T> {}

impl<T> JoinHandle<T> {
    pub(super) fn new(raw: RawTask) -> JoinHandle<T> {
        JoinHandle {
            raw,
            _p: PhantomData,
        }
    }

    /// Abort the task associated with the handle.
    ///
    /// Awaiting a cancelled task might complete as usual if the task was
    /// already completed at the time it was cancelled, but most likely it
    /// will fail with a [cancelled] `JoinError`.
    ///
    /// ```rust
    /// use tokio::time;
    ///
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn main() {
    /// # tokio::time::pause();
    /// let mut handles = Vec::new();
    ///
    /// handles.push(tokio::spawn(async {
    ///    time::sleep(time::Duration::from_secs(10)).await;
    ///    true
    /// }));
    ///
    /// handles.push(tokio::spawn(async {
    ///    time::sleep(time::Duration::from_secs(10)).await;
    ///    false
    /// }));
    ///
    /// for handle in &handles {
    ///     handle.abort();
    /// }
    ///
    /// for handle in handles {
    ///     assert!(handle.await.unwrap_err().is_cancelled());
    /// }
    /// # }
    /// ```
    /// [cancelled]: method@super::error::JoinError::is_cancelled
    pub fn abort(&self) {
        self.raw.remote_abort();
    }

Review comment (suggestion):
Now that we're hiding the main macro, we can do this instead:
```suggestion
    /// # #[tokio::main(flavor = "current_thread", start_paused = true)]
    /// # async fn main() {
```

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

impl<T> UnwindSafe for JoinHandle<T> {}
impl<T> RefUnwindSafe for JoinHandle<T> {}

impl<T> JoinHandle<T> {
    pub(super) fn new(raw: RawTask) -> JoinHandle<T> {
        JoinHandle {
            raw,
            _p: PhantomData,
        }
    }

    /// Abort the task associated with the handle.
    ///
    /// Awaiting a cancelled task might complete as usual if the task was
    /// already completed at the time it was cancelled, but most likely it
    /// will fail with a [cancelled] `JoinError`.
    ///
    /// ```rust
    /// use tokio::time;
    ///
    /// # #[tokio::main(flavor = "current_thread", start_paused = true)]
    /// # async fn main() {
    /// let mut handles = Vec::new();
    ///
    /// handles.push(tokio::spawn(async {
    ///    time::sleep(time::Duration::from_secs(10)).await;
    ///    true
    /// }));
    ///
    /// handles.push(tokio::spawn(async {
    ///    time::sleep(time::Duration::from_secs(10)).await;
    ///    false
    /// }));
    ///
    /// for handle in &handles {
    ///     handle.abort();
    /// }
    ///
    /// for handle in handles {
    ///     assert!(handle.await.unwrap_err().is_cancelled());
    /// }
    /// # }
    /// ```
    /// [cancelled]: method@super::error::JoinError::is_cancelled
    pub fn abort(&self) {
        self.raw.remote_abort();
    }
Repository: tokio-rs/tokio
File: tokio/tests/time_interval.rs (Rust)

Code before:
                Poll::Pending => {
                    println!(
                        "Timer returned Poll::Pending after {:?}",
                        this.start.elapsed()
                    );
                    Poll::Pending
                }
                Poll::Ready(_) => {
                    println!(
                        "Timer returned Poll::Ready after {:?}",
                        this.start.elapsed()
                    );
                    this.counter += 1;
                    if this.counter % 4 == 0 {
                        Poll::Ready(Some(this.counter))
                    } else {
                        // Schedule this task for wake-up
                        cx.waker().wake_by_ref();
                        Poll::Pending
                    }
                }
            }
        }
    }

    #[tokio::test]
    async fn reset_without_reregister() {
        let stream = IntervalStreamer {
            start: Instant::now(),
            counter: 0,
            timer: crate::time::interval(std::time::Duration::from_millis(10)),
        };

        pin_mut!(stream);

        let mut results = Vec::with_capacity(4);
        while let Some(item) = stream.next().await {
            println!("Stream yielded an item: {}", item);
            results.push(item);
        }

        dbg!(&results);
        assert_eq!(results, vec![4, 8, 12]);
    }
}


Review comment (suggestion):
This should make the test execute much faster by having it instantly skip forward to the next timer whenever a sleep happens.
```suggestion
    #[tokio::test(start_paused = true)]
```

Code after:

        match this.timer.poll_tick(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(_) => {
                this.counter += 1;
                if this.counter % 4 == 0 {
                    Poll::Ready(Some(this.counter))
                } else {
                    if this.wake_on_pending {
                        // Schedule this task for wake-up
                        cx.waker().wake_by_ref();
                    }
                    Poll::Pending
                }
            }
        }
    }
}

#[tokio::test(start_paused = true)]
async fn stream_with_interval_poll_tick_self_waking() {
    let stream = IntervalStreamer {
        counter: 0,
        timer: tokio::time::interval(tokio::time::Duration::from_millis(10)),
        wake_on_pending: true,
    };

    let (res_tx, mut res_rx) = tokio::sync::mpsc::channel(12);

    // Wrap task in timeout so that it will finish eventually even if the stream
    // stalls.
    tokio::spawn(tokio::time::timeout(
        tokio::time::Duration::from_millis(150),
        async move {
            tokio::pin!(stream);

            while let Some(item) = stream.next().await {
                res_tx.send(item).await.ok();
            }
        },
    ));

    let mut items = Vec::with_capacity(3);
    while let Some(result) = res_rx.recv().await {
        items.push(result);
    }

    // We expect the stream to yield normally and thus three items.
    assert_eq!(items, vec![4, 8, 12]);
}
Repository: tokio-rs/tokio
File: tokio-test/tests/io.rs (Rust)

Code before:
#![warn(rust_2018_idioms)]

use std::{
    io,
    time::{Duration, Instant},
};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio_test::io::Builder;

#[tokio::test]
async fn read() {
    let mut mock = Builder::new().read(b"hello ").read(b"world!").build();

    let mut buf = [0; 256];

    let n = mock.read(&mut buf).await.expect("read 1");
    assert_eq!(&buf[..n], b"hello ");

    let n = mock.read(&mut buf).await.expect("read 2");
    assert_eq!(&buf[..n], b"world!");
}

#[tokio::test]
async fn read_error() {
    let error = io::Error::new(io::ErrorKind::Other, "cruel");
    let mut mock = Builder::new()
        .read(b"hello ")
        .read_error(error)
        .read(b"world!")
        .build();
    let mut buf = [0; 256];

Review comment (suggestion):
We don't use that import style in Tokio.
```suggestion
use std::io;
use std::time::{Duration, Instant};
```

Code after:
#![warn(rust_2018_idioms)]

use std::io;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::{Duration, Instant};
use tokio_test::io::Builder;

#[tokio::test]
async fn read() {
    let mut mock = Builder::new().read(b"hello ").read(b"world!").build();

    let mut buf = [0; 256];

    let n = mock.read(&mut buf).await.expect("read 1");
    assert_eq!(&buf[..n], b"hello ");

    let n = mock.read(&mut buf).await.expect("read 2");
    assert_eq!(&buf[..n], b"world!");
}

#[tokio::test]
async fn read_error() {
    let error = io::Error::new(io::ErrorKind::Other, "cruel");
    let mut mock = Builder::new()
        .read(b"hello ")
        .read_error(error)
        .read(b"world!")
        .build();
    let mut buf = [0; 256];

    let n = mock.read(&mut buf).await.expect("read 1");
Repository: tokio-rs/tokio
File: tokio/tests/io_read_buf.rs (Rust)

Code before:
    }

    impl AsyncRead for Rd {
        fn poll_read(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &mut ReadBuf<'_>,
        ) -> Poll<io::Result<()>> {
            self.cnt += 1;
            buf.put_slice(b"hello world");
            Poll::Ready(Ok(()))
        }
    }

    let mut buf = vec![];
    let mut rd = Rd { cnt: 0 };

    let n = assert_ok!(rd.read_buf(&mut buf).await);
    assert_eq!(1, rd.cnt);
    assert_eq!(n, 11);
    assert_eq!(buf[..], b"hello world"[..]);
}

#[tokio::test]
#[cfg(feature = "io-util")]
#[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
async fn issue_5588() {
    use bytes::BufMut;

    // steps to zero
    let mut buf = [0; 8];
    let mut read_buf = ReadBuf::new(&mut buf);
    assert_eq!(read_buf.remaining_mut(), 8);
    assert_eq!(read_buf.remaining_mut(), 8);
    unsafe {
        read_buf.advance_mut(1);
    }
    assert_eq!(read_buf.remaining_mut(), 7);
    assert_eq!(read_buf.remaining_mut(), 7);
    unsafe {
        read_buf.advance_mut(5);
    }
    assert_eq!(read_buf.remaining_mut(), 2);
    assert_eq!(read_buf.remaining_mut(), 2);
    unsafe {
        read_buf.advance_mut(2);
    }
    assert_eq!(read_buf.remaining_mut(), 0);
    assert_eq!(read_buf.remaining_mut(), 0);

    // directly to zero

Review comment (suggestion):
This annotation is used only for documentation, so it isn't used in tests.
```suggestion
```

Code after:
    }

    impl AsyncRead for Rd {
        fn poll_read(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &mut ReadBuf<'_>,
        ) -> Poll<io::Result<()>> {
            self.cnt += 1;
            buf.put_slice(b"hello world");
            Poll::Ready(Ok(()))
        }
    }

    let mut buf = vec![];
    let mut rd = Rd { cnt: 0 };

    let n = assert_ok!(rd.read_buf(&mut buf).await);
    assert_eq!(1, rd.cnt);
    assert_eq!(n, 11);
    assert_eq!(buf[..], b"hello world"[..]);
}

#[tokio::test]
#[cfg(feature = "io-util")]
async fn issue_5588() {
    use bytes::BufMut;

    // steps to zero
    let mut buf = [0; 8];
    let mut read_buf = ReadBuf::new(&mut buf);
    assert_eq!(read_buf.remaining_mut(), 8);
    assert_eq!(read_buf.chunk_mut().len(), 8);
    unsafe {
        read_buf.advance_mut(1);
    }
    assert_eq!(read_buf.remaining_mut(), 7);
    assert_eq!(read_buf.chunk_mut().len(), 7);
    unsafe {
        read_buf.advance_mut(5);
    }
    assert_eq!(read_buf.remaining_mut(), 2);
    assert_eq!(read_buf.chunk_mut().len(), 2);
    unsafe {
        read_buf.advance_mut(2);
    }
    assert_eq!(read_buf.remaining_mut(), 0);
    assert_eq!(read_buf.chunk_mut().len(), 0);

    // directly to zero
    let mut buf = [0; 8];
Repository: tokio-rs/tokio
File: tokio/src/sync/tests/loom_watch.rs (Rust)

Code before:

        block_on(rx3.changed()).unwrap();
        assert_eq!(*rx3.borrow(), 2);

        block_on(rx4.changed()).unwrap();
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(0);
        
        // here we repeatedly send values to the channel
        // to trigger its loop
        let th = thread::spawn(move || {
            for i in 0..10 {
                tx.send(i).unwrap();
            }
        });

        // here we block the main thread until the
        // value is 9
        let result = block_on(rx.wait_for(|x| *x == 9));
        assert_eq!(result.unwrap(), true);

        th.join().unwrap();
    });
}

Review comment (suggestion):
I would like to see two threads modify it in parallel instead.

Code after:
        assert_eq!(*rx2.borrow(), 2);

        block_on(rx3.changed()).unwrap();
        assert_eq!(*rx3.borrow(), 2);

        block_on(rx4.changed()).unwrap();
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..2 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        assert_eq!(*block_on(rx.wait_for(|x| *x)).unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}

#[test]
fn wait_for_returns_correct_value() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(0);

        let jh = thread::spawn(move || {
            tx.send(1).unwrap();
            tx.send(2).unwrap();
            tx.send(3).unwrap();
        });
Repository: tokio-rs/tokio
File: tokio/src/sync/tests/loom_watch.rs (Rust)

Code before:
        assert_eq!(*rx3.borrow(), 2);

        block_on(rx4.changed()).unwrap();
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..10 {
                tx1.send(false).unwrap();
                std::thread::sleep(Duration::from_millis(10));
            }
        });

        let th2 = thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(10));
            tx2.send(true).unwrap();
        });

        let result = block_on(rx.wait_for(|x| *x));
        assert_eq!(result.unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}


Review comment (suggestion):
Please reduce the loop to 3 iterations. Also, you can use `send_modify` here to avoid unsetting the boolean when the other thread has set it to `true`.

Code after:

        block_on(rx4.changed()).unwrap();
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..2 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        assert_eq!(*block_on(rx.wait_for(|x| *x)).unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}

#[test]
fn wait_for_returns_correct_value() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(0);

        let jh = thread::spawn(move || {
            tx.send(1).unwrap();
            tx.send(2).unwrap();
            tx.send(3).unwrap();
        });

        // Stop at the first value we are called at.
        let mut stopped_at = usize::MAX;
        let returned = *block_on(rx.wait_for(|x| {