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

Code before:
use crate::io::{AsyncRead, ReadBuf};

use pin_project_lite::pin_project;
use std::future::Future;
use std::io;
use std::marker::PhantomPinned;
use std::marker::Unpin;
use std::pin::Pin;
use std::task::{Context, Poll};

/// A future which can be used to easily read exactly enough bytes to fill
/// a buffer.
///
/// Created by the [`read_exact`].
/// [`read_exact`]: crate::io::AsyncReadExt::read_exact
pub(crate) fn read_exact<'a, A>(reader: &'a mut A, buf: &'a mut [u8]) -> ReadExact<'a, A>
where
    A: AsyncRead + Unpin + ?Sized,
{
    ReadExact {
        reader,
        buf: ReadBuf::new(buf),
        _pin: PhantomPinned,
    }
}

pin_project! {
    /// Creates a future which will read exactly enough bytes to fill `buf`,
    /// returning an error if EOF is hit sooner.
    ///
    /// On success the number of bytes is returned
    #[derive(Debug)]
    #[must_use = "futures do nothing unless you `.await` or poll them"]
    pub struct ReadExact<'a, A: ?Sized> {
        reader: &'a mut A,
        buf: ReadBuf<'a>,
        // Make this future `!Unpin` for compatibility with async trait methods.
        #[pin]
        _pin: PhantomPinned,
    }

Review comment (suggestion):
This just makes the comment harder to understand. We explicitly write out `AsyncReadExt` because there are multiple methods of the same name.

Code after:
use crate::io::{AsyncRead, ReadBuf};

use pin_project_lite::pin_project;
use std::future::Future;
use std::io;
use std::marker::PhantomPinned;
use std::marker::Unpin;
use std::pin::Pin;
use std::task::{Context, Poll};

/// A future which can be used to easily read exactly enough bytes to fill
/// a buffer.
///
/// Created by the [`AsyncReadExt::read_exact`][read_exact].
/// [read_exact]: [crate::io::AsyncReadExt::read_exact]
pub(crate) fn read_exact<'a, A>(reader: &'a mut A, buf: &'a mut [u8]) -> ReadExact<'a, A>
where
    A: AsyncRead + Unpin + ?Sized,
{
    ReadExact {
        reader,
        buf: ReadBuf::new(buf),
        _pin: PhantomPinned,
    }
}

pin_project! {
    /// Creates a future which will read exactly enough bytes to fill `buf`,
    /// returning an error if EOF is hit sooner.
    ///
    /// On success the number of bytes is returned
    #[derive(Debug)]
    #[must_use = "futures do nothing unless you `.await` or poll them"]
    pub struct ReadExact<'a, A: ?Sized> {
        reader: &'a mut A,
        buf: ReadBuf<'a>,
        // Make this future `!Unpin` for compatibility with async trait methods.
        #[pin]
        _pin: PhantomPinned,
    }
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_ext/peekable.rs (Rust)

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

use futures_core::Stream;
use pin_project_lite::pin_project;

use crate::StreamExt;

pin_project! {
    /// Stream returned by the [`chain`](super::StreamExt::peekable) method.
    pub struct Peekable<T: Stream> {
        peek: Option<T::Item>,
        #[pin]
        stream: T,
    }
}

impl<T: Stream> Peekable<T> {
    pub(crate) fn new(stream: T) -> Self {
        Self { peek: None, stream }
    }

    /// Peek at the next item in the stream.
    pub async fn peek(&mut self) -> Option<&T::Item>
    where
        T: Unpin,
    {
        if let Some(ref it) = self.peek {
            Some(it)
        } else {
            self.peek = self.next().await;
            self.peek.as_ref()
        }
    }
}


Review comment (suggestion):
```suggestion
    /// Stream returned by the [`peekable`](super::StreamExt::peekable) method.
```

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

use futures_core::Stream;
use pin_project_lite::pin_project;

use crate::stream_ext::Fuse;
use crate::StreamExt;

