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

Code before:
    /// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
    ///
    /// This is the same as [`SyncIoBridge::new`], but allows passing an arbitrary handle and hence may
    /// be initially invoked outside of an asynchronous context.
    pub fn new_with_handle(src: T, rt: tokio::runtime::Handle) -> Self {
        Self { src, rt }
    }

    /// Consume this bridge, returning the underlying stream.
    pub fn into_inner(self) -> T {
        self.src
    }

    /// Returns a shared reference to the inner resource.
    ///
    /// It is inadvisable to directly read/write from the underlying resource.
    pub fn get_ref(&self) -> &T {
        &self.src
    }

    /// Returns a mutable reference to the inner resource.
    ///
    /// It is inadvisable to directly read/write from the underlying resource.
    pub fn get_mut(&mut self) -> &mut T {
        &mut self.src
    }
}


Review comment (suggestion):
We probably should use the AsRef/AsMut traits instead? I think we do that in most places where we provide accessor to the inner value.

Code after:
    /// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`].
    ///
    /// This is the same as [`SyncIoBridge::new`], but allows passing an arbitrary handle and hence may
    /// be initially invoked outside of an asynchronous context.
    pub fn new_with_handle(src: T, rt: tokio::runtime::Handle) -> Self {
        Self { src, rt }
    }

    /// Consume this bridge, returning the underlying stream.
    pub fn into_inner(self) -> T {
        self.src
    }
}

impl<T> AsMut<T> for SyncIoBridge<T> {
    fn as_mut(&mut self) -> &mut T {
        &mut self.src
    }
}

impl<T> AsRef<T> for SyncIoBridge<T> {
    fn as_ref(&self) -> &T {
        &self.src
    }
}
Repository: tokio-rs/tokio
File: tokio/tests/process_kill_after_wait.rs (Rust)

Code before:

use tokio::process::Command;

#[tokio::test]
async fn kill_after_wait() {
    let mut cmd;

    if cfg!(windows) {
        cmd = Command::new("cmd");
        cmd.arg("/c");
    } else {
        cmd = Command::new("sh");
        cmd.arg("-c");
    }

    let mut child = cmd.arg("exit 2").spawn().unwrap();

    child.start_kill().unwrap();

    child.wait().await.unwrap();

    // Kill after `wait` is fine.
    child.start_kill().unwrap();
    child.kill().await.unwrap();
    child.kill().await.unwrap();
    child.kill().await.unwrap();
}


Review comment (suggestion):
Since `wait` is called and returned successfully, `child` is already fused. So calling `child.kill()` once should suffice.

Code after:

use tokio::process::Command;

#[tokio::test]
async fn kill_after_wait() {
    let mut cmd;

    if cfg!(windows) {
        cmd = Command::new("cmd");
        cmd.arg("/c");
    } else {
        cmd = Command::new("sh");
        cmd.arg("-c");
    }

    let mut child = cmd.arg("exit 2").spawn().unwrap();

    child.start_kill().unwrap();

    child.wait().await.unwrap();

    // Kill after `wait` is fine.
    child.start_kill().unwrap();
    child.kill().await.unwrap();
}
Repository: tokio-rs/tokio
File: tokio/src/macros/support.rs (Rust)

Code before:
cfg_macros! {
    pub use crate::future::maybe_done::maybe_done;

    pub use std::future::poll_fn;

    #[doc(hidden)]
    pub fn thread_rng_n(n: u32) -> u32 {
        crate::runtime::context::thread_rng_n(n)
    }

    #[doc(hidden)]
    pub fn has_budget_remaining() -> bool {
        #[cfg(feature = "rt")]
        { crate::task::coop::has_budget_remaining() }
        #[cfg(not(feature = "rt"))]
        { true }
    }
}

pub use std::future::{Future, IntoFuture};
pub use std::pin::Pin;
pub use std::task::Poll;


Review comment (suggestion):
```suggestion
    #[doc(hidden)]
    #[inline]
    pub fn has_budget_remaining() -> bool {
```

