ronantakizawa/codereview-bench huggingface dataset:
Repository: tokio-rs/tokio
File: tokio/src/loom/std/parking_lot.rs (Rust)

Code before:
//! A minimal adaption of the `parking_lot` synchronization primitives to the
//! equivalent `std::sync` types.
//!
//! This can be extended to additional types/methods as required.

use std::marker::PhantomData;
use std::ops::{Deref, DerefMut};
use std::sync::LockResult;
use std::time::Duration;

// Types that do not need wrapping
pub(crate) use parking_lot::WaitTimeoutResult;

#[derive(Debug)]
pub(crate) struct Mutex<T: ?Sized>(PhantomData<std::sync::Mutex<T>>, parking_lot::Mutex<T>);

#[derive(Debug)]
pub(crate) struct RwLock<T>(PhantomData<std::sync::RwLock<T>>, parking_lot::RwLock<T>);

#[derive(Debug)]
pub(crate) struct Condvar(PhantomData<std::sync::Condvar>, parking_lot::Condvar);

#[derive(Debug)]
pub(crate) struct MutexGuard<'a, T: ?Sized>(
    PhantomData<std::sync::MutexGuard<'a, T>>,
    parking_lot::MutexGuard<'a, T>,
);

#[derive(Debug)]
pub(crate) struct RwLockReadGuard<'a, T: ?Sized>(
    PhantomData<std::sync::RwLockReadGuard<'a, T>>,
    parking_lot::RwLockReadGuard<'a, T>,
);

#[derive(Debug)]
pub(crate) struct RwLockWriteGuard<'a, T: ?Sized>(
    PhantomData<std::sync::RwLockWriteGuard<'a, T>>,
    parking_lot::RwLockWriteGuard<'a, T>,
);

impl<T> Mutex<T> {
    #[inline]
    pub(crate) fn new(t: T) -> Mutex<T> {

Review comment (suggestion):
i have a slight preference for either making these named fields, or breaking them across multiple lines like we do with the guard types...this feels like a _lot_ for one line. not a blocker though.

Code after:
//! A minimal adaption of the `parking_lot` synchronization primitives to the
//! equivalent `std::sync` types.
//!
//! This can be extended to additional types/methods as required.

use std::fmt;
use std::marker::PhantomData;
use std::ops::{Deref, DerefMut};
use std::sync::LockResult;
use std::time::Duration;

// All types in this file are marked with PhantomData to ensure that
// parking_lot's send_guard feature does not leak through and affect when Tokio
// types are Send.
//
// See <https://github.com/tokio-rs/tokio/pull/4359> for more info.

// Types that do not need wrapping
pub(crate) use parking_lot::WaitTimeoutResult;

#[derive(Debug)]
pub(crate) struct Mutex<T: ?Sized>(PhantomData<std::sync::Mutex<T>>, parking_lot::Mutex<T>);

#[derive(Debug)]
pub(crate) struct RwLock<T>(PhantomData<std::sync::RwLock<T>>, parking_lot::RwLock<T>);

#[derive(Debug)]
pub(crate) struct Condvar(PhantomData<std::sync::Condvar>, parking_lot::Condvar);

#[derive(Debug)]
pub(crate) struct MutexGuard<'a, T: ?Sized>(
    PhantomData<std::sync::MutexGuard<'a, T>>,
    parking_lot::MutexGuard<'a, T>,
);

#[derive(Debug)]
pub(crate) struct RwLockReadGuard<'a, T: ?Sized>(
    PhantomData<std::sync::RwLockReadGuard<'a, T>>,
    parking_lot::RwLockReadGuard<'a, T>,
);

#[derive(Debug)]
pub(crate) struct RwLockWriteGuard<'a, T: ?Sized>(
Repository: tokio-rs/tokio
File: tokio/src/runtime/basic_scheduler.rs (Rust)

Code before:
use std::future::Future;
use std::sync::atomic::Ordering::{AcqRel, Release};
use std::sync::Arc;
use std::task::Poll::{Pending, Ready};
use std::time::Duration;

/// Executes tasks on the current thread
pub(crate) struct BasicScheduler {
    /// Core scheduler data is acquired by a thread entering `block_on`.
    core: AtomicCell<Core>,