pin_project! {
    /// Stream returned by the [`chain`](super::StreamExt::peekable) method.
    pub struct Peekable<T: Stream> {
        peek: Option<T::Item>,
        #[pin]
        stream: Fuse<T>,
    }
}

impl<T: Stream> Peekable<T> {
    pub(crate) fn new(stream: T) -> Self {
        let stream = stream.fuse();
        Self { peek: None, stream }
    }

    /// Peek at the next item in the stream.
    pub async fn peek(&mut self) -> Option<&T::Item>
    where
        T: Unpin,
    {
        if let Some(ref it) = self.peek {
            Some(it)
        } else {
            self.peek = self.next().await;
            self.peek.as_ref()
        }
    }
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_ext/peekable.rs (Rust)

Code before:
    /// Stream returned by the [`chain`](super::StreamExt::peekable) method.
    pub struct Peekable<T: Stream> {
        peek: Option<T::Item>,
        #[pin]
        stream: T,
    }
}

impl<T: Stream> Peekable<T> {
    pub(crate) fn new(stream: T) -> Self {
        Self { peek: None, stream }
    }

    /// Peek at the next item in the stream.
    pub async fn peek(&mut self) -> Option<&T::Item>
    where
        T: Unpin,
    {
        if let Some(ref it) = self.peek {
            Some(it)
        } else {
            self.peek = self.next().await;
            self.peek.as_ref()
        }
    }
}

impl<T: Stream> Stream for Peekable<T> {
    type Item = T::Item;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();
        if let Some(it) = this.peek.take() {
            Poll::Ready(Some(it))
        } else {
            this.stream.poll_next(cx)
        }
    }
}


Review comment (suggestion):
This is only correct if the inner stream is fused. Some usages of `peek` can end up calling `next` on it after it has returned `None`.

Code after:
pin_project! {
    /// Stream returned by the [`chain`](super::StreamExt::peekable) method.
    pub struct Peekable<T: Stream> {
        peek: Option<T::Item>,
        #[pin]
        stream: Fuse<T>,
    }
}

impl<T: Stream> Peekable<T> {
    pub(crate) fn new(stream: T) -> Self {
        let stream = stream.fuse();
        Self { peek: None, stream }
    }

    /// Peek at the next item in the stream.
    pub async fn peek(&mut self) -> Option<&T::Item>
    where
        T: Unpin,
    {
        if let Some(ref it) = self.peek {
            Some(it)
        } else {
            self.peek = self.next().await;
            self.peek.as_ref()
        }
    }
}

impl<T: Stream> Stream for Peekable<T> {
    type Item = T::Item;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();
        if let Some(it) = this.peek.take() {
            Poll::Ready(Some(it))
        } else {
            this.stream.poll_next(cx)
        }
    }
}
Repository: tokio-rs/tokio
File: tokio/tests/rt_metrics.rs (Rust)

Code before:
    assert_eq!(0, metrics.stop_tasks_count());

    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.start_tasks_count());
        assert_eq!(0, metrics.stop_tasks_count());
    }))
    .unwrap();

    assert_eq!(1, rt.metrics().start_tasks_count());
    assert_eq!(1, rt.metrics().stop_tasks_count());

    let rt = threaded();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.start_tasks_count());
    assert_eq!(0, metrics.stop_tasks_count());

    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.start_tasks_count());
        assert_eq!(0, metrics.stop_tasks_count());
    }))
    .unwrap();

    // for some reason, sometimes the stop count doesn't get a chance to incremenet before we get here.
    // Only observed on single-cpu systems. Most likely the worker thread doesn't a chance to clean up
    // the spawned task yet. We yield to give it an opportunity.
    std::thread::yield_now();

    assert_eq!(1, rt.metrics().start_tasks_count());
    assert_eq!(1, rt.metrics().stop_tasks_count());
}