Code after:
cfg_macros! {
    pub use crate::future::maybe_done::maybe_done;

    pub use std::future::poll_fn;

    #[doc(hidden)]
    pub fn thread_rng_n(n: u32) -> u32 {
        crate::runtime::context::thread_rng_n(n)
    }

    cfg_coop! {
        #[doc(hidden)]
        #[inline]
        pub fn poll_budget_available(cx: &mut Context<'_>) -> Poll<()> {
            crate::task::coop::poll_budget_available(cx)
        }
    }

    cfg_not_coop! {
        #[doc(hidden)]
        #[inline]
        pub fn poll_budget_available(_: &mut Context<'_>) -> Poll<()> {
            Poll::Ready(())
        }
    }
}

pub use std::future::{Future, IntoFuture};
pub use std::pin::Pin;
pub use std::task::{Context, Poll};
Repository: tokio-rs/tokio
File: tokio-util/src/io/read_arc.rs (Rust)

Code before:
/// # async fn main() -> std::io::Result<()> {
/// use tokio_util::io::read_exact_arc;
///
/// let read = tokio::io::repeat(42);
///
/// let arc = read_exact_arc(read, 4).await?;
///
/// assert_eq!(&arc[..], &[42; 4]);
/// # Ok(())
/// # }
/// ```
pub async fn read_exact_arc<R>(mut read: R, len: usize) -> io::Result<Arc<[u8]>>
where
    R: AsyncRead + Unpin,
{
    let mut arc = Arc::new_uninit_slice(len);
    let mut buf = unsafe { Arc::get_mut(&mut arc).unwrap_unchecked() };
    let mut bytes_remaining = len;
    while bytes_remaining != 0 {
        let bytes_read = read.read_buf(&mut buf).await?;
        if bytes_read == 0 {
            return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof"));
        }
        bytes_remaining -= bytes_read;
    }
    Ok(unsafe { arc.assume_init() })
}


Review comment (suggestion):
You can avoid bumping the MSRV by implementing `MaybeUninit`-related operations yourself.

Instead of using `Arc::new_uninit_slice`, you can `collect` into an uninitialized slice:
```rs
let mut arc: Arc<[MaybeUninit<u8>]>
    = (0..len).map(|_| MaybeUninit::uninit()).collect();
```
This generates the same assembly code as `new_uninit_slice` (see [playground](https://play.rust-lang.org/?version=stable&mode=release&edition=2024&gist=78125210880d661484ff18d52f94d2fd)).

And `assume_init` would be:
```rs
Arc::from_raw(Arc::into_raw(arc) as *const [u8])
```

Code after:
/// # #[tokio::main]
/// # async fn main() -> std::io::Result<()> {
/// use tokio_util::io::read_exact_arc;
///
/// let read = tokio::io::repeat(42);
///
/// let arc = read_exact_arc(read, 4).await?;
///
/// assert_eq!(&arc[..], &[42; 4]);
/// # Ok(())
/// # }
/// ```
pub async fn read_exact_arc<R: AsyncRead>(read: R, len: usize) -> io::Result<Arc<[u8]>> {
    tokio::pin!(read);
    // TODO(MSRV 1.82): When bumping MSRV, switch to `Arc::new_uninit_slice(len)`. The following is
    // equivalent, and generates the same assembly, but works without requiring MSRV 1.82.
    let arc: Arc<[MaybeUninit<u8>]> = (0..len).map(|_| MaybeUninit::uninit()).collect();
    // TODO(MSRV future): Use `Arc::get_mut_unchecked` once it's stabilized.
    // SAFETY: We're the only owner of the `Arc`, and we keep the `Arc` valid throughout this loop
    // as we write through this reference.
    let mut buf = unsafe { &mut *(Arc::as_ptr(&arc) as *mut [MaybeUninit<u8>]) };
    while !buf.is_empty() {
        if read.read_buf(&mut buf).await? == 0 {
            return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof"));
        }
    }
    // TODO(MSRV 1.82): When bumping MSRV, switch to `arc.assume_init()`. The following is
    // equivalent, and generates the same assembly, but works without requiring MSRV 1.82.
    // SAFETY: This changes `[MaybeUninit<u8>]` to `[u8]`, and we've initialized all the bytes in
    // the loop above.
    Ok(unsafe { Arc::from_raw(Arc::into_raw(arc) as *const [u8]) })
}
Repository: tokio-rs/tokio
File: tokio-util/src/io/read_arc.rs (Rust)

Code before:
use std::io;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt};

/// Read data from an `AsyncRead` into an `Arc`.
///
/// This uses `Arc::new_uninit_slice` and reads into the resulting uninitialized `Arc`.
///
/// # Example
///
/// ```
/// # #[tokio::main]
/// # async fn main() -> std::io::Result<()> {
/// use tokio_util::io::read_exact_arc;
///
/// let read = tokio::io::repeat(42);
///
/// let arc = read_exact_arc(read, 4).await?;
///
/// assert_eq!(&arc[..], &[42; 4]);
/// # Ok(())
/// # }
/// ```
pub async fn read_exact_arc<R>(mut read: R, len: usize) -> io::Result<Arc<[u8]>>
where
    R: AsyncRead + Unpin,
{
    let mut arc = Arc::new_uninit_slice(len);
    let mut buf = unsafe { Arc::get_mut(&mut arc).unwrap_unchecked() };
    let mut bytes_remaining = len;
    while bytes_remaining != 0 {
        let bytes_read = read.read_buf(&mut buf).await?;
        if bytes_read == 0 {
            return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof"));
        }
        bytes_remaining -= bytes_read;
    }
    Ok(unsafe { arc.assume_init() })
}


Review comment (suggestion):
You don't need the `R: Unpin` when you take it by value. You should be able to add `tokio::pin!(read)` as the first line of the function call. Or just take `&mut R`.

Code after:
use std::io;
use std::mem::MaybeUninit;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt};

