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

Code before:

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

        /// Returns a reference to the inner reader.
        pub fn reader(&self) -> &R {
            &self.reader
        }

        /// Returns a reference to the inner writer.
        pub fn writer(&self) -> &W {
            &self.writer
        }

        /// Returns a mutable reference to the inner reader.
        pub fn reader_mut(&mut self) -> &mut R {
            &mut self.reader
        }

        /// Returns a mutable reference to the inner writer.
        pub fn writer_mut(&mut self) -> &mut W {
            &mut self.writer
        }
    }

    impl<R, W> AsyncRead for Join<R, W>
    where
        R: AsyncRead + Unpin,
    {
        fn poll_read(
            self: Pin<&mut Self>,
            cx: &mut Context<'_>,
            buf: &mut ReadBuf<'_>,
        ) -> Poll<Result<(), io::Error>> {
            self.project().reader.poll_read(cx, buf)
        }
    }

    impl<R, W> AsyncWrite for Join<R, W>
    where
        W: AsyncWrite + Unpin,
    {
        fn poll_write(
            self: Pin<&mut Self>,
            cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<Result<usize, io::Error>> {
            self.project().writer.poll_write(cx, buf)

Review comment (suggestion):
I would probably also add `reader_pin_mut`/`writer_pin_mut` with this signature:
```
pub fn reader_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R>;
```

Code after:
where
    R: AsyncRead,
    W: AsyncWrite,
{
    /// Splits this `Join` back into its `AsyncRead` and `AsyncWrite`
    /// components.
    pub fn into_inner(self) -> (R, W) {
        (self.reader, self.writer)
    }

    /// Returns a reference to the inner reader.
    pub fn reader(&self) -> &R {
        &self.reader
    }

    /// Returns a reference to the inner writer.
    pub fn writer(&self) -> &W {
        &self.writer
    }

    /// Returns a mutable reference to the inner reader.
    pub fn reader_mut(&mut self) -> &mut R {
        &mut self.reader
    }

    /// Returns a mutable reference to the inner writer.
    pub fn writer_mut(&mut self) -> &mut W {
        &mut self.writer
    }

    /// Returns a pinned mutable reference to the inner reader.
    pub fn reader_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R> {
        self.project().reader
    }

    /// Returns a pinned mutable reference to the inner writer.
    pub fn writer_pin_mut(self: Pin<&mut Self>) -> Pin<&mut W> {
        self.project().writer
    }
}

impl<R, W> AsyncRead for Join<R, W>
where
    R: AsyncRead,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<Result<(), io::Error>> {
        self.project().reader.poll_read(cx, buf)
Repository: tokio-rs/tokio
File: tokio/src/io/mod.rs (Rust)

Code before:
    mod async_fd;

    pub mod unix {
        //! Asynchronous IO structures specific to Unix-like operating systems.
        pub use super::async_fd::{AsyncFd, AsyncFdReadyGuard, AsyncFdReadyMutGuard, TryIoError};
    }
}

cfg_io_std! {
    mod stdio_common;

    mod stderr;
    pub use stderr::{stderr, Stderr};

    mod stdin;
    pub use stdin::{stdin, Stdin};

    mod stdout;
    pub use stdout::{stdout, Stdout};
}

cfg_io_util! {
    mod split;
    pub use split::{split, ReadHalf, WriteHalf};
    mod join;
    pub use join::Join;

    pub(crate) mod seek;
    pub(crate) mod util;
    pub use util::{
        copy, copy_bidirectional, copy_buf, duplex, empty, repeat, sink, AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt,
        BufReader, BufStream, BufWriter, DuplexStream, Empty, Lines, Repeat, Sink, Split, Take,
    };
}

cfg_not_io_util! {
    cfg_process! {
        pub(crate) mod util;
    }
}

cfg_io_blocking! {
    /// Types in this module can be mocked out in tests.
    mod sys {
        // TODO: don't rename
        pub(crate) use crate::blocking::spawn_blocking as run;
        pub(crate) use crate::blocking::JoinHandle as Blocking;
    }
}


Review comment (suggestion):
Similar to `split`, I would also introduce a top-level function for constructing this type.

Code after:
    mod async_fd;

    pub mod unix {
        //! Asynchronous IO structures specific to Unix-like operating systems.
        pub use super::async_fd::{AsyncFd, AsyncFdReadyGuard, AsyncFdReadyMutGuard, TryIoError};
    }
}

cfg_io_std! {
    mod stdio_common;

    mod stderr;
    pub use stderr::{stderr, Stderr};

    mod stdin;
    pub use stdin::{stdin, Stdin};

    mod stdout;
    pub use stdout::{stdout, Stdout};
}

cfg_io_util! {
    mod split;
    pub use split::{split, ReadHalf, WriteHalf};
    mod join;
    pub use join::{join, Join};

    pub(crate) mod seek;
    pub(crate) mod util;
    pub use util::{
        copy, copy_bidirectional, copy_buf, duplex, empty, repeat, sink, AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt,
        BufReader, BufStream, BufWriter, DuplexStream, Empty, Lines, Repeat, Sink, Split, Take,
    };
}

cfg_not_io_util! {
    cfg_process! {
        pub(crate) mod util;
    }
}

cfg_io_blocking! {
    /// Types in this module can be mocked out in tests.
    mod sys {
        // TODO: don't rename
        pub(crate) use crate::blocking::spawn_blocking as run;
        pub(crate) use crate::blocking::JoinHandle as Blocking;
    }
}
Repository: tokio-rs/tokio
File: tokio/tests/io_join.rs (Rust)

Code before:
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support panic recovery

use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, Join, ReadBuf};

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

struct R;

impl AsyncRead for R {
    fn poll_read(
        self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        buf.put_slice(&[b'z']);
        Poll::Ready(Ok(()))
    }
}

struct W;

impl AsyncWrite for W {
    fn poll_write(
        self: Pin<&mut Self>,

Review comment (suggestion):
```suggestion
#![cfg(feature = "full")]
```
You're not using panics or threads anywhere.

Code after:
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]

use tokio::io::{join, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, Join, ReadBuf};

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

struct R;

impl AsyncRead for R {
    fn poll_read(
        self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        buf.put_slice(&[b'z']);
        Poll::Ready(Ok(()))
    }
}

struct W;

impl AsyncWrite for W {
    fn poll_write(
        self: Pin<&mut Self>,
Repository: tokio-rs/tokio
File: tokio-util/src/either.rs (Rust)

Code before:
        delegate_call!(self.poll_flush(cx))
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
        delegate_call!(self.poll_shutdown(cx))
    }
}

impl<L, R> futures_core::stream::Stream for Either<L, R>
where
    L: futures_core::stream::Stream,
    R: futures_core::stream::Stream<Item = L::Item>,
{
    type Item = L::Item;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        delegate_call!(self.poll_next(cx))
    }
}

impl<L, R, Item> futures_sink::Sink<Item> for Either<L, R>
where
    L: futures_sink::Sink<Item>,
    R: futures_sink::Sink<Item>,
{
    type Error = L::Error;

    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
        delegate_call!(self.poll_ready(cx))
    }

    fn start_send(self: Pin<&mut Self>, item: Item) -> Result<()> {
        delegate_call!(self.start_send(item))
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
        delegate_call!(self.poll_flush(cx))
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
        delegate_call!(self.poll_close(cx))
    }
}