#[test]
fn remote_schedule_count() {
    use std::thread;

    let rt = current_thread();
    let handle = rt.handle().clone();
    let task = thread::spawn(move || {
        handle.spawn(async {
            // DO nothing
        })
    })
    .join()
    .unwrap();

    rt.block_on(task).unwrap();

    assert_eq!(1, rt.metrics().remote_schedule_count());

    let rt = threaded();
    let handle = rt.handle().clone();

Review comment (bug):
I would strongly prefer that we don't add flaky tests. If it's not possible to fix this, then please add a loop that runs this test many times and only fails it if it is failing repeatedly.

Code after:
    let metrics = rt.metrics();
    assert_eq!(0, metrics.num_active_tasks());
    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.num_active_tasks());
    }))
    .unwrap();

    // try for 10 seconds to see if this eventually succeeds.
    // wake_join() is called before the task is released, so in multithreaded
    // code, this means we sometimes exit the block_on before the counter decrements.
    for _ in 0..100 {
        if rt.metrics().num_active_tasks() == 0 {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    assert_eq!(0, rt.metrics().num_active_tasks());
}

#[test]
fn spawned_tasks_count() {
    let rt = current_thread();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.spawned_tasks_count());

    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.spawned_tasks_count());
    }))
    .unwrap();

    assert_eq!(1, rt.metrics().spawned_tasks_count());

    let rt = threaded();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.spawned_tasks_count());

    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.spawned_tasks_count());
    }))
    .unwrap();

    assert_eq!(1, rt.metrics().spawned_tasks_count());
}

#[test]
fn remote_schedule_count() {
    use std::thread;

    let rt = current_thread();
    let handle = rt.handle().clone();
    let task = thread::spawn(move || {
Repository: tokio-rs/tokio
File: tokio/src/process/unix/pidfd_reaper.rs (Rust)

Code before:

impl Pidfd {
    fn open(pid: u32) -> Option<Pidfd> {
        // Store false (0) to reduce executable size
        static NO_PIDFD_SUPPORT: AtomicBool = AtomicBool::new(false);

        if NO_PIDFD_SUPPORT.load(Relaxed) {
            return None;
        }

        unsafe {
            let fd = syscall(SYS_pidfd_open, pid, PIDFD_NONBLOCK);
            if fd == -1 {
                let errno = *__errno_location();

                if errno == ENOSYS {
                    NO_PIDFD_SUPPORT.store(true, Relaxed)
                }

                None
            } else {
                Some(Pidfd {
                    fd: File::from_raw_fd(fd as i32),
                })
            }
        }
    }
}

impl AsRawFd for Pidfd {
    fn as_raw_fd(&self) -> RawFd {
        self.fd.as_raw_fd()
    }
}

impl Source for Pidfd {
    fn register(
        &mut self,
        registry: &mio::Registry,
        token: mio::Token,
        interest: mio::Interest,
    ) -> io::Result<()> {
        SourceFd(&self.as_raw_fd()).register(registry, token, interest)
    }

    fn reregister(
        &mut self,
        registry: &mio::Registry,
        token: mio::Token,
        interest: mio::Interest,
    ) -> io::Result<()> {

Review comment (suggestion):
Generally, it is preferred to have a separate unsafe block for each unsafe operation, and to annotate each block with a `// SAFETY:` comment justifying its correctness.

Code after:

impl Pidfd {
    fn open(pid: u32) -> Option<Pidfd> {
        // Store false (0) to reduce executable size
        static NO_PIDFD_SUPPORT: AtomicBool = AtomicBool::new(false);

        if NO_PIDFD_SUPPORT.load(Relaxed) {
            return None;
        }

        // Safety: The following function calls invovkes syscall pidfd_open,
        // which takes two parameter: pidfd_open(fd: c_int, flag: c_int)
        let fd = unsafe { syscall(SYS_pidfd_open, pid, PIDFD_NONBLOCK) };
        if fd == -1 {
            let errno = io::Error::last_os_error().raw_os_error().unwrap();

            if errno == ENOSYS {
                NO_PIDFD_SUPPORT.store(true, Relaxed)
            }

            None
        } else {
            // Safety: pidfd_open returns -1 on error or a valid fd with ownership.
            Some(Pidfd {
                fd: unsafe { File::from_raw_fd(fd as i32) },
            })
        }
    }
}

impl AsRawFd for Pidfd {
    fn as_raw_fd(&self) -> RawFd {
        self.fd.as_raw_fd()
    }
}

impl Source for Pidfd {
    fn register(
        &mut self,
        registry: &mio::Registry,
        token: mio::Token,
        interest: mio::Interest,
    ) -> io::Result<()> {
        SourceFd(&self.as_raw_fd()).register(registry, token, interest)
    }

    fn reregister(
        &mut self,
        registry: &mio::Registry,
        token: mio::Token,
        interest: mio::Interest,
Repository: tokio-rs/tokio
File: tokio/src/sync/mpsc/bounded.rs (Rust)

Code before:
        }
    }
}

impl<T> fmt::Debug for Permit<'_, T> {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("Permit")
            .field("chan", &self.chan)
            .finish()
    }
}

// ===== impl PermitIterator =====

impl<'a, T> Iterator for PermitIterator<'a, T> {
    type Item = Permit<'a, T>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.n == 0 {
            return None;
        }

        self.n -= 1;
        Some(Permit { chan: self.chan })
    }
}