/// Read data from an `AsyncRead` into an `Arc`.
///
/// This uses `Arc::new_uninit_slice` and reads into the resulting uninitialized `Arc`.
///
/// # Example
///
/// ```
/// # #[tokio::main]
/// # async fn main() -> std::io::Result<()> {
/// use tokio_util::io::read_exact_arc;
///
/// let read = tokio::io::repeat(42);
///
/// let arc = read_exact_arc(read, 4).await?;
///
/// assert_eq!(&arc[..], &[42; 4]);
/// # Ok(())
/// # }
/// ```
pub async fn read_exact_arc<R: AsyncRead>(read: R, len: usize) -> io::Result<Arc<[u8]>> {
    tokio::pin!(read);
    // TODO(MSRV 1.82): When bumping MSRV, switch to `Arc::new_uninit_slice(len)`. The following is
    // equivalent, and generates the same assembly, but works without requiring MSRV 1.82.
    let arc: Arc<[MaybeUninit<u8>]> = (0..len).map(|_| MaybeUninit::uninit()).collect();
    // TODO(MSRV future): Use `Arc::get_mut_unchecked` once it's stabilized.
    // SAFETY: We're the only owner of the `Arc`, and we keep the `Arc` valid throughout this loop
    // as we write through this reference.
    let mut buf = unsafe { &mut *(Arc::as_ptr(&arc) as *mut [MaybeUninit<u8>]) };
    while !buf.is_empty() {
        if read.read_buf(&mut buf).await? == 0 {
            return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof"));
        }
    }
    // TODO(MSRV 1.82): When bumping MSRV, switch to `arc.assume_init()`. The following is
    // equivalent, and generates the same assembly, but works without requiring MSRV 1.82.
    // SAFETY: This changes `[MaybeUninit<u8>]` to `[u8]`, and we've initialized all the bytes in
    // the loop above.
    Ok(unsafe { Arc::from_raw(Arc::into_raw(arc) as *const [u8]) })
}
Repository: tokio-rs/tokio
File: tokio/src/util/blocking_check.rs (Rust)

Code before:
#[cfg(unix)]
use std::os::fd::AsFd;

#[cfg(unix)]
#[allow(unused_variables)]
pub(crate) fn check_socket_for_blocking<S: AsFd>(s: &S) -> crate::io::Result<()> {
    #[cfg(debug_assertions)]
    {
        let sock = socket2::SockRef::from(s);

        if !sock.nonblocking()? {
            eprintln!("Warning: registering a blocking socket, this may be a bug!");
        }
    }

    Ok(())
}

#[cfg(not(unix))]
#[allow(unused_variables)]
pub(crate) fn check_socket_for_blocking<S>(s: &S) -> crate::io::Result<()> {
    // we cannot retrieve the nonblocking status on windows
    // and i dont know how to support wasi yet
    Ok(())
}


Review comment (suggestion):
Is it possible/likely that users are already using stderr for other things? Emitting stderr from within a library feels kinda gross; it might be easier to spot if we just panic here.

Code after:
#[cfg(unix)]
use std::os::fd::AsFd;

