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

Code before:
    feature = "time",
))]
pub(crate) mod linked_list;

#[cfg(any(feature = "rt", feature = "macros", feature = "stream"))]
mod rand;

cfg_rt! {
    mod idle_notified_set;
    pub(crate) use idle_notified_set::IdleNotifiedSet;

    mod wake;
    pub(crate) use wake::WakerRef;
    pub(crate) use wake::{waker_ref, Wake};

    mod sync_wrapper;
    pub(crate) use sync_wrapper::SyncWrapper;

    mod vec_deque_cell;
    pub(crate) use vec_deque_cell::VecDequeCell;

    mod rc_cell;
    pub(crate) use rc_cell::RcCell;
}

#[allow(unreachable_pub)]
pub use self::rand::RngSeed;
pub(crate) use self::rand::{replace_thread_rng, RngSeedGenerator};

#[cfg(any(feature = "macros"))]
#[cfg_attr(not(feature = "macros"), allow(unreachable_pub))]
pub use self::rand::thread_rng_n;

cfg_rt_multi_thread! {
    pub(crate) use self::rand::FastRand;

    mod try_lock;
    pub(crate) use try_lock::TryLock;
}

pub(crate) mod trace;

pub(crate) mod error;


Review comment (suggestion):
```suggestion
#[cfg_attr(not(tokio_unstable), allow(unreachable_pub))]
```

Code after:
    feature = "time",
))]
pub(crate) mod linked_list;

#[cfg(any(feature = "rt", feature = "macros"))]
mod rand;

cfg_rt! {
    mod idle_notified_set;
    pub(crate) use idle_notified_set::IdleNotifiedSet;

    pub(crate) use self::rand::{RngSeedGenerator,replace_thread_rng};

    mod wake;
    pub(crate) use wake::WakerRef;
    pub(crate) use wake::{waker_ref, Wake};

    mod sync_wrapper;
    pub(crate) use sync_wrapper::SyncWrapper;

    mod vec_deque_cell;
    pub(crate) use vec_deque_cell::VecDequeCell;

    mod rc_cell;
    pub(crate) use rc_cell::RcCell;
}

#[cfg_attr(not(tokio_unstable), allow(unreachable_pub))]
#[cfg(feature = "rt")]
pub use self::rand::RngSeed;

#[cfg(any(feature = "macros"))]
#[cfg_attr(not(feature = "macros"), allow(unreachable_pub))]
pub use self::rand::thread_rng_n;

cfg_rt_multi_thread! {
    pub(crate) use self::rand::FastRand;

    mod try_lock;
    pub(crate) use try_lock::TryLock;
}

pub(crate) mod trace;

pub(crate) mod error;
Repository: tokio-rs/tokio
File: tokio/src/signal/windows/sys.rs (Rust)

Code before:
    ctrl_close: EventInfo,
    ctrl_c: EventInfo,
    ctrl_logoff: EventInfo,
    ctrl_shutdown: EventInfo,
}

impl Init for OsStorage {
    fn init() -> Self {
        Self {
            ctrl_break: Default::default(),
            ctrl_close: Default::default(),
            ctrl_c: Default::default(),
            ctrl_logoff: Default::default(),
            ctrl_shutdown: Default::default(),
        }
    }
}

impl Storage for OsStorage {
    fn event_info(&self, id: EventId) -> Option<&EventInfo> {
        match DWORD::try_from(id) {
            Ok(CTRL_BREAK_EVENT) => Some(&self.ctrl_break),
            Ok(CTRL_CLOSE_EVENT) => Some(&self.ctrl_close),
            Ok(CTRL_C_EVENT) => Some(&self.ctrl_c),
            Ok(CTRL_LOGOFF_EVENT) => Some(&self.ctrl_logoff),
            Ok(CTRL_SHUTDOWN_EVENT) => Some(&self.ctrl_shutdown),
            _ => None,
        }
    }

    fn for_each<'a, F>(&'a self, mut f: F)
    where
        F: FnMut(&'a EventInfo),
    {
        f(&self.ctrl_break);
        f(&self.ctrl_close);
        f(&self.ctrl_c);
        f(&self.ctrl_logoff);
        f(&self.ctrl_shutdown);
    }
}