impl<T> Drop for PermitIterator<'_, T> {
    fn drop(&mut self) {
        use chan::Semaphore;

        let semaphore = self.chan.semaphore();

        // Add the remaining permits back to the semaphore
        semaphore.add_permits(self.n as usize);

        // If this is the last sender for this channel, wake the receiver so
        // that it can be notified that the channel is closed.
        if semaphore.is_closed() && semaphore.is_idle() {
            self.chan.wake_rx();
        }
    }
}

impl<T> fmt::Debug for PermitIterator<'_, T> {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("Permit")
            .field("chan", &self.chan)
            .field("capacity", &self.n)
            .finish()
    }

Review comment (suggestion):
Please add an implementation of `size_hint` as well. I think it also makes sense to implement `ExactSizeIterator`, since we know the exact length.

Code after:
    ///     assert_eq!(tx.capacity(), 4);
    /// }
    /// ```
    ///
    /// [`channel`]: channel
    /// [`max_capacity`]: Sender::max_capacity
    /// [`capacity`]: Sender::capacity
    pub fn max_capacity(&self) -> usize {
        self.chan.semaphore().bound
    }
}

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Self {
        Sender {
            chan: self.chan.clone(),
        }
    }
}

impl<T> fmt::Debug for Sender<T> {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("Sender")
            .field("chan", &self.chan)
            .finish()
    }
}

impl<T> Clone for WeakSender<T> {
    fn clone(&self) -> Self {
        WeakSender {
            chan: self.chan.clone(),
        }
    }
}

impl<T> WeakSender<T> {
    /// Tries to convert a `WeakSender` into a [`Sender`]. This will return `Some`
    /// if there are other `Sender` instances alive and the channel wasn't
    /// previously dropped, otherwise `None` is returned.
    pub fn upgrade(&self) -> Option<Sender<T>> {
        chan::Tx::upgrade(self.chan.clone()).map(Sender::new)
    }
}

impl<T> fmt::Debug for WeakSender<T> {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("WeakSender").finish()
    }
}
Repository: tokio-rs/tokio
File: tokio/tests/sync_mpsc.rs (Rust)

Code before:

    let permit = tx.try_reserve().unwrap();

    // This should fail
    match assert_err!(tx.try_reserve()) {
        TrySendError::Full(()) => {}
        _ => panic!(),
    }

    permit.send("foo");

    assert_eq!(rx.recv().await, Some("foo"));

    // Dropping permit releases the slot.
    let permit = tx.try_reserve().unwrap();
    drop(permit);

    let _permit = tx.try_reserve().unwrap();
}