#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::{repeat, AsyncReadExt, Repeat};
    use tokio_stream::{once, Once, StreamExt};


Review comment (suggestion):
```suggestion
impl<L, R, Item, Error> futures_sink::Sink<Item> for Either<L, R>
where
    L: futures_sink::Sink<Item, Error = Error>,
    R: futures_sink::Sink<Item, Error = Error>,
{
    type Error = Error;
```

Code after:
        delegate_call!(self.poll_flush(cx))
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
        delegate_call!(self.poll_shutdown(cx))
    }
}

impl<L, R> futures_core::stream::Stream for Either<L, R>
where
    L: futures_core::stream::Stream,
    R: futures_core::stream::Stream<Item = L::Item>,
{
    type Item = L::Item;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        delegate_call!(self.poll_next(cx))
    }
}

impl<L, R, Item, Error> futures_sink::Sink<Item> for Either<L, R>
where
    L: futures_sink::Sink<Item, Error = Error>,
    R: futures_sink::Sink<Item, Error = Error>,
{
    type Error = Error;

    fn poll_ready(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<std::result::Result<(), Self::Error>> {
        delegate_call!(self.poll_ready(cx))
    }

    fn start_send(self: Pin<&mut Self>, item: Item) -> std::result::Result<(), Self::Error> {
        delegate_call!(self.start_send(item))
    }

    fn poll_flush(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<std::result::Result<(), Self::Error>> {
        delegate_call!(self.poll_flush(cx))
    }

    fn poll_close(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<std::result::Result<(), Self::Error>> {
        delegate_call!(self.poll_close(cx))
    }
Repository: tokio-rs/tokio
File: tokio-util/src/either.rs (Rust)

Code before:

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
        delegate_call!(self.poll_shutdown(cx))
    }
}

impl<L, R> futures_core::stream::Stream for Either<L, R>
where
    L: futures_core::stream::Stream,
    R: futures_core::stream::Stream<Item = L::Item>,
{
    type Item = L::Item;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        delegate_call!(self.poll_next(cx))
    }
}

impl<L, R, Item, Error> futures_sink::Sink<Item> for Either<L, R>
where
    L: futures_sink::Sink<Item, Error = Error>,
    R: futures_sink::Sink<Item, Error = Error>,
{
    type Error = Error;

    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
        delegate_call!(self.poll_ready(cx))
    }

    fn start_send(self: Pin<&mut Self>, item: Item) -> Result<()> {
        delegate_call!(self.start_send(item))
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
        delegate_call!(self.poll_flush(cx))
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
        delegate_call!(self.poll_close(cx))
    }
}