    /// Notifier for waking up other threads to steal the
    /// parker.
    notify: Notify,

    /// Sendable task spawner
    spawner: Spawner,

    /// This is usually None, but right before dropping the BasicScheduler, it
    /// is changed to `Some` with the context being the runtime's own context.
    /// This ensures that any tasks dropped in the `BasicScheduler`s destructor
    /// run in that runtime's context.
    context_guard: Option<EnterGuard>,
}

/// Core data
struct Core {
    /// Scheduler run queue
    tasks: VecDeque<task::Notified<Arc<Shared>>>,

    /// Sendable task spawner
    spawner: Spawner,

    /// Current tick
    tick: u8,

    /// Thread park handle
    ///
    /// The parker is removed before starting to park
    park: Option<Driver>,

    /// Stats batcher
    stats: WorkerStatsBatcher,
}

#[derive(Clone)]
pub(crate) struct Spawner {
    shared: Arc<Shared>,
}

/// A remote scheduler entry.

Review comment (refactor):
It would probably be helpful to have better docs here with this refactor. Core data isn't super helpful lol

Code after:
use std::future::Future;
use std::sync::atomic::Ordering::{AcqRel, Release};
use std::sync::Arc;
use std::task::Poll::{Pending, Ready};
use std::time::Duration;

/// Executes tasks on the current thread
pub(crate) struct BasicScheduler {
    /// Core scheduler data is acquired by a thread entering `block_on`.
    core: AtomicCell<Core>,

    /// Notifier for waking up other threads to steal the
    /// driver.
    notify: Notify,

    /// Sendable task spawner
    spawner: Spawner,

    /// This is usually None, but right before dropping the BasicScheduler, it
    /// is changed to `Some` with the context being the runtime's own context.
    /// This ensures that any tasks dropped in the `BasicScheduler`s destructor
    /// run in that runtime's context.
    context_guard: Option<EnterGuard>,
}

/// Data required for executing the scheduler. The struct is passed around to
/// a function that will perform the scheduling work and acts as a capability token.
struct Core {
    /// Scheduler run queue
    tasks: VecDeque<task::Notified<Arc<Shared>>>,

    /// Sendable task spawner
    spawner: Spawner,

    /// Current tick
    tick: u8,

    /// Runtime driver
    ///
    /// The driver is removed before starting to park the thread
    driver: Option<Driver>,

    /// Stats batcher
    stats: WorkerStatsBatcher,
}

#[derive(Clone)]
pub(crate) struct Spawner {
    shared: Arc<Shared>,
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/basic_scheduler.rs (Rust)

Code before:
use std::sync::atomic::Ordering::{AcqRel, Release};
use std::sync::Arc;
use std::task::Poll::{Pending, Ready};
use std::time::Duration;

/// Executes tasks on the current thread
pub(crate) struct BasicScheduler {
    /// Core scheduler data is acquired by a thread entering `block_on`.
    core: AtomicCell<Core>,

    /// Notifier for waking up other threads to steal the
    /// driver.
    notify: Notify,

    /// Sendable task spawner
    spawner: Spawner,

    /// This is usually None, but right before dropping the BasicScheduler, it
    /// is changed to `Some` with the context being the runtime's own context.
    /// This ensures that any tasks dropped in the `BasicScheduler`s destructor
    /// run in that runtime's context.
    context_guard: Option<EnterGuard>,
}

/// Data required for executing the scheduler. The struct is passed around to
/// function that perform scheduling work and acts as a capability token.
struct Core {
    /// Scheduler run queue
    tasks: VecDeque<task::Notified<Arc<Shared>>>,

    /// Sendable task spawner
    spawner: Spawner,

    /// Current tick
    tick: u8,

    /// Runtime driver
    ///
    /// The driver is removed before starting to park the thread
    driver: Option<Driver>,

