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

Code before:
                    self.steal_operations += 1;
                }
            }
        }

        cfg_metrics_variant! {
            stable: {
                pub(crate) fn incr_overflow_count(&mut self) {}
            },
            unstable: {
                pub(crate) fn incr_overflow_count(&mut self) {
                    self.overflow_count += 1;
                }
            }
        }
    }
}

pub(crate) fn duration_as_u64(dur: Duration) -> u64 {
    u64::try_from(dur.as_nanos()).unwrap_or(u64::MAX)
}

/// Gate time metrics as platforms like `WASM` do not typically support [`std::time`]
/// <https://github.com/tokio-rs/tokio/issues/7319>
fn now() -> Option<Instant> {
    if cfg!(target_family = "wasm") && !cfg!(feature = "time") {
        None
    } else {
        Some(Instant::now())
    }
}


Review comment (bug):
You should do an exact match: `cfg!(all(target_arch = "wasm", target_os = "unknown", target_vendor = "unknown"))`. Instant::now should be available just fine on all other Wasm targets that have std.

It is not expected that any new broken targets like this will be added.

Code after:
                    self.steal_operations += 1;
                }
            }
        }

        cfg_metrics_variant! {
            stable: {
                pub(crate) fn incr_overflow_count(&mut self) {}
            },
            unstable: {
                pub(crate) fn incr_overflow_count(&mut self) {
                    self.overflow_count += 1;
                }
            }
        }
    }
}

pub(crate) fn duration_as_u64(dur: Duration) -> u64 {
    u64::try_from(dur.as_nanos()).unwrap_or(u64::MAX)
}

/// Gate unsupported time metrics for `wasm32-unknown-unknown`
/// <https://github.com/tokio-rs/tokio/issues/7319>
fn now() -> Option<Instant> {
    if cfg!(all(
        target_arch = "wasm32",
        target_os = "unknown",
        target_vendor = "unknown"
    )) {
        None
    } else {
        Some(Instant::now())
    }
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/io/driver/uring.rs (Rust)

Code before:
        while unsafe { ctx.ring_mut().submission().push(&entry).is_err() } {
            // If the submission queue is full, flush it to the kernel
            submit_or_remove(ctx)?;
        }

        // Note: For now, we submit the entry immediately without utilizing batching.
        submit_or_remove(ctx)?;

        Ok(index)
    }

    // TODO: Remove this annotation when operations are actually supported
    #[allow(unused_variables, unreachable_code)]
    pub(crate) fn cancel_op<T: Cancellable>(&self, index: usize, data: Option<T>) {
        let mut guard = self.get_uring().lock();
        let ctx = &mut *guard;
        let ops = &mut ctx.ops;
        let Some(lifecycle) = ops.get_mut(index) else {
            // The corresponding index doesn't exist anymore, so this Op is already complete.
            return;
        };

        // This Op will be cancelled. Here, we don't remove the lifecycle from the slab to keep
        // uring data alive until the operation completes.

        let cancell_data = data.expect("Data should be present").cancel();
        match mem::replace(lifecycle, Lifecycle::Cancelled(cancell_data)) {
            Lifecycle::Submitted | Lifecycle::Waiting(_) => (),
            // The driver saw the completion, but it was never polled.
            Lifecycle::Completed(_) => (),
            prev => panic!("Unexpected state: {:?}", prev),
        };
    }
}


Review comment (suggestion):
```suggestion
        let cancel_data = data.expect("Data should be present").cancel();
```

Code after:
            // If the submission queue is full, flush it to the kernel
            submit_or_remove(ctx)?;
        }

        // Note: For now, we submit the entry immediately without utilizing batching.
        submit_or_remove(ctx)?;

        Ok(index)
    }

    // TODO: Remove this annotation when operations are actually supported
    #[allow(unused_variables, unreachable_code)]
    pub(crate) fn cancel_op<T: Cancellable>(&self, index: usize, data: Option<T>) {
        let mut guard = self.get_uring().lock();
        let ctx = &mut *guard;
        let ops = &mut ctx.ops;
        let Some(lifecycle) = ops.get_mut(index) else {
            // The corresponding index doesn't exist anymore, so this Op is already complete.
            return;
        };

        // This Op will be cancelled. Here, we don't remove the lifecycle from the slab to keep
        // uring data alive until the operation completes.

        let cancel_data = data.expect("Data should be present").cancel();
        match mem::replace(lifecycle, Lifecycle::Cancelled(cancel_data)) {
            Lifecycle::Submitted | Lifecycle::Waiting(_) => (),
            // The driver saw the completion, but it was never polled.
            Lifecycle::Completed(_) => (),
            prev => panic!("Unexpected state: {:?}", prev),
        };
    }
}
Repository: tokio-rs/tokio
File: tokio-util/src/sync/cancellation_token/guard_ref.rs (Rust)