#[cfg(unix)]
#[allow(unused_variables)]
#[track_caller]
pub(crate) fn check_socket_for_blocking<S: AsFd>(s: &S) -> crate::io::Result<()> {
    #[cfg(not(tokio_allow_from_blocking_fd))]
    {
        let sock = socket2::SockRef::from(s);

        debug_assert!(
            sock.nonblocking()?,
            "Registering a blocking socket with the tokio runtime is unsupported. \
            If you wish to do anyways, please add `--cfg tokio_allow_from_blocking_fd` to your \
            RUSTFLAGS. See github.com/tokio-rs/tokio/issues/7172 for details."
        );
    }

    Ok(())
}

#[cfg(not(unix))]
#[allow(unused_variables)]
pub(crate) fn check_socket_for_blocking<S>(s: &S) -> crate::io::Result<()> {
    // we cannot retrieve the nonblocking status on windows
    // and i dont know how to support wasi yet
    Ok(())
}
Repository: tokio-rs/tokio
File: tokio/src/util/mod.rs (Rust)

Code before:
cfg_io_driver! {
    pub(crate) mod bit;
}

#[cfg(feature = "rt")]
pub(crate) mod atomic_cell;

pub(crate) mod metric_atomics;

#[cfg(any(feature = "rt", feature = "signal", feature = "process"))]
pub(crate) mod once_cell;

#[cfg(feature = "net")]
pub(crate) mod blocking_check;

#[cfg(any(
    // io driver uses `WakeList` directly
    feature = "net",
    feature = "process",
    // `sync` enables `Notify` and `batch_semaphore`, which require `WakeList`.
    feature = "sync",
    // `fs` uses `batch_semaphore`, which requires `WakeList`.
    feature = "fs",
    // rt and signal use `Notify`, which requires `WakeList`.
    feature = "rt",
    feature = "signal",
    // time driver uses `WakeList` in `Handle::process_at_time`.
    feature = "time",
))]
mod wake_list;
#[cfg(any(
    feature = "net",
    feature = "process",
    feature = "sync",
    feature = "fs",
    feature = "rt",
    feature = "signal",
    feature = "time",
))]

Review comment (suggestion):
Consider re-exporting `check_socket_for_blocking` here to make the imports more nice.

Code after:
cfg_io_driver! {
    pub(crate) mod bit;
}

#[cfg(feature = "rt")]
pub(crate) mod atomic_cell;

#[cfg(feature = "net")]
mod blocking_check;
#[cfg(feature = "net")]
#[allow(unused_imports)]
pub(crate) use blocking_check::check_socket_for_blocking;

pub(crate) mod metric_atomics;

#[cfg(any(feature = "rt", feature = "signal", feature = "process"))]
pub(crate) mod once_cell;

#[cfg(any(
    // io driver uses `WakeList` directly
    feature = "net",
    feature = "process",
    // `sync` enables `Notify` and `batch_semaphore`, which require `WakeList`.
    feature = "sync",
    // `fs` uses `batch_semaphore`, which requires `WakeList`.
    feature = "fs",
    // rt and signal use `Notify`, which requires `WakeList`.
    feature = "rt",
    feature = "signal",
    // time driver uses `WakeList` in `Handle::process_at_time`.
    feature = "time",
))]
mod wake_list;
#[cfg(any(
    feature = "net",
    feature = "process",
    feature = "sync",
    feature = "fs",
    feature = "rt",
Repository: tokio-rs/tokio
File: tokio/src/util/blocking_check.rs (Rust)

Code before:
#[cfg(unix)]
use std::os::fd::AsFd;

#[cfg(unix)]
#[allow(unused_variables)]
pub(crate) fn check_socket_for_blocking<S: AsFd>(s: &S) -> crate::io::Result<()> {
    #[cfg(debug_assertions)]
    {
        let sock = socket2::SockRef::from(s);

        if !sock.nonblocking()? {
            eprintln!("Warning: registering a blocking socket, this may be a bug!");
        }
    }

    Ok(())
}

#[cfg(not(unix))]
#[allow(unused_variables)]
pub(crate) fn check_socket_for_blocking<S>(s: &S) -> crate::io::Result<()> {
    // we cannot retrieve the nonblocking status on windows
    // and i dont know how to support wasi yet
    Ok(())
}


Review comment (suggestion):
The warning should say that it comes from Tokio. Perhaps include `Location::caller` via `#[track_caller]` to print the caller location.