    /// Stats batcher
    stats: WorkerStatsBatcher,
}

#[derive(Clone)]
pub(crate) struct Spawner {
    shared: Arc<Shared>,
}

/// A remote scheduler entry.

Review comment (suggestion):
```suggestion
/// Data required for executing the scheduler. The struct is passed around to
/// a function that will perform the scheduling work and acts as a capability token.
```

Code after:
use std::sync::atomic::Ordering::{AcqRel, Release};
use std::sync::Arc;
use std::task::Poll::{Pending, Ready};
use std::time::Duration;

/// Executes tasks on the current thread
pub(crate) struct BasicScheduler {
    /// Core scheduler data is acquired by a thread entering `block_on`.
    core: AtomicCell<Core>,

    /// Notifier for waking up other threads to steal the
    /// driver.
    notify: Notify,

    /// Sendable task spawner
    spawner: Spawner,

    /// This is usually None, but right before dropping the BasicScheduler, it
    /// is changed to `Some` with the context being the runtime's own context.
    /// This ensures that any tasks dropped in the `BasicScheduler`s destructor
    /// run in that runtime's context.
    context_guard: Option<EnterGuard>,
}

/// Data required for executing the scheduler. The struct is passed around to
/// a function that will perform the scheduling work and acts as a capability token.
struct Core {
    /// Scheduler run queue
    tasks: VecDeque<task::Notified<Arc<Shared>>>,

    /// Sendable task spawner
    spawner: Spawner,

    /// Current tick
    tick: u8,

    /// Runtime driver
    ///
    /// The driver is removed before starting to park the thread
    driver: Option<Driver>,

    /// Stats batcher
    stats: WorkerStatsBatcher,
}

#[derive(Clone)]
pub(crate) struct Spawner {
    shared: Arc<Shared>,
}

/// A remote scheduler entry.
Repository: tokio-rs/tokio
File: tokio-util/src/net/unix/mod.rs (Rust)

Code before:
//! Unix domain socket helpers.

use std::future::Future;
use std::io::Result;
use std::pin::Pin;

/// Listening address.
#[derive(Debug)]
pub enum ListenAddr {
    /// Socket address.
    SocketAddr(std::net::SocketAddr),
    /// Unix socket.
    UnixSocket(tokio::net::unix::SocketAddr),
}

impl From<std::net::SocketAddr> for ListenAddr {
    fn from(addr: std::net::SocketAddr) -> Self {
        Self::SocketAddr(addr)
    }
}

impl From<tokio::net::unix::SocketAddr> for ListenAddr {
    fn from(addr: tokio::net::unix::SocketAddr) -> Self {
        Self::UnixSocket(addr)
    }
}

/// A trait for a listener: `TcpListener` and `UnixListener`.
pub trait Listener: Send + Unpin {
    /// The stream's type of this listener.
    type Io: tokio::io::AsyncRead + tokio::io::AsyncWrite;

    /// Accepts a new incoming connection from this listener.
    fn accept<'a>(
        &'a self,
    ) -> Pin<Box<dyn Future<Output = Result<(Self::Io, ListenAddr)>> + Send + 'a>>;

    /// Returns the local address that this listener is bound to.
    fn local_addr(&self) -> Result<ListenAddr>;

Review comment (suggestion):
This should probably be an associated type of the `Listener` trait like how `Io` is set up.

Code after:
//! Unix domain socket helpers.

use super::Listener;
use std::io::Result;
use std::task::{Context, Poll};

impl Listener for tokio::net::UnixListener {
    type Io = tokio::net::UnixStream;
    type Addr = tokio::net::unix::SocketAddr;

    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>> {
        Self::poll_accept(self, cx)
    }

    fn local_addr(&self) -> Result<Self::Addr> {
        self.local_addr().map(Into::into)
    }
}
Repository: tokio-rs/tokio
File: tokio-util/src/net/unix/mod.rs (Rust)

Code before:

impl From<std::net::SocketAddr> for ListenAddr {
    fn from(addr: std::net::SocketAddr) -> Self {
        Self::SocketAddr(addr)
    }
}

impl From<tokio::net::unix::SocketAddr> for ListenAddr {
    fn from(addr: tokio::net::unix::SocketAddr) -> Self {
        Self::UnixSocket(addr)
    }
}

/// A trait for a listener: `TcpListener` and `UnixListener`.
pub trait Listener: Send + Unpin {
    /// The stream's type of this listener.
    type Io: tokio::io::AsyncRead + tokio::io::AsyncWrite;