#[derive(Debug)]
pub(crate) struct OsExtraData {}

impl Init for OsExtraData {
    fn init() -> Self {
        Self {}
    }
}


Review comment (suggestion):
These need to be qualified now

```suggestion
            Ok(wincon::CTRL_BREAK_EVENT) => Some(&self.ctrl_break),
            Ok(wincon::CTRL_CLOSE_EVENT) => Some(&self.ctrl_close),
            Ok(wincon::CTRL_C_EVENT) => Some(&self.ctrl_c),
            Ok(wincon::CTRL_LOGOFF_EVENT) => Some(&self.ctrl_logoff),
            Ok(wincon::CTRL_SHUTDOWN_EVENT) => Some(&self.ctrl_shutdown),
```

Code after:
    ctrl_close: EventInfo,
    ctrl_c: EventInfo,
    ctrl_logoff: EventInfo,
    ctrl_shutdown: EventInfo,
}

impl Init for OsStorage {
    fn init() -> Self {
        Self {
            ctrl_break: Default::default(),
            ctrl_close: Default::default(),
            ctrl_c: Default::default(),
            ctrl_logoff: Default::default(),
            ctrl_shutdown: Default::default(),
        }
    }
}

impl Storage for OsStorage {
    fn event_info(&self, id: EventId) -> Option<&EventInfo> {
        match DWORD::try_from(id) {
            Ok(wincon::CTRL_BREAK_EVENT) => Some(&self.ctrl_break),
            Ok(wincon::CTRL_CLOSE_EVENT) => Some(&self.ctrl_close),
            Ok(wincon::CTRL_C_EVENT) => Some(&self.ctrl_c),
            Ok(wincon::CTRL_LOGOFF_EVENT) => Some(&self.ctrl_logoff),
            Ok(wincon::CTRL_SHUTDOWN_EVENT) => Some(&self.ctrl_shutdown),
            _ => None,
        }
    }

    fn for_each<'a, F>(&'a self, mut f: F)
    where
        F: FnMut(&'a EventInfo),
    {
        f(&self.ctrl_break);
        f(&self.ctrl_close);
        f(&self.ctrl_c);
        f(&self.ctrl_logoff);
        f(&self.ctrl_shutdown);
    }
}

#[derive(Debug)]
pub(crate) struct OsExtraData {}

impl Init for OsExtraData {
    fn init() -> Self {
        Self {}
    }
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/driver.rs (Rust)

Code before:
// ===== process driver =====

cfg_process_driver! {
    type ProcessDriver = crate::process::unix::driver::Driver;

    fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
        crate::process::unix::driver::Driver::new(signal_driver)
    }
}

cfg_not_process_driver! {
    cfg_io_driver! {
        type ProcessDriver = SignalDriver;

        fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
            signal_driver
        }
    }
}

// ===== time driver =====

