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

Code before:
use crate::Stream;
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};

pin_project! {
    /// Stream that returns None when the stream is exhausted. To be used with StreamMap to know when a stream ends.
    /// # Example
    /// Using `StreamNotifyClose` to handle closed streams with `StreamMap`
    /// ```no_run
    /// use tokio_stream::{Stream, StreamExt, StreamMap, StreamNotifyClose};

    /// struct Test {
    ///     state: i32,
    /// }

    /// impl Stream for Test {
    ///     type Item = i32;

    ///     fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<i32>> {
    ///         let val = self.state;
    ///         let res = match val {
    ///             0 => Poll::Ready(Some(0)),
    ///             1 => Poll::Ready(Some(1)),
    ///             2 => Poll::Ready(None),
    ///             _ => unreachable!("shouldn't get here"),
    ///         };

    ///         self.state += 1;
    ///         res
    ///     }
    /// }

    /// #[tokio::main]
    /// async fn main() {
    ///     let mut map = StreamMap::new();
    ///     let mut stream = StreamNotifyClose::new(Test { state: 0 });
    ///     let mut stream2 = StreamNotifyClose::new(Test { state: 0 });

Review comment (suggestion):
Please fix these holes in all of your documentation comments.
```suggestion
    /// use tokio_stream::{Stream, StreamExt, StreamMap, StreamNotifyClose};
    ///
    /// struct Test {
```

Code after:
use crate::Stream;
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};