Code before:
use crate::sync::CancellationToken;

/// A wrapper for cancellation token which automatically cancels
/// it on drop. It is created using [`drop_guard_ref`] method on the [`CancellationToken`].
///
/// This is a `ref` version of [`DropGuard`]
///
/// [`drop_guard_ref`]: CancellationToken::drop_guard_ref
/// [`DropGuard`]: super::DropGuard
#[derive(Debug)]
pub struct DropGuardRef<'a> {
    pub(super) inner: Option<&'a CancellationToken>,
}

impl<'a> DropGuardRef<'a> {
    /// Returns stored cancellation token and removes this drop guard instance
    /// (i.e. it will no longer cancel token). Other guards for this token
    /// are not affected.
    pub fn disarm(mut self) -> &'a CancellationToken {
        self.inner
            .take()
            .expect("`inner` can be only None in a destructor")
    }
}

impl Drop for DropGuardRef<'_> {
    fn drop(&mut self) {
        if let Some(inner) = self.inner {
            inner.cancel();
        }
    }

Review comment (suggestion):
```suggestion
/// This is a borrowed version of [`DropGuard`].
```

Code after:
use crate::sync::CancellationToken;

/// A wrapper for cancellation token which automatically cancels
/// it on drop. It is created using [`drop_guard_ref`] method on the [`CancellationToken`].
///
/// This is a borrowed version of [`DropGuard`].
///
/// [`drop_guard_ref`]: CancellationToken::drop_guard_ref
/// [`DropGuard`]: super::DropGuard
#[derive(Debug)]
pub struct DropGuardRef<'a> {
    pub(super) inner: Option<&'a CancellationToken>,
}

impl<'a> DropGuardRef<'a> {
    /// Returns stored cancellation token and removes this drop guard instance
    /// (i.e. it will no longer cancel token). Other guards for this token
    /// are not affected.
    pub fn disarm(mut self) -> &'a CancellationToken {
        self.inner
            .take()
            .expect("`inner` can be only None in a destructor")
    }
}

impl Drop for DropGuardRef<'_> {
    fn drop(&mut self) {
        if let Some(inner) = self.inner {
            inner.cancel();
        }
    }
Repository: tokio-rs/tokio
File: tokio/src/runtime/io/scheduled_io.rs (Rust)

Code before:
            &self.pointers
        }
    }
}

/// Future returned by `readiness()`.
struct Readiness<'a> {
    scheduled_io: &'a ScheduledIo,

    state: State,

    /// Entry in the waiter `LinkedList`.
    waiter: UnsafeCell<Waiter>,
}

enum State {
    Init,
    Waiting,
    Done,
}

// The `ScheduledIo::readiness` (`AtomicUsize`) is packed full of goodness.
//
// | shutdown | driver tick | readiness |
// |----------+-------------+-----------|
// |   1 bit  |   15 bit    |  16 bits  |

const READINESS: bit::Pack = bit::Pack::least_significant(16);

const TICK: bit::Pack = READINESS.then(15);

const SHUTDOWN: bit::Pack = TICK.then(1);

// ===== impl ScheduledIo =====

impl Default for ScheduledIo {
    fn default() -> ScheduledIo {
        ScheduledIo {
            linked_list_pointers: UnsafeCell::new(linked_list::Pointers::new()),
            readiness: AtomicUsize::new(0),
            waiters: Mutex::new(Waiters::default()),
        }
    }
}

impl ScheduledIo {
    pub(crate) fn token(&self) -> mio::Token {
        mio::Token(super::EXPOSE_IO.expose_provenance(self))
    }

    /// Invoked when the IO driver is shut down; forces this `ScheduledIo` into a

Review comment (suggestion):
```suggestion
// |   1 bit  |   15 bits   |  16 bits  |
```

Code after:
            &self.pointers
        }
    }
}