    /// Accepts a new incoming connection from this listener.
    fn accept<'a>(
        &'a self,
    ) -> Pin<Box<dyn Future<Output = Result<(Self::Io, ListenAddr)>> + Send + 'a>>;

    /// Returns the local address that this listener is bound to.
    fn local_addr(&self) -> Result<ListenAddr>;
}

impl Listener for tokio::net::TcpListener {
    type Io = tokio::net::TcpStream;

    fn accept<'a>(
        &'a self,
    ) -> Pin<Box<dyn Future<Output = Result<(Self::Io, ListenAddr)>> + Send + 'a>> {
        let accept = self.accept();
        Box::pin(async {
            let (stream, addr) = accept.await?;
            Ok((stream, addr.into()))
        })
    }

    fn local_addr(&self) -> Result<ListenAddr> {
        self.local_addr().map(Into::into)
    }
}

impl Listener for tokio::net::UnixListener {
    type Io = tokio::net::UnixStream;

    fn accept<'a>(
        &'a self,
    ) -> Pin<Box<dyn Future<Output = Result<(Self::Io, ListenAddr)>> + Send + 'a>> {

Review comment (suggestion):
I think the trait should look more like this:
```suggestion
/// A trait for a listener: `TcpListener` and `UnixListener`.
pub trait Listener {
    /// The stream's type of this listener.
    type Io: tokio::io::AsyncRead + tokio::io::AsyncWrite;
    /// The address of the new connection.
    type Addr;

    fn poll_accept(self: Pin<&mut self>) -> Poll<(Self::Io, Self::ListenAddr)>;

    /// Returns the local address that this listener is bound to.
    fn local_addr(&self) -> Result<ListenAddr>;

    /// Accepts a new incoming connection from this listener.
    fn accept<'a>(&'a mut self) -> ListenerAcceptFut<'a, Self>
    where
        Self: Sized + Unpin,
    {
         ListenerAcceptFut::new(self)
    }
}
```
Then, the crate would define a `ListenAcceptFut` type that implements `Future` with a manual impl similar to the return types in the `Ext` traits.

Code after:
//! Unix domain socket helpers.

use super::Listener;
use std::io::Result;
use std::task::{Context, Poll};

impl Listener for tokio::net::UnixListener {
    type Io = tokio::net::UnixStream;
    type Addr = tokio::net::unix::SocketAddr;

    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>> {
        Self::poll_accept(self, cx)
    }

    fn local_addr(&self) -> Result<Self::Addr> {
        self.local_addr().map(Into::into)
    }
}
Repository: tokio-rs/tokio
File: tokio-util/src/net/mod.rs (Rust)

Code before:
//! TCP/UDP/Unix helpers for tokio.

use crate::either::Either;
use std::future::Future;
use std::io::Result;
use std::pin::Pin;
use std::task::{Context, Poll};

#[cfg(unix)]
pub mod unix;

/// A trait for a listener: `TcpListener` and `UnixListener`.
pub trait Listener {
    /// The stream's type of this listener.
    type Io: tokio::io::AsyncRead + tokio::io::AsyncWrite;
    /// The socket address type of this listener.
    type Addr;

    /// Polls to accept a new incoming connection to this listener.
    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>>;

    /// Accepts a new incoming connection from this listener.
    fn accept(&mut self) -> ListenerAcceptFut<'_, Self>
    where
        Self: Sized + Unpin,
    {
        ListenerAcceptFut::new(self)
    }

    /// Returns the local address that this listener is bound to.
    fn local_addr(&self) -> Result<Self::Addr>;
}

impl Listener for tokio::net::TcpListener {
    type Io = tokio::net::TcpStream;
    type Addr = std::net::SocketAddr;

    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>> {
        Self::poll_accept(self, cx)
    }

    fn local_addr(&self) -> Result<Self::Addr> {
        self.local_addr().map(Into::into)
    }
}

/// Future for accepting a new connection from a listener.
#[derive(Debug)]
pub struct ListenerAcceptFut<'a, L> {
    listener: &'a mut L,
}

