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

Code before:
    pub(crate) fn expose_provenance(&self, ptr: *const T) -> usize {
        #[cfg(miri)]
        {
            // FIXME: Use `pointer:addr` when it is stable.
            // SAFETY: Equivalent to `pointer::addr` which is safe.
            let addr: usize = unsafe { std::mem::transmute(ptr) };
            self.map.lock().insert(addr, ptr);
            addr
        }

        #[cfg(not(miri))]
        {
            ptr as usize
        }
    }

    #[inline]
    #[allow(clippy::wrong_self_convention)] // mirrors std name
    pub(crate) fn from_exposed_addr(&self, addr: usize) -> *const T {
        #[cfg(miri)]
        {
            *self
                .map
                .lock()
                .get(&addr)
                .expect("Provided address is not exposed.")
        }

        #[cfg(not(miri))]
        {
            addr as *const T
        }
    }

    #[inline]
    pub(crate) fn unexpose_provenance(&self, _ptr: *const T) {
        #[cfg(miri)]
        {
            // SAFETY: Equivalent to `pointer::addr` which is safe.
            let addr: usize = unsafe { std::mem::transmute(_ptr) };
            self.map
                .lock()
                .remove(&addr)
                .expect("Provided address is not exposed.");
        }
    }
}


Review comment (bug):
@RalfJung Can I trigger UB with a custom message instead of just panicking? Panics may not fail the test in all cases.

Code after:
    pub(crate) fn expose_provenance(&self, ptr: *const T) -> usize {
        #[cfg(miri)]
        {
            // FIXME: Use `pointer:addr` when it is stable.
            // SAFETY: Equivalent to `pointer::addr` which is safe.
            let addr: usize = unsafe { std::mem::transmute(ptr) };
            self.map.lock().insert(addr, ptr);
            addr
        }

        #[cfg(not(miri))]
        {
            ptr as usize
        }
    }

    #[inline]
    #[allow(clippy::wrong_self_convention)] // mirrors std name
    pub(crate) fn from_exposed_addr(&self, addr: usize) -> *const T {
        #[cfg(miri)]
        {
            let maybe_ptr = self.map.lock().get(&addr).copied();

            // SAFETY: Intentionally trigger a miri failure if the provenance we want is not
            // exposed.
            unsafe { maybe_ptr.unwrap_unchecked() }
        }

        #[cfg(not(miri))]
        {
            addr as *const T
        }
    }

    #[inline]
    pub(crate) fn unexpose_provenance(&self, _ptr: *const T) {
        #[cfg(miri)]
        {
            // SAFETY: Equivalent to `pointer::addr` which is safe.
            let addr: usize = unsafe { std::mem::transmute(_ptr) };
            let maybe_ptr = self.map.lock().remove(&addr);

            // SAFETY: Intentionally trigger a miri failure if the provenance we want is not
            // exposed.
            unsafe { maybe_ptr.unwrap_unchecked() };
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/net/unix/socketaddr.rs (Rust)

Code before:
use std::fmt;
use std::path::Path;

/// An address associated with a Tokio Unix socket.
///
/// This type is a think wrapper around
/// [`std::os::unix::net::SocketAddr`];
/// you can use [`From`] to wrap and unwrap
/// instances of [`std::os::unix::net::SocketAddr`] as this type.
pub struct SocketAddr(pub(super) std::os::unix::net::SocketAddr);

impl SocketAddr {
    /// Returns `true` if the address is unnamed.
    ///
    /// Documentation reflected in [`SocketAddr`]
    ///
    /// [`SocketAddr`]: std::os::unix::net::SocketAddr
    pub fn is_unnamed(&self) -> bool {
        self.0.is_unnamed()
    }

    /// Returns the contents of this address if it is a `pathname` address.
    ///
    /// Documentation reflected in [`SocketAddr`]
    ///
    /// [`SocketAddr`]: std::os::unix::net::SocketAddr
    pub fn as_pathname(&self) -> Option<&Path> {
        self.0.as_pathname()
    }
}

impl fmt::Debug for SocketAddr {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(fmt)

Review comment (suggestion):
```suggestion
/// This type is a thin wrapper around [`std::os::unix::net::SocketAddr`]. You
/// can convert to and from the standard library `SocketAddr` type using the
/// [`From`] trait.
```


Code after:
use std::fmt;
use std::path::Path;

/// An address associated with a Tokio Unix socket.
///
/// This type is a thin wrapper around [`std::os::unix::net::SocketAddr`]. You
/// can convert to and from the standard library `SocketAddr` type using the
/// [`From`] trait.
pub struct SocketAddr(pub(super) std::os::unix::net::SocketAddr);

impl SocketAddr {
    /// Returns `true` if the address is unnamed.
    ///
    /// Documentation reflected in [`SocketAddr`]
    ///
    /// [`SocketAddr`]: std::os::unix::net::SocketAddr
    pub fn is_unnamed(&self) -> bool {
        self.0.is_unnamed()
    }

    /// Returns the contents of this address if it is a `pathname` address.
    ///
    /// Documentation reflected in [`SocketAddr`]
    ///
    /// [`SocketAddr`]: std::os::unix::net::SocketAddr
    pub fn as_pathname(&self) -> Option<&Path> {
        self.0.as_pathname()
    }
}

impl fmt::Debug for SocketAddr {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(fmt)
    }
Repository: tokio-rs/tokio
File: tokio/tests/task_trace_self.rs (Rust)

Code before:
#![allow(unknown_lints, unexpected_cfgs)]
#![cfg(all(
    tokio_unstable,
    tokio_taskdump,
    target_os = "linux",
    any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")
))]