/// Future returned by `readiness()`.
struct Readiness<'a> {
    scheduled_io: &'a ScheduledIo,

    state: State,

    /// Entry in the waiter `LinkedList`.
    waiter: UnsafeCell<Waiter>,
}

enum State {
    Init,
    Waiting,
    Done,
}

// The `ScheduledIo::readiness` (`AtomicUsize`) is packed full of goodness.
//
// | shutdown | driver tick | readiness |
// |----------+-------------+-----------|
// |   1 bit  |   15 bits   |  16 bits  |

const READINESS: bit::Pack = bit::Pack::least_significant(16);

const TICK: bit::Pack = READINESS.then(15);

const SHUTDOWN: bit::Pack = TICK.then(1);

// ===== impl ScheduledIo =====

impl Default for ScheduledIo {
    fn default() -> ScheduledIo {
        ScheduledIo {
            linked_list_pointers: UnsafeCell::new(linked_list::Pointers::new()),
            readiness: AtomicUsize::new(0),
            waiters: Mutex::new(Waiters::default()),
        }
    }
}

impl ScheduledIo {
    pub(crate) fn token(&self) -> mio::Token {
        mio::Token(super::EXPOSE_IO.expose_provenance(self))
    }

    /// Invoked when the IO driver is shut down; forces this `ScheduledIo` into a
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))
        );
    }

    // Does not poll the future when token is already cancelled.
    {
        let token = CancellationToken::new();

        let fut = token.run_until_cancelled(async { panic!("fut polled after cancellation") });
        pin!(fut);

        token.cancel();

        assert_eq!(
            Poll::Ready(None),
            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::<()>())

Review comment (suggestion):
```suggestion
    // Do not poll the future when token is already cancelled.
```

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))
        );
    }

    // Do not poll the future when token is already cancelled.
    {
        let token = CancellationToken::new();

        let fut = token.run_until_cancelled(async { panic!("fut polled after cancellation") });
        pin!(fut);

        token.cancel();

        assert_eq!(
            Poll::Ready(None),
            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::<()>())
Repository: tokio-rs/tokio
File: tokio-stream/src/wrappers/mpsc_bounded.rs (Rust)

Code before:
    type Item = T;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.inner.poll_recv(cx)
    }

    /// Returns the bounds of the stream based on the underlying receiver.
    ///
    /// For open channels, it returns `(receiver.len(), None)`.
    ///
    /// For closed channels, it returns `(receiver.len(), Some(used_capacity))`
    /// where `used_capacity` is calculated as `receiver.max_capacity() -
    /// receiver.capacity()`. This accounts for any [`Permit`] that is still
    /// able to send a message.
    ///
    /// [`Permit`]: struct@tokio::sync::mpsc::Permit
    fn size_hint(&self) -> (usize, Option<usize>) {
        let upper = if self.inner.is_closed() {
            let used_capacity = self.inner.max_capacity() - self.inner.capacity();
            Some(used_capacity)
        } else {
            None
        };

        (self.inner.len(), upper)
    }
}

impl<T> AsRef<Receiver<T>> for ReceiverStream<T> {
    fn as_ref(&self) -> &Receiver<T> {
        &self.inner
    }
}

impl<T> AsMut<Receiver<T>> for ReceiverStream<T> {
    fn as_mut(&mut self) -> &mut Receiver<T> {
        &mut self.inner
    }
}

impl<T> From<Receiver<T>> for ReceiverStream<T> {
    fn from(recv: Receiver<T>) -> Self {
        Self::new(recv)
    }
}


Review comment (suggestion):
Downstream may not send anything through the `Permit`, and then drop it. We can't assume every `Permit` always send a value.

Code after:
    type Item = T;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.inner.poll_recv(cx)
    }

    /// Returns the bounds of the stream based on the underlying receiver.
    ///
    /// For open channels, it returns `(receiver.len(), None)`.
    ///
    /// For closed channels, it returns `(receiver.len(), Some(used_capacity))`
    /// where `used_capacity` is calculated as `receiver.max_capacity() -
    /// receiver.capacity()`. This accounts for any [`Permit`] that is still
    /// able to send a message.
    ///
    /// [`Permit`]: struct@tokio::sync::mpsc::Permit
    fn size_hint(&self) -> (usize, Option<usize>) {
        if self.inner.is_closed() {
            let used_capacity = self.inner.max_capacity() - self.inner.capacity();
            (self.inner.len(), Some(used_capacity))
        } else {
            (self.inner.len(), None)
        }
    }
}