pin_project! {
    /// A `Stream` that wraps the values in an `Option`.
    ///
    /// Whenever the wrapped stream yields an item, this stream yields that item
    /// wrapped in `Some`. When the inner stream ends, then this stream first
    /// yields a `None` item, and then this stream will also end.
    ///
    /// # Example
    ///
    /// Using `StreamNotifyClose` to handle closed streams with `StreamMap`.
    ///
    /// ```
    /// use tokio_stream::{StreamExt, StreamMap, StreamNotifyClose};
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let mut map = StreamMap::new();
    ///     let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
    ///     let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
    ///     map.insert(0, stream);
    ///     map.insert(1, stream2);
    ///     while let Some((key, val)) = map.next().await {
    ///         match val {
    ///             Some(val) => println!("got {val:?} from stream {key:?}"),
    ///             None => println!("stream {key:?} closed"),
    ///         }
    ///     }
    /// }
    /// ```
    #[must_use = "streams do nothing unless polled"]
    pub struct StreamNotifyClose<S> {
        #[pin]
        inner: Option<S>,
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_close.rs (Rust)

Code before:
use crate::Stream;
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};

pin_project! {
    /// Stream that returns None when the stream is exhausted. To be used with StreamMap to know when a stream ends.
    /// # Example
    /// Using `StreamNotifyClose` to handle closed streams with `StreamMap`
    /// ```no_run
    /// use tokio_stream::{Stream, StreamExt, StreamMap, StreamNotifyClose};

    /// struct Test {
    ///     state: i32,
    /// }

    /// impl Stream for Test {
    ///     type Item = i32;

    ///     fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<i32>> {
    ///         let val = self.state;
    ///         let res = match val {
    ///             0 => Poll::Ready(Some(0)),
    ///             1 => Poll::Ready(Some(1)),
    ///             2 => Poll::Ready(None),
    ///             _ => unreachable!("shouldn't get here"),
    ///         };

    ///         self.state += 1;
    ///         res
    ///     }
    /// }

    /// #[tokio::main]

Review comment (suggestion):
The grammar here is bad. Additionally, we have empty lines before and after headings.
```suggestion
    /// A `Stream` that wraps the values in an `Option`.
    ///
    /// Whenever the wrapped stream yields an item, this stream yields that item
    /// wrapped in `Some`. When the inner stream ends, then this stream first
    /// yields a `None` item, and then this stream will also end.
    ///
    /// # Example
    ///
    /// Using `StreamNotifyClose` to handle closed streams with `StreamMap`.
```

Code after:
use crate::Stream;
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};

pin_project! {
    /// A `Stream` that wraps the values in an `Option`.
    ///
    /// Whenever the wrapped stream yields an item, this stream yields that item
    /// wrapped in `Some`. When the inner stream ends, then this stream first
    /// yields a `None` item, and then this stream will also end.
    ///
    /// # Example
    ///
    /// Using `StreamNotifyClose` to handle closed streams with `StreamMap`.
    ///
    /// ```
    /// use tokio_stream::{StreamExt, StreamMap, StreamNotifyClose};
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let mut map = StreamMap::new();
    ///     let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
    ///     let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
    ///     map.insert(0, stream);
    ///     map.insert(1, stream2);
    ///     while let Some((key, val)) = map.next().await {
    ///         match val {
    ///             Some(val) => println!("got {val:?} from stream {key:?}"),
    ///             None => println!("stream {key:?} closed"),
    ///         }
    ///     }
    /// }
    /// ```
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_close.rs (Rust)

Code before:
use crate::Stream;
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};

pin_project! {
    /// Stream that returns None when the stream is exhausted. To be used with StreamMap to know when a stream ends.
    /// # Example
    /// Using `StreamNotifyClose` to handle closed streams with `StreamMap`
    /// ```no_run
    /// use tokio_stream::{Stream, StreamExt, StreamMap, StreamNotifyClose};

    /// struct Test {
    ///     state: i32,
    /// }

    /// impl Stream for Test {
    ///     type Item = i32;

    ///     fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<i32>> {
    ///         let val = self.state;
    ///         let res = match val {
    ///             0 => Poll::Ready(Some(0)),
    ///             1 => Poll::Ready(Some(1)),
    ///             2 => Poll::Ready(None),
    ///             _ => unreachable!("shouldn't get here"),
    ///         };

    ///         self.state += 1;
    ///         res
    ///     }
    /// }

    /// #[tokio::main]
    /// async fn main() {

Review comment (suggestion):
Documentation examples should usually not be marked `no_run` if they contain a main function. In this case, the annotation hides the mistake where this example will panic if you run it.
```suggestion
    /// ```
```

Code after:
use crate::Stream;
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};

pin_project! {
    /// A `Stream` that wraps the values in an `Option`.
    ///
    /// Whenever the wrapped stream yields an item, this stream yields that item
    /// wrapped in `Some`. When the inner stream ends, then this stream first
    /// yields a `None` item, and then this stream will also end.
    ///
    /// # Example
    ///
    /// Using `StreamNotifyClose` to handle closed streams with `StreamMap`.
    ///
    /// ```
    /// use tokio_stream::{StreamExt, StreamMap, StreamNotifyClose};
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let mut map = StreamMap::new();
    ///     let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
    ///     let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
    ///     map.insert(0, stream);
    ///     map.insert(1, stream2);
    ///     while let Some((key, val)) = map.next().await {
    ///         match val {
    ///             Some(val) => println!("got {val:?} from stream {key:?}"),
    ///             None => println!("stream {key:?} closed"),
    ///         }
    ///     }
    /// }
    /// ```
    #[must_use = "streams do nothing unless polled"]
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_close.rs (Rust)

Code before:
    ///         };

    ///         self.state += 1;
    ///         res
    ///     }
    /// }

    /// #[tokio::main]
    /// async fn main() {
    ///     let mut map = StreamMap::new();
    ///     let mut stream = StreamNotifyClose::new(Test { state: 0 });
    ///     let mut stream2 = StreamNotifyClose::new(Test { state: 0 });
    ///     map.insert(0, stream);
    ///     map.insert(1, stream2);
    ///     loop {
    ///         tokio::select! {
    ///           Some((key, val)) = map.next() => {
    ///              match val {
    ///                 Some(val) => println!("got {val:?} from stream {key:?}"),
    ///                 None => println!("stream {key:?} closed")
    ///              }
    ///             }
    ///         }
    ///     }
    /// }
    /// ```
    #[must_use = "streams do nothing unless polled"]
    pub struct StreamNotifyClose<S> {
        #[pin]
        inner: Option<S>,
    }
}

impl<S> StreamNotifyClose<S> {
    /// Create a new `StreamNotifyClose`.
    pub fn new(stream: S) -> Self {
        Self {
            inner: Some(stream),
        }
    }

    /// Get back the inner `Stream`.
    pub fn into_inner(self) -> Option<S> {
        self.inner
    }
}

impl<S> Stream for StreamNotifyClose<S>
where
    S: Stream,
{

Review comment (suggestion):
The documentation should explain that `into_inner` returns `None` if the stream has reached its end.

Code after:
    ///     while let Some((key, val)) = map.next().await {
    ///         match val {
    ///             Some(val) => println!("got {val:?} from stream {key:?}"),
    ///             None => println!("stream {key:?} closed"),
    ///         }
    ///     }
    /// }
    /// ```
    #[must_use = "streams do nothing unless polled"]
    pub struct StreamNotifyClose<S> {
        #[pin]
        inner: Option<S>,
    }
}

impl<S> StreamNotifyClose<S> {
    /// Create a new `StreamNotifyClose`.
    pub fn new(stream: S) -> Self {
        Self {
            inner: Some(stream),
        }
    }

    /// Get back the inner `Stream`.
    ///
    /// Returns `None` if the stream has reached its end.
    pub fn into_inner(self) -> Option<S> {
        self.inner
    }
}

impl<S> Stream for StreamNotifyClose<S>
where
    S: Stream,
{
    type Item = Option<S::Item>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // We can't invoke poll_next after it ended, so we unset the inner stream as a marker.
        match self
            .as_mut()
            .project()
            .inner
            .as_pin_mut()
            .map(|stream| S::poll_next(stream, cx))
        {
            Some(Poll::Ready(Some(item))) => Poll::Ready(Some(Some(item))),
            Some(Poll::Ready(None)) => {
                self.project().inner.set(None);
                Poll::Ready(Some(None))
            }
Repository: tokio-rs/tokio
File: tokio-stream/tests/stream_close.rs (Rust)

Code before:
use tokio_stream::{Stream, StreamExt, StreamNotifyClose};

use std::pin::Pin;
use std::task::{Context, Poll};

struct Test {
    state: i32,
}

impl Stream for Test {
    type Item = i32;

    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<i32>> {
        let val = self.state;
        let res = match val {
            0 => Poll::Ready(Some(0)),
            1 => Poll::Ready(Some(1)),
            2 => Poll::Ready(None),
            _ => unreachable!("shouldn't get here"),
        };

        self.state += 1;
        res
    }
}

#[tokio::test]
async fn basic_usage() {
    let mut stream = StreamNotifyClose::new(Test { state: 0 });

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

Review comment (suggestion):
Your custom stream is the same as `tokio_stream::iter(vec![0,1])`.
```suggestion
struct Test {
    state: i32,
}
```

Code after:
use tokio_stream::{StreamExt, StreamNotifyClose};

#[tokio::test]
async fn basic_usage() {
    let mut stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));

    assert_eq!(stream.next().await, Some(Some(0)));
    assert_eq!(stream.next().await, Some(Some(1)));
    assert_eq!(stream.next().await, Some(None));
    assert_eq!(stream.next().await, None);
}
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_close.rs (Rust)

Code before:
        // We can't invoke poll_next after it ended, so we unset the inner stream as a marker.
        match self
            .as_mut()
            .project()
            .inner
            .as_pin_mut()
            .map(|stream| S::poll_next(stream, cx))
        {
            Some(Poll::Ready(Some(item))) => Poll::Ready(Some(Some(item))),
            Some(Poll::Ready(None)) => {
                self.project().inner.set(None);
                Poll::Ready(Some(None))
            }
            Some(Poll::Pending) => Poll::Pending,
            None => Poll::Ready(None),
        }
    }

    #[inline]
    fn size_hint(&self) -> (usize, Option<usize>) {
        if let Some(inner) = &self.inner {
            // We always return +1 because when there's stream there's atleast one more item.
            match inner.size_hint() {
                (lower, Some(upper)) => (lower + 1, Some(upper + 1)),
                (lower, None) => (lower + 1, None),
            }
        } else {
            (0, Some(0))
        }
    }
}