Code after:
#[cfg(unix)]
use std::os::fd::AsFd;

#[cfg(unix)]
#[allow(unused_variables)]
#[track_caller]
pub(crate) fn check_socket_for_blocking<S: AsFd>(s: &S) -> crate::io::Result<()> {
    #[cfg(not(tokio_allow_from_blocking_fd))]
    {
        let sock = socket2::SockRef::from(s);

        debug_assert!(
            sock.nonblocking()?,
            "Registering a blocking socket with the tokio runtime is unsupported. \
            If you wish to do anyways, please add `--cfg tokio_allow_from_blocking_fd` to your \
            RUSTFLAGS. See github.com/tokio-rs/tokio/issues/7172 for details."
        );
    }

    Ok(())
}

#[cfg(not(unix))]
#[allow(unused_variables)]
pub(crate) fn check_socket_for_blocking<S>(s: &S) -> crate::io::Result<()> {
    // we cannot retrieve the nonblocking status on windows
    // and i dont know how to support wasi yet
    Ok(())
}
Repository: tokio-rs/tokio
File: tokio/src/util/typeid.rs (Rust)

Code before:
use std::{
    any::TypeId,
    marker::PhantomData,
    mem::{self, ManuallyDrop},
};

// SAFETY: this function does not compare lifetimes. Values returned as `Ok`
// may have their lifetimes extended.
pub(super) unsafe fn try_transmute<Src, Target: 'static>(x: Src) -> Result<Target, Src> {
    if nonstatic_typeid::<Src>() == nonstatic_typeid::<Target>() {
        let x = ManuallyDrop::new(x);
        Ok(mem::transmute_copy::<Src, Target>(&x))
    } else {
        Err(x)
    }
}

// https://github.com/dtolnay/typeid/blob/b06a3c08a0eaccc7df6091ade1ae4e3fb53609d5/src/lib.rs#L197-L222
#[inline(always)]
fn nonstatic_typeid<T>() -> TypeId
where
    T: ?Sized,
{
    trait NonStaticAny {
        fn get_type_id(&self) -> TypeId
        where
            Self: 'static;
    }

    impl<T: ?Sized> NonStaticAny for PhantomData<T> {
        #[inline(always)]
        fn get_type_id(&self) -> TypeId
        where
            Self: 'static,
        {

Review comment (suggestion):
Let's only use the hack where required.
```suggestion
    if nonstatic_typeid::<Src>() == TypeId::of::<Target>() {
```

Code after:
use std::{
    any::TypeId,
    marker::PhantomData,
    mem::{self, ManuallyDrop},
};

// SAFETY: this function does not compare lifetimes. Values returned as `Ok`
// may have their lifetimes extended.
pub(super) unsafe fn try_transmute<Src, Target: 'static>(x: Src) -> Result<Target, Src> {
    if nonstatic_typeid::<Src>() == TypeId::of::<Target>() {
        let x = ManuallyDrop::new(x);
        Ok(mem::transmute_copy::<Src, Target>(&x))
    } else {
        Err(x)
    }
}

// https://github.com/dtolnay/typeid/blob/b06a3c08a0eaccc7df6091ade1ae4e3fb53609d5/src/lib.rs#L197-L222
#[inline(always)]
fn nonstatic_typeid<T>() -> TypeId
where
    T: ?Sized,
{
    trait NonStaticAny {
        fn get_type_id(&self) -> TypeId
        where
            Self: 'static;
    }

    impl<T: ?Sized> NonStaticAny for PhantomData<T> {
        #[inline(always)]
        fn get_type_id(&self) -> TypeId
        where
            Self: 'static,
        {
Repository: tokio-rs/tokio
File: tokio/tests/rt_threaded.rs (Rust)

Code before:
            .unwrap()
        });
    }
}

#[test]
fn yield_now_in_block_in_place() {
    let rt = runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .build()
        .unwrap();

    rt.block_on(async {
        tokio::spawn(async {
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(tokio::task::yield_now());
            })
        })
        .await
        .unwrap()
    })
}

#[test]
fn mutex_in_block_in_place() {
    const N: usize = 128;

    let rt = runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .build()
        .unwrap();

    rt.block_on(async {
        let lock = tokio::sync::Mutex::new(0);

        tokio::spawn(async move {
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(async move {
                    for i in 0..(N + 1) {
                        let mut guard = lock.lock().await;
                        *guard = i;
                    }
                });
            })
        })
        .await
        .unwrap();
    })
}