Review comment (suggestion):
Either `poll_accept` should take `self: Pin<&mut Self>`, or the `Unpin` bound should be removed from `accept`.

Code after:
//! TCP/UDP/Unix helpers for tokio.

use crate::either::Either;
use std::future::Future;
use std::io::Result;
use std::pin::Pin;
use std::task::{Context, Poll};

#[cfg(unix)]
pub mod unix;

/// A trait for a listener: `TcpListener` and `UnixListener`.
pub trait Listener {
    /// The stream's type of this listener.
    type Io: tokio::io::AsyncRead + tokio::io::AsyncWrite;
    /// The socket address type of this listener.
    type Addr;

    /// Polls to accept a new incoming connection to this listener.
    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>>;

    /// Accepts a new incoming connection from this listener.
    fn accept(&mut self) -> ListenerAcceptFut<'_, Self>
    where
        Self: Sized,
    {
        ListenerAcceptFut { listener: self }
    }

    /// Returns the local address that this listener is bound to.
    fn local_addr(&self) -> Result<Self::Addr>;
}

impl Listener for tokio::net::TcpListener {
    type Io = tokio::net::TcpStream;
    type Addr = std::net::SocketAddr;

    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>> {
        Self::poll_accept(self, cx)
    }

    fn local_addr(&self) -> Result<Self::Addr> {
        self.local_addr().map(Into::into)
    }
}

/// Future for accepting a new connection from a listener.
#[derive(Debug)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct ListenerAcceptFut<'a, L> {
    listener: &'a mut L,
Repository: tokio-rs/tokio
File: tokio-util/src/net/mod.rs (Rust)

Code before:
    where
        Self: Sized,
    {
        ListenerAcceptFut::new(self)
    }

    /// Returns the local address that this listener is bound to.
    fn local_addr(&self) -> Result<Self::Addr>;
}

impl Listener for tokio::net::TcpListener {
    type Io = tokio::net::TcpStream;
    type Addr = std::net::SocketAddr;

    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>> {
        Self::poll_accept(self, cx)
    }

    fn local_addr(&self) -> Result<Self::Addr> {
        self.local_addr().map(Into::into)
    }
}

/// Future for accepting a new connection from a listener.
#[derive(Debug)]
pub struct ListenerAcceptFut<'a, L> {
    listener: &'a mut L,
}

impl<'a, L> ListenerAcceptFut<'a, L>
where
    L: Listener,
{
    fn new(listener: &'a mut L) -> Self {
        Self { listener }
    }
}

impl<'a, L> Future for ListenerAcceptFut<'a, L>
where
    L: Listener,
{
    type Output = Result<(L::Io, L::Addr)>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.listener.poll_accept(cx)
    }
}

impl<L, R> Listener for Either<L, R>
where

Review comment (bug):
```suggestion
#[derive(Debug)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct ListenerAcceptFut<'a, L> {
```

Code after:
    where
        Self: Sized,
    {
        ListenerAcceptFut { listener: self }
    }

    /// Returns the local address that this listener is bound to.
    fn local_addr(&self) -> Result<Self::Addr>;
}

impl Listener for tokio::net::TcpListener {
    type Io = tokio::net::TcpStream;
    type Addr = std::net::SocketAddr;

    fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<(Self::Io, Self::Addr)>> {
        Self::poll_accept(self, cx)
    }

    fn local_addr(&self) -> Result<Self::Addr> {
        self.local_addr().map(Into::into)
    }
}

/// Future for accepting a new connection from a listener.
#[derive(Debug)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct ListenerAcceptFut<'a, L> {
    listener: &'a mut L,
}

impl<'a, L> Future for ListenerAcceptFut<'a, L>
where
    L: Listener,
{
    type Output = Result<(L::Io, L::Addr)>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.listener.poll_accept(cx)
    }
}