impl<T> AsRef<Receiver<T>> for ReceiverStream<T> {
    fn as_ref(&self) -> &Receiver<T> {
        &self.inner
    }
}

impl<T> AsMut<Receiver<T>> for ReceiverStream<T> {
    fn as_mut(&mut self) -> &mut Receiver<T> {
        &mut self.inner
    }
}

impl<T> From<Receiver<T>> for ReceiverStream<T> {
    fn from(recv: Receiver<T>) -> Self {
        Self::new(recv)
    }
}
Repository: tokio-rs/tokio
File: tokio-stream/tests/mpsc_bounded_stream.rs (Rust)

Code before:

#[tokio::test]
async fn size_hint_stream_open() {
    let (tx, rx) = mpsc::channel(4);

    tx.send(1).await.unwrap();
    tx.send(2).await.unwrap();

    let mut stream = ReceiverStream::new(rx);

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

#[tokio::test]
async fn size_hint_stream_closed() {
    let (tx, rx) = mpsc::channel(4);

    tx.send(1).await.unwrap();
    tx.send(2).await.unwrap();

    let mut stream = ReceiverStream::new(rx);
    stream.close();

    assert_eq!(stream.size_hint(), (2, Some(2)));
    stream.next().await;
    assert_eq!(stream.size_hint(), (1, Some(1)));
    stream.next().await;
    assert_eq!(stream.size_hint(), (0, Some(0)));
}

#[test]
fn size_hint_stream_instantly_closed() {
    let (_tx, rx) = mpsc::channel::<i32>(4);

    let mut stream = ReceiverStream::new(rx);
    stream.close();

    assert_eq!(stream.size_hint(), (0, Some(0)));
}

#[tokio::test]
async fn size_hint_stream_closed_permits() {
    let (tx, rx) = mpsc::channel(4);

    tx.send(1).await.unwrap();
    let permit1 = tx.reserve().await.unwrap();
    let permit2 = tx.reserve().await.unwrap();

Review comment (suggestion):
A stream may be closed both by calling `Receiver::close` and by dropping all senders. These use different mechanisms internally, so it would be good to test both scenarios.

Code after:

#[tokio::test]
async fn size_hint_stream_open() {
    let (tx, rx) = mpsc::channel(4);

    tx.send(1).await.unwrap();
    tx.send(2).await.unwrap();

    let mut stream = ReceiverStream::new(rx);

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

#[tokio::test]
async fn size_hint_stream_closed() {
    let (tx, rx) = mpsc::channel(4);

    tx.send(1).await.unwrap();
    tx.send(2).await.unwrap();

    let mut stream = ReceiverStream::new(rx);
    stream.close();

    assert_eq!(stream.size_hint(), (2, Some(2)));
    stream.next().await;
    assert_eq!(stream.size_hint(), (1, Some(1)));
    stream.next().await;
    assert_eq!(stream.size_hint(), (0, Some(0)));
}

#[tokio::test]
async fn size_hint_sender_dropped() {
    let (tx, rx) = mpsc::channel(4);

    tx.send(1).await.unwrap();
    tx.send(2).await.unwrap();

    let mut stream = ReceiverStream::new(rx);
    drop(tx);

    assert_eq!(stream.size_hint(), (2, Some(2)));
    stream.next().await;
    assert_eq!(stream.size_hint(), (1, Some(1)));
    stream.next().await;
    assert_eq!(stream.size_hint(), (0, Some(0)));
}
Repository: tokio-rs/tokio
File: tokio-stream/src/wrappers/mpsc_unbounded.rs (Rust)

Code before:
    /// This prevents any further messages from being sent on the channel while
    /// still enabling the receiver to drain messages that are buffered.
    pub fn close(&mut self) {
        self.inner.close();
    }
}

impl<T> Stream for UnboundedReceiverStream<T> {
    type Item = T;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.inner.poll_recv(cx)
    }

    /// Returns the bounds of the stream based on the underlying receiver.
    ///
    /// For open channels, it returns `(receiver.len(), None)`.
    ///
    /// For closed channels, it returns `(receiver.len(), receiver.len())`.
    fn size_hint(&self) -> (usize, Option<usize>) {
        let len = self.inner.len();
        let upper = if self.inner.is_closed() {
            Some(len)
        } else {
            None
        };

        (len, upper)
    }
}

impl<T> AsRef<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
    fn as_ref(&self) -> &UnboundedReceiver<T> {
        &self.inner
    }
}