#[maybe_tokio_test]
async fn try_reserve_many_fails() {
    for i in 4..20 {
        let (tx, mut rx) = mpsc::channel(i);

        let mut permit1 = assert_ok!(tx.try_reserve_many(i - 2));

        // This should fail, there is only two remaining permits
        match assert_err!(tx.try_reserve_many(3)) {
            TrySendError::Full(()) => {}
            _ => panic!(),
        }

        permit1.next().unwrap().send("foo");
        permit1.next().unwrap().send("foo");

        assert_eq!(rx.recv().await, Some("foo"));
        assert_eq!(rx.recv().await, Some("foo"));

        // There are now 4 remaining permits because of the 2 sends/recv
        let permit2 = assert_ok!(tx.try_reserve_many(4));

        // Dropping permit iterator releases the remaining slots.
        drop(permit1);
        drop(permit2);

        // It is possible to reserve all the permits
        assert_ok!(tx.try_reserve_many(i));

        // This should fail because the channel is closed
        drop(rx);

Review comment (suggestion):
This test does a lot of stuff, but it's not clear to me that you're covering all the things I mentioned. Instead of putting everything in the same test, I would like to see several really small and easy-to-understand tests for each case that I mentioned.

Code after:
    let (tx, _rx) = mpsc::channel::<()>(1);

    assert_err!(tx.reserve_many(2).await);
    assert_err!(tx.reserve_many(MAX_PERMITS + 1).await);
    assert_err!(tx.reserve_many(usize::MAX).await);
}

#[test]
fn try_reserve_many_zero() {
    let (tx, rx) = mpsc::channel::<()>(1);

    // Succeeds when not closed.
    assert!(assert_ok!(tx.try_reserve_many(0)).next().is_none());

    // Even when channel is full.
    tx.try_send(()).unwrap();
    assert!(assert_ok!(tx.try_reserve_many(0)).next().is_none());

    drop(rx);

    // Closed error when closed.
    assert_eq!(
        assert_err!(tx.try_reserve_many(0)),
        TrySendError::Closed(())
    );
}

#[maybe_tokio_test]
async fn reserve_many_zero() {
    let (tx, rx) = mpsc::channel::<()>(1);

    // Succeeds when not closed.
    assert!(assert_ok!(tx.reserve_many(0).await).next().is_none());

    // Even when channel is full.
    tx.send(()).await.unwrap();
    assert!(assert_ok!(tx.reserve_many(0).await).next().is_none());

    drop(rx);

    // Closed error when closed.
    assert_err!(tx.reserve_many(0).await);
}

#[maybe_tokio_test]
async fn try_reserve_many_edge_cases() {
    const MAX_PERMITS: usize = tokio::sync::Semaphore::MAX_PERMITS;

    let (tx, rx) = mpsc::channel::<()>(1);

    let mut permit = assert_ok!(tx.try_reserve_many(0));
Repository: tokio-rs/tokio
File: tokio/tests/sync_mpsc.rs (Rust)

Code before:
        assert_eq!(rx.recv().await, Some("foo"));
        assert_eq!(rx.recv().await, Some("foo"));

        // There are now 4 remaining permits because of the 2 sends/recv
        let permit2 = assert_ok!(tx.try_reserve_many(4));

        // Dropping permit iterator releases the remaining slots.
        drop(permit1);
        drop(permit2);

        // It is possible to reserve all the permits
        assert_ok!(tx.try_reserve_many(i));

        // This should fail because the channel is closed
        drop(rx);
        match assert_err!(tx.try_reserve_many(3)) {
            TrySendError::Closed(()) => {}
            _ => panic!(),
        };
    }
}

#[maybe_tokio_test]
async fn reserve_many_and_send() {
    let (tx, mut rx) = mpsc::channel(100);
    for i in 1..100 {
        for permit in assert_ok!(tx.reserve_many(i).await) {
            permit.send("foo");
            assert_eq!(rx.recv().await, Some("foo"));
        }
        assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
    }
}