cfg_time! {
    #[derive(Debug)]
    pub(crate) enum TimeDriver {
        Enabled(crate::runtime::time::Driver),
        Disabled(IoStack),
    }

    pub(crate) enum TimerUnpark {
        Enabled(crate::runtime::time::TimerUnpark),
        Disabled(IoUnpark),
    }

    pub(crate) type Clock = crate::time::Clock;
    pub(crate) type TimeHandle = Option<crate::runtime::time::Handle>;

    fn create_clock(enable_pausing: bool, start_paused: bool) -> Clock {
        crate::time::Clock::new(enable_pausing, start_paused)
    }

    fn create_time_driver(
        enable: bool,
        io_stack: IoStack,
        clock: Clock,
    ) -> (TimeDriver, TimeHandle) {
        if enable {
            let driver = crate::runtime::time::Driver::new(io_stack, clock);
            let handle = driver.handle();

            (TimeDriver::Enabled(driver), Some(handle))

Review comment (suggestion):
The `TimeDriver` enum is temporary and just used to help make the PRs more approachable.

Code after:
// ===== process driver =====

cfg_process_driver! {
    type ProcessDriver = crate::process::unix::driver::Driver;

    fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
        crate::process::unix::driver::Driver::new(signal_driver)
    }
}

cfg_not_process_driver! {
    cfg_io_driver! {
        type ProcessDriver = SignalDriver;

        fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
            signal_driver
        }
    }
}

// ===== time driver =====

cfg_time! {
    #[derive(Debug)]
    pub(crate) enum TimeDriver {
        Enabled {
            driver: crate::runtime::time::Driver,
            handle: crate::runtime::time::Handle,
        },
        Disabled(IoStack),
    }

    pub(crate) enum TimerUnpark {
        Enabled(crate::runtime::time::TimerUnpark),
        Disabled(IoUnpark),
    }

    pub(crate) type Clock = crate::time::Clock;
    pub(crate) type TimeHandle = Option<crate::runtime::time::Handle>;

    fn create_clock(enable_pausing: bool, start_paused: bool) -> Clock {
        crate::time::Clock::new(enable_pausing, start_paused)
    }

    fn create_time_driver(
        enable: bool,
        io_stack: IoStack,
        clock: Clock,
    ) -> (TimeDriver, TimeHandle) {
        if enable {
            let (driver, handle) = crate::runtime::time::Driver::new(io_stack, clock);
Repository: tokio-rs/tokio
File: tokio-util/src/io/inspect.rs (Rust)

Code before:
use futures_core::ready;
use pin_project_lite::pin_project;
use std::{
    io::{IoSlice, Result},
    pin::Pin,
    task::{Context, Poll},
};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};

pin_project! {
    /// An adapter that lets you inspect the data that's being read.
    ///
    /// This is useful for things like hashing data as it's read in.
    pub struct InspectReader<R: AsyncRead, F: FnMut(&[u8])> {
        #[pin]
        reader: R,
        f: F,
    }
}

impl<R: AsyncRead, F: FnMut(&[u8])> InspectReader<R, F> {
    /// Create a new InspectReader, wrapping `reader` and calling `f` for the
    /// new data supplied by each read call.
    pub fn new(reader: R, f: F) -> InspectReader<R, F> {
        InspectReader { reader, f }
    }

    /// Consumes the `InspectReader`, returning the wrapped reader
    pub fn into_inner(self) -> R {
        self.reader
    }
}

Review comment (suggestion):
We don't use multi-line imports in Tokio.
```suggestion
use std::io::{IoSlice, Result};
use std::pin::Pin;
use std::task::{Context, Poll};
```

Code after:
use futures_core::ready;
use pin_project_lite::pin_project;
use std::io::{IoSlice, Result};
use std::pin::Pin;
use std::task::{Context, Poll};

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

pin_project! {
    /// An adapter that lets you inspect the data that's being read.
    ///
    /// This is useful for things like hashing data as it's read in.
    pub struct InspectReader<R, F> {
        #[pin]
        reader: R,
        f: F,
    }
}

impl<R, F> InspectReader<R, F> {
    /// Create a new InspectReader, wrapping `reader` and calling `f` for the
    /// new data supplied by each read call.
    ///
    /// The closure will only be called with an empty slice if the inner reader
    /// returns without reading data into the buffer. This happens at EOF, or if
    /// `poll_read` is called with a zero-size buffer.
    pub fn new(reader: R, f: F) -> InspectReader<R, F>
    where
        R: AsyncRead,
        F: FnMut(&[u8]),
    {
        InspectReader { reader, f }
Repository: tokio-rs/tokio
File: tokio-util/src/io/inspect.rs (Rust)

Code before:
use futures_core::ready;
use pin_project_lite::pin_project;
use std::{
    io::{IoSlice, Result},
    pin::Pin,
    task::{Context, Poll},
};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};

pin_project! {
    /// An adapter that lets you inspect the data that's being read.
    ///
    /// This is useful for things like hashing data as it's read in.
    pub struct InspectReader<R: AsyncRead, F: FnMut(&[u8])> {
        #[pin]
        reader: R,
        f: F,
    }
}

impl<R: AsyncRead, F: FnMut(&[u8])> InspectReader<R, F> {
    /// Create a new InspectReader, wrapping `reader` and calling `f` for the
    /// new data supplied by each read call.
    pub fn new(reader: R, f: F) -> InspectReader<R, F> {
        InspectReader { reader, f }
    }

    /// Consumes the `InspectReader`, returning the wrapped reader
    pub fn into_inner(self) -> R {
        self.reader
    }
}

impl<R: AsyncRead, F: FnMut(&[u8])> AsyncRead for InspectReader<R, F> {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<Result<()>> {

Review comment (suggestion):
This is unnecessary.
```suggestion
    pub struct InspectReader<R, F> {
```

Code after:
use futures_core::ready;
use pin_project_lite::pin_project;
use std::io::{IoSlice, Result};
use std::pin::Pin;
use std::task::{Context, Poll};

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

pin_project! {
    /// An adapter that lets you inspect the data that's being read.
    ///
    /// This is useful for things like hashing data as it's read in.
    pub struct InspectReader<R, F> {
        #[pin]
        reader: R,
        f: F,
    }
}

impl<R, F> InspectReader<R, F> {
    /// Create a new InspectReader, wrapping `reader` and calling `f` for the
    /// new data supplied by each read call.
    ///
    /// The closure will only be called with an empty slice if the inner reader
    /// returns without reading data into the buffer. This happens at EOF, or if
    /// `poll_read` is called with a zero-size buffer.
    pub fn new(reader: R, f: F) -> InspectReader<R, F>
    where
        R: AsyncRead,
        F: FnMut(&[u8]),
    {
        InspectReader { reader, f }
    }

    /// Consumes the `InspectReader`, returning the wrapped reader
    pub fn into_inner(self) -> R {
        self.reader
    }
}
Repository: tokio-rs/tokio
File: tokio-util/src/io/sink_writer.rs (Rust)

Code before:

    /// Gets a reference to the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn get_ref(&self) -> &S {
        &self.inner
    }

    /// Gets a mutable reference to the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consumes this `SinkWriter`, returning the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn into_inner(self) -> S {
        self.inner
    }
}

impl<S, E> AsyncWrite for SinkWriter<S>
where
    S: Sink<Vec<u8>, Error = E>,
    E: Into<io::Error>
{
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, io::Error>> {
        match self.as_mut().project().inner.poll_ready(cx) {
            Poll::Ready(Ok(())) => {
                if let Err(e) = self.as_mut().project().inner.start_send(buf.to_vec()) {
                    Poll::Ready(Err(e.into()))
                } else {
                    Poll::Ready(Ok(buf.len()))
                }
            }
            Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
            Poll::Pending => {
                cx.waker().wake_by_ref();
                Poll::Pending
            }
        }
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
        self.project().inner.poll_flush(cx).map_err(Into::into)

Review comment (suggestion):
```suggestion
    for<'a> S: Sink<&'a [u8], Error = E>,
```

Code after:
    pub struct SinkWriter<S> {
        #[pin]
        inner: S,
    }
}

impl<S> SinkWriter<S> {
    /// Creates a new [`SinkWriter`].
    pub fn new(sink: S) -> Self {
        Self { inner: sink }
    }

    /// Gets a reference to the underlying sink.
    pub fn get_ref(&self) -> &S {
        &self.inner
    }

    /// Gets a mutable reference to the underlying sink.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consumes this [`SinkWriter`], returning the underlying sink.
    pub fn into_inner(self) -> S {
        self.inner
    }
}
impl<S, E> AsyncWrite for SinkWriter<S>
where
    for<'a> S: Sink<&'a [u8], Error = E>,
    E: Into<io::Error>,
{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, io::Error>> {
        let mut this = self.project();
        match this.inner.as_mut().poll_ready(cx) {
            Poll::Ready(Ok(())) => {
                if let Err(e) = this.inner.as_mut().start_send(buf) {
                    Poll::Ready(Err(e.into()))
                } else {
                    Poll::Ready(Ok(buf.len()))
                }
            }
            Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
            Poll::Pending => {
                cx.waker().wake_by_ref();
                Poll::Pending
            }
Repository: tokio-rs/tokio
File: tokio-util/src/io/sink_writer.rs (Rust)

Code before:
    }
}

impl<S> SinkWriter<S> {
    /// Creates a new [`SinkWriter<S>`].
    pub fn new(sink: S) -> Self {
        Self { inner: sink }
    }

    /// Gets a reference to the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn get_ref(&self) -> &S {
        &self.inner
    }

    /// Gets a mutable reference to the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consumes this `SinkWriter`, returning the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn into_inner(self) -> S {
        self.inner
    }
}
pin_project! {
    /// A helper structure, converting a sink of byte slices into a
    /// sink of owned [`Bytes`].
    ///
    ///
    /// [`Bytes`]: bytes::Bytes
    #[derive(Debug)]
    pub struct CopyToBytes<S> {
        #[pin]
        inner: S,
    }
}

impl<S> CopyToBytes<S> {
    /// Creates a new [`CopyToBytes<S>`].
    pub fn new(inner: S) -> Self {
        Self { inner }
    }

    /// Gets a reference to the underlying sink.
    ///

Review comment (bug):
This is definitely wrong - once you've consumed the `SInkWriter` and got the `Sink` back, it should be safe to write to the underlying `Sink`. If it's not, this method is very hard to use correctly.

Code after:
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`AsyncWrite`]: tokio::io::AsyncWrite
    /// [`CopyToBytes`]: crate::io::CopyToBytes
    /// [`Encoder`]: crate::codec::Encoder
    /// [`Sink`]: futures_sink::Sink
    /// [`codec`]: tokio_util::codec
    #[derive(Debug)]
    pub struct SinkWriter<S> {
        #[pin]
        inner: S,
    }
}

impl<S> SinkWriter<S> {
    /// Creates a new [`SinkWriter`].
    pub fn new(sink: S) -> Self {
        Self { inner: sink }
    }

    /// Gets a reference to the underlying sink.
    pub fn get_ref(&self) -> &S {
        &self.inner
    }

    /// Gets a mutable reference to the underlying sink.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consumes this [`SinkWriter`], returning the underlying sink.
    pub fn into_inner(self) -> S {
        self.inner
    }
}
impl<S, E> AsyncWrite for SinkWriter<S>
where
    for<'a> S: Sink<&'a [u8], Error = E>,
    E: Into<io::Error>,
{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, io::Error>> {
        let mut this = self.project();
        match this.inner.as_mut().poll_ready(cx) {
            Poll::Ready(Ok(())) => {
                if let Err(e) = this.inner.as_mut().start_send(buf) {
Repository: tokio-rs/tokio
File: tokio-util/tests/io_sink_writer.rs (Rust)

Code before:
    ));

    // Write data to our interface...
    let data: [u8; 4] = [1, 2, 3, 4];
    let _ = writer.write(&data).await;

    // ... and receive it.
    assert_eq!(data.to_vec(), rx.recv().await.unwrap());

    Ok(())
}

/// A trivial encoder.
struct SliceEncoder;

impl SliceEncoder {
    fn new() -> Self {
        Self {}
    }
}

impl<'a> Encoder<&'a [u8]> for SliceEncoder {
    type Error = Error;

    fn encode(&mut self, item: &'a [u8], dst: &mut bytes::BytesMut) -> Result<(), Self::Error> {
        // We pretend there is something important going on here.
        dst.extend_from_slice(item);
        Ok(())
    }
}

#[tokio::test]
async fn test_direct_sink_writer() -> Result<(), Error> {
    // We define a framed writer which accepts bytes slices
    // and 'reverse' this construction immediately.
    let framed_byte_lc = FramedWrite::new(Vec::new(), SliceEncoder::new());
    let mut writer = SinkWriter::new(framed_byte_lc);

    // Write multiple slices to the sink...
    let _ = writer.write(&[1, 2, 3]).await;
    let _ = writer.write(&[4, 5, 6]).await;

    // ... and compare it with the buffer.
    assert_eq!(
        writer.into_inner().write_buffer().to_vec().as_slice(),
        &[1, 2, 3, 4, 5, 6]
    );

    Ok(())
}


Review comment (suggestion):
```suggestion
        // This is where we'd write packet headers, lengths etc in a real encoder
```

Code after:
    ));

    // Write data to our interface...
    let data: [u8; 4] = [1, 2, 3, 4];
    let _ = writer.write(&data).await;

    // ... and receive it.
    assert_eq!(data.to_vec(), rx.recv().await.unwrap().to_vec());

    Ok(())
}

/// A trivial encoder.
struct SliceEncoder;

impl SliceEncoder {
    fn new() -> Self {
        Self {}
    }
}

impl<'a> Encoder<&'a [u8]> for SliceEncoder {
    type Error = Error;

    fn encode(&mut self, item: &'a [u8], dst: &mut bytes::BytesMut) -> Result<(), Self::Error> {
        // This is where we'd write packet headers, lengths, etc. in a real encoder.
        // For simplicity and demonstration purposes, we just pack a copy of
        // the slice at the end of a buffer.
        dst.extend_from_slice(item);
        Ok(())
    }
}

#[tokio::test]
async fn test_direct_sink_writer() -> Result<(), Error> {
    // We define a framed writer which accepts byte slices
    // and 'reverse' this construction immediately.
    let framed_byte_lc = FramedWrite::new(Vec::new(), SliceEncoder::new());
    let mut writer = SinkWriter::new(framed_byte_lc);

    // Write multiple slices to the sink...
    let _ = writer.write(&[1, 2, 3]).await;
    let _ = writer.write(&[4, 5, 6]).await;

    // ... and compare it with the buffer.
    assert_eq!(
        writer.into_inner().write_buffer().to_vec().as_slice(),
        &[1, 2, 3, 4, 5, 6]
    );

    Ok(())
Repository: tokio-rs/tokio
File: tokio-util/src/io/sink_writer.rs (Rust)

Code before:
    }

    /// Gets a reference to the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn get_ref(&self) -> &S {
        &self.inner
    }

    /// Gets a mutable reference to the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consumes this `SinkWriter`, returning the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn into_inner(self) -> S {
        self.inner
    }
}
pin_project! {
    /// A helper structure, converting a sink of byte slices into a
    /// sink of owned [`Bytes`].
    ///
    ///
    /// [`Bytes`]: bytes::Bytes
    #[derive(Debug)]
    pub struct CopyToBytes<S> {
        #[pin]
        inner: S,
    }
}