use std::{
    future::Future,
    pin::Pin,
    sync::{Arc, Mutex},
    task::{Context, Poll},
    time::{Duration, Instant},
};

use tokio::runtime::dump::{Root, Trace};

pin_project_lite::pin_project! { pub struct PrettyFuture<F: Future> {
    #[pin]
    f: Root<F>,
    t_last: State,
    logs: Arc<Mutex<Vec<Trace>>>,
}
}

enum State {
    NotStarted,
    Running { since: Instant },
    Alerted,
}

impl<F: Future> PrettyFuture<F> {
    pub fn pretty(f: F, logs: Arc<Mutex<Vec<Trace>>>) -> Self {
        PrettyFuture {
            f: Trace::root(f),
            t_last: State::NotStarted,
            logs,
        }
    }

Review comment (suggestion):
We don't use this syntax for imports. Generally, we have one `use std::` per module.

Code after:
#![allow(unknown_lints, unexpected_cfgs)]
#![cfg(all(
    tokio_unstable,
    tokio_taskdump,
    target_os = "linux",
    any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")
))]

use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use tokio::runtime::dump::{Root, Trace};

pin_project_lite::pin_project! {
    pub struct PrettyFuture<F: Future> {
        #[pin]
        f: Root<F>,
        t_last: State,
        logs: Arc<Mutex<Vec<Trace>>>,
    }
}

enum State {
    NotStarted,
    Running { since: Instant },
    Alerted,
}