Review comment (bug):
Your code does not handle overflow. (And it is not uncommon - infinite streams have `lower = usize::MAX`.)
```suggestion
            let (l, u) = inner.size_hint();
            (l.saturating_add(1), u.and_then(|u| u.checked_add(1)))
```

Code after:
    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // We can't invoke poll_next after it ended, so we unset the inner stream as a marker.
        match self
            .as_mut()
            .project()
            .inner
            .as_pin_mut()
            .map(|stream| S::poll_next(stream, cx))
        {
            Some(Poll::Ready(Some(item))) => Poll::Ready(Some(Some(item))),
            Some(Poll::Ready(None)) => {
                self.project().inner.set(None);
                Poll::Ready(Some(None))
            }
            Some(Poll::Pending) => Poll::Pending,
            None => Poll::Ready(None),
        }
    }

    #[inline]
    fn size_hint(&self) -> (usize, Option<usize>) {
        if let Some(inner) = &self.inner {
            // We always return +1 because when there's stream there's atleast one more item.
            let (l, u) = inner.size_hint();
            (l.saturating_add(1), u.and_then(|u| u.checked_add(1)))
        } else {
            (0, Some(0))
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/build.rs (Rust)

Code before:
                e
            );
        }
    }

    if !enable_const_thread_local {
        // To disable this feature on compilers that support it, you can
        // explicitly pass this flag with the following environment variable:
        //
        // RUSTFLAGS="--cfg tokio_no_const_thread_local"
        autocfg::emit("tokio_no_const_thread_local")
    }

    if !enable_addr_of {
        // To disable this feature on compilers that support it, you can
        // explicitly pass this flag with the following environment variable:
        //
        // RUSTFLAGS="--cfg tokio_no_addr_of"
        autocfg::emit("tokio_no_addr_of")
    }

    let target = ::std::env::var("TARGET").unwrap_or_default();

    if target.starts_with("wasm") {
        autocfg::emit("tokio_wasm");
        if target.contains("wasi") {
            autocfg::emit("tokio_wasi");
        } else {
            autocfg::emit("tokio_wasm_not_wasi");
        }
    }
}