#[tokio::test]
#[cfg(feature = "full")]
async fn drop_permit_releases_permit() {
    // poll_ready reserves capacity, ensure that the capacity is released if tx
    // is dropped w/o sending a value.
    let (tx1, _rx) = mpsc::channel::<i32>(1);
    let tx2 = tx1.clone();

    let permit = assert_ok!(tx1.reserve().await);

    let mut reserve2 = tokio_test::task::spawn(tx2.reserve());
    assert_pending!(reserve2.poll());

    drop(permit);

    assert!(reserve2.is_woken());
    assert_ready_ok!(reserve2.poll());

Review comment (suggestion):
This test is really nice because it is so small. 
```suggestion
    for i in 0..100 {
```

Code after:
    assert!(assert_ok!(tx.reserve_many(0).await).next().is_none());

    drop(rx);

    // Closed error when closed.
    assert_err!(tx.reserve_many(0).await);
}

#[maybe_tokio_test]
async fn try_reserve_many_edge_cases() {
    const MAX_PERMITS: usize = tokio::sync::Semaphore::MAX_PERMITS;

    let (tx, rx) = mpsc::channel::<()>(1);

    let mut permit = assert_ok!(tx.try_reserve_many(0));
    assert!(permit.next().is_none());

    let permit = tx.try_reserve_many(MAX_PERMITS + 1);
    match assert_err!(permit) {
        TrySendError::Full(..) => {}
        _ => panic!(),
    }

    let permit = tx.try_reserve_many(usize::MAX);
    match assert_err!(permit) {
        TrySendError::Full(..) => {}
        _ => panic!(),
    }

    // Dropping the receiver should close the channel
    drop(rx);
    assert_err!(tx.reserve_many(0).await);
}

#[maybe_tokio_test]
async fn try_reserve_fails() {
    let (tx, mut rx) = mpsc::channel(1);

    let permit = tx.try_reserve().unwrap();

    // This should fail
    match assert_err!(tx.try_reserve()) {
        TrySendError::Full(()) => {}
        _ => panic!(),
    }

    permit.send("foo");

    assert_eq!(rx.recv().await, Some("foo"));

    // Dropping permit releases the slot.
Repository: tokio-rs/tokio
File: tokio/src/io/join.rs (Rust)

Code before:
//! Join two values implementing `AsyncRead` and `AsyncWrite` into a single one.

use crate::io::{AsyncRead, AsyncWrite, ReadBuf};

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

cfg_io_util! {
    pin_project_lite::pin_project! {
        /// Joins two values implementing `AsyncRead` and `AsyncWrite` into a
        /// single handle.
        pub struct Join<R, W> {
            #[pin]
            reader: R,
            #[pin]
            writer: W,
        }
    }

    impl<R, W> Join<R, W>
    where
        R: AsyncRead + Unpin,
        W: AsyncWrite + Unpin,
    {
        /// Join two values implementing `AsyncRead` and `AsyncWrite` into a
        /// single handle.
        pub fn new(reader: R, writer: W) -> Self {
            Self { reader, writer }
        }

        /// Splits this `Join` back into its `AsyncRead` and `AsyncWrite`
        /// components.
        pub fn split(self) -> (R, W) {

Review comment (refactor):
Please remove this `cfg_*` block. You're already wrapping the `mod` statement.

Code after:
//! Join two values implementing `AsyncRead` and `AsyncWrite` into a single one.

use crate::io::{AsyncRead, AsyncWrite, ReadBuf};

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

/// Join two values implementing `AsyncRead` and `AsyncWrite` into a
/// single handle.
pub fn join<R, W>(reader: R, writer: W) -> Join<R, W>
where
    R: AsyncRead,
    W: AsyncWrite,
{
    Join { reader, writer }
}

pin_project_lite::pin_project! {
    /// Joins two values implementing `AsyncRead` and `AsyncWrite` into a
    /// single handle.
    #[derive(Debug)]
    pub struct Join<R, W> {
        #[pin]
        reader: R,
        #[pin]
        writer: W,
    }
}

impl<R, W> Join<R, W>
where
    R: AsyncRead,
    W: AsyncWrite,
{