impl<F: Future> PrettyFuture<F> {
    pub fn pretty(f: F, logs: Arc<Mutex<Vec<Trace>>>) -> Self {
        PrettyFuture {
            f: Trace::root(f),
            t_last: State::NotStarted,
            logs,
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/dump.rs (Rust)

Code before:
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Tasks {
    tasks: Vec<Task>,
}

/// A snapshot of a task.
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Task {
    id: Id,
    trace: Trace,
}

/// A backtrace symbol. This is similar to [backtrace::BacktraceSymbol],
/// but is a separate struct to avoid public dependency issues.
///
/// This struct is guaranteed to be pure data and operations involving
/// it will not call platform functions that take an unpredictable amount
/// of time to finish.
#[derive(Clone, Debug)]
pub struct BacktraceSymbol {
    name: Option<Vec<u8>>,
    name_demangled: Option<String>,
    addr: Option<*mut std::ffi::c_void>,
    filename: Option<std::path::PathBuf>,
    lineno: Option<u32>,
    colno: Option<u32>,
}

impl BacktraceSymbol {
    pub(crate) fn from_backtrace_symbol(sym: &backtrace::BacktraceSymbol) -> Self {
        let name = sym.name();
        Self {
            name: name.as_ref().map(|name| name.as_bytes().into()),
            name_demangled: name.map(|name| format!("{}", name)),
            addr: sym.addr(),
            filename: sym.filename().map(From::from),
            lineno: sym.lineno(),
            colno: sym.colno(),
        }
    }

    /// Return the raw name of the symbol.
    pub fn name_raw(&self) -> Option<&[u8]> {
        self.name.as_deref()
    }

    /// Return the demangled name of the symbol.

Review comment (suggestion):
I think a boxed slice would be enough here.

```suggestion
    name: Option<Box<[u8]>>,
    name_demangled: Option<Box<str>>,
```

Code after:
///
/// See [`Handle::dump`][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Tasks {
    tasks: Vec<Task>,
}

/// A snapshot of a task.
///
/// See [`Handle::dump`][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Task {
    id: Id,
    trace: Trace,
}

/// Represents an address that should not be dereferenced.
///
/// This type exists to get the auto traits correct, the public API
/// uses raw pointers to make life easier for users.
#[derive(Copy, Clone, Debug)]
struct Address(*mut std::ffi::c_void);

// Safe since Address should not be dereferenced
unsafe impl Send for Address {}
unsafe impl Sync for Address {}

/// A backtrace symbol.
///
/// This struct provides accessors for backtrace symbols, similar to [`backtrace::BacktraceSymbol`].
#[derive(Clone, Debug)]
pub struct BacktraceSymbol {
    name: Option<Box<[u8]>>,
    name_demangled: Option<Box<str>>,
    addr: Option<Address>,
    filename: Option<std::path::PathBuf>,
    lineno: Option<u32>,
    colno: Option<u32>,
}

impl BacktraceSymbol {
    pub(crate) fn from_backtrace_symbol(sym: &backtrace::BacktraceSymbol) -> Self {
        let name = sym.name();
        Self {
            name: name.as_ref().map(|name| name.as_bytes().into()),
            name_demangled: name.map(|name| format!("{}", name).into()),
            addr: sym.addr().map(Address),
            filename: sym.filename().map(From::from),
            lineno: sym.lineno(),
            colno: sym.colno(),
        }
Repository: tokio-rs/tokio
File: tokio/src/runtime/dump.rs (Rust)

Code before:
/// A snapshot of a runtime's state.
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Dump {
    tasks: Tasks,
}

/// Snapshots of tasks.
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Tasks {
    tasks: Vec<Task>,
}

/// A snapshot of a task.
///
/// See [Handle::dump][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Task {
    id: Id,
    trace: Trace,
}

/// A backtrace symbol. This is similar to [backtrace::BacktraceSymbol],
/// but is a separate struct to avoid public dependency issues.
///
/// This struct is guaranteed to be pure data and operations involving
/// it will not call platform functions that take an unpredictable amount
/// of time to finish.
#[derive(Clone, Debug)]
pub struct BacktraceSymbol {
    name: Option<Box<[u8]>>,
    name_demangled: Option<Box<str>>,
    addr: Option<*mut std::ffi::c_void>,
    filename: Option<std::path::PathBuf>,
    lineno: Option<u32>,
    colno: Option<u32>,
}

impl BacktraceSymbol {
    pub(crate) fn from_backtrace_symbol(sym: &backtrace::BacktraceSymbol) -> Self {
        let name = sym.name();
        Self {
            name: name.as_ref().map(|name| name.as_bytes().into()),
            name_demangled: name.map(|name| format!("{}", name).into()),
            addr: sym.addr(),
            filename: sym.filename().map(From::from),
            lineno: sym.lineno(),
            colno: sym.colno(),

Review comment (suggestion):
Rust types should be in backticks. This way, they don't need to be in the spellcheck dictionary.
```suggestion
/// A backtrace symbol. This is similar to [`backtrace::BacktraceSymbol`],
```

Code after:
/// A snapshot of a runtime's state.
///
/// See [`Handle::dump`][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Dump {
    tasks: Tasks,
}

/// Snapshots of tasks.
///
/// See [`Handle::dump`][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Tasks {
    tasks: Vec<Task>,
}

/// A snapshot of a task.
///
/// See [`Handle::dump`][crate::runtime::Handle::dump].
#[derive(Debug)]
pub struct Task {
    id: Id,
    trace: Trace,
}

/// Represents an address that should not be dereferenced.
///
/// This type exists to get the auto traits correct, the public API
/// uses raw pointers to make life easier for users.
#[derive(Copy, Clone, Debug)]
struct Address(*mut std::ffi::c_void);

// Safe since Address should not be dereferenced
unsafe impl Send for Address {}
unsafe impl Sync for Address {}

/// A backtrace symbol.
///
/// This struct provides accessors for backtrace symbols, similar to [`backtrace::BacktraceSymbol`].
#[derive(Clone, Debug)]
pub struct BacktraceSymbol {
    name: Option<Box<[u8]>>,
    name_demangled: Option<Box<str>>,
    addr: Option<Address>,
    filename: Option<std::path::PathBuf>,
    lineno: Option<u32>,
    colno: Option<u32>,
}

impl BacktraceSymbol {
    pub(crate) fn from_backtrace_symbol(sym: &backtrace::BacktraceSymbol) -> Self {
Repository: tokio-rs/tokio
File: tokio/src/runtime/tests/task.rs (Rust)

Code before:
}

struct AssertDrop {
    is_dropped: Arc<AtomicBool>,
}
impl AssertDrop {
    fn new() -> (Self, AssertDropHandle) {
        let shared = Arc::new(AtomicBool::new(false));
        (
            AssertDrop {
                is_dropped: shared.clone(),
            },
            AssertDropHandle {
                is_dropped: shared.clone(),
            },
        )
    }
}
impl Drop for AssertDrop {
    fn drop(&mut self) {
        self.is_dropped.store(true, Ordering::SeqCst);
    }
}

#[test]
fn drop_tasks_with_reference_cycle() {
    let rt = runtime::Builder::new_current_thread().build().unwrap();

    rt.block_on(async {
        let (tx, mut rx) = mpsc::channel(1);

        let barrier = Arc::new(Barrier::new(3));
        let barrier_a = barrier.clone();
        let barrier_b = barrier.clone();

        let a = spawn(async move {
            let b = rx.recv().await.unwrap();

            // Poll the JoinHandle once. This registers the waker.
            // The other task cannot have finished at this point due to the barrier below.
            futures::future::select(b, std::future::ready(())).await;

            barrier_a.wait().await;
        });

        let b = spawn(async move {
            // Poll the JoinHandle once. This registers the waker.
            // The other task cannot have finished at this point due to the barrier below.
            futures::future::select(a, std::future::ready(())).await;

            barrier_b.wait().await;

Review comment (suggestion):
This does not use any Tokio internals, so it does not need to be inside `src/`. It should be in `tokio/tests/` instead.

Code after:
struct AssertDrop {
    is_dropped: Arc<AtomicBool>,
}
impl AssertDrop {
    fn new() -> (Self, AssertDropHandle) {
        let shared = Arc::new(AtomicBool::new(false));
        (
            AssertDrop {
                is_dropped: shared.clone(),
            },
            AssertDropHandle {
                is_dropped: shared.clone(),
            },
        )
    }
}
impl Drop for AssertDrop {
    fn drop(&mut self) {
        self.is_dropped.store(true, Ordering::SeqCst);
    }
}

// A Notified does not shut down on drop, but it is dropped once the ref-count
// hits zero.
#[test]
fn create_drop1() {
    let (ad, handle) = AssertDrop::new();
    let (notified, join) = unowned(
        async {
            drop(ad);
            unreachable!()
        },
        NoopSchedule,
        Id::next(),
    );
    drop(notified);
    handle.assert_not_dropped();
    drop(join);
    handle.assert_dropped();
}

#[test]
fn create_drop2() {
    let (ad, handle) = AssertDrop::new();
    let (notified, join) = unowned(
        async {
            drop(ad);
            unreachable!()
        },
        NoopSchedule,
        Id::next(),
Repository: tokio-rs/tokio
File: tokio/tests/signal_info.rs (Rust)

Code before:

mod support {
    pub mod signal;
}
use support::signal::send_signal;

use tokio::signal;
use tokio::signal::unix::SignalKind;
use tokio::sync::oneshot;

#[tokio::test]
async fn siginfo() {
    let mut sig = signal::unix::signal(SignalKind::info()).expect("installed signal handler");

    let (fire, wait) = oneshot::channel();

    // NB: simulate a signal coming in by exercising our signal handler
    // to avoid complications with sending SIGINFO to the test process
    tokio::spawn(async {
        wait.await.expect("wait failed");
        send_signal(libc::SIGINFO);
    });

    let _ = fire.send(());

    sig.recv().await.expect("received SIGINFO signal");
}


Review comment (bug):
Should we consider setting a (very long) timeout on this, so that the test failure mode if the signal is not received is a panic, rather than running forever? Just a thought.

Code after:

mod support {
    pub mod signal;
}
use support::signal::send_signal;

use tokio::signal;
use tokio::signal::unix::SignalKind;
use tokio::sync::oneshot;
use tokio::time::{timeout, Duration};

#[tokio::test]
async fn siginfo() {
    let mut sig = signal::unix::signal(SignalKind::info()).expect("installed signal handler");

    let (fire, wait) = oneshot::channel();

    // NB: simulate a signal coming in by exercising our signal handler
    // to avoid complications with sending SIGINFO to the test process
    tokio::spawn(async {
        wait.await.expect("wait failed");
        send_signal(libc::SIGINFO);
    });

    let _ = fire.send(());

    // Add a timeout to ensure the test doesn't hang.
    timeout(Duration::from_secs(5), sig.recv())
        .await
        .expect("received SIGINFO signal in time")
        .expect("received SIGINFO signal");
}
Repository: tokio-rs/tokio
File: tokio-stream/src/wrappers/broadcast.rs (Rust)

Code before:
use std::pin::Pin;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::broadcast::Receiver;

use futures_core::Stream;
use tokio_util::sync::ReusableBoxFuture;

use std::fmt;
use std::task::{ready, Context, Poll};

/// A wrapper around [`tokio::sync::broadcast::Receiver`] that implements [`Stream`].
///
/// # Example
///
/// ```
/// use tokio::sync::broadcast;
/// use tokio_stream::wrappers::BroadcastStream;
/// use tokio_stream::StreamExt;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
///     let (tx, rx) = broadcast::channel(16);
///     tx.send(10)?;
///     tx.send(20)?;
///     drop(tx);
///     
///     let mut stream = BroadcastStream::new(rx);
///     assert_eq!(stream.next().await, Some(Ok(10)));
///     assert_eq!(stream.next().await, Some(Ok(20)));
///     assert_eq!(stream.next().await, None);
///     Ok(())
/// }
/// ```
///
/// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
/// [`Stream`]: trait@futures_core::Stream
#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
pub struct BroadcastStream<T> {
    inner: ReusableBoxFuture<'static, (Result<T, RecvError>, Receiver<T>)>,
}

/// An error returned from the inner stream of a [`BroadcastStream`].
#[derive(Debug, PartialEq, Eq, Clone)]
pub enum BroadcastStreamRecvError {
    /// The receiver lagged too far behind. Attempting to receive again will
    /// return the oldest message still retained by the channel.

Review comment (suggestion):
I like to avoid this error type in examples because it is not `Send` which means it only works in the `block_on` task.

Code after:
use std::pin::Pin;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::broadcast::Receiver;

use futures_core::Stream;
use tokio_util::sync::ReusableBoxFuture;

use std::fmt;
use std::task::{ready, Context, Poll};

/// A wrapper around [`tokio::sync::broadcast::Receiver`] that implements [`Stream`].
///
/// # Example
///
/// ```
/// use tokio::sync::broadcast;
/// use tokio_stream::wrappers::BroadcastStream;
/// use tokio_stream::StreamExt;
///
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() -> Result<(), tokio::sync::broadcast::error::SendError<u8>> {
/// let (tx, rx) = broadcast::channel(16);
/// tx.send(10)?;
/// tx.send(20)?;
/// # // prevent the doc test from hanging
/// drop(tx);
///
/// let mut stream = BroadcastStream::new(rx);
/// assert_eq!(stream.next().await, Some(Ok(10)));
/// assert_eq!(stream.next().await, Some(Ok(20)));
/// assert_eq!(stream.next().await, None);
/// # Ok(())
/// # }
/// ```
///
/// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
/// [`Stream`]: trait@futures_core::Stream
#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
pub struct BroadcastStream<T> {
    inner: ReusableBoxFuture<'static, (Result<T, RecvError>, Receiver<T>)>,
}

/// An error returned from the inner stream of a [`BroadcastStream`].
#[derive(Debug, PartialEq, Eq, Clone)]
pub enum BroadcastStreamRecvError {
    /// The receiver lagged too far behind. Attempting to receive again will
Repository: tokio-rs/tokio
File: tokio-stream/src/wrappers/broadcast.rs (Rust)

Code before:
use std::pin::Pin;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::broadcast::Receiver;

use futures_core::Stream;
use tokio_util::sync::ReusableBoxFuture;

use std::fmt;
use std::task::{ready, Context, Poll};

/// A wrapper around [`tokio::sync::broadcast::Receiver`] that implements [`Stream`].
///
/// # Example
///
/// ```
/// use tokio::sync::broadcast;
/// use tokio_stream::wrappers::BroadcastStream;
/// use tokio_stream::StreamExt;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), tokio::sync::broadcast::error::SendError<u8>> {
/// let (tx, rx) = broadcast::channel(16);
/// tx.send(10)?;
/// tx.send(20)?;
/// # // prevent the doc test from hanging
/// drop(tx);
///
/// let mut stream = BroadcastStream::new(rx);
/// assert_eq!(stream.next().await, Some(Ok(10)));
/// assert_eq!(stream.next().await, Some(Ok(20)));
/// assert_eq!(stream.next().await, None);
/// # Ok(())
/// # }
/// ```
///
/// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
/// [`Stream`]: trait@futures_core::Stream
#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
pub struct BroadcastStream<T> {
    inner: ReusableBoxFuture<'static, (Result<T, RecvError>, Receiver<T>)>,
}

/// An error returned from the inner stream of a [`BroadcastStream`].
#[derive(Debug, PartialEq, Eq, Clone)]
pub enum BroadcastStreamRecvError {

Review comment (suggestion):
If you're hiding this part, let's mark them with the cheaper runtime:
```suggestion
/// # #[tokio::main(flavor = "current_thread")]
```

Code after:
use std::pin::Pin;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::broadcast::Receiver;

use futures_core::Stream;
use tokio_util::sync::ReusableBoxFuture;

use std::fmt;
use std::task::{ready, Context, Poll};

/// A wrapper around [`tokio::sync::broadcast::Receiver`] that implements [`Stream`].
///
/// # Example
///
/// ```
/// use tokio::sync::broadcast;
/// use tokio_stream::wrappers::BroadcastStream;
/// use tokio_stream::StreamExt;
///
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() -> Result<(), tokio::sync::broadcast::error::SendError<u8>> {
/// let (tx, rx) = broadcast::channel(16);
/// tx.send(10)?;
/// tx.send(20)?;
/// # // prevent the doc test from hanging
/// drop(tx);
///
/// let mut stream = BroadcastStream::new(rx);
/// assert_eq!(stream.next().await, Some(Ok(10)));
/// assert_eq!(stream.next().await, Some(Ok(20)));
/// assert_eq!(stream.next().await, None);
/// # Ok(())
/// # }
/// ```
///
/// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
/// [`Stream`]: trait@futures_core::Stream
#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
pub struct BroadcastStream<T> {
    inner: ReusableBoxFuture<'static, (Result<T, RecvError>, Receiver<T>)>,
}

/// An error returned from the inner stream of a [`BroadcastStream`].
#[derive(Debug, PartialEq, Eq, Clone)]
pub enum BroadcastStreamRecvError {