// Testing the tuning logic is tricky as it is inherently timing based, and more

Review comment (suggestion):
To match other constants containing the budget and make them easier to find.
```suggestion
    const BUDGET: usize = 128;
```

Code after:
            .unwrap()
        });
    }
}

#[test]
fn yield_now_in_block_in_place() {
    let rt = runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .build()
        .unwrap();

    rt.block_on(async {
        tokio::spawn(async {
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(tokio::task::yield_now());
            })
        })
        .await
        .unwrap()
    })
}

#[test]
fn mutex_in_block_in_place() {
    const BUDGET: usize = 128;

    let rt = runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .build()
        .unwrap();

    rt.block_on(async {
        let lock = tokio::sync::Mutex::new(0);

        tokio::spawn(async move {
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(async move {
                    for i in 0..(BUDGET + 1) {
                        let mut guard = lock.lock().await;
                        *guard = i;
                    }
                });
            })
        })
        .await
        .unwrap();
    })
}

// Testing the tuning logic is tricky as it is inherently timing based, and more
Repository: tokio-rs/tokio
File: tokio-util/tests/task_join_map.rs (Rust)

Code before:
    map.spawn(1, async {
        recv.await.unwrap();
        2
    });

    assert_eq!(map.len(), 1);

    tokio::select! {
        biased;
        res = map.join_next() => match res {
            Some((_key, res)) => panic!("Task {res:?} exited."),
            None => panic!("Phantom task completion."),
        },
        () = tokio::task::yield_now() => {},
    }

    send.send(()).unwrap();

    let (key, res) = map.join_next().await.unwrap();
    assert_eq!(key, 1);
    assert_eq!(res.unwrap(), 2);

    assert!(map.join_next().await.is_none());
}

#[cfg(not(target_arch = "wasm32"))]
#[tokio::test]
async fn duplicate_keys_drop() {
    #[derive(Hash, Debug, PartialEq, Eq)]
    struct Key;
    impl Drop for Key {
        fn drop(&mut self) {
            panic!("drop called for key");
        }
    }

    let (send, recv) = oneshot::channel::<()>();

    let mut map = JoinMap::new();

    map.spawn(Key, async { recv.await.unwrap() });

    // replace the task, force it to drop the key and abort the task
    // we should expect it to panic when dropping the key.
    let _ = std::panic::catch_unwind(AssertUnwindSafe(|| map.spawn(Key, async {}))).unwrap_err();

    // don't panic when this key drops.
    let (key, _) = map.join_next().await.unwrap();
    std::mem::forget(key);

    // original task should have been aborted, so the sender should be dangling.

Review comment (suggestion):
There are downstream users of Tokio who compile with panic=abort in cases other than wasm.
```suggestion
#[cfg_attr(not(panic = "unwind"), ignore)]
```

Code after:
    map.spawn(1, async {
        recv.await.unwrap();
        2
    });

    assert_eq!(map.len(), 1);

    tokio::select! {
        biased;
        res = map.join_next() => match res {
            Some((_key, res)) => panic!("Task {res:?} exited."),
            None => panic!("Phantom task completion."),
        },
        () = tokio::task::yield_now() => {},
    }

    send.send(()).unwrap();

    let (key, res) = map.join_next().await.unwrap();
    assert_eq!(key, 1);
    assert_eq!(res.unwrap(), 2);

    assert!(map.join_next().await.is_none());
}

#[cfg_attr(not(panic = "unwind"), ignore)]
#[tokio::test]
async fn duplicate_keys_drop() {
    #[derive(Hash, Debug, PartialEq, Eq)]
    struct Key;
    impl Drop for Key {
        fn drop(&mut self) {
            panic!("drop called for key");
        }
    }

    let (send, recv) = oneshot::channel::<()>();

    let mut map = JoinMap::new();

    map.spawn(Key, async { recv.await.unwrap() });

    // replace the task, force it to drop the key and abort the task
    // we should expect it to panic when dropping the key.
    let _ = std::panic::catch_unwind(AssertUnwindSafe(|| map.spawn(Key, async {}))).unwrap_err();

    // don't panic when this key drops.
    let (key, _) = map.join_next().await.unwrap();
    std::mem::forget(key);

    // original task should have been aborted, so the sender should be dangling.