#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::{repeat, AsyncReadExt, Repeat};
    use tokio_stream::{once, Once, StreamExt};

    #[tokio::test]
    async fn either_is_stream() {

Review comment (bug):
In each method, you need this:
```suggestion
    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
```
Right now, you're using `std::io::Error` instead, which is the wrong error type.

Code after:

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
        delegate_call!(self.poll_shutdown(cx))
    }
}

impl<L, R> futures_core::stream::Stream for Either<L, R>
where
    L: futures_core::stream::Stream,
    R: futures_core::stream::Stream<Item = L::Item>,
{
    type Item = L::Item;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        delegate_call!(self.poll_next(cx))
    }
}

impl<L, R, Item, Error> futures_sink::Sink<Item> for Either<L, R>
where
    L: futures_sink::Sink<Item, Error = Error>,
    R: futures_sink::Sink<Item, Error = Error>,
{
    type Error = Error;

    fn poll_ready(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<std::result::Result<(), Self::Error>> {
        delegate_call!(self.poll_ready(cx))
    }

    fn start_send(self: Pin<&mut Self>, item: Item) -> std::result::Result<(), Self::Error> {
        delegate_call!(self.start_send(item))
    }

    fn poll_flush(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<std::result::Result<(), Self::Error>> {
        delegate_call!(self.poll_flush(cx))
    }

    fn poll_close(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<std::result::Result<(), Self::Error>> {
        delegate_call!(self.poll_close(cx))
    }
}
Repository: tokio-rs/tokio
File: tokio/src/io/util/repeat.rs (Rust)

Code before:
    /// }
    /// ```
    pub fn repeat(byte: u8) -> Repeat {
        Repeat { byte }
    }
}

impl AsyncRead for Repeat {
    #[inline]
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        ready!(crate::trace::trace_leaf(cx));
        ready!(poll_proceed_and_make_progress(cx));
        // TODO: could be faster, but should we unsafe it?
        while buf.remaining() != 0 {
            buf.put_slice(&[self.byte]);
        }
        Poll::Ready(Ok(()))
    }
}

cfg_coop! {
    fn poll_proceed_and_make_progress(cx: &mut Context<'_>) -> Poll<()> {
        let coop = ready!(crate::runtime::coop::poll_proceed(cx));
        coop.made_progress();
        Poll::Ready(())
    }
}

cfg_not_coop! {
    fn poll_proceed_and_make_progress(_: &mut Context<'_>) -> Poll<()> {
        Poll::Ready(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn assert_unpin() {
        crate::is_unpin::<Repeat>();
    }
}


Review comment (refactor):
I belive these codes are duplicates from file `tokio/src/io/util/empty.rs`, we can move these code to `tokio/src/io/util/mod.rs`.

Code after:
    ///     assert_eq!(buffer, [0b101, 0b101, 0b101]);
    /// }
    /// ```
    pub fn repeat(byte: u8) -> Repeat {
        Repeat { byte }
    }
}

impl AsyncRead for Repeat {
    #[inline]
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        ready!(crate::trace::trace_leaf(cx));
        ready!(poll_proceed_and_make_progress(cx));
        // TODO: could be faster, but should we unsafe it?
        while buf.remaining() != 0 {
            buf.put_slice(&[self.byte]);
        }
        Poll::Ready(Ok(()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn assert_unpin() {
        crate::is_unpin::<Repeat>();
    }
}
Repository: tokio-rs/tokio
File: tokio/tests/io_sink.rs (Rust)

Code before:
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full"))]

use tokio::io::AsyncWriteExt;

#[tokio::test]
async fn sink_is_cooperative() {
    tokio::select! {
        biased;
        _ = async {
            loop {
                let buf= vec![1, 2, 3];
                let _ = tokio::io::sink().write(&buf).await.unwrap();
            }
        } => {},
        _ = tokio::task::yield_now() => {}
    }
}


Review comment (suggestion):
```suggestion
                let buf = vec![1, 2, 3];
```

Code after:
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full"))]

use tokio::io::AsyncWriteExt;

#[tokio::test]
async fn sink_poll_write_is_cooperative() {
    tokio::select! {
        biased;
        _ = async {
            loop {
                let buf = vec![1, 2, 3];
                tokio::io::sink().write_all(&buf).await.unwrap();
            }
        } => {},
        _ = tokio::task::yield_now() => {}
    }
}

#[tokio::test]
async fn sink_poll_flush_is_cooperative() {
    tokio::select! {
        biased;
        _ = async {
            loop {
                tokio::io::sink().flush().await.unwrap();
            }
        } => {},
        _ = tokio::task::yield_now() => {}
    }
}

#[tokio::test]
async fn sink_poll_shutdown_is_cooperative() {
    tokio::select! {
        biased;
        _ = async {
Repository: tokio-rs/tokio
File: tokio/tests/io_sink.rs (Rust)

Code before:
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full"))]

use tokio::io::AsyncWriteExt;

#[tokio::test]
async fn sink_is_cooperative() {
    tokio::select! {
        biased;
        _ = async {
            loop {
                let buf= vec![1, 2, 3];
                let _ = tokio::io::sink().write(&buf).await.unwrap();
            }
        } => {},
        _ = tokio::task::yield_now() => {}
    }
}


Review comment (suggestion):
I don't think this is ignoring anything?
```suggestion
                tokio::io::sink().write(&buf).await.unwrap();
```

Code after:
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full"))]

use tokio::io::AsyncWriteExt;

#[tokio::test]
async fn sink_poll_write_is_cooperative() {
    tokio::select! {
        biased;
        _ = async {
            loop {
                let buf = vec![1, 2, 3];
                tokio::io::sink().write_all(&buf).await.unwrap();
            }
        } => {},
        _ = tokio::task::yield_now() => {}
    }
}

#[tokio::test]
async fn sink_poll_flush_is_cooperative() {
    tokio::select! {
        biased;
        _ = async {
            loop {
                tokio::io::sink().flush().await.unwrap();
            }
        } => {},
        _ = tokio::task::yield_now() => {}
    }
}

#[tokio::test]
async fn sink_poll_shutdown_is_cooperative() {
    tokio::select! {
        biased;
        _ = async {
            loop {
Repository: tokio-rs/tokio
File: tokio-util/src/time/mod.rs (Rust)

Code before:
#[doc(inline)]
pub use delay_queue::DelayQueue;
use futures_core::Future;
use tokio::time::Timeout;

/// A trait which contains a variety of conevenient adapters and utilities for `Future`s.
pub trait FutureExt: Future + Sized {
    /// A wrapper around [`tokio::time::timeout`], with the advantage that it is easier to write
    /// fluent call chains.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tokio::{sync::oneshot, time::Duration};
    /// use tokio_util::time::FutureExt;
    ///
    /// # async fn dox() {
    /// let (tx, rx) = oneshot::channel::<()>();
    ///
    /// let res = rx.timeout(Duration::from_millis(10)).await;
    /// assert!(res.is_err());
    /// # }
    /// ```
    fn timeout(self, timeout: Duration) -> Timeout<Self> {
        tokio::time::timeout(timeout, self)
    }
}

impl<T: Future + Sized> FutureExt for T {}

// ===== Internal utils =====

enum Round {
    Up,
    Down,
}

/// Convert a `Duration` to milliseconds, rounding up and saturating at
/// `u64::MAX`.
///
/// The saturating is fine because `u64::MAX` milliseconds are still many
/// million years.
#[inline]
fn ms(duration: Duration, round: Round) -> u64 {
    const NANOS_PER_MILLI: u32 = 1_000_000;
    const MILLIS_PER_SEC: u64 = 1_000;

    // Round up.
    let millis = match round {
        Round::Up => (duration.subsec_nanos() + NANOS_PER_MILLI - 1) / NANOS_PER_MILLI,
        Round::Down => duration.subsec_millis(),

Review comment (suggestion):
Please use this mechanism instead of specifying `Sized` as a sub-trait.
```suggestion
    fn timeout(self, timeout: Duration) -> Timeout<Self>
    where
        Self: Sized,
    {
        tokio::time::timeout(timeout, self)
    }
```

Code after:
pub mod delay_queue;

#[doc(inline)]
pub use delay_queue::DelayQueue;

/// A trait which contains a variety of convenient adapters and utilities for `Future`s.
pub trait FutureExt: Future {
    /// A wrapper around [`tokio::time::timeout`], with the advantage that it is easier to write
    /// fluent call chains.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tokio::{sync::oneshot, time::Duration};
    /// use tokio_util::time::FutureExt;
    ///
    /// # async fn dox() {
    /// let (tx, rx) = oneshot::channel::<()>();
    ///
    /// let res = rx.timeout(Duration::from_millis(10)).await;
    /// assert!(res.is_err());
    /// # }
    /// ```
    fn timeout(self, timeout: Duration) -> Timeout<Self>
    where
        Self: Sized,
    {
        tokio::time::timeout(timeout, self)
    }
}

impl<T: Future + ?Sized> FutureExt for T {}

// ===== Internal utils =====

enum Round {
    Up,
    Down,
}

/// Convert a `Duration` to milliseconds, rounding up and saturating at
/// `u64::MAX`.
///
/// The saturating is fine because `u64::MAX` milliseconds are still many
/// million years.
#[inline]
fn ms(duration: Duration, round: Round) -> u64 {
    const NANOS_PER_MILLI: u32 = 1_000_000;
    const MILLIS_PER_SEC: u64 = 1_000;

    // Round up.
Repository: tokio-rs/tokio
File: tokio-util/src/time/mod.rs (Rust)

Code before:
pub trait FutureExt: Future {
    /// A wrapper around [`tokio::time::timeout`], with the advantage that it is easier to write
    /// fluent call chains.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tokio::{sync::oneshot, time::Duration};
    /// use tokio_util::time::FutureExt;
    ///
    /// # async fn dox() {
    /// let (tx, rx) = oneshot::channel::<()>();
    ///
    /// let res = rx.timeout(Duration::from_millis(10)).await;
    /// assert!(res.is_err());
    /// # }
    /// ```
    fn timeout(self, timeout: Duration) -> Timeout<Self>
    where
        Self: Sized,
    {
        tokio::time::timeout(timeout, self)
    }
}

impl<T: Future + Sized> FutureExt for T {}

// ===== Internal utils =====

enum Round {
    Up,
    Down,
}

/// Convert a `Duration` to milliseconds, rounding up and saturating at
/// `u64::MAX`.
///
/// The saturating is fine because `u64::MAX` milliseconds are still many
/// million years.
#[inline]
fn ms(duration: Duration, round: Round) -> u64 {
    const NANOS_PER_MILLI: u32 = 1_000_000;
    const MILLIS_PER_SEC: u64 = 1_000;

    // Round up.
    let millis = match round {
        Round::Up => (duration.subsec_nanos() + NANOS_PER_MILLI - 1) / NANOS_PER_MILLI,
        Round::Down => duration.subsec_millis(),
    };

    duration

Review comment (suggestion):
```suggestion
impl<T: Future + ?Sized> FutureExt for T {}
```

Code after:
pub trait FutureExt: Future {
    /// A wrapper around [`tokio::time::timeout`], with the advantage that it is easier to write
    /// fluent call chains.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use tokio::{sync::oneshot, time::Duration};
    /// use tokio_util::time::FutureExt;
    ///
    /// # async fn dox() {
    /// let (tx, rx) = oneshot::channel::<()>();
    ///
    /// let res = rx.timeout(Duration::from_millis(10)).await;
    /// assert!(res.is_err());
    /// # }
    /// ```
    fn timeout(self, timeout: Duration) -> Timeout<Self>
    where
        Self: Sized,
    {
        tokio::time::timeout(timeout, self)
    }
}

impl<T: Future + ?Sized> FutureExt for T {}

// ===== Internal utils =====

enum Round {
    Up,
    Down,
}

/// Convert a `Duration` to milliseconds, rounding up and saturating at
/// `u64::MAX`.
///
/// The saturating is fine because `u64::MAX` milliseconds are still many
/// million years.
#[inline]
fn ms(duration: Duration, round: Round) -> u64 {
    const NANOS_PER_MILLI: u32 = 1_000_000;
    const MILLIS_PER_SEC: u64 = 1_000;

    // Round up.
    let millis = match round {
        Round::Up => (duration.subsec_nanos() + NANOS_PER_MILLI - 1) / NANOS_PER_MILLI,
        Round::Down => duration.subsec_millis(),
    };

    duration