Review comment (suggestion):
```suggestion
    // We emit cfgs instead of using `target_family = "wasm"` that requires Rust 1.54.
    // Note that these cfgs are unavailable in `Cargo.toml`.
    if target.starts_with("wasm") {
```

Code after:
                e
            );
        }
    }

    if !enable_const_thread_local {
        // To disable this feature on compilers that support it, you can
        // explicitly pass this flag with the following environment variable:
        //
        // RUSTFLAGS="--cfg tokio_no_const_thread_local"
        autocfg::emit("tokio_no_const_thread_local")
    }

    if !enable_addr_of {
        // To disable this feature on compilers that support it, you can
        // explicitly pass this flag with the following environment variable:
        //
        // RUSTFLAGS="--cfg tokio_no_addr_of"
        autocfg::emit("tokio_no_addr_of")
    }

    let target = ::std::env::var("TARGET").unwrap_or_default();

    // We emit cfgs instead of using `target_family = "wasm"` that requires Rust 1.54.
    // Note that these cfgs are unavailable in `Cargo.toml`.
    if target.starts_with("wasm") {
        autocfg::emit("tokio_wasm");
        if target.contains("wasi") {
            autocfg::emit("tokio_wasi");
        } else {
            autocfg::emit("tokio_wasm_not_wasi");
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/sync/broadcast.rs (Rust)

Code before:
    pub fn send(&self, value: T) -> Result<usize, SendError<T>> {
        let mut tail = self.shared.tail.lock();

        if tail.rx_cnt == 0 {
            return Err(SendError(value));
        }

        // Position to write into
        let pos = tail.pos;
        let rem = tail.rx_cnt;
        let idx = (pos & self.shared.mask as u64) as usize;

        // Update the tail position
        tail.pos = tail.pos.wrapping_add(1);

        // Get the slot
        let mut slot = self.shared.buffer[idx].write().unwrap();

        // Track the position
        slot.pos = pos;

        // Set remaining receivers
        slot.rem.with_mut(|v| *v = rem);

        // Write the value
        slot.val.with_mut(|ptr| unsafe { *ptr = Some(value) });

        // Release the slot lock before notifying the receivers.
        drop(slot);

        tail.notify_rx();

        // Release the mutex. This must happen after the slot lock is released,
        // otherwise the writer lock bit could be cleared while another thread
        // is in the critical section.
        drop(tail);

        Ok(rem)
    }

    /// Creates a new [`Receiver`] handle that will receive values sent **after**
    /// this call to `subscribe`.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::broadcast;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, _rx) = broadcast::channel(16);

Review comment (suggestion):
This doesn't require unsafe since you have mutable access to the value.
```suggestion
        // Write the value
        slot.val = UnsafeCell::new(Some(value));
```

Code after:
    pub fn send(&self, value: T) -> Result<usize, SendError<T>> {
        let mut tail = self.shared.tail.lock();

        if tail.rx_cnt == 0 {
            return Err(SendError(value));
        }

        // Position to write into
        let pos = tail.pos;
        let rem = tail.rx_cnt;
        let idx = (pos & self.shared.mask as u64) as usize;

        // Update the tail position
        tail.pos = tail.pos.wrapping_add(1);

        // Get the slot
        let mut slot = self.shared.buffer[idx].write().unwrap();

        // Track the position
        slot.pos = pos;

        // Set remaining receivers
        slot.rem.with_mut(|v| *v = rem);

        // Write the value
        slot.val = UnsafeCell::new(Some(value));

        // Release the slot lock before notifying the receivers.
        drop(slot);

        tail.notify_rx();

        // Release the mutex. This must happen after the slot lock is released,
        // otherwise the writer lock bit could be cleared while another thread
        // is in the critical section.
        drop(tail);

        Ok(rem)
    }

    /// Creates a new [`Receiver`] handle that will receive values sent **after**
    /// this call to `subscribe`.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::broadcast;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, _rx) = broadcast::channel(16);
Repository: tokio-rs/tokio
File: tokio/tests/sync_mpsc.rs (Rust)

Code before:
    assert!(weak_tx.upgrade().is_some());
}

// Tests that `downgrade` does not change the `tx_count` of the channel.
#[tokio::test]
async fn test_tx_count_weak_sender() {
    let (tx, _rx) = mpsc::channel::<i32>(1);
    let tx_weak = tx.downgrade();
    let tx_weak2 = tx.downgrade();
    drop(tx);

    assert!(tx_weak.upgrade().is_none() && tx_weak2.upgrade().is_none());
}

// Tests that channel `capacity` changes and `max_capacity` stays the same
#[tokio::test]
async fn test_tx_capacity() {
    let (tx, _rx) = mpsc::channel::<()>(10);
    // both capacities are same before
    assert_eq!(tx.capacity(), 10);
    assert_eq!(tx.max_capacity(), 10);

    let _permit = tx.reserve().await.unwrap();
    // after reserve, only capacity should drop by one
    assert_eq!(tx.capacity(), 9);
    assert_eq!(tx.max_capacity(), 10);
}


Review comment (suggestion):
let's maybe also test this using `send`? although the behavior should be the same...

Code after:
    assert!(weak_tx.upgrade().is_some());
}

// Tests that `downgrade` does not change the `tx_count` of the channel.
#[tokio::test]
async fn test_tx_count_weak_sender() {
    let (tx, _rx) = mpsc::channel::<i32>(1);
    let tx_weak = tx.downgrade();
    let tx_weak2 = tx.downgrade();
    drop(tx);

    assert!(tx_weak.upgrade().is_none() && tx_weak2.upgrade().is_none());
}

// Tests that channel `capacity` changes and `max_capacity` stays the same
#[tokio::test]
async fn test_tx_capacity() {
    let (tx, _rx) = mpsc::channel::<()>(10);
    // both capacities are same before
    assert_eq!(tx.capacity(), 10);
    assert_eq!(tx.max_capacity(), 10);

    let _permit = tx.reserve().await.unwrap();
    // after reserve, only capacity should drop by one
    assert_eq!(tx.capacity(), 9);
    assert_eq!(tx.max_capacity(), 10);

    let _sent = tx.send(()).await.unwrap();
    // after send, capacity should drop by one again
    assert_eq!(tx.capacity(), 8);
    assert_eq!(tx.max_capacity(), 10);
}