impl<L, R> Either<L, R>
where
    L: Listener,
    R: Listener,
{
    /// Accepts a new incoming connection from this listener.
    pub async fn accept(&mut self) -> Result<Either<(L::Io, L::Addr), (R::Io, R::Addr)>> {
        match self {
            Either::Left(listener) => {
                let (stream, addr) = listener.accept().await?;
Repository: tokio-rs/tokio
File: tokio/tests/io_mem_stream.rs (Rust)

Code before:
        let n = a.write(&[0u8; 64]).await.unwrap();
        assert_eq!(n, 4);
    });

    let mut buf = [0u8; 4];
    b.read_exact(&mut buf).await.unwrap();

    t1.await.unwrap();

    // drop b only after task t1 finishes writing
    drop(b);
}

#[tokio::test]
async fn duplex_is_cooperative() {
    let (mut tx, mut rx) = tokio::io::duplex(1024 * 8);

    tokio::select! {
        biased;

        _ = async {
            loop {
                let buf = [3u8; 4096];
                let _ = tx.write_all(&buf).await;
                let mut buf = [0u8; 4096];
                let _ = rx.read(&mut buf).await;
            }
        } => {},
        _ = tokio::task::yield_now() => {}
    }
}


Review comment (suggestion):
```suggestion
                let buf = [3u8; 4096];
                tx.write_all(&buf).await.unwrap();
                let mut buf = [0u8; 4096];
                rx.read(&mut buf).await.unwrap();
```

Code after:
        let n = a.write(&[0u8; 64]).await.unwrap();
        assert_eq!(n, 4);
    });

    let mut buf = [0u8; 4];
    b.read_exact(&mut buf).await.unwrap();

    t1.await.unwrap();

    // drop b only after task t1 finishes writing
    drop(b);
}

#[tokio::test]
async fn duplex_is_cooperative() {
    let (mut tx, mut rx) = tokio::io::duplex(1024 * 8);

    tokio::select! {
        biased;

        _ = async {
            loop {
                let buf = [3u8; 4096];
                tx.write_all(&buf).await.unwrap();
                let mut buf = [0u8; 4096];
                rx.read(&mut buf).await.unwrap();
            }
        } => {},
        _ = tokio::task::yield_now() => {}
    }
}
Repository: tokio-rs/tokio
File: tokio/src/io/util/mem.rs (Rust)

Code before:
        ) -> Poll<std::io::Result<()>> {
            self.poll_read_internal(cx, buf)
        }
    }
}

impl AsyncWrite for Pipe {
    cfg_coop! {
        fn poll_write(
            self: Pin<&mut Self>,
            cx: &mut task::Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            let coop = ready!(crate::coop::poll_proceed(cx));

            let ret = self.poll_write_internal(cx, buf);
            if ret.is_ready() {
                coop.made_progress();
            }
            ret
        }
    }

    cfg_not_coop! {
        fn poll_write(
            mut self: Pin<&mut Self>,
            cx: &mut task::Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            self.poll_write_internal(cx, buf)
        }
    }

    fn poll_flush(self: Pin<&mut Self>, _: &mut task::Context<'_>) -> Poll<std::io::Result<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(
        mut self: Pin<&mut Self>,
        _: &mut task::Context<'_>,
    ) -> Poll<std::io::Result<()>> {
        self.close_write();
        Poll::Ready(Ok(()))
    }
}


Review comment (bug):
You're getting a CI failure due to some warnings about these being unnecessary:
```suggestion
            self: Pin<&mut Self>,
```

Code after:
        ) -> Poll<std::io::Result<()>> {
            self.poll_read_internal(cx, buf)
        }
    }
}

impl AsyncWrite for Pipe {
    cfg_coop! {
        fn poll_write(
            self: Pin<&mut Self>,
            cx: &mut task::Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            let coop = ready!(crate::coop::poll_proceed(cx));

            let ret = self.poll_write_internal(cx, buf);
            if ret.is_ready() {
                coop.made_progress();
            }
            ret
        }
    }

    cfg_not_coop! {
        fn poll_write(
            self: Pin<&mut Self>,
            cx: &mut task::Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            self.poll_write_internal(cx, buf)
        }
    }

    fn poll_flush(self: Pin<&mut Self>, _: &mut task::Context<'_>) -> Poll<std::io::Result<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(
        mut self: Pin<&mut Self>,
        _: &mut task::Context<'_>,
    ) -> Poll<std::io::Result<()>> {
        self.close_write();
        Poll::Ready(Ok(()))
    }
}