impl<S> CopyToBytes<S> {
    /// Creates a new [`CopyToBytes<S>`].
    pub fn new(inner: S) -> Self {
        Self { inner }
    }

    /// Gets a reference to the underlying sink.
    ///
    /// It is inadvisable to directly write to the underlying sink.
    pub fn get_ref(&self) -> &S {
        &self.inner
    }

    /// Gets a mutable reference to the underlying sink.
    ///

Review comment (suggestion):
The documentation here isn't sufficient. Can you add some more details? You should explain that this is a _wrapper_ and that it can be used to convert a `Sink<Bytes>` into a `Sink<&'a [u8]>`.

Code after:
    /// [`Sink`]: futures_sink::Sink
    /// [`codec`]: tokio_util::codec
    #[derive(Debug)]
    pub struct SinkWriter<S> {
        #[pin]
        inner: S,
    }
}

impl<S> SinkWriter<S> {
    /// Creates a new [`SinkWriter`].
    pub fn new(sink: S) -> Self {
        Self { inner: sink }
    }

    /// Gets a reference to the underlying sink.
    pub fn get_ref(&self) -> &S {
        &self.inner
    }

    /// Gets a mutable reference to the underlying sink.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consumes this [`SinkWriter`], returning the underlying sink.
    pub fn into_inner(self) -> S {
        self.inner
    }
}
impl<S, E> AsyncWrite for SinkWriter<S>
where
    for<'a> S: Sink<&'a [u8], Error = E>,
    E: Into<io::Error>,
{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, io::Error>> {
        let mut this = self.project();
        match this.inner.as_mut().poll_ready(cx) {
            Poll::Ready(Ok(())) => {
                if let Err(e) = this.inner.as_mut().start_send(buf) {
                    Poll::Ready(Err(e.into()))
                } else {
                    Poll::Ready(Ok(buf.len()))
                }
            }
            Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
            Poll::Pending => {