impl<T> AsMut<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
    fn as_mut(&mut self) -> &mut UnboundedReceiver<T> {
        &mut self.inner
    }
}

impl<T> From<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
    fn from(recv: UnboundedReceiver<T>) -> Self {
        Self::new(recv)
    }
}


Review comment (bug):
Consider this scenario:

1. We read `self.inner.len()`.
2. Another thread sends a message.
3. Another thread drops the last sender.
4. We read `self.inner.is_closed()` and get true.

Now the value we got from `len` is an incorrect upper bound. Please call `is_closed()` before calling `len()`. This ensures that dropping the last sender happens-before the call to `len()`.

Code after:
    /// This prevents any further messages from being sent on the channel while
    /// still enabling the receiver to drain messages that are buffered.
    pub fn close(&mut self) {
        self.inner.close();
    }
}

impl<T> Stream for UnboundedReceiverStream<T> {
    type Item = T;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.inner.poll_recv(cx)
    }

    /// Returns the bounds of the stream based on the underlying receiver.
    ///
    /// For open channels, it returns `(receiver.len(), None)`.
    ///
    /// For closed channels, it returns `(receiver.len(), receiver.len())`.
    fn size_hint(&self) -> (usize, Option<usize>) {
        if self.inner.is_closed() {
            let len = self.inner.len();
            (len, Some(len))
        } else {
            (self.inner.len(), None)
        }
    }
}

impl<T> AsRef<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
    fn as_ref(&self) -> &UnboundedReceiver<T> {
        &self.inner
    }
}

impl<T> AsMut<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
    fn as_mut(&mut self) -> &mut UnboundedReceiver<T> {
        &mut self.inner
    }
}

impl<T> From<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
    fn from(recv: UnboundedReceiver<T>) -> Self {
        Self::new(recv)
    }
}
Repository: tokio-rs/tokio
File: tokio/tests/time_wasm.rs (Rust)

Code before:
#![warn(rust_2018_idioms)]
#![cfg(all(target_arch = "wasm32", not(target_os = "wasi")))]

use wasm_bindgen_test::wasm_bindgen_test;

#[wasm_bindgen_test]
#[should_panic]
fn instant_now_panics() {
    let _ = tokio::time::Instant::now();
}

#[cfg(all(feature = "rt", not(feature = "time")))]
#[wasm_bindgen_test]
fn runtime_without_time_does_not_panic() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    rt.block_on(async {});
}

#[cfg(all(feature = "rt", feature = "time"))]
#[wasm_bindgen_test]
#[should_panic]
fn sleep_panics_on_unknown_unknown() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_time()
        .build()
        .unwrap();
    rt.block_on(async { tokio::time::sleep(core::time::Duration::from_millis(1)).await });
}


Review comment (suggestion):
This test should also run if the time feature is enabled. Enabling time must not cause a panic here.

Code after:
#![warn(rust_2018_idioms)]
#![cfg(all(target_arch = "wasm32", not(target_os = "wasi")))]

use wasm_bindgen_test::wasm_bindgen_test;

#[wasm_bindgen_test]
#[should_panic]
fn instant_now_panics() {
    let _ = tokio::time::Instant::now();
}

#[cfg(all(feature = "rt", not(feature = "time")))]
#[wasm_bindgen_test]
fn runtime_without_time_does_not_panic() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    rt.block_on(async {});
}

#[cfg(all(feature = "rt", feature = "time"))]
#[wasm_bindgen_test]
#[should_panic] // should remove this once time is supported
fn runtime_with_time_does_not_panic() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    rt.block_on(async {});
}

#[cfg(all(feature = "rt", feature = "time"))]
#[wasm_bindgen_test]
#[should_panic]
fn sleep_panics_on_unknown_unknown() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_time()
        .build()
        .unwrap();
    rt.block_on(async { tokio::time::sleep(core::time::Duration::from_millis(1)).await });
}