Adding upstream version 115.8.0esr.upstream/115.8.0esr

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

37 files changed, 6891 insertions, 0 deletions

diff --git a/third_party/rust/tokio/src/io/util/async_buf_read_ext.rs b/third_party/rust/tokio/src/io/util/async_buf_read_ext.rs
new file mode 100644
index 0000000000..b241e354ba
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/async_buf_read_ext.rs
@@ -0,0 +1,351 @@
+use crate::io::util::fill_buf::{fill_buf, FillBuf};
+use crate::io::util::lines::{lines, Lines};
+use crate::io::util::read_line::{read_line, ReadLine};
+use crate::io::util::read_until::{read_until, ReadUntil};
+use crate::io::util::split::{split, Split};
+use crate::io::AsyncBufRead;
+
+cfg_io_util! {
+    /// An extension trait which adds utility methods to [`AsyncBufRead`] types.
+    ///
+    /// [`AsyncBufRead`]: crate::io::AsyncBufRead
+    pub trait AsyncBufReadExt: AsyncBufRead {
+        /// Reads all bytes into `buf` until the delimiter `byte` or EOF is reached.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn read_until(&mut self, byte: u8, buf: &mut Vec<u8>) -> io::Result<usize>;
+        /// ```
+        ///
+        /// This function will read bytes from the underlying stream until the
+        /// delimiter or EOF is found. Once found, all bytes up to, and including,
+        /// the delimiter (if found) will be appended to `buf`.
+        ///
+        /// If successful, this function will return the total number of bytes read.
+        ///
+        /// If this function returns `Ok(0)`, the stream has reached EOF.
+        ///
+        /// # Errors
+        ///
+        /// This function will ignore all instances of [`ErrorKind::Interrupted`] and
+        /// will otherwise return any errors returned by [`fill_buf`].
+        ///
+        /// If an I/O error is encountered then all bytes read so far will be
+        /// present in `buf` and its length will have been adjusted appropriately.
+        ///
+        /// [`fill_buf`]: AsyncBufRead::poll_fill_buf
+        /// [`ErrorKind::Interrupted`]: std::io::ErrorKind::Interrupted
+        ///
+        /// # Cancel safety
+        ///
+        /// If the method is used as the event in a
+        /// [`tokio::select!`](crate::select) statement and some other branch
+        /// completes first, then some data may have been partially read. Any
+        /// partially read bytes are appended to `buf`, and the method can be
+        /// called again to continue reading until `byte`.
+        ///
+        /// This method returns the total number of bytes read. If you cancel
+        /// the call to `read_until` and then call it again to continue reading,
+        /// the counter is reset.
+        ///
+        /// # Examples
+        ///
+        /// [`std::io::Cursor`][`Cursor`] is a type that implements `BufRead`. In
+        /// this example, we use [`Cursor`] to read all the bytes in a byte slice
+        /// in hyphen delimited segments:
+        ///
+        /// [`Cursor`]: std::io::Cursor
+        ///
+        /// ```
+        /// use tokio::io::AsyncBufReadExt;
+        ///
+        /// use std::io::Cursor;
+        ///
+        /// #[tokio::main]
+        /// async fn main() {
+        ///     let mut cursor = Cursor::new(b"lorem-ipsum");
+        ///     let mut buf = vec![];
+        ///
+        ///     // cursor is at 'l'
+        ///     let num_bytes = cursor.read_until(b'-', &mut buf)
+        ///         .await
+        ///         .expect("reading from cursor won't fail");
+        ///
+        ///     assert_eq!(num_bytes, 6);
+        ///     assert_eq!(buf, b"lorem-");
+        ///     buf.clear();
+        ///
+        ///     // cursor is at 'i'
+        ///     let num_bytes = cursor.read_until(b'-', &mut buf)
+        ///         .await
+        ///         .expect("reading from cursor won't fail");
+        ///
+        ///     assert_eq!(num_bytes, 5);
+        ///     assert_eq!(buf, b"ipsum");
+        ///     buf.clear();
+        ///
+        ///     // cursor is at EOF
+        ///     let num_bytes = cursor.read_until(b'-', &mut buf)
+        ///         .await
+        ///         .expect("reading from cursor won't fail");
+        ///     assert_eq!(num_bytes, 0);
+        ///     assert_eq!(buf, b"");
+        /// }
+        /// ```
+        fn read_until<'a>(&'a mut self, byte: u8, buf: &'a mut Vec<u8>) -> ReadUntil<'a, Self>
+        where
+            Self: Unpin,
+        {
+            read_until(self, byte, buf)
+        }
+
+        /// Reads all bytes until a newline (the 0xA byte) is reached, and append
+        /// them to the provided buffer.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn read_line(&mut self, buf: &mut String) -> io::Result<usize>;
+        /// ```
+        ///
+        /// This function will read bytes from the underlying stream until the
+        /// newline delimiter (the 0xA byte) or EOF is found. Once found, all bytes
+        /// up to, and including, the delimiter (if found) will be appended to
+        /// `buf`.
+        ///
+        /// If successful, this function will return the total number of bytes read.
+        ///
+        /// If this function returns `Ok(0)`, the stream has reached EOF.
+        ///
+        /// # Errors
+        ///
+        /// This function has the same error semantics as [`read_until`] and will
+        /// also return an error if the read bytes are not valid UTF-8. If an I/O
+        /// error is encountered then `buf` may contain some bytes already read in
+        /// the event that all data read so far was valid UTF-8.
+        ///
+        /// [`read_until`]: AsyncBufReadExt::read_until
+        ///
+        /// # Cancel safety
+        ///
+        /// This method is not cancellation safe. If the method is used as the
+        /// event in a [`tokio::select!`](crate::select) statement and some
+        /// other branch completes first, then some data may have been partially
+        /// read, and this data is lost. There are no guarantees regarding the
+        /// contents of `buf` when the call is cancelled. The current
+        /// implementation replaces `buf` with the empty string, but this may
+        /// change in the future.
+        ///
+        /// This function does not behave like [`read_until`] because of the
+        /// requirement that a string contains only valid utf-8. If you need a
+        /// cancellation safe `read_line`, there are three options:
+        ///
+        ///  * Call [`read_until`] with a newline character and manually perform the utf-8 check.
+        ///  * The stream returned by [`lines`] has a cancellation safe
+        ///    [`next_line`] method.
+        ///  * Use [`tokio_util::codec::LinesCodec`][LinesCodec].
+        ///
+        /// [LinesCodec]: https://docs.rs/tokio-util/0.6/tokio_util/codec/struct.LinesCodec.html
+        /// [`read_until`]: Self::read_until
+        /// [`lines`]: Self::lines
+        /// [`next_line`]: crate::io::Lines::next_line
+        ///
+        /// # Examples
+        ///
+        /// [`std::io::Cursor`][`Cursor`] is a type that implements
+        /// `AsyncBufRead`. In this example, we use [`Cursor`] to read all the
+        /// lines in a byte slice:
+        ///
+        /// [`Cursor`]: std::io::Cursor
+        ///
+        /// ```
+        /// use tokio::io::AsyncBufReadExt;
+        ///
+        /// use std::io::Cursor;
+        ///
+        /// #[tokio::main]
+        /// async fn main() {
+        ///     let mut cursor = Cursor::new(b"foo\nbar");
+        ///     let mut buf = String::new();
+        ///
+        ///     // cursor is at 'f'
+        ///     let num_bytes = cursor.read_line(&mut buf)
+        ///         .await
+        ///         .expect("reading from cursor won't fail");
+        ///
+        ///     assert_eq!(num_bytes, 4);
+        ///     assert_eq!(buf, "foo\n");
+        ///     buf.clear();
+        ///
+        ///     // cursor is at 'b'
+        ///     let num_bytes = cursor.read_line(&mut buf)
+        ///         .await
+        ///         .expect("reading from cursor won't fail");
+        ///
+        ///     assert_eq!(num_bytes, 3);
+        ///     assert_eq!(buf, "bar");
+        ///     buf.clear();
+        ///
+        ///     // cursor is at EOF
+        ///     let num_bytes = cursor.read_line(&mut buf)
+        ///         .await
+        ///         .expect("reading from cursor won't fail");
+        ///
+        ///     assert_eq!(num_bytes, 0);
+        ///     assert_eq!(buf, "");
+        /// }
+        /// ```
+        fn read_line<'a>(&'a mut self, buf: &'a mut String) -> ReadLine<'a, Self>
+        where
+            Self: Unpin,
+        {
+            read_line(self, buf)
+        }
+
+        /// Returns a stream of the contents of this reader split on the byte
+        /// `byte`.
+        ///
+        /// This method is the asynchronous equivalent to
+        /// [`BufRead::split`](std::io::BufRead::split).
+        ///
+        /// The stream returned from this function will yield instances of
+        /// [`io::Result`]`<`[`Option`]`<`[`Vec<u8>`]`>>`. Each vector returned will *not* have
+        /// the delimiter byte at the end.
+        ///
+        /// [`io::Result`]: std::io::Result
+        /// [`Option`]: core::option::Option
+        /// [`Vec<u8>`]: std::vec::Vec
+        ///
+        /// # Errors
+        ///
+        /// Each item of the stream has the same error semantics as
+        /// [`AsyncBufReadExt::read_until`](AsyncBufReadExt::read_until).
+        ///
+        /// # Examples
+        ///
+        /// ```
+        /// # use tokio::io::AsyncBufRead;
+        /// use tokio::io::AsyncBufReadExt;
+        ///
+        /// # async fn dox(my_buf_read: impl AsyncBufRead + Unpin) -> std::io::Result<()> {
+        /// let mut segments = my_buf_read.split(b'f');
+        ///
+        /// while let Some(segment) = segments.next_segment().await? {
+        ///     println!("length = {}", segment.len())
+        /// }
+        /// # Ok(())
+        /// # }
+        /// ```
+        fn split(self, byte: u8) -> Split<Self>
+        where
+            Self: Sized + Unpin,
+        {
+            split(self, byte)
+        }
+
+        /// Returns the contents of the internal buffer, filling it with more
+        /// data from the inner reader if it is empty.
+        ///
+        /// This function is a lower-level call. It needs to be paired with the
+        /// [`consume`] method to function properly. When calling this method,
+        /// none of the contents will be "read" in the sense that later calling
+        /// `read` may return the same contents. As such, [`consume`] must be
+        /// called with the number of bytes that are consumed from this buffer
+        /// to ensure that the bytes are never returned twice.
+        ///
+        /// An empty buffer returned indicates that the stream has reached EOF.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn fill_buf(&mut self) -> io::Result<&[u8]>;
+        /// ```
+        ///
+        /// # Errors
+        ///
+        /// This function will return an I/O error if the underlying reader was
+        /// read, but returned an error.
+        ///
+        /// [`consume`]: crate::io::AsyncBufReadExt::consume
+        fn fill_buf(&mut self) -> FillBuf<'_, Self>
+        where
+            Self: Unpin,
+        {
+            fill_buf(self)
+        }
+
+        /// Tells this buffer that `amt` bytes have been consumed from the
+        /// buffer, so they should no longer be returned in calls to [`read`].
+        ///
+        /// This function is a lower-level call. It needs to be paired with the
+        /// [`fill_buf`] method to function properly. This function does not
+        /// perform any I/O, it simply informs this object that some amount of
+        /// its buffer, returned from [`fill_buf`], has been consumed and should
+        /// no longer be returned. As such, this function may do odd things if
+        /// [`fill_buf`] isn't called before calling it.
+        ///
+        /// The `amt` must be less than the number of bytes in the buffer
+        /// returned by [`fill_buf`].
+        ///
+        /// [`read`]: crate::io::AsyncReadExt::read
+        /// [`fill_buf`]: crate::io::AsyncBufReadExt::fill_buf
+        fn consume(&mut self, amt: usize)
+        where
+            Self: Unpin,
+        {
+            std::pin::Pin::new(self).consume(amt)
+        }
+
+        /// Returns a stream over the lines of this reader.
+        /// This method is the async equivalent to [`BufRead::lines`](std::io::BufRead::lines).
+        ///
+        /// The stream returned from this function will yield instances of
+        /// [`io::Result`]`<`[`Option`]`<`[`String`]`>>`. Each string returned will *not* have a newline
+        /// byte (the 0xA byte) or CRLF (0xD, 0xA bytes) at the end.
+        ///
+        /// [`io::Result`]: std::io::Result
+        /// [`Option`]: core::option::Option
+        /// [`String`]: String
+        ///
+        /// # Errors
+        ///
+        /// Each line of the stream has the same error semantics as [`AsyncBufReadExt::read_line`].
+        ///
+        /// # Examples
+        ///
+        /// [`std::io::Cursor`][`Cursor`] is a type that implements `BufRead`. In
+        /// this example, we use [`Cursor`] to iterate over all the lines in a byte
+        /// slice.
+        ///
+        /// [`Cursor`]: std::io::Cursor
+        ///
+        /// ```
+        /// use tokio::io::AsyncBufReadExt;
+        ///
+        /// use std::io::Cursor;
+        ///
+        /// #[tokio::main]
+        /// async fn main() {
+        ///     let cursor = Cursor::new(b"lorem\nipsum\r\ndolor");
+        ///
+        ///     let mut lines = cursor.lines();
+        ///
+        ///     assert_eq!(lines.next_line().await.unwrap(), Some(String::from("lorem")));
+        ///     assert_eq!(lines.next_line().await.unwrap(), Some(String::from("ipsum")));
+        ///     assert_eq!(lines.next_line().await.unwrap(), Some(String::from("dolor")));
+        ///     assert_eq!(lines.next_line().await.unwrap(), None);
+        /// }
+        /// ```
+        ///
+        /// [`AsyncBufReadExt::read_line`]: AsyncBufReadExt::read_line
+        fn lines(self) -> Lines<Self>
+        where
+            Self: Sized,
+        {
+            lines(self)
+        }
+    }
+}
+
+impl<R: AsyncBufRead + ?Sized> AsyncBufReadExt for R {}
diff --git a/third_party/rust/tokio/src/io/util/async_read_ext.rs b/third_party/rust/tokio/src/io/util/async_read_ext.rs
new file mode 100644
index 0000000000..df5445c2c6
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/async_read_ext.rs
@@ -0,0 +1,1294 @@
+use crate::io::util::chain::{chain, Chain};
+use crate::io::util::read::{read, Read};
+use crate::io::util::read_buf::{read_buf, ReadBuf};
+use crate::io::util::read_exact::{read_exact, ReadExact};
+use crate::io::util::read_int::{ReadF32, ReadF32Le, ReadF64, ReadF64Le};
+use crate::io::util::read_int::{
+    ReadI128, ReadI128Le, ReadI16, ReadI16Le, ReadI32, ReadI32Le, ReadI64, ReadI64Le, ReadI8,
+};
+use crate::io::util::read_int::{
+    ReadU128, ReadU128Le, ReadU16, ReadU16Le, ReadU32, ReadU32Le, ReadU64, ReadU64Le, ReadU8,
+};
+use crate::io::util::read_to_end::{read_to_end, ReadToEnd};
+use crate::io::util::read_to_string::{read_to_string, ReadToString};
+use crate::io::util::take::{take, Take};
+use crate::io::AsyncRead;
+
+use bytes::BufMut;
+
+cfg_io_util! {
+    /// Defines numeric reader
+    macro_rules! read_impl {
+        (
+            $(
+                $(#[$outer:meta])*
+                fn $name:ident(&mut self) -> $($fut:ident)*;
+            )*
+        ) => {
+            $(
+                $(#[$outer])*
+                fn $name<'a>(&'a mut self) -> $($fut)*<&'a mut Self> where Self: Unpin {
+                    $($fut)*::new(self)
+                }
+            )*
+        }
+    }
+
+    /// Reads bytes from a source.
+    ///
+    /// Implemented as an extension trait, adding utility methods to all
+    /// [`AsyncRead`] types. Callers will tend to import this trait instead of
+    /// [`AsyncRead`].
+    ///
+    /// ```no_run
+    /// use tokio::fs::File;
+    /// use tokio::io::{self, AsyncReadExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> io::Result<()> {
+    ///     let mut f = File::open("foo.txt").await?;
+    ///     let mut buffer = [0; 10];
+    ///
+    ///     // The `read` method is defined by this trait.
+    ///     let n = f.read(&mut buffer[..]).await?;
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    ///
+    /// See [module][crate::io] documentation for more details.
+    ///
+    /// [`AsyncRead`]: AsyncRead
+    pub trait AsyncReadExt: AsyncRead {
+        /// Creates a new `AsyncRead` instance that chains this stream with
+        /// `next`.
+        ///
+        /// The returned `AsyncRead` instance will first read all bytes from this object
+        /// until EOF is encountered. Afterwards the output is equivalent to the
+        /// output of `next`.
+        ///
+        /// # Examples
+        ///
+        /// [`File`][crate::fs::File]s implement `AsyncRead`:
+        ///
+        /// ```no_run
+        /// use tokio::fs::File;
+        /// use tokio::io::{self, AsyncReadExt};
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let f1 = File::open("foo.txt").await?;
+        ///     let f2 = File::open("bar.txt").await?;
+        ///
+        ///     let mut handle = f1.chain(f2);
+        ///     let mut buffer = String::new();
+        ///
+        ///     // read the value into a String. We could use any AsyncRead
+        ///     // method here, this is just one example.
+        ///     handle.read_to_string(&mut buffer).await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        fn chain<R>(self, next: R) -> Chain<Self, R>
+        where
+            Self: Sized,
+            R: AsyncRead,
+        {
+            chain(self, next)
+        }
+
+        /// Pulls some bytes from this source into the specified buffer,
+        /// returning how many bytes were read.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize>;
+        /// ```
+        ///
+        /// This method does not provide any guarantees about whether it
+        /// completes immediately or asynchronously.
+        ///
+        /// # Return
+        ///
+        /// If the return value of this method is `Ok(n)`, then it must be
+        /// guaranteed that `0 <= n <= buf.len()`. A nonzero `n` value indicates
+        /// that the buffer `buf` has been filled in with `n` bytes of data from
+        /// this source. If `n` is `0`, then it can indicate one of two
+        /// scenarios:
+        ///
+        /// 1. This reader has reached its "end of file" and will likely no longer
+        ///    be able to produce bytes. Note that this does not mean that the
+        ///    reader will *always* no longer be able to produce bytes.
+        /// 2. The buffer specified was 0 bytes in length.
+        ///
+        /// No guarantees are provided about the contents of `buf` when this
+        /// function is called, implementations cannot rely on any property of the
+        /// contents of `buf` being `true`. It is recommended that *implementations*
+        /// only write data to `buf` instead of reading its contents.
+        ///
+        /// Correspondingly, however, *callers* of this method may not assume
+        /// any guarantees about how the implementation uses `buf`. It is
+        /// possible that the code that's supposed to write to the buffer might
+        /// also read from it. It is your responsibility to make sure that `buf`
+        /// is initialized before calling `read`.
+        ///
+        /// # Errors
+        ///
+        /// If this function encounters any form of I/O or other error, an error
+        /// variant will be returned. If an error is returned then it must be
+        /// guaranteed that no bytes were read.
+        ///
+        /// # Cancel safety
+        ///
+        /// This method is cancel safe. If you use it as the event in a
+        /// [`tokio::select!`](crate::select) statement and some other branch
+        /// completes first, then it is guaranteed that no data was read.
+        ///
+        /// # Examples
+        ///
+        /// [`File`][crate::fs::File]s implement `Read`:
+        ///
+        /// ```no_run
+        /// use tokio::fs::File;
+        /// use tokio::io::{self, AsyncReadExt};
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut f = File::open("foo.txt").await?;
+        ///     let mut buffer = [0; 10];
+        ///
+        ///     // read up to 10 bytes
+        ///     let n = f.read(&mut buffer[..]).await?;
+        ///
+        ///     println!("The bytes: {:?}", &buffer[..n]);
+        ///     Ok(())
+        /// }
+        /// ```
+        fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Read<'a, Self>
+        where
+            Self: Unpin,
+        {
+            read(self, buf)
+        }
+
+        /// Pulls some bytes from this source into the specified buffer,
+        /// advancing the buffer's internal cursor.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> io::Result<usize>;
+        /// ```
+        ///
+        /// Usually, only a single `read` syscall is issued, even if there is
+        /// more space in the supplied buffer.
+        ///
+        /// This method does not provide any guarantees about whether it
+        /// completes immediately or asynchronously.
+        ///
+        /// # Return
+        ///
+        /// A nonzero `n` value indicates that the buffer `buf` has been filled
+        /// in with `n` bytes of data from this source. If `n` is `0`, then it
+        /// can indicate one of two scenarios:
+        ///
+        /// 1. This reader has reached its "end of file" and will likely no longer
+        ///    be able to produce bytes. Note that this does not mean that the
+        ///    reader will *always* no longer be able to produce bytes.
+        /// 2. The buffer specified had a remaining capacity of zero.
+        ///
+        /// # Errors
+        ///
+        /// If this function encounters any form of I/O or other error, an error
+        /// variant will be returned. If an error is returned then it must be
+        /// guaranteed that no bytes were read.
+        ///
+        /// # Cancel safety
+        ///
+        /// This method is cancel safe. If you use it as the event in a
+        /// [`tokio::select!`](crate::select) statement and some other branch
+        /// completes first, then it is guaranteed that no data was read.
+        ///
+        /// # Examples
+        ///
+        /// [`File`] implements `Read` and [`BytesMut`] implements [`BufMut`]:
+        ///
+        /// [`File`]: crate::fs::File
+        /// [`BytesMut`]: bytes::BytesMut
+        /// [`BufMut`]: bytes::BufMut
+        ///
+        /// ```no_run
+        /// use tokio::fs::File;
+        /// use tokio::io::{self, AsyncReadExt};
+        ///
+        /// use bytes::BytesMut;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut f = File::open("foo.txt").await?;
+        ///     let mut buffer = BytesMut::with_capacity(10);
+        ///
+        ///     assert!(buffer.is_empty());
+        ///
+        ///     // read up to 10 bytes, note that the return value is not needed
+        ///     // to access the data that was read as `buffer`'s internal
+        ///     // cursor is updated.
+        ///     f.read_buf(&mut buffer).await?;
+        ///
+        ///     println!("The bytes: {:?}", &buffer[..]);
+        ///     Ok(())
+        /// }
+        /// ```
+        fn read_buf<'a, B>(&'a mut self, buf: &'a mut B) -> ReadBuf<'a, Self, B>
+        where
+            Self: Sized + Unpin,
+            B: BufMut,
+        {
+            read_buf(self, buf)
+        }
+
+        /// Reads the exact number of bytes required to fill `buf`.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn read_exact(&mut self, buf: &mut [u8]) -> io::Result<usize>;
+        /// ```
+        ///
+        /// This function reads as many bytes as necessary to completely fill
+        /// the specified buffer `buf`.
+        ///
+        /// # Errors
+        ///
+        /// If the operation encounters an "end of file" before completely
+        /// filling the buffer, it returns an error of the kind
+        /// [`ErrorKind::UnexpectedEof`]. The contents of `buf` are unspecified
+        /// in this case.
+        ///
+        /// If any other read error is encountered then the operation
+        /// immediately returns. The contents of `buf` are unspecified in this
+        /// case.
+        ///
+        /// If this operation returns an error, it is unspecified how many bytes
+        /// it has read, but it will never read more than would be necessary to
+        /// completely fill the buffer.
+        ///
+        /// # Cancel safety
+        ///
+        /// This method is not cancellation safe. If the method is used as the
+        /// event in a [`tokio::select!`](crate::select) statement and some
+        /// other branch completes first, then some data may already have been
+        /// read into `buf`.
+        ///
+        /// # Examples
+        ///
+        /// [`File`][crate::fs::File]s implement `Read`:
+        ///
+        /// ```no_run
+        /// use tokio::fs::File;
+        /// use tokio::io::{self, AsyncReadExt};
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut f = File::open("foo.txt").await?;
+        ///     let mut buffer = [0; 10];
+        ///
+        ///     // read exactly 10 bytes
+        ///     f.read_exact(&mut buffer).await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        ///
+        /// [`ErrorKind::UnexpectedEof`]: std::io::ErrorKind::UnexpectedEof
+        fn read_exact<'a>(&'a mut self, buf: &'a mut [u8]) -> ReadExact<'a, Self>
+        where
+            Self: Unpin,
+        {
+            read_exact(self, buf)
+        }
+
+        read_impl! {
+            /// Reads an unsigned 8 bit integer from the underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u8(&mut self) -> io::Result<u8>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 8 bit integers from an `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![2, 5]);
+            ///
+            ///     assert_eq!(2, reader.read_u8().await?);
+            ///     assert_eq!(5, reader.read_u8().await?);
+            ///
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u8(&mut self) -> ReadU8;
+
+            /// Reads a signed 8 bit integer from the underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i8(&mut self) -> io::Result<i8>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 8 bit integers from an `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0x02, 0xfb]);
+            ///
+            ///     assert_eq!(2, reader.read_i8().await?);
+            ///     assert_eq!(-5, reader.read_i8().await?);
+            ///
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i8(&mut self) -> ReadI8;
+
+            /// Reads an unsigned 16-bit integer in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u16(&mut self) -> io::Result<u16>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 16 bit big-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![2, 5, 3, 0]);
+            ///
+            ///     assert_eq!(517, reader.read_u16().await?);
+            ///     assert_eq!(768, reader.read_u16().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u16(&mut self) -> ReadU16;
+
+            /// Reads a signed 16-bit integer in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i16(&mut self) -> io::Result<i16>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read signed 16 bit big-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0x00, 0xc1, 0xff, 0x7c]);
+            ///
+            ///     assert_eq!(193, reader.read_i16().await?);
+            ///     assert_eq!(-132, reader.read_i16().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i16(&mut self) -> ReadI16;
+
+            /// Reads an unsigned 32-bit integer in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u32(&mut self) -> io::Result<u32>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 32-bit big-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0x00, 0x00, 0x01, 0x0b]);
+            ///
+            ///     assert_eq!(267, reader.read_u32().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u32(&mut self) -> ReadU32;
+
+            /// Reads a signed 32-bit integer in big-endian order from the
+            /// underlying reader.
+            ///
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i32(&mut self) -> io::Result<i32>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read signed 32-bit big-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0xff, 0xff, 0x7a, 0x33]);
+            ///
+            ///     assert_eq!(-34253, reader.read_i32().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i32(&mut self) -> ReadI32;
+
+            /// Reads an unsigned 64-bit integer in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u64(&mut self) -> io::Result<u64>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 64-bit big-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![
+            ///         0x00, 0x03, 0x43, 0x95, 0x4d, 0x60, 0x86, 0x83
+            ///     ]);
+            ///
+            ///     assert_eq!(918733457491587, reader.read_u64().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u64(&mut self) -> ReadU64;
+
+            /// Reads an signed 64-bit integer in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i64(&mut self) -> io::Result<i64>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read signed 64-bit big-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0x80, 0, 0, 0, 0, 0, 0, 0]);
+            ///
+            ///     assert_eq!(i64::MIN, reader.read_i64().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i64(&mut self) -> ReadI64;
+
+            /// Reads an unsigned 128-bit integer in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u128(&mut self) -> io::Result<u128>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 128-bit big-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![
+            ///         0x00, 0x03, 0x43, 0x95, 0x4d, 0x60, 0x86, 0x83,
+            ///         0x00, 0x03, 0x43, 0x95, 0x4d, 0x60, 0x86, 0x83
+            ///     ]);
+            ///
+            ///     assert_eq!(16947640962301618749969007319746179, reader.read_u128().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u128(&mut self) -> ReadU128;
+
+            /// Reads an signed 128-bit integer in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i128(&mut self) -> io::Result<i128>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read signed 128-bit big-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![
+            ///         0x80, 0, 0, 0, 0, 0, 0, 0,
+            ///         0, 0, 0, 0, 0, 0, 0, 0
+            ///     ]);
+            ///
+            ///     assert_eq!(i128::MIN, reader.read_i128().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i128(&mut self) -> ReadI128;
+
+            /// Reads an 32-bit floating point type in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_f32(&mut self) -> io::Result<f32>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read 32-bit floating point type from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0xff, 0x7f, 0xff, 0xff]);
+            ///
+            ///     assert_eq!(f32::MIN, reader.read_f32().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_f32(&mut self) -> ReadF32;
+
+            /// Reads an 64-bit floating point type in big-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_f64(&mut self) -> io::Result<f64>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read 64-bit floating point type from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![
+            ///         0xff, 0xef, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff
+            ///     ]);
+            ///
+            ///     assert_eq!(f64::MIN, reader.read_f64().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_f64(&mut self) -> ReadF64;
+
+            /// Reads an unsigned 16-bit integer in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u16_le(&mut self) -> io::Result<u16>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 16 bit little-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![2, 5, 3, 0]);
+            ///
+            ///     assert_eq!(1282, reader.read_u16_le().await?);
+            ///     assert_eq!(3, reader.read_u16_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u16_le(&mut self) -> ReadU16Le;
+
+            /// Reads a signed 16-bit integer in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i16_le(&mut self) -> io::Result<i16>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read signed 16 bit little-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0x00, 0xc1, 0xff, 0x7c]);
+            ///
+            ///     assert_eq!(-16128, reader.read_i16_le().await?);
+            ///     assert_eq!(31999, reader.read_i16_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i16_le(&mut self) -> ReadI16Le;
+
+            /// Reads an unsigned 32-bit integer in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u32_le(&mut self) -> io::Result<u32>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 32-bit little-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0x00, 0x00, 0x01, 0x0b]);
+            ///
+            ///     assert_eq!(184614912, reader.read_u32_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u32_le(&mut self) -> ReadU32Le;
+
+            /// Reads a signed 32-bit integer in little-endian order from the
+            /// underlying reader.
+            ///
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i32_le(&mut self) -> io::Result<i32>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read signed 32-bit little-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0xff, 0xff, 0x7a, 0x33]);
+            ///
+            ///     assert_eq!(863698943, reader.read_i32_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i32_le(&mut self) -> ReadI32Le;
+
+            /// Reads an unsigned 64-bit integer in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u64_le(&mut self) -> io::Result<u64>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 64-bit little-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![
+            ///         0x00, 0x03, 0x43, 0x95, 0x4d, 0x60, 0x86, 0x83
+            ///     ]);
+            ///
+            ///     assert_eq!(9477368352180732672, reader.read_u64_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u64_le(&mut self) -> ReadU64Le;
+
+            /// Reads an signed 64-bit integer in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i64_le(&mut self) -> io::Result<i64>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read signed 64-bit little-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0x80, 0, 0, 0, 0, 0, 0, 0]);
+            ///
+            ///     assert_eq!(128, reader.read_i64_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i64_le(&mut self) -> ReadI64Le;
+
+            /// Reads an unsigned 128-bit integer in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_u128_le(&mut self) -> io::Result<u128>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read unsigned 128-bit little-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![
+            ///         0x00, 0x03, 0x43, 0x95, 0x4d, 0x60, 0x86, 0x83,
+            ///         0x00, 0x03, 0x43, 0x95, 0x4d, 0x60, 0x86, 0x83
+            ///     ]);
+            ///
+            ///     assert_eq!(174826588484952389081207917399662330624, reader.read_u128_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_u128_le(&mut self) -> ReadU128Le;
+
+            /// Reads an signed 128-bit integer in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_i128_le(&mut self) -> io::Result<i128>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read signed 128-bit little-endian integers from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![
+            ///         0x80, 0, 0, 0, 0, 0, 0, 0,
+            ///         0, 0, 0, 0, 0, 0, 0, 0
+            ///     ]);
+            ///
+            ///     assert_eq!(128, reader.read_i128_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i128_le(&mut self) -> ReadI128Le;
+
+            /// Reads an 32-bit floating point type in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_f32_le(&mut self) -> io::Result<f32>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read 32-bit floating point type from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![0xff, 0xff, 0x7f, 0xff]);
+            ///
+            ///     assert_eq!(f32::MIN, reader.read_f32_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_f32_le(&mut self) -> ReadF32Le;
+
+            /// Reads an 64-bit floating point type in little-endian order from the
+            /// underlying reader.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn read_f64_le(&mut self) -> io::Result<f64>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered reader to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncReadExt::read_exact`].
+            ///
+            /// [`AsyncReadExt::read_exact`]: AsyncReadExt::read_exact
+            ///
+            /// # Examples
+            ///
+            /// Read 64-bit floating point type from a `AsyncRead`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncReadExt};
+            ///
+            /// use std::io::Cursor;
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut reader = Cursor::new(vec![
+            ///         0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xef, 0xff
+            ///     ]);
+            ///
+            ///     assert_eq!(f64::MIN, reader.read_f64_le().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_f64_le(&mut self) -> ReadF64Le;
+        }
+
+        /// Reads all bytes until EOF in this source, placing them into `buf`.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize>;
+        /// ```
+        ///
+        /// All bytes read from this source will be appended to the specified
+        /// buffer `buf`. This function will continuously call [`read()`] to
+        /// append more data to `buf` until [`read()`] returns `Ok(0)`.
+        ///
+        /// If successful, the total number of bytes read is returned.
+        ///
+        /// [`read()`]: AsyncReadExt::read
+        ///
+        /// # Errors
+        ///
+        /// If a read error is encountered then the `read_to_end` operation
+        /// immediately completes. Any bytes which have already been read will
+        /// be appended to `buf`.
+        ///
+        /// # Examples
+        ///
+        /// [`File`][crate::fs::File]s implement `Read`:
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, AsyncReadExt};
+        /// use tokio::fs::File;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut f = File::open("foo.txt").await?;
+        ///     let mut buffer = Vec::new();
+        ///
+        ///     // read the whole file
+        ///     f.read_to_end(&mut buffer).await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        ///
+        /// (See also the [`tokio::fs::read`] convenience function for reading from a
+        /// file.)
+        ///
+        /// [`tokio::fs::read`]: fn@crate::fs::read
+        fn read_to_end<'a>(&'a mut self, buf: &'a mut Vec<u8>) -> ReadToEnd<'a, Self>
+        where
+            Self: Unpin,
+        {
+            read_to_end(self, buf)
+        }
+
+        /// Reads all bytes until EOF in this source, appending them to `buf`.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn read_to_string(&mut self, buf: &mut String) -> io::Result<usize>;
+        /// ```
+        ///
+        /// If successful, the number of bytes which were read and appended to
+        /// `buf` is returned.
+        ///
+        /// # Errors
+        ///
+        /// If the data in this stream is *not* valid UTF-8 then an error is
+        /// returned and `buf` is unchanged.
+        ///
+        /// See [`read_to_end`][AsyncReadExt::read_to_end] for other error semantics.
+        ///
+        /// # Examples
+        ///
+        /// [`File`][crate::fs::File]s implement `Read`:
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, AsyncReadExt};
+        /// use tokio::fs::File;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut f = File::open("foo.txt").await?;
+        ///     let mut buffer = String::new();
+        ///
+        ///     f.read_to_string(&mut buffer).await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        ///
+        /// (See also the [`crate::fs::read_to_string`] convenience function for
+        /// reading from a file.)
+        ///
+        /// [`crate::fs::read_to_string`]: fn@crate::fs::read_to_string
+        fn read_to_string<'a>(&'a mut self, dst: &'a mut String) -> ReadToString<'a, Self>
+        where
+            Self: Unpin,
+        {
+            read_to_string(self, dst)
+        }
+
+        /// Creates an adaptor which reads at most `limit` bytes from it.
+        ///
+        /// This function returns a new instance of `AsyncRead` which will read
+        /// at most `limit` bytes, after which it will always return EOF
+        /// (`Ok(0)`). Any read errors will not count towards the number of
+        /// bytes read and future calls to [`read()`] may succeed.
+        ///
+        /// [`read()`]: fn@crate::io::AsyncReadExt::read
+        ///
+        /// [read]: AsyncReadExt::read
+        ///
+        /// # Examples
+        ///
+        /// [`File`][crate::fs::File]s implement `Read`:
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, AsyncReadExt};
+        /// use tokio::fs::File;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let f = File::open("foo.txt").await?;
+        ///     let mut buffer = [0; 5];
+        ///
+        ///     // read at most five bytes
+        ///     let mut handle = f.take(5);
+        ///
+        ///     handle.read(&mut buffer).await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        fn take(self, limit: u64) -> Take<Self>
+        where
+            Self: Sized,
+        {
+            take(self, limit)
+        }
+    }
+}
+
+impl<R: AsyncRead + ?Sized> AsyncReadExt for R {}
diff --git a/third_party/rust/tokio/src/io/util/async_seek_ext.rs b/third_party/rust/tokio/src/io/util/async_seek_ext.rs
new file mode 100644
index 0000000000..46b3e6c0d3
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/async_seek_ext.rs
@@ -0,0 +1,93 @@
+use crate::io::seek::{seek, Seek};
+use crate::io::AsyncSeek;
+use std::io::SeekFrom;
+
+cfg_io_util! {
+    /// An extension trait that adds utility methods to [`AsyncSeek`] types.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use std::io::{self, Cursor, SeekFrom};
+    /// use tokio::io::{AsyncSeekExt, AsyncReadExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> io::Result<()> {
+    ///     let mut cursor = Cursor::new(b"abcdefg");
+    ///
+    ///     // the `seek` method is defined by this trait
+    ///     cursor.seek(SeekFrom::Start(3)).await?;
+    ///
+    ///     let mut buf = [0; 1];
+    ///     let n = cursor.read(&mut buf).await?;
+    ///     assert_eq!(n, 1);
+    ///     assert_eq!(buf, [b'd']);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    ///
+    /// See [module][crate::io] documentation for more details.
+    ///
+    /// [`AsyncSeek`]: AsyncSeek
+    pub trait AsyncSeekExt: AsyncSeek {
+        /// Creates a future which will seek an IO object, and then yield the
+        /// new position in the object and the object itself.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn seek(&mut self, pos: SeekFrom) -> io::Result<u64>;
+        /// ```
+        ///
+        /// In the case of an error the buffer and the object will be discarded, with
+        /// the error yielded.
+        ///
+        /// # Examples
+        ///
+        /// ```no_run
+        /// use tokio::fs::File;
+        /// use tokio::io::{AsyncSeekExt, AsyncReadExt};
+        ///
+        /// use std::io::SeekFrom;
+        ///
+        /// # async fn dox() -> std::io::Result<()> {
+        /// let mut file = File::open("foo.txt").await?;
+        /// file.seek(SeekFrom::Start(6)).await?;
+        ///
+        /// let mut contents = vec![0u8; 10];
+        /// file.read_exact(&mut contents).await?;
+        /// # Ok(())
+        /// # }
+        /// ```
+        fn seek(&mut self, pos: SeekFrom) -> Seek<'_, Self>
+        where
+            Self: Unpin,
+        {
+            seek(self, pos)
+        }
+
+        /// Creates a future which will rewind to the beginning of the stream.
+        ///
+        /// This is convenience method, equivalent to to `self.seek(SeekFrom::Start(0))`.
+        fn rewind(&mut self) -> Seek<'_, Self>
+        where
+            Self: Unpin,
+        {
+            self.seek(SeekFrom::Start(0))
+        }
+
+        /// Creates a future which will return the current seek position from the
+        /// start of the stream.
+        ///
+        /// This is equivalent to `self.seek(SeekFrom::Current(0))`.
+        fn stream_position(&mut self) -> Seek<'_, Self>
+        where
+            Self: Unpin,
+        {
+            self.seek(SeekFrom::Current(0))
+        }
+    }
+}
+
+impl<S: AsyncSeek + ?Sized> AsyncSeekExt for S {}
diff --git a/third_party/rust/tokio/src/io/util/async_write_ext.rs b/third_party/rust/tokio/src/io/util/async_write_ext.rs
new file mode 100644
index 0000000000..93a318315e
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/async_write_ext.rs
@@ -0,0 +1,1293 @@
+use crate::io::util::flush::{flush, Flush};
+use crate::io::util::shutdown::{shutdown, Shutdown};
+use crate::io::util::write::{write, Write};
+use crate::io::util::write_all::{write_all, WriteAll};
+use crate::io::util::write_all_buf::{write_all_buf, WriteAllBuf};
+use crate::io::util::write_buf::{write_buf, WriteBuf};
+use crate::io::util::write_int::{WriteF32, WriteF32Le, WriteF64, WriteF64Le};
+use crate::io::util::write_int::{
+    WriteI128, WriteI128Le, WriteI16, WriteI16Le, WriteI32, WriteI32Le, WriteI64, WriteI64Le,
+    WriteI8,
+};
+use crate::io::util::write_int::{
+    WriteU128, WriteU128Le, WriteU16, WriteU16Le, WriteU32, WriteU32Le, WriteU64, WriteU64Le,
+    WriteU8,
+};
+use crate::io::util::write_vectored::{write_vectored, WriteVectored};
+use crate::io::AsyncWrite;
+use std::io::IoSlice;
+
+use bytes::Buf;
+
+cfg_io_util! {
+    /// Defines numeric writer.
+    macro_rules! write_impl {
+        (
+            $(
+                $(#[$outer:meta])*
+                fn $name:ident(&mut self, n: $ty:ty) -> $($fut:ident)*;
+            )*
+        ) => {
+            $(
+                $(#[$outer])*
+                fn $name<'a>(&'a mut self, n: $ty) -> $($fut)*<&'a mut Self> where Self: Unpin {
+                    $($fut)*::new(self, n)
+                }
+            )*
+        }
+    }
+
+    /// Writes bytes to a sink.
+    ///
+    /// Implemented as an extension trait, adding utility methods to all
+    /// [`AsyncWrite`] types. Callers will tend to import this trait instead of
+    /// [`AsyncWrite`].
+    ///
+    /// ```no_run
+    /// use tokio::io::{self, AsyncWriteExt};
+    /// use tokio::fs::File;
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> io::Result<()> {
+    ///     let data = b"some bytes";
+    ///
+    ///     let mut pos = 0;
+    ///     let mut buffer = File::create("foo.txt").await?;
+    ///
+    ///     while pos < data.len() {
+    ///         let bytes_written = buffer.write(&data[pos..]).await?;
+    ///         pos += bytes_written;
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    ///
+    /// See [module][crate::io] documentation for more details.
+    ///
+    /// [`AsyncWrite`]: AsyncWrite
+    pub trait AsyncWriteExt: AsyncWrite {
+        /// Writes a buffer into this writer, returning how many bytes were
+        /// written.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn write(&mut self, buf: &[u8]) -> io::Result<usize>;
+        /// ```
+        ///
+        /// This function will attempt to write the entire contents of `buf`, but
+        /// the entire write may not succeed, or the write may also generate an
+        /// error. A call to `write` represents *at most one* attempt to write to
+        /// any wrapped object.
+        ///
+        /// # Return
+        ///
+        /// If the return value is `Ok(n)` then it must be guaranteed that `n <=
+        /// buf.len()`. A return value of `0` typically means that the
+        /// underlying object is no longer able to accept bytes and will likely
+        /// not be able to in the future as well, or that the buffer provided is
+        /// empty.
+        ///
+        /// # Errors
+        ///
+        /// Each call to `write` may generate an I/O error indicating that the
+        /// operation could not be completed. If an error is returned then no bytes
+        /// in the buffer were written to this writer.
+        ///
+        /// It is **not** considered an error if the entire buffer could not be
+        /// written to this writer.
+        ///
+        /// # Cancel safety
+        ///
+        /// This method is cancellation safe in the sense that if it is used as
+        /// the event in a [`tokio::select!`](crate::select) statement and some
+        /// other branch completes first, then it is guaranteed that no data was
+        /// written to this `AsyncWrite`.
+        ///
+        /// # Examples
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, AsyncWriteExt};
+        /// use tokio::fs::File;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut file = File::create("foo.txt").await?;
+        ///
+        ///     // Writes some prefix of the byte string, not necessarily all of it.
+        ///     file.write(b"some bytes").await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        fn write<'a>(&'a mut self, src: &'a [u8]) -> Write<'a, Self>
+        where
+            Self: Unpin,
+        {
+            write(self, src)
+        }
+
+        /// Like [`write`], except that it writes from a slice of buffers.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> io::Result<usize>;
+        /// ```
+        ///
+        /// See [`AsyncWrite::poll_write_vectored`] for more details.
+        ///
+        /// # Cancel safety
+        ///
+        /// This method is cancellation safe in the sense that if it is used as
+        /// the event in a [`tokio::select!`](crate::select) statement and some
+        /// other branch completes first, then it is guaranteed that no data was
+        /// written to this `AsyncWrite`.
+        ///
+        /// # Examples
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, AsyncWriteExt};
+        /// use tokio::fs::File;
+        /// use std::io::IoSlice;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut file = File::create("foo.txt").await?;
+        ///
+        ///     let bufs: &[_] = &[
+        ///         IoSlice::new(b"hello"),
+        ///         IoSlice::new(b" "),
+        ///         IoSlice::new(b"world"),
+        ///     ];
+        ///
+        ///     file.write_vectored(&bufs).await?;
+        ///
+        ///     Ok(())
+        /// }
+        /// ```
+        ///
+        /// [`write`]: AsyncWriteExt::write
+        fn write_vectored<'a, 'b>(&'a mut self, bufs: &'a [IoSlice<'b>]) -> WriteVectored<'a, 'b, Self>
+        where
+            Self: Unpin,
+        {
+            write_vectored(self, bufs)
+        }
+
+        /// Writes a buffer into this writer, advancing the buffer's internal
+        /// cursor.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn write_buf<B: Buf>(&mut self, buf: &mut B) -> io::Result<usize>;
+        /// ```
+        ///
+        /// This function will attempt to write the entire contents of `buf`, but
+        /// the entire write may not succeed, or the write may also generate an
+        /// error. After the operation completes, the buffer's
+        /// internal cursor is advanced by the number of bytes written. A
+        /// subsequent call to `write_buf` using the **same** `buf` value will
+        /// resume from the point that the first call to `write_buf` completed.
+        /// A call to `write_buf` represents *at most one* attempt to write to any
+        /// wrapped object.
+        ///
+        /// # Return
+        ///
+        /// If the return value is `Ok(n)` then it must be guaranteed that `n <=
+        /// buf.len()`. A return value of `0` typically means that the
+        /// underlying object is no longer able to accept bytes and will likely
+        /// not be able to in the future as well, or that the buffer provided is
+        /// empty.
+        ///
+        /// # Errors
+        ///
+        /// Each call to `write` may generate an I/O error indicating that the
+        /// operation could not be completed. If an error is returned then no bytes
+        /// in the buffer were written to this writer.
+        ///
+        /// It is **not** considered an error if the entire buffer could not be
+        /// written to this writer.
+        ///
+        /// # Cancel safety
+        ///
+        /// This method is cancellation safe in the sense that if it is used as
+        /// the event in a [`tokio::select!`](crate::select) statement and some
+        /// other branch completes first, then it is guaranteed that no data was
+        /// written to this `AsyncWrite`.
+        ///
+        /// # Examples
+        ///
+        /// [`File`] implements [`AsyncWrite`] and [`Cursor`]`<&[u8]>` implements [`Buf`]:
+        ///
+        /// [`File`]: crate::fs::File
+        /// [`Buf`]: bytes::Buf
+        /// [`Cursor`]: std::io::Cursor
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, AsyncWriteExt};
+        /// use tokio::fs::File;
+        ///
+        /// use bytes::Buf;
+        /// use std::io::Cursor;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut file = File::create("foo.txt").await?;
+        ///     let mut buffer = Cursor::new(b"data to write");
+        ///
+        ///     // Loop until the entire contents of the buffer are written to
+        ///     // the file.
+        ///     while buffer.has_remaining() {
+        ///         // Writes some prefix of the byte string, not necessarily
+        ///         // all of it.
+        ///         file.write_buf(&mut buffer).await?;
+        ///     }
+        ///
+        ///     Ok(())
+        /// }
+        /// ```
+        fn write_buf<'a, B>(&'a mut self, src: &'a mut B) -> WriteBuf<'a, Self, B>
+        where
+            Self: Sized + Unpin,
+            B: Buf,
+        {
+            write_buf(self, src)
+        }
+
+        /// Attempts to write an entire buffer into this writer.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn write_all_buf(&mut self, buf: impl Buf) -> Result<(), io::Error> {
+        ///     while buf.has_remaining() {
+        ///         self.write_buf(&mut buf).await?;
+        ///     }
+        ///     Ok(())
+        /// }
+        /// ```
+        ///
+        /// This method will continuously call [`write`] until
+        /// [`buf.has_remaining()`](bytes::Buf::has_remaining) returns false. This method will not
+        /// return until the entire buffer has been successfully written or an error occurs. The
+        /// first error generated will be returned.
+        ///
+        /// The buffer is advanced after each chunk is successfully written. After failure,
+        /// `src.chunk()` will return the chunk that failed to write.
+        ///
+        /// # Cancel safety
+        ///
+        /// If `write_all_buf` is used as the event in a
+        /// [`tokio::select!`](crate::select) statement and some other branch
+        /// completes first, then the data in the provided buffer may have been
+        /// partially written. However, it is guaranteed that the provided
+        /// buffer has been [advanced] by the amount of bytes that have been
+        /// partially written.
+        ///
+        /// # Examples
+        ///
+        /// [`File`] implements [`AsyncWrite`] and [`Cursor`]`<&[u8]>` implements [`Buf`]:
+        ///
+        /// [`File`]: crate::fs::File
+        /// [`Buf`]: bytes::Buf
+        /// [`Cursor`]: std::io::Cursor
+        /// [advanced]: bytes::Buf::advance
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, AsyncWriteExt};
+        /// use tokio::fs::File;
+        ///
+        /// use std::io::Cursor;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut file = File::create("foo.txt").await?;
+        ///     let mut buffer = Cursor::new(b"data to write");
+        ///
+        ///     file.write_all_buf(&mut buffer).await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        ///
+        /// [`write`]: AsyncWriteExt::write
+        fn write_all_buf<'a, B>(&'a mut self, src: &'a mut B) -> WriteAllBuf<'a, Self, B>
+        where
+            Self: Sized + Unpin,
+            B: Buf,
+        {
+            write_all_buf(self, src)
+        }
+
+        /// Attempts to write an entire buffer into this writer.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn write_all(&mut self, buf: &[u8]) -> io::Result<()>;
+        /// ```
+        ///
+        /// This method will continuously call [`write`] until there is no more data
+        /// to be written. This method will not return until the entire buffer
+        /// has been successfully written or such an error occurs. The first
+        /// error generated from this method will be returned.
+        ///
+        /// # Cancel safety
+        ///
+        /// This method is not cancellation safe. If it is used as the event
+        /// in a [`tokio::select!`](crate::select) statement and some other
+        /// branch completes first, then the provided buffer may have been
+        /// partially written, but future calls to `write_all` will start over
+        /// from the beginning of the buffer.
+        ///
+        /// # Errors
+        ///
+        /// This function will return the first error that [`write`] returns.
+        ///
+        /// # Examples
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, AsyncWriteExt};
+        /// use tokio::fs::File;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let mut file = File::create("foo.txt").await?;
+        ///
+        ///     file.write_all(b"some bytes").await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        ///
+        /// [`write`]: AsyncWriteExt::write
+        fn write_all<'a>(&'a mut self, src: &'a [u8]) -> WriteAll<'a, Self>
+        where
+            Self: Unpin,
+        {
+            write_all(self, src)
+        }
+
+        write_impl! {
+            /// Writes an unsigned 8-bit integer to the underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u8(&mut self, n: u8) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 8 bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u8(2).await?;
+            ///     writer.write_u8(5).await?;
+            ///
+            ///     assert_eq!(writer, b"\x02\x05");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u8(&mut self, n: u8) -> WriteU8;
+
+            /// Writes an unsigned 8-bit integer to the underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i8(&mut self, n: i8) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 8 bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u8(2).await?;
+            ///     writer.write_u8(5).await?;
+            ///
+            ///     assert_eq!(writer, b"\x02\x05");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i8(&mut self, n: i8) -> WriteI8;
+
+            /// Writes an unsigned 16-bit integer in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u16(&mut self, n: u16) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 16-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u16(517).await?;
+            ///     writer.write_u16(768).await?;
+            ///
+            ///     assert_eq!(writer, b"\x02\x05\x03\x00");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u16(&mut self, n: u16) -> WriteU16;
+
+            /// Writes a signed 16-bit integer in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i16(&mut self, n: i16) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write signed 16-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_i16(193).await?;
+            ///     writer.write_i16(-132).await?;
+            ///
+            ///     assert_eq!(writer, b"\x00\xc1\xff\x7c");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i16(&mut self, n: i16) -> WriteI16;
+
+            /// Writes an unsigned 32-bit integer in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u32(&mut self, n: u32) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 32-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u32(267).await?;
+            ///     writer.write_u32(1205419366).await?;
+            ///
+            ///     assert_eq!(writer, b"\x00\x00\x01\x0b\x47\xd9\x3d\x66");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u32(&mut self, n: u32) -> WriteU32;
+
+            /// Writes a signed 32-bit integer in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i32(&mut self, n: i32) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write signed 32-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_i32(267).await?;
+            ///     writer.write_i32(1205419366).await?;
+            ///
+            ///     assert_eq!(writer, b"\x00\x00\x01\x0b\x47\xd9\x3d\x66");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i32(&mut self, n: i32) -> WriteI32;
+
+            /// Writes an unsigned 64-bit integer in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u64(&mut self, n: u64) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 64-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u64(918733457491587).await?;
+            ///     writer.write_u64(143).await?;
+            ///
+            ///     assert_eq!(writer, b"\x00\x03\x43\x95\x4d\x60\x86\x83\x00\x00\x00\x00\x00\x00\x00\x8f");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u64(&mut self, n: u64) -> WriteU64;
+
+            /// Writes an signed 64-bit integer in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i64(&mut self, n: i64) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write signed 64-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_i64(i64::MIN).await?;
+            ///     writer.write_i64(i64::MAX).await?;
+            ///
+            ///     assert_eq!(writer, b"\x80\x00\x00\x00\x00\x00\x00\x00\x7f\xff\xff\xff\xff\xff\xff\xff");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i64(&mut self, n: i64) -> WriteI64;
+
+            /// Writes an unsigned 128-bit integer in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u128(&mut self, n: u128) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 128-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u128(16947640962301618749969007319746179).await?;
+            ///
+            ///     assert_eq!(writer, vec![
+            ///         0x00, 0x03, 0x43, 0x95, 0x4d, 0x60, 0x86, 0x83,
+            ///         0x00, 0x03, 0x43, 0x95, 0x4d, 0x60, 0x86, 0x83
+            ///     ]);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u128(&mut self, n: u128) -> WriteU128;
+
+            /// Writes an signed 128-bit integer in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i128(&mut self, n: i128) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write signed 128-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_i128(i128::MIN).await?;
+            ///
+            ///     assert_eq!(writer, vec![
+            ///         0x80, 0, 0, 0, 0, 0, 0, 0,
+            ///         0, 0, 0, 0, 0, 0, 0, 0
+            ///     ]);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i128(&mut self, n: i128) -> WriteI128;
+
+            /// Writes an 32-bit floating point type in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_f32(&mut self, n: f32) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write 32-bit floating point type to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_f32(f32::MIN).await?;
+            ///
+            ///     assert_eq!(writer, vec![0xff, 0x7f, 0xff, 0xff]);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_f32(&mut self, n: f32) -> WriteF32;
+
+            /// Writes an 64-bit floating point type in big-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_f64(&mut self, n: f64) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write 64-bit floating point type to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_f64(f64::MIN).await?;
+            ///
+            ///     assert_eq!(writer, vec![
+            ///         0xff, 0xef, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff
+            ///     ]);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_f64(&mut self, n: f64) -> WriteF64;
+
+            /// Writes an unsigned 16-bit integer in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u16_le(&mut self, n: u16) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 16-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u16_le(517).await?;
+            ///     writer.write_u16_le(768).await?;
+            ///
+            ///     assert_eq!(writer, b"\x05\x02\x00\x03");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u16_le(&mut self, n: u16) -> WriteU16Le;
+
+            /// Writes a signed 16-bit integer in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i16_le(&mut self, n: i16) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write signed 16-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_i16_le(193).await?;
+            ///     writer.write_i16_le(-132).await?;
+            ///
+            ///     assert_eq!(writer, b"\xc1\x00\x7c\xff");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i16_le(&mut self, n: i16) -> WriteI16Le;
+
+            /// Writes an unsigned 32-bit integer in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u32_le(&mut self, n: u32) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 32-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u32_le(267).await?;
+            ///     writer.write_u32_le(1205419366).await?;
+            ///
+            ///     assert_eq!(writer, b"\x0b\x01\x00\x00\x66\x3d\xd9\x47");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u32_le(&mut self, n: u32) -> WriteU32Le;
+
+            /// Writes a signed 32-bit integer in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i32_le(&mut self, n: i32) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write signed 32-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_i32_le(267).await?;
+            ///     writer.write_i32_le(1205419366).await?;
+            ///
+            ///     assert_eq!(writer, b"\x0b\x01\x00\x00\x66\x3d\xd9\x47");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i32_le(&mut self, n: i32) -> WriteI32Le;
+
+            /// Writes an unsigned 64-bit integer in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u64_le(&mut self, n: u64) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 64-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u64_le(918733457491587).await?;
+            ///     writer.write_u64_le(143).await?;
+            ///
+            ///     assert_eq!(writer, b"\x83\x86\x60\x4d\x95\x43\x03\x00\x8f\x00\x00\x00\x00\x00\x00\x00");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u64_le(&mut self, n: u64) -> WriteU64Le;
+
+            /// Writes an signed 64-bit integer in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i64_le(&mut self, n: i64) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write signed 64-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_i64_le(i64::MIN).await?;
+            ///     writer.write_i64_le(i64::MAX).await?;
+            ///
+            ///     assert_eq!(writer, b"\x00\x00\x00\x00\x00\x00\x00\x80\xff\xff\xff\xff\xff\xff\xff\x7f");
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i64_le(&mut self, n: i64) -> WriteI64Le;
+
+            /// Writes an unsigned 128-bit integer in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_u128_le(&mut self, n: u128) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write unsigned 128-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_u128_le(16947640962301618749969007319746179).await?;
+            ///
+            ///     assert_eq!(writer, vec![
+            ///         0x83, 0x86, 0x60, 0x4d, 0x95, 0x43, 0x03, 0x00,
+            ///         0x83, 0x86, 0x60, 0x4d, 0x95, 0x43, 0x03, 0x00,
+            ///     ]);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_u128_le(&mut self, n: u128) -> WriteU128Le;
+
+            /// Writes an signed 128-bit integer in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_i128_le(&mut self, n: i128) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write signed 128-bit integers to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_i128_le(i128::MIN).await?;
+            ///
+            ///     assert_eq!(writer, vec![
+            ///          0, 0, 0, 0, 0, 0, 0,
+            ///         0, 0, 0, 0, 0, 0, 0, 0, 0x80
+            ///     ]);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_i128_le(&mut self, n: i128) -> WriteI128Le;
+
+            /// Writes an 32-bit floating point type in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_f32_le(&mut self, n: f32) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write 32-bit floating point type to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_f32_le(f32::MIN).await?;
+            ///
+            ///     assert_eq!(writer, vec![0xff, 0xff, 0x7f, 0xff]);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_f32_le(&mut self, n: f32) -> WriteF32Le;
+
+            /// Writes an 64-bit floating point type in little-endian order to the
+            /// underlying writer.
+            ///
+            /// Equivalent to:
+            ///
+            /// ```ignore
+            /// async fn write_f64_le(&mut self, n: f64) -> io::Result<()>;
+            /// ```
+            ///
+            /// It is recommended to use a buffered writer to avoid excessive
+            /// syscalls.
+            ///
+            /// # Errors
+            ///
+            /// This method returns the same errors as [`AsyncWriteExt::write_all`].
+            ///
+            /// [`AsyncWriteExt::write_all`]: AsyncWriteExt::write_all
+            ///
+            /// # Examples
+            ///
+            /// Write 64-bit floating point type to a `AsyncWrite`:
+            ///
+            /// ```rust
+            /// use tokio::io::{self, AsyncWriteExt};
+            ///
+            /// #[tokio::main]
+            /// async fn main() -> io::Result<()> {
+            ///     let mut writer = Vec::new();
+            ///
+            ///     writer.write_f64_le(f64::MIN).await?;
+            ///
+            ///     assert_eq!(writer, vec![
+            ///         0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xef, 0xff
+            ///     ]);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn write_f64_le(&mut self, n: f64) -> WriteF64Le;
+        }
+
+        /// Flushes this output stream, ensuring that all intermediately buffered
+        /// contents reach their destination.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn flush(&mut self) -> io::Result<()>;
+        /// ```
+        ///
+        /// # Errors
+        ///
+        /// It is considered an error if not all bytes could be written due to
+        /// I/O errors or EOF being reached.
+        ///
+        /// # Examples
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, BufWriter, AsyncWriteExt};
+        /// use tokio::fs::File;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let f = File::create("foo.txt").await?;
+        ///     let mut buffer = BufWriter::new(f);
+        ///
+        ///     buffer.write_all(b"some bytes").await?;
+        ///     buffer.flush().await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        fn flush(&mut self) -> Flush<'_, Self>
+        where
+            Self: Unpin,
+        {
+            flush(self)
+        }
+
+        /// Shuts down the output stream, ensuring that the value can be dropped
+        /// cleanly.
+        ///
+        /// Equivalent to:
+        ///
+        /// ```ignore
+        /// async fn shutdown(&mut self) -> io::Result<()>;
+        /// ```
+        ///
+        /// Similar to [`flush`], all intermediately buffered is written to the
+        /// underlying stream. Once the operation completes, the caller should
+        /// no longer attempt to write to the stream. For example, the
+        /// `TcpStream` implementation will issue a `shutdown(Write)` sys call.
+        ///
+        /// [`flush`]: fn@crate::io::AsyncWriteExt::flush
+        ///
+        /// # Examples
+        ///
+        /// ```no_run
+        /// use tokio::io::{self, BufWriter, AsyncWriteExt};
+        /// use tokio::fs::File;
+        ///
+        /// #[tokio::main]
+        /// async fn main() -> io::Result<()> {
+        ///     let f = File::create("foo.txt").await?;
+        ///     let mut buffer = BufWriter::new(f);
+        ///
+        ///     buffer.write_all(b"some bytes").await?;
+        ///     buffer.shutdown().await?;
+        ///     Ok(())
+        /// }
+        /// ```
+        fn shutdown(&mut self) -> Shutdown<'_, Self>
+        where
+            Self: Unpin,
+        {
+            shutdown(self)
+        }
+    }
+}
+
+impl<W: AsyncWrite + ?Sized> AsyncWriteExt for W {}
diff --git a/third_party/rust/tokio/src/io/util/buf_reader.rs b/third_party/rust/tokio/src/io/util/buf_reader.rs
new file mode 100644
index 0000000000..60879c0fdc
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/buf_reader.rs
@@ -0,0 +1,311 @@
+use crate::io::util::DEFAULT_BUF_SIZE;
+use crate::io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
+
+use pin_project_lite::pin_project;
+use std::io::{self, IoSlice, SeekFrom};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::{cmp, fmt, mem};
+
+pin_project! {
+    /// The `BufReader` struct adds buffering to any reader.
+    ///
+    /// It can be excessively inefficient to work directly with a [`AsyncRead`]
+    /// instance. A `BufReader` performs large, infrequent reads on the underlying
+    /// [`AsyncRead`] and maintains an in-memory buffer of the results.
+    ///
+    /// `BufReader` can improve the speed of programs that make *small* and
+    /// *repeated* read calls to the same file or network socket. It does not
+    /// help when reading very large amounts at once, or reading just one or a few
+    /// times. It also provides no advantage when reading from a source that is
+    /// already in memory, like a `Vec<u8>`.
+    ///
+    /// When the `BufReader` is dropped, the contents of its buffer will be
+    /// discarded. Creating multiple instances of a `BufReader` on the same
+    /// stream can cause data loss.
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+    pub struct BufReader<R> {
+        #[pin]
+        pub(super) inner: R,
+        pub(super) buf: Box<[u8]>,
+        pub(super) pos: usize,
+        pub(super) cap: usize,
+        pub(super) seek_state: SeekState,
+    }
+}
+
+impl<R: AsyncRead> BufReader<R> {
+    /// Creates a new `BufReader` with a default buffer capacity. The default is currently 8 KB,
+    /// but may change in the future.
+    pub fn new(inner: R) -> Self {
+        Self::with_capacity(DEFAULT_BUF_SIZE, inner)
+    }
+
+    /// Creates a new `BufReader` with the specified buffer capacity.
+    pub fn with_capacity(capacity: usize, inner: R) -> Self {
+        let buffer = vec![0; capacity];
+        Self {
+            inner,
+            buf: buffer.into_boxed_slice(),
+            pos: 0,
+            cap: 0,
+            seek_state: SeekState::Init,
+        }
+    }
+
+    /// Gets a reference to the underlying reader.
+    ///
+    /// It is inadvisable to directly read from the underlying reader.
+    pub fn get_ref(&self) -> &R {
+        &self.inner
+    }
+
+    /// Gets a mutable reference to the underlying reader.
+    ///
+    /// It is inadvisable to directly read from the underlying reader.
+    pub fn get_mut(&mut self) -> &mut R {
+        &mut self.inner
+    }
+
+    /// Gets a pinned mutable reference to the underlying reader.
+    ///
+    /// It is inadvisable to directly read from the underlying reader.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R> {
+        self.project().inner
+    }
+
+    /// Consumes this `BufReader`, returning the underlying reader.
+    ///
+    /// Note that any leftover data in the internal buffer is lost.
+    pub fn into_inner(self) -> R {
+        self.inner
+    }
+
+    /// Returns a reference to the internally buffered data.
+    ///
+    /// Unlike `fill_buf`, this will not attempt to fill the buffer if it is empty.
+    pub fn buffer(&self) -> &[u8] {
+        &self.buf[self.pos..self.cap]
+    }
+
+    /// Invalidates all data in the internal buffer.
+    #[inline]
+    fn discard_buffer(self: Pin<&mut Self>) {
+        let me = self.project();
+        *me.pos = 0;
+        *me.cap = 0;
+    }
+}
+
+impl<R: AsyncRead> AsyncRead for BufReader<R> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        // If we don't have any buffered data and we're doing a massive read
+        // (larger than our internal buffer), bypass our internal buffer
+        // entirely.
+        if self.pos == self.cap && buf.remaining() >= self.buf.len() {
+            let res = ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
+            self.discard_buffer();
+            return Poll::Ready(res);
+        }
+        let rem = ready!(self.as_mut().poll_fill_buf(cx))?;
+        let amt = std::cmp::min(rem.len(), buf.remaining());
+        buf.put_slice(&rem[..amt]);
+        self.consume(amt);
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<R: AsyncRead> AsyncBufRead for BufReader<R> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        let me = self.project();
+
+        // If we've reached the end of our internal buffer then we need to fetch
+        // some more data from the underlying reader.
+        // Branch using `>=` instead of the more correct `==`
+        // to tell the compiler that the pos..cap slice is always valid.
+        if *me.pos >= *me.cap {
+            debug_assert!(*me.pos == *me.cap);
+            let mut buf = ReadBuf::new(me.buf);
+            ready!(me.inner.poll_read(cx, &mut buf))?;
+            *me.cap = buf.filled().len();
+            *me.pos = 0;
+        }
+        Poll::Ready(Ok(&me.buf[*me.pos..*me.cap]))
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        let me = self.project();
+        *me.pos = cmp::min(*me.pos + amt, *me.cap);
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub(super) enum SeekState {
+    /// start_seek has not been called.
+    Init,
+    /// start_seek has been called, but poll_complete has not yet been called.
+    Start(SeekFrom),
+    /// Waiting for completion of the first poll_complete in the `n.checked_sub(remainder).is_none()` branch.
+    PendingOverflowed(i64),
+    /// Waiting for completion of poll_complete.
+    Pending,
+}
+
+/// Seeks to an offset, in bytes, in the underlying reader.
+///
+/// The position used for seeking with `SeekFrom::Current(_)` is the
+/// position the underlying reader would be at if the `BufReader` had no
+/// internal buffer.
+///
+/// Seeking always discards the internal buffer, even if the seek position
+/// would otherwise fall within it. This guarantees that calling
+/// `.into_inner()` immediately after a seek yields the underlying reader
+/// at the same position.
+///
+/// See [`AsyncSeek`] for more details.
+///
+/// Note: In the edge case where you're seeking with `SeekFrom::Current(n)`
+/// where `n` minus the internal buffer length overflows an `i64`, two
+/// seeks will be performed instead of one. If the second seek returns
+/// `Err`, the underlying reader will be left at the same position it would
+/// have if you called `seek` with `SeekFrom::Current(0)`.
+impl<R: AsyncRead + AsyncSeek> AsyncSeek for BufReader<R> {
+    fn start_seek(self: Pin<&mut Self>, pos: SeekFrom) -> io::Result<()> {
+        // We needs to call seek operation multiple times.
+        // And we should always call both start_seek and poll_complete,
+        // as start_seek alone cannot guarantee that the operation will be completed.
+        // poll_complete receives a Context and returns a Poll, so it cannot be called
+        // inside start_seek.
+        *self.project().seek_state = SeekState::Start(pos);
+        Ok(())
+    }
+
+    fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
+        let res = match mem::replace(self.as_mut().project().seek_state, SeekState::Init) {
+            SeekState::Init => {
+                // 1.x AsyncSeek recommends calling poll_complete before start_seek.
+                // We don't have to guarantee that the value returned by
+                // poll_complete called without start_seek is correct,
+                // so we'll return 0.
+                return Poll::Ready(Ok(0));
+            }
+            SeekState::Start(SeekFrom::Current(n)) => {
+                let remainder = (self.cap - self.pos) as i64;
+                // it should be safe to assume that remainder fits within an i64 as the alternative
+                // means we managed to allocate 8 exbibytes and that's absurd.
+                // But it's not out of the realm of possibility for some weird underlying reader to
+                // support seeking by i64::MIN so we need to handle underflow when subtracting
+                // remainder.
+                if let Some(offset) = n.checked_sub(remainder) {
+                    self.as_mut()
+                        .get_pin_mut()
+                        .start_seek(SeekFrom::Current(offset))?;
+                } else {
+                    // seek backwards by our remainder, and then by the offset
+                    self.as_mut()
+                        .get_pin_mut()
+                        .start_seek(SeekFrom::Current(-remainder))?;
+                    if self.as_mut().get_pin_mut().poll_complete(cx)?.is_pending() {
+                        *self.as_mut().project().seek_state = SeekState::PendingOverflowed(n);
+                        return Poll::Pending;
+                    }
+
+                    // https://github.com/rust-lang/rust/pull/61157#issuecomment-495932676
+                    self.as_mut().discard_buffer();
+
+                    self.as_mut()
+                        .get_pin_mut()
+                        .start_seek(SeekFrom::Current(n))?;
+                }
+                self.as_mut().get_pin_mut().poll_complete(cx)?
+            }
+            SeekState::PendingOverflowed(n) => {
+                if self.as_mut().get_pin_mut().poll_complete(cx)?.is_pending() {
+                    *self.as_mut().project().seek_state = SeekState::PendingOverflowed(n);
+                    return Poll::Pending;
+                }
+
+                // https://github.com/rust-lang/rust/pull/61157#issuecomment-495932676
+                self.as_mut().discard_buffer();
+
+                self.as_mut()
+                    .get_pin_mut()
+                    .start_seek(SeekFrom::Current(n))?;
+                self.as_mut().get_pin_mut().poll_complete(cx)?
+            }
+            SeekState::Start(pos) => {
+                // Seeking with Start/End doesn't care about our buffer length.
+                self.as_mut().get_pin_mut().start_seek(pos)?;
+                self.as_mut().get_pin_mut().poll_complete(cx)?
+            }
+            SeekState::Pending => self.as_mut().get_pin_mut().poll_complete(cx)?,
+        };
+
+        match res {
+            Poll::Ready(res) => {
+                self.discard_buffer();
+                Poll::Ready(Ok(res))
+            }
+            Poll::Pending => {
+                *self.as_mut().project().seek_state = SeekState::Pending;
+                Poll::Pending
+            }
+        }
+    }
+}
+
+impl<R: AsyncRead + AsyncWrite> AsyncWrite for BufReader<R> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        self.get_pin_mut().poll_write(cx, buf)
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        self.get_pin_mut().poll_write_vectored(cx, bufs)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        self.get_ref().is_write_vectored()
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.get_pin_mut().poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.get_pin_mut().poll_shutdown(cx)
+    }
+}
+
+impl<R: fmt::Debug> fmt::Debug for BufReader<R> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("BufReader")
+            .field("reader", &self.inner)
+            .field(
+                "buffer",
+                &format_args!("{}/{}", self.cap - self.pos, self.buf.len()),
+            )
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<BufReader<()>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/buf_stream.rs b/third_party/rust/tokio/src/io/util/buf_stream.rs
new file mode 100644
index 0000000000..595c142aca
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/buf_stream.rs
@@ -0,0 +1,207 @@
+use crate::io::util::{BufReader, BufWriter};
+use crate::io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
+
+use pin_project_lite::pin_project;
+use std::io::{self, IoSlice, SeekFrom};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Wraps a type that is [`AsyncWrite`] and [`AsyncRead`], and buffers its input and output.
+    ///
+    /// It can be excessively inefficient to work directly with something that implements [`AsyncWrite`]
+    /// and [`AsyncRead`]. For example, every `write`, however small, has to traverse the syscall
+    /// interface, and similarly, every read has to do the same. The [`BufWriter`] and [`BufReader`]
+    /// types aid with these problems respectively, but do so in only one direction. `BufStream` wraps
+    /// one in the other so that both directions are buffered. See their documentation for details.
+    #[derive(Debug)]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+    pub struct BufStream<RW> {
+        #[pin]
+        inner: BufReader<BufWriter<RW>>,
+    }
+}
+
+impl<RW: AsyncRead + AsyncWrite> BufStream<RW> {
+    /// Wraps a type in both [`BufWriter`] and [`BufReader`].
+    ///
+    /// See the documentation for those types and [`BufStream`] for details.
+    pub fn new(stream: RW) -> BufStream<RW> {
+        BufStream {
+            inner: BufReader::new(BufWriter::new(stream)),
+        }
+    }
+
+    /// Creates a `BufStream` with the specified [`BufReader`] capacity and [`BufWriter`]
+    /// capacity.
+    ///
+    /// See the documentation for those types and [`BufStream`] for details.
+    pub fn with_capacity(
+        reader_capacity: usize,
+        writer_capacity: usize,
+        stream: RW,
+    ) -> BufStream<RW> {
+        BufStream {
+            inner: BufReader::with_capacity(
+                reader_capacity,
+                BufWriter::with_capacity(writer_capacity, stream),
+            ),
+        }
+    }
+
+    /// Gets a reference to the underlying I/O object.
+    ///
+    /// It is inadvisable to directly read from the underlying I/O object.
+    pub fn get_ref(&self) -> &RW {
+        self.inner.get_ref().get_ref()
+    }
+
+    /// Gets a mutable reference to the underlying I/O object.
+    ///
+    /// It is inadvisable to directly read from the underlying I/O object.
+    pub fn get_mut(&mut self) -> &mut RW {
+        self.inner.get_mut().get_mut()
+    }
+
+    /// Gets a pinned mutable reference to the underlying I/O object.
+    ///
+    /// It is inadvisable to directly read from the underlying I/O object.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut RW> {
+        self.project().inner.get_pin_mut().get_pin_mut()
+    }
+
+    /// Consumes this `BufStream`, returning the underlying I/O object.
+    ///
+    /// Note that any leftover data in the internal buffer is lost.
+    pub fn into_inner(self) -> RW {
+        self.inner.into_inner().into_inner()
+    }
+}
+
+impl<RW> From<BufReader<BufWriter<RW>>> for BufStream<RW> {
+    fn from(b: BufReader<BufWriter<RW>>) -> Self {
+        BufStream { inner: b }
+    }
+}
+
+impl<RW> From<BufWriter<BufReader<RW>>> for BufStream<RW> {
+    fn from(b: BufWriter<BufReader<RW>>) -> Self {
+        // we need to "invert" the reader and writer
+        let BufWriter {
+            inner:
+                BufReader {
+                    inner,
+                    buf: rbuf,
+                    pos,
+                    cap,
+                    seek_state: rseek_state,
+                },
+            buf: wbuf,
+            written,
+            seek_state: wseek_state,
+        } = b;
+
+        BufStream {
+            inner: BufReader {
+                inner: BufWriter {
+                    inner,
+                    buf: wbuf,
+                    written,
+                    seek_state: wseek_state,
+                },
+                buf: rbuf,
+                pos,
+                cap,
+                seek_state: rseek_state,
+            },
+        }
+    }
+}
+
+impl<RW: AsyncRead + AsyncWrite> AsyncWrite for BufStream<RW> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        self.project().inner.poll_write(cx, buf)
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        self.project().inner.poll_write_vectored(cx, bufs)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        self.inner.is_write_vectored()
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.project().inner.poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.project().inner.poll_shutdown(cx)
+    }
+}
+
+impl<RW: AsyncRead + AsyncWrite> AsyncRead for BufStream<RW> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        self.project().inner.poll_read(cx, buf)
+    }
+}
+
+/// Seek to an offset, in bytes, in the underlying stream.
+///
+/// The position used for seeking with `SeekFrom::Current(_)` is the
+/// position the underlying stream would be at if the `BufStream` had no
+/// internal buffer.
+///
+/// Seeking always discards the internal buffer, even if the seek position
+/// would otherwise fall within it. This guarantees that calling
+/// `.into_inner()` immediately after a seek yields the underlying reader
+/// at the same position.
+///
+/// See [`AsyncSeek`] for more details.
+///
+/// Note: In the edge case where you're seeking with `SeekFrom::Current(n)`
+/// where `n` minus the internal buffer length overflows an `i64`, two
+/// seeks will be performed instead of one. If the second seek returns
+/// `Err`, the underlying reader will be left at the same position it would
+/// have if you called `seek` with `SeekFrom::Current(0)`.
+impl<RW: AsyncRead + AsyncWrite + AsyncSeek> AsyncSeek for BufStream<RW> {
+    fn start_seek(self: Pin<&mut Self>, position: SeekFrom) -> io::Result<()> {
+        self.project().inner.start_seek(position)
+    }
+
+    fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
+        self.project().inner.poll_complete(cx)
+    }
+}
+
+impl<RW: AsyncRead + AsyncWrite> AsyncBufRead for BufStream<RW> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        self.project().inner.poll_fill_buf(cx)
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        self.project().inner.consume(amt)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<BufStream<()>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/buf_writer.rs b/third_party/rust/tokio/src/io/util/buf_writer.rs
new file mode 100644
index 0000000000..8dd1bba60a
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/buf_writer.rs
@@ -0,0 +1,310 @@
+use crate::io::util::DEFAULT_BUF_SIZE;
+use crate::io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
+
+use pin_project_lite::pin_project;
+use std::fmt;
+use std::io::{self, IoSlice, SeekFrom, Write};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Wraps a writer and buffers its output.
+    ///
+    /// It can be excessively inefficient to work directly with something that
+    /// implements [`AsyncWrite`]. A `BufWriter` keeps an in-memory buffer of data and
+    /// writes it to an underlying writer in large, infrequent batches.
+    ///
+    /// `BufWriter` can improve the speed of programs that make *small* and
+    /// *repeated* write calls to the same file or network socket. It does not
+    /// help when writing very large amounts at once, or writing just one or a few
+    /// times. It also provides no advantage when writing to a destination that is
+    /// in memory, like a `Vec<u8>`.
+    ///
+    /// When the `BufWriter` is dropped, the contents of its buffer will be
+    /// discarded. Creating multiple instances of a `BufWriter` on the same
+    /// stream can cause data loss. If you need to write out the contents of its
+    /// buffer, you must manually call flush before the writer is dropped.
+    ///
+    /// [`AsyncWrite`]: AsyncWrite
+    /// [`flush`]: super::AsyncWriteExt::flush
+    ///
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+    pub struct BufWriter<W> {
+        #[pin]
+        pub(super) inner: W,
+        pub(super) buf: Vec<u8>,
+        pub(super) written: usize,
+        pub(super) seek_state: SeekState,
+    }
+}
+
+impl<W: AsyncWrite> BufWriter<W> {
+    /// Creates a new `BufWriter` with a default buffer capacity. The default is currently 8 KB,
+    /// but may change in the future.
+    pub fn new(inner: W) -> Self {
+        Self::with_capacity(DEFAULT_BUF_SIZE, inner)
+    }
+
+    /// Creates a new `BufWriter` with the specified buffer capacity.
+    pub fn with_capacity(cap: usize, inner: W) -> Self {
+        Self {
+            inner,
+            buf: Vec::with_capacity(cap),
+            written: 0,
+            seek_state: SeekState::Init,
+        }
+    }
+
+    fn flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let mut me = self.project();
+
+        let len = me.buf.len();
+        let mut ret = Ok(());
+        while *me.written < len {
+            match ready!(me.inner.as_mut().poll_write(cx, &me.buf[*me.written..])) {
+                Ok(0) => {
+                    ret = Err(io::Error::new(
+                        io::ErrorKind::WriteZero,
+                        "failed to write the buffered data",
+                    ));
+                    break;
+                }
+                Ok(n) => *me.written += n,
+                Err(e) => {
+                    ret = Err(e);
+                    break;
+                }
+            }
+        }
+        if *me.written > 0 {
+            me.buf.drain(..*me.written);
+        }
+        *me.written = 0;
+        Poll::Ready(ret)
+    }
+
+    /// Gets a reference to the underlying writer.
+    pub fn get_ref(&self) -> &W {
+        &self.inner
+    }
+
+    /// Gets a mutable reference to the underlying writer.
+    ///
+    /// It is inadvisable to directly write to the underlying writer.
+    pub fn get_mut(&mut self) -> &mut W {
+        &mut self.inner
+    }
+
+    /// Gets a pinned mutable reference to the underlying writer.
+    ///
+    /// It is inadvisable to directly write to the underlying writer.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut W> {
+        self.project().inner
+    }
+
+    /// Consumes this `BufWriter`, returning the underlying writer.
+    ///
+    /// Note that any leftover data in the internal buffer is lost.
+    pub fn into_inner(self) -> W {
+        self.inner
+    }
+
+    /// Returns a reference to the internally buffered data.
+    pub fn buffer(&self) -> &[u8] {
+        &self.buf
+    }
+}
+
+impl<W: AsyncWrite> AsyncWrite for BufWriter<W> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        if self.buf.len() + buf.len() > self.buf.capacity() {
+            ready!(self.as_mut().flush_buf(cx))?;
+        }
+
+        let me = self.project();
+        if buf.len() >= me.buf.capacity() {
+            me.inner.poll_write(cx, buf)
+        } else {
+            Poll::Ready(me.buf.write(buf))
+        }
+    }
+
+    fn poll_write_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        mut bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        if self.inner.is_write_vectored() {
+            let total_len = bufs
+                .iter()
+                .fold(0usize, |acc, b| acc.saturating_add(b.len()));
+            if total_len > self.buf.capacity() - self.buf.len() {
+                ready!(self.as_mut().flush_buf(cx))?;
+            }
+            let me = self.as_mut().project();
+            if total_len >= me.buf.capacity() {
+                // It's more efficient to pass the slices directly to the
+                // underlying writer than to buffer them.
+                // The case when the total_len calculation saturates at
+                // usize::MAX is also handled here.
+                me.inner.poll_write_vectored(cx, bufs)
+            } else {
+                bufs.iter().for_each(|b| me.buf.extend_from_slice(b));
+                Poll::Ready(Ok(total_len))
+            }
+        } else {
+            // Remove empty buffers at the beginning of bufs.
+            while bufs.first().map(|buf| buf.len()) == Some(0) {
+                bufs = &bufs[1..];
+            }
+            if bufs.is_empty() {
+                return Poll::Ready(Ok(0));
+            }
+            // Flush if the first buffer doesn't fit.
+            let first_len = bufs[0].len();
+            if first_len > self.buf.capacity() - self.buf.len() {
+                ready!(self.as_mut().flush_buf(cx))?;
+                debug_assert!(self.buf.is_empty());
+            }
+            let me = self.as_mut().project();
+            if first_len >= me.buf.capacity() {
+                // The slice is at least as large as the buffering capacity,
+                // so it's better to write it directly, bypassing the buffer.
+                debug_assert!(me.buf.is_empty());
+                return me.inner.poll_write(cx, &bufs[0]);
+            } else {
+                me.buf.extend_from_slice(&bufs[0]);
+                bufs = &bufs[1..];
+            }
+            let mut total_written = first_len;
+            debug_assert!(total_written != 0);
+            // Append the buffers that fit in the internal buffer.
+            for buf in bufs {
+                if buf.len() > me.buf.capacity() - me.buf.len() {
+                    break;
+                } else {
+                    me.buf.extend_from_slice(buf);
+                    total_written += buf.len();
+                }
+            }
+            Poll::Ready(Ok(total_written))
+        }
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        true
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        ready!(self.as_mut().flush_buf(cx))?;
+        self.get_pin_mut().poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        ready!(self.as_mut().flush_buf(cx))?;
+        self.get_pin_mut().poll_shutdown(cx)
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub(super) enum SeekState {
+    /// start_seek has not been called.
+    Init,
+    /// start_seek has been called, but poll_complete has not yet been called.
+    Start(SeekFrom),
+    /// Waiting for completion of poll_complete.
+    Pending,
+}
+
+/// Seek to the offset, in bytes, in the underlying writer.
+///
+/// Seeking always writes out the internal buffer before seeking.
+impl<W: AsyncWrite + AsyncSeek> AsyncSeek for BufWriter<W> {
+    fn start_seek(self: Pin<&mut Self>, pos: SeekFrom) -> io::Result<()> {
+        // We need to flush the internal buffer before seeking.
+        // It receives a `Context` and returns a `Poll`, so it cannot be called
+        // inside `start_seek`.
+        *self.project().seek_state = SeekState::Start(pos);
+        Ok(())
+    }
+
+    fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
+        let pos = match self.seek_state {
+            SeekState::Init => {
+                return self.project().inner.poll_complete(cx);
+            }
+            SeekState::Start(pos) => Some(pos),
+            SeekState::Pending => None,
+        };
+
+        // Flush the internal buffer before seeking.
+        ready!(self.as_mut().flush_buf(cx))?;
+
+        let mut me = self.project();
+        if let Some(pos) = pos {
+            // Ensure previous seeks have finished before starting a new one
+            ready!(me.inner.as_mut().poll_complete(cx))?;
+            if let Err(e) = me.inner.as_mut().start_seek(pos) {
+                *me.seek_state = SeekState::Init;
+                return Poll::Ready(Err(e));
+            }
+        }
+        match me.inner.poll_complete(cx) {
+            Poll::Ready(res) => {
+                *me.seek_state = SeekState::Init;
+                Poll::Ready(res)
+            }
+            Poll::Pending => {
+                *me.seek_state = SeekState::Pending;
+                Poll::Pending
+            }
+        }
+    }
+}
+
+impl<W: AsyncWrite + AsyncRead> AsyncRead for BufWriter<W> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        self.get_pin_mut().poll_read(cx, buf)
+    }
+}
+
+impl<W: AsyncWrite + AsyncBufRead> AsyncBufRead for BufWriter<W> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        self.get_pin_mut().poll_fill_buf(cx)
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        self.get_pin_mut().consume(amt)
+    }
+}
+
+impl<W: fmt::Debug> fmt::Debug for BufWriter<W> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("BufWriter")
+            .field("writer", &self.inner)
+            .field(
+                "buffer",
+                &format_args!("{}/{}", self.buf.len(), self.buf.capacity()),
+            )
+            .field("written", &self.written)
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<BufWriter<()>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/chain.rs b/third_party/rust/tokio/src/io/util/chain.rs
new file mode 100644
index 0000000000..84f37fc7d4
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/chain.rs
@@ -0,0 +1,144 @@
+use crate::io::{AsyncBufRead, AsyncRead, ReadBuf};
+
+use pin_project_lite::pin_project;
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Stream for the [`chain`](super::AsyncReadExt::chain) method.
+    #[must_use = "streams do nothing unless polled"]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+    pub struct Chain<T, U> {
+        #[pin]
+        first: T,
+        #[pin]
+        second: U,
+        done_first: bool,
+    }
+}
+
+pub(super) fn chain<T, U>(first: T, second: U) -> Chain<T, U>
+where
+    T: AsyncRead,
+    U: AsyncRead,
+{
+    Chain {
+        first,
+        second,
+        done_first: false,
+    }
+}
+
+impl<T, U> Chain<T, U>
+where
+    T: AsyncRead,
+    U: AsyncRead,
+{
+    /// Gets references to the underlying readers in this `Chain`.
+    pub fn get_ref(&self) -> (&T, &U) {
+        (&self.first, &self.second)
+    }
+
+    /// Gets mutable references to the underlying readers in this `Chain`.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying readers as doing so may corrupt the internal state of this
+    /// `Chain`.
+    pub fn get_mut(&mut self) -> (&mut T, &mut U) {
+        (&mut self.first, &mut self.second)
+    }
+
+    /// Gets pinned mutable references to the underlying readers in this `Chain`.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying readers as doing so may corrupt the internal state of this
+    /// `Chain`.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> (Pin<&mut T>, Pin<&mut U>) {
+        let me = self.project();
+        (me.first, me.second)
+    }
+
+    /// Consumes the `Chain`, returning the wrapped readers.
+    pub fn into_inner(self) -> (T, U) {
+        (self.first, self.second)
+    }
+}
+
+impl<T, U> fmt::Debug for Chain<T, U>
+where
+    T: fmt::Debug,
+    U: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Chain")
+            .field("t", &self.first)
+            .field("u", &self.second)
+            .finish()
+    }
+}
+
+impl<T, U> AsyncRead for Chain<T, U>
+where
+    T: AsyncRead,
+    U: AsyncRead,
+{
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        let me = self.project();
+
+        if !*me.done_first {
+            let rem = buf.remaining();
+            ready!(me.first.poll_read(cx, buf))?;
+            if buf.remaining() == rem {
+                *me.done_first = true;
+            } else {
+                return Poll::Ready(Ok(()));
+            }
+        }
+        me.second.poll_read(cx, buf)
+    }
+}
+
+impl<T, U> AsyncBufRead for Chain<T, U>
+where
+    T: AsyncBufRead,
+    U: AsyncBufRead,
+{
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        let me = self.project();
+
+        if !*me.done_first {
+            match ready!(me.first.poll_fill_buf(cx)?) {
+                buf if buf.is_empty() => {
+                    *me.done_first = true;
+                }
+                buf => return Poll::Ready(Ok(buf)),
+            }
+        }
+        me.second.poll_fill_buf(cx)
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        let me = self.project();
+        if !*me.done_first {
+            me.first.consume(amt)
+        } else {
+            me.second.consume(amt)
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Chain<(), ()>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/copy.rs b/third_party/rust/tokio/src/io/util/copy.rs
new file mode 100644
index 0000000000..d0ab7cb140
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/copy.rs
@@ -0,0 +1,175 @@
+use crate::io::{AsyncRead, AsyncWrite, ReadBuf};
+
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[derive(Debug)]
+pub(super) struct CopyBuffer {
+    read_done: bool,
+    need_flush: bool,
+    pos: usize,
+    cap: usize,
+    amt: u64,
+    buf: Box<[u8]>,
+}
+
+impl CopyBuffer {
+    pub(super) fn new() -> Self {
+        Self {
+            read_done: false,
+            need_flush: false,
+            pos: 0,
+            cap: 0,
+            amt: 0,
+            buf: vec![0; super::DEFAULT_BUF_SIZE].into_boxed_slice(),
+        }
+    }
+
+    pub(super) fn poll_copy<R, W>(
+        &mut self,
+        cx: &mut Context<'_>,
+        mut reader: Pin<&mut R>,
+        mut writer: Pin<&mut W>,
+    ) -> Poll<io::Result<u64>>
+    where
+        R: AsyncRead + ?Sized,
+        W: AsyncWrite + ?Sized,
+    {
+        loop {
+            // If our buffer is empty, then we need to read some data to
+            // continue.
+            if self.pos == self.cap && !self.read_done {
+                let me = &mut *self;
+                let mut buf = ReadBuf::new(&mut me.buf);
+
+                match reader.as_mut().poll_read(cx, &mut buf) {
+                    Poll::Ready(Ok(_)) => (),
+                    Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+                    Poll::Pending => {
+                        // Try flushing when the reader has no progress to avoid deadlock
+                        // when the reader depends on buffered writer.
+                        if self.need_flush {
+                            ready!(writer.as_mut().poll_flush(cx))?;
+                            self.need_flush = false;
+                        }
+
+                        return Poll::Pending;
+                    }
+                }
+
+                let n = buf.filled().len();
+                if n == 0 {
+                    self.read_done = true;
+                } else {
+                    self.pos = 0;
+                    self.cap = n;
+                }
+            }
+
+            // If our buffer has some data, let's write it out!
+            while self.pos < self.cap {
+                let me = &mut *self;
+                let i = ready!(writer.as_mut().poll_write(cx, &me.buf[me.pos..me.cap]))?;
+                if i == 0 {
+                    return Poll::Ready(Err(io::Error::new(
+                        io::ErrorKind::WriteZero,
+                        "write zero byte into writer",
+                    )));
+                } else {
+                    self.pos += i;
+                    self.amt += i as u64;
+                    self.need_flush = true;
+                }
+            }
+
+            // If pos larger than cap, this loop will never stop.
+            // In particular, user's wrong poll_write implementation returning
+            // incorrect written length may lead to thread blocking.
+            debug_assert!(
+                self.pos <= self.cap,
+                "writer returned length larger than input slice"
+            );
+
+            // If we've written all the data and we've seen EOF, flush out the
+            // data and finish the transfer.
+            if self.pos == self.cap && self.read_done {
+                ready!(writer.as_mut().poll_flush(cx))?;
+                return Poll::Ready(Ok(self.amt));
+            }
+        }
+    }
+}
+
+/// A future that asynchronously copies the entire contents of a reader into a
+/// writer.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+struct Copy<'a, R: ?Sized, W: ?Sized> {
+    reader: &'a mut R,
+    writer: &'a mut W,
+    buf: CopyBuffer,
+}
+
+cfg_io_util! {
+    /// Asynchronously copies the entire contents of a reader into a writer.
+    ///
+    /// This function returns a future that will continuously read data from
+    /// `reader` and then write it into `writer` in a streaming fashion until
+    /// `reader` returns EOF.
+    ///
+    /// On success, the total number of bytes that were copied from `reader` to
+    /// `writer` is returned.
+    ///
+    /// This is an asynchronous version of [`std::io::copy`][std].
+    ///
+    /// [std]: std::io::copy
+    ///
+    /// # Errors
+    ///
+    /// The returned future will return an error immediately if any call to
+    /// `poll_read` or `poll_write` returns an error.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io;
+    ///
+    /// # async fn dox() -> std::io::Result<()> {
+    /// let mut reader: &[u8] = b"hello";
+    /// let mut writer: Vec<u8> = vec![];
+    ///
+    /// io::copy(&mut reader, &mut writer).await?;
+    ///
+    /// assert_eq!(&b"hello"[..], &writer[..]);
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn copy<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> io::Result<u64>
+    where
+        R: AsyncRead + Unpin + ?Sized,
+        W: AsyncWrite + Unpin + ?Sized,
+    {
+        Copy {
+            reader,
+            writer,
+            buf: CopyBuffer::new()
+        }.await
+    }
+}
+
+impl<R, W> Future for Copy<'_, R, W>
+where
+    R: AsyncRead + Unpin + ?Sized,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<u64>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
+        let me = &mut *self;
+
+        me.buf
+            .poll_copy(cx, Pin::new(&mut *me.reader), Pin::new(&mut *me.writer))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/copy_bidirectional.rs b/third_party/rust/tokio/src/io/util/copy_bidirectional.rs
new file mode 100644
index 0000000000..c93060b361
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/copy_bidirectional.rs
@@ -0,0 +1,120 @@
+use super::copy::CopyBuffer;
+
+use crate::io::{AsyncRead, AsyncWrite};
+
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+enum TransferState {
+    Running(CopyBuffer),
+    ShuttingDown(u64),
+    Done(u64),
+}
+
+struct CopyBidirectional<'a, A: ?Sized, B: ?Sized> {
+    a: &'a mut A,
+    b: &'a mut B,
+    a_to_b: TransferState,
+    b_to_a: TransferState,
+}
+
+fn transfer_one_direction<A, B>(
+    cx: &mut Context<'_>,
+    state: &mut TransferState,
+    r: &mut A,
+    w: &mut B,
+) -> Poll<io::Result<u64>>
+where
+    A: AsyncRead + AsyncWrite + Unpin + ?Sized,
+    B: AsyncRead + AsyncWrite + Unpin + ?Sized,
+{
+    let mut r = Pin::new(r);
+    let mut w = Pin::new(w);
+
+    loop {
+        match state {
+            TransferState::Running(buf) => {
+                let count = ready!(buf.poll_copy(cx, r.as_mut(), w.as_mut()))?;
+                *state = TransferState::ShuttingDown(count);
+            }
+            TransferState::ShuttingDown(count) => {
+                ready!(w.as_mut().poll_shutdown(cx))?;
+
+                *state = TransferState::Done(*count);
+            }
+            TransferState::Done(count) => return Poll::Ready(Ok(*count)),
+        }
+    }
+}
+
+impl<'a, A, B> Future for CopyBidirectional<'a, A, B>
+where
+    A: AsyncRead + AsyncWrite + Unpin + ?Sized,
+    B: AsyncRead + AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<(u64, u64)>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        // Unpack self into mut refs to each field to avoid borrow check issues.
+        let CopyBidirectional {
+            a,
+            b,
+            a_to_b,
+            b_to_a,
+        } = &mut *self;
+
+        let a_to_b = transfer_one_direction(cx, a_to_b, &mut *a, &mut *b)?;
+        let b_to_a = transfer_one_direction(cx, b_to_a, &mut *b, &mut *a)?;
+
+        // It is not a problem if ready! returns early because transfer_one_direction for the
+        // other direction will keep returning TransferState::Done(count) in future calls to poll
+        let a_to_b = ready!(a_to_b);
+        let b_to_a = ready!(b_to_a);
+
+        Poll::Ready(Ok((a_to_b, b_to_a)))
+    }
+}
+
+/// Copies data in both directions between `a` and `b`.
+///
+/// This function returns a future that will read from both streams,
+/// writing any data read to the opposing stream.
+/// This happens in both directions concurrently.
+///
+/// If an EOF is observed on one stream, [`shutdown()`] will be invoked on
+/// the other, and reading from that stream will stop. Copying of data in
+/// the other direction will continue.
+///
+/// The future will complete successfully once both directions of communication has been shut down.
+/// A direction is shut down when the reader reports EOF,
+/// at which point [`shutdown()`] is called on the corresponding writer. When finished,
+/// it will return a tuple of the number of bytes copied from a to b
+/// and the number of bytes copied from b to a, in that order.
+///
+/// [`shutdown()`]: crate::io::AsyncWriteExt::shutdown
+///
+/// # Errors
+///
+/// The future will immediately return an error if any IO operation on `a`
+/// or `b` returns an error. Some data read from either stream may be lost (not
+/// written to the other stream) in this case.
+///
+/// # Return value
+///
+/// Returns a tuple of bytes copied `a` to `b` and bytes copied `b` to `a`.
+#[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+pub async fn copy_bidirectional<A, B>(a: &mut A, b: &mut B) -> Result<(u64, u64), std::io::Error>
+where
+    A: AsyncRead + AsyncWrite + Unpin + ?Sized,
+    B: AsyncRead + AsyncWrite + Unpin + ?Sized,
+{
+    CopyBidirectional {
+        a,
+        b,
+        a_to_b: TransferState::Running(CopyBuffer::new()),
+        b_to_a: TransferState::Running(CopyBuffer::new()),
+    }
+    .await
+}
diff --git a/third_party/rust/tokio/src/io/util/copy_buf.rs b/third_party/rust/tokio/src/io/util/copy_buf.rs
new file mode 100644
index 0000000000..6831580b40
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/copy_buf.rs
@@ -0,0 +1,102 @@
+use crate::io::{AsyncBufRead, AsyncWrite};
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// A future that asynchronously copies the entire contents of a reader into a
+    /// writer.
+    ///
+    /// This struct is generally created by calling [`copy_buf`][copy_buf]. Please
+    /// see the documentation of `copy_buf()` for more details.
+    ///
+    /// [copy_buf]: copy_buf()
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    struct CopyBuf<'a, R: ?Sized, W: ?Sized> {
+        reader: &'a mut R,
+        writer: &'a mut W,
+        amt: u64,
+    }
+
+    /// Asynchronously copies the entire contents of a reader into a writer.
+    ///
+    /// This function returns a future that will continuously read data from
+    /// `reader` and then write it into `writer` in a streaming fashion until
+    /// `reader` returns EOF.
+    ///
+    /// On success, the total number of bytes that were copied from `reader` to
+    /// `writer` is returned.
+    ///
+    ///
+    /// # Errors
+    ///
+    /// The returned future will finish with an error will return an error
+    /// immediately if any call to `poll_fill_buf` or `poll_write` returns an
+    /// error.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io;
+    ///
+    /// # async fn dox() -> std::io::Result<()> {
+    /// let mut reader: &[u8] = b"hello";
+    /// let mut writer: Vec<u8> = vec![];
+    ///
+    /// io::copy_buf(&mut reader, &mut writer).await?;
+    ///
+    /// assert_eq!(b"hello", &writer[..]);
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn copy_buf<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> io::Result<u64>
+    where
+        R: AsyncBufRead + Unpin + ?Sized,
+        W: AsyncWrite + Unpin + ?Sized,
+    {
+        CopyBuf {
+            reader,
+            writer,
+            amt: 0,
+        }.await
+    }
+}
+
+impl<R, W> Future for CopyBuf<'_, R, W>
+where
+    R: AsyncBufRead + Unpin + ?Sized,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<u64>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        loop {
+            let me = &mut *self;
+            let buffer = ready!(Pin::new(&mut *me.reader).poll_fill_buf(cx))?;
+            if buffer.is_empty() {
+                ready!(Pin::new(&mut self.writer).poll_flush(cx))?;
+                return Poll::Ready(Ok(self.amt));
+            }
+
+            let i = ready!(Pin::new(&mut *me.writer).poll_write(cx, buffer))?;
+            if i == 0 {
+                return Poll::Ready(Err(std::io::ErrorKind::WriteZero.into()));
+            }
+            self.amt += i as u64;
+            Pin::new(&mut *self.reader).consume(i);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<CopyBuf<'_, PhantomPinned, PhantomPinned>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/empty.rs b/third_party/rust/tokio/src/io/util/empty.rs
new file mode 100644
index 0000000000..77db60e40b
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/empty.rs
@@ -0,0 +1,100 @@
+use crate::io::{AsyncBufRead, AsyncRead, ReadBuf};
+
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// An async reader which is always at EOF.
+    ///
+    /// This struct is generally created by calling [`empty`]. Please see
+    /// the documentation of [`empty()`][`empty`] for more details.
+    ///
+    /// This is an asynchronous version of [`std::io::empty`][std].
+    ///
+    /// [`empty`]: fn@empty
+    /// [std]: std::io::empty
+    pub struct Empty {
+        _p: (),
+    }
+
+    /// Creates a new empty async reader.
+    ///
+    /// All reads from the returned reader will return `Poll::Ready(Ok(0))`.
+    ///
+    /// This is an asynchronous version of [`std::io::empty`][std].
+    ///
+    /// [std]: std::io::empty
+    ///
+    /// # Examples
+    ///
+    /// A slightly sad example of not reading anything into a buffer:
+    ///
+    /// ```
+    /// use tokio::io::{self, AsyncReadExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let mut buffer = String::new();
+    ///     io::empty().read_to_string(&mut buffer).await.unwrap();
+    ///     assert!(buffer.is_empty());
+    /// }
+    /// ```
+    pub fn empty() -> Empty {
+        Empty { _p: () }
+    }
+}
+
+impl AsyncRead for Empty {
+    #[inline]
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        _: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        ready!(poll_proceed_and_make_progress(cx));
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl AsyncBufRead for Empty {
+    #[inline]
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        ready!(poll_proceed_and_make_progress(cx));
+        Poll::Ready(Ok(&[]))
+    }
+
+    #[inline]
+    fn consume(self: Pin<&mut Self>, _: usize) {}
+}
+
+impl fmt::Debug for Empty {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("Empty { .. }")
+    }
+}
+
+cfg_coop! {
+    fn poll_proceed_and_make_progress(cx: &mut Context<'_>) -> Poll<()> {
+        let coop = ready!(crate::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::<Empty>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/fill_buf.rs b/third_party/rust/tokio/src/io/util/fill_buf.rs
new file mode 100644
index 0000000000..bb07c766e2
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/fill_buf.rs
@@ -0,0 +1,59 @@
+use crate::io::AsyncBufRead;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Future for the [`fill_buf`](crate::io::AsyncBufReadExt::fill_buf) method.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct FillBuf<'a, R: ?Sized> {
+        reader: Option<&'a mut R>,
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+pub(crate) fn fill_buf<R>(reader: &mut R) -> FillBuf<'_, R>
+where
+    R: AsyncBufRead + ?Sized + Unpin,
+{
+    FillBuf {
+        reader: Some(reader),
+        _pin: PhantomPinned,
+    }
+}
+
+impl<'a, R: AsyncBufRead + ?Sized + Unpin> Future for FillBuf<'a, R> {
+    type Output = io::Result<&'a [u8]>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+
+        let reader = me.reader.take().expect("Polled after completion.");
+        match Pin::new(&mut *reader).poll_fill_buf(cx) {
+            Poll::Ready(Ok(slice)) => unsafe {
+                // Safety: This is necessary only due to a limitation in the
+                // borrow checker. Once Rust starts using the polonius borrow
+                // checker, this can be simplified.
+                //
+                // The safety of this transmute relies on the fact that the
+                // value of `reader` is `None` when we return in this branch.
+                // Otherwise the caller could poll us again after
+                // completion, and access the mutable reference while the
+                // returned immutable reference still exists.
+                let slice = std::mem::transmute::<&[u8], &'a [u8]>(slice);
+                Poll::Ready(Ok(slice))
+            },
+            Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
+            Poll::Pending => {
+                *me.reader = Some(reader);
+                Poll::Pending
+            }
+        }
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/flush.rs b/third_party/rust/tokio/src/io/util/flush.rs
new file mode 100644
index 0000000000..88d60b868d
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/flush.rs
@@ -0,0 +1,46 @@
+use crate::io::AsyncWrite;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// A future used to fully flush an I/O object.
+    ///
+    /// Created by the [`AsyncWriteExt::flush`][flush] function.
+    /// [flush]: crate::io::AsyncWriteExt::flush
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct Flush<'a, A: ?Sized> {
+        a: &'a mut A,
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+/// Creates a future which will entirely flush an I/O object.
+pub(super) fn flush<A>(a: &mut A) -> Flush<'_, A>
+where
+    A: AsyncWrite + Unpin + ?Sized,
+{
+    Flush {
+        a,
+        _pin: PhantomPinned,
+    }
+}
+
+impl<A> Future for Flush<'_, A>
+where
+    A: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+        Pin::new(&mut *me.a).poll_flush(cx)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/lines.rs b/third_party/rust/tokio/src/io/util/lines.rs
new file mode 100644
index 0000000000..717f633f95
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/lines.rs
@@ -0,0 +1,145 @@
+use crate::io::util::read_line::read_line_internal;
+use crate::io::AsyncBufRead;
+
+use pin_project_lite::pin_project;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Reads lines from an [`AsyncBufRead`].
+    ///
+    /// A `Lines` can be turned into a `Stream` with [`LinesStream`].
+    ///
+    /// This type is usually created using the [`lines`] method.
+    ///
+    /// [`AsyncBufRead`]: crate::io::AsyncBufRead
+    /// [`LinesStream`]: https://docs.rs/tokio-stream/0.1/tokio_stream/wrappers/struct.LinesStream.html
+    /// [`lines`]: crate::io::AsyncBufReadExt::lines
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+    pub struct Lines<R> {
+        #[pin]
+        reader: R,
+        buf: String,
+        bytes: Vec<u8>,
+        read: usize,
+    }
+}
+
+pub(crate) fn lines<R>(reader: R) -> Lines<R>
+where
+    R: AsyncBufRead,
+{
+    Lines {
+        reader,
+        buf: String::new(),
+        bytes: Vec::new(),
+        read: 0,
+    }
+}
+
+impl<R> Lines<R>
+where
+    R: AsyncBufRead + Unpin,
+{
+    /// Returns the next line in the stream.
+    ///
+    /// # Cancel safety
+    ///
+    /// This method is cancellation safe.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use tokio::io::AsyncBufRead;
+    /// use tokio::io::AsyncBufReadExt;
+    ///
+    /// # async fn dox(my_buf_read: impl AsyncBufRead + Unpin) -> std::io::Result<()> {
+    /// let mut lines = my_buf_read.lines();
+    ///
+    /// while let Some(line) = lines.next_line().await? {
+    ///     println!("length = {}", line.len())
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn next_line(&mut self) -> io::Result<Option<String>> {
+        use crate::future::poll_fn;
+
+        poll_fn(|cx| Pin::new(&mut *self).poll_next_line(cx)).await
+    }
+
+    /// Obtains a mutable reference to the underlying reader.
+    pub fn get_mut(&mut self) -> &mut R {
+        &mut self.reader
+    }
+
+    /// Obtains a reference to the underlying reader.
+    pub fn get_ref(&mut self) -> &R {
+        &self.reader
+    }
+
+    /// Unwraps this `Lines<R>`, returning the underlying reader.
+    ///
+    /// Note that any leftover data in the internal buffer is lost.
+    /// Therefore, a following read from the underlying reader may lead to data loss.
+    pub fn into_inner(self) -> R {
+        self.reader
+    }
+}
+
+impl<R> Lines<R>
+where
+    R: AsyncBufRead,
+{
+    /// Polls for the next line in the stream.
+    ///
+    /// This method returns:
+    ///
+    ///  * `Poll::Pending` if the next line is not yet available.
+    ///  * `Poll::Ready(Ok(Some(line)))` if the next line is available.
+    ///  * `Poll::Ready(Ok(None))` if there are no more lines in this stream.
+    ///  * `Poll::Ready(Err(err))` if an IO error occurred while reading the next line.
+    ///
+    /// When the method returns `Poll::Pending`, the `Waker` in the provided
+    /// `Context` is scheduled to receive a wakeup when more bytes become
+    /// available on the underlying IO resource.  Note that on multiple calls to
+    /// `poll_next_line`, only the `Waker` from the `Context` passed to the most
+    /// recent call is scheduled to receive a wakeup.
+    pub fn poll_next_line(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<Option<String>>> {
+        let me = self.project();
+
+        let n = ready!(read_line_internal(me.reader, cx, me.buf, me.bytes, me.read))?;
+        debug_assert_eq!(*me.read, 0);
+
+        if n == 0 && me.buf.is_empty() {
+            return Poll::Ready(Ok(None));
+        }
+
+        if me.buf.ends_with('\n') {
+            me.buf.pop();
+
+            if me.buf.ends_with('\r') {
+                me.buf.pop();
+            }
+        }
+
+        Poll::Ready(Ok(Some(mem::take(me.buf))))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Lines<()>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/mem.rs b/third_party/rust/tokio/src/io/util/mem.rs
new file mode 100644
index 0000000000..4019db56ff
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/mem.rs
@@ -0,0 +1,295 @@
+//! In-process memory IO types.
+
+use crate::io::{AsyncRead, AsyncWrite, ReadBuf};
+use crate::loom::sync::Mutex;
+
+use bytes::{Buf, BytesMut};
+use std::{
+    pin::Pin,
+    sync::Arc,
+    task::{self, Poll, Waker},
+};
+
+/// A bidirectional pipe to read and write bytes in memory.
+///
+/// A pair of `DuplexStream`s are created together, and they act as a "channel"
+/// that can be used as in-memory IO types. Writing to one of the pairs will
+/// allow that data to be read from the other, and vice versa.
+///
+/// # Closing a `DuplexStream`
+///
+/// If one end of the `DuplexStream` channel is dropped, any pending reads on
+/// the other side will continue to read data until the buffer is drained, then
+/// they will signal EOF by returning 0 bytes. Any writes to the other side,
+/// including pending ones (that are waiting for free space in the buffer) will
+/// return `Err(BrokenPipe)` immediately.
+///
+/// # Example
+///
+/// ```
+/// # async fn ex() -> std::io::Result<()> {
+/// # use tokio::io::{AsyncReadExt, AsyncWriteExt};
+/// let (mut client, mut server) = tokio::io::duplex(64);
+///
+/// client.write_all(b"ping").await?;
+///
+/// let mut buf = [0u8; 4];
+/// server.read_exact(&mut buf).await?;
+/// assert_eq!(&buf, b"ping");
+///
+/// server.write_all(b"pong").await?;
+///
+/// client.read_exact(&mut buf).await?;
+/// assert_eq!(&buf, b"pong");
+/// # Ok(())
+/// # }
+/// ```
+#[derive(Debug)]
+#[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+pub struct DuplexStream {
+    read: Arc<Mutex<Pipe>>,
+    write: Arc<Mutex<Pipe>>,
+}
+
+/// A unidirectional IO over a piece of memory.
+///
+/// Data can be written to the pipe, and reading will return that data.
+#[derive(Debug)]
+struct Pipe {
+    /// The buffer storing the bytes written, also read from.
+    ///
+    /// Using a `BytesMut` because it has efficient `Buf` and `BufMut`
+    /// functionality already. Additionally, it can try to copy data in the
+    /// same buffer if there read index has advanced far enough.
+    buffer: BytesMut,
+    /// Determines if the write side has been closed.
+    is_closed: bool,
+    /// The maximum amount of bytes that can be written before returning
+    /// `Poll::Pending`.
+    max_buf_size: usize,
+    /// If the `read` side has been polled and is pending, this is the waker
+    /// for that parked task.
+    read_waker: Option<Waker>,
+    /// If the `write` side has filled the `max_buf_size` and returned
+    /// `Poll::Pending`, this is the waker for that parked task.
+    write_waker: Option<Waker>,
+}
+
+// ===== impl DuplexStream =====
+
+/// Create a new pair of `DuplexStream`s that act like a pair of connected sockets.
+///
+/// The `max_buf_size` argument is the maximum amount of bytes that can be
+/// written to a side before the write returns `Poll::Pending`.
+#[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+pub fn duplex(max_buf_size: usize) -> (DuplexStream, DuplexStream) {
+    let one = Arc::new(Mutex::new(Pipe::new(max_buf_size)));
+    let two = Arc::new(Mutex::new(Pipe::new(max_buf_size)));
+
+    (
+        DuplexStream {
+            read: one.clone(),
+            write: two.clone(),
+        },
+        DuplexStream {
+            read: two,
+            write: one,
+        },
+    )
+}
+
+impl AsyncRead for DuplexStream {
+    // Previous rustc required this `self` to be `mut`, even though newer
+    // versions recognize it isn't needed to call `lock()`. So for
+    // compatibility, we include the `mut` and `allow` the lint.
+    //
+    // See https://github.com/rust-lang/rust/issues/73592
+    #[allow(unused_mut)]
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut *self.read.lock()).poll_read(cx, buf)
+    }
+}
+
+impl AsyncWrite for DuplexStream {
+    #[allow(unused_mut)]
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+        buf: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        Pin::new(&mut *self.write.lock()).poll_write(cx, buf)
+    }
+
+    #[allow(unused_mut)]
+    fn poll_flush(
+        mut self: Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut *self.write.lock()).poll_flush(cx)
+    }
+
+    #[allow(unused_mut)]
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut *self.write.lock()).poll_shutdown(cx)
+    }
+}
+
+impl Drop for DuplexStream {
+    fn drop(&mut self) {
+        // notify the other side of the closure
+        self.write.lock().close_write();
+        self.read.lock().close_read();
+    }
+}
+
+// ===== impl Pipe =====
+
+impl Pipe {
+    fn new(max_buf_size: usize) -> Self {
+        Pipe {
+            buffer: BytesMut::new(),
+            is_closed: false,
+            max_buf_size,
+            read_waker: None,
+            write_waker: None,
+        }
+    }
+
+    fn close_write(&mut self) {
+        self.is_closed = true;
+        // needs to notify any readers that no more data will come
+        if let Some(waker) = self.read_waker.take() {
+            waker.wake();
+        }
+    }
+
+    fn close_read(&mut self) {
+        self.is_closed = true;
+        // needs to notify any writers that they have to abort
+        if let Some(waker) = self.write_waker.take() {
+            waker.wake();
+        }
+    }
+
+    fn poll_read_internal(
+        mut self: Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        if self.buffer.has_remaining() {
+            let max = self.buffer.remaining().min(buf.remaining());
+            buf.put_slice(&self.buffer[..max]);
+            self.buffer.advance(max);
+            if max > 0 {
+                // The passed `buf` might have been empty, don't wake up if
+                // no bytes have been moved.
+                if let Some(waker) = self.write_waker.take() {
+                    waker.wake();
+                }
+            }
+            Poll::Ready(Ok(()))
+        } else if self.is_closed {
+            Poll::Ready(Ok(()))
+        } else {
+            self.read_waker = Some(cx.waker().clone());
+            Poll::Pending
+        }
+    }
+
+    fn poll_write_internal(
+        mut self: Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+        buf: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        if self.is_closed {
+            return Poll::Ready(Err(std::io::ErrorKind::BrokenPipe.into()));
+        }
+        let avail = self.max_buf_size - self.buffer.len();
+        if avail == 0 {
+            self.write_waker = Some(cx.waker().clone());
+            return Poll::Pending;
+        }
+
+        let len = buf.len().min(avail);
+        self.buffer.extend_from_slice(&buf[..len]);
+        if let Some(waker) = self.read_waker.take() {
+            waker.wake();
+        }
+        Poll::Ready(Ok(len))
+    }
+}
+
+impl AsyncRead for Pipe {
+    cfg_coop! {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            cx: &mut task::Context<'_>,
+            buf: &mut ReadBuf<'_>,
+        ) -> Poll<std::io::Result<()>> {
+            let coop = ready!(crate::coop::poll_proceed(cx));
+
+            let ret = self.poll_read_internal(cx, buf);
+            if ret.is_ready() {
+                coop.made_progress();
+            }
+            ret
+        }
+    }
+
+    cfg_not_coop! {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            cx: &mut task::Context<'_>,
+            buf: &mut ReadBuf<'_>,
+        ) -> 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(()))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/mod.rs b/third_party/rust/tokio/src/io/util/mod.rs
new file mode 100644
index 0000000000..21199d0be8
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/mod.rs
@@ -0,0 +1,97 @@
+#![allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
+
+cfg_io_util! {
+    mod async_buf_read_ext;
+    pub use async_buf_read_ext::AsyncBufReadExt;
+
+    mod async_read_ext;
+    pub use async_read_ext::AsyncReadExt;
+
+    mod async_seek_ext;
+    pub use async_seek_ext::AsyncSeekExt;
+
+    mod async_write_ext;
+    pub use async_write_ext::AsyncWriteExt;
+
+    mod buf_reader;
+    pub use buf_reader::BufReader;
+
+    mod buf_stream;
+    pub use buf_stream::BufStream;
+
+    mod buf_writer;
+    pub use buf_writer::BufWriter;
+
+    mod chain;
+
+    mod copy;
+    pub use copy::copy;
+
+    mod copy_bidirectional;
+    pub use copy_bidirectional::copy_bidirectional;
+
+    mod copy_buf;
+    pub use copy_buf::copy_buf;
+
+    mod empty;
+    pub use empty::{empty, Empty};
+
+    mod flush;
+
+    mod lines;
+    pub use lines::Lines;
+
+    mod mem;
+    pub use mem::{duplex, DuplexStream};
+
+    mod read;
+    mod read_buf;
+    mod read_exact;
+    mod read_int;
+    mod read_line;
+    mod fill_buf;
+
+    mod read_to_end;
+    mod vec_with_initialized;
+    cfg_process! {
+        pub(crate) use read_to_end::read_to_end;
+    }
+
+    mod read_to_string;
+    mod read_until;
+
+    mod repeat;
+    pub use repeat::{repeat, Repeat};
+
+    mod shutdown;
+
+    mod sink;
+    pub use sink::{sink, Sink};
+
+    mod split;
+    pub use split::Split;
+
+    mod take;
+    pub use take::Take;
+
+    mod write;
+    mod write_vectored;
+    mod write_all;
+    mod write_buf;
+    mod write_all_buf;
+    mod write_int;
+
+
+    // used by `BufReader` and `BufWriter`
+    // https://github.com/rust-lang/rust/blob/master/library/std/src/sys_common/io.rs#L1
+    const DEFAULT_BUF_SIZE: usize = 8 * 1024;
+}
+
+cfg_not_io_util! {
+    cfg_process! {
+        mod vec_with_initialized;
+        mod read_to_end;
+        // Used by process
+        pub(crate) use read_to_end::read_to_end;
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/read.rs b/third_party/rust/tokio/src/io/util/read.rs
new file mode 100644
index 0000000000..edc9d5a9e6
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read.rs
@@ -0,0 +1,55 @@
+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};
+
+/// Tries to read some bytes directly into the given `buf` in asynchronous
+/// manner, returning a future type.
+///
+/// The returned future will resolve to both the I/O stream and the buffer
+/// as well as the number of bytes read once the read operation is completed.
+pub(crate) fn read<'a, R>(reader: &'a mut R, buf: &'a mut [u8]) -> Read<'a, R>
+where
+    R: AsyncRead + Unpin + ?Sized,
+{
+    Read {
+        reader,
+        buf,
+        _pin: PhantomPinned,
+    }
+}
+
+pin_project! {
+    /// A future which can be used to easily read available number of bytes to fill
+    /// a buffer.
+    ///
+    /// Created by the [`read`] function.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct Read<'a, R: ?Sized> {
+        reader: &'a mut R,
+        buf: &'a mut [u8],
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+impl<R> Future for Read<'_, R>
+where
+    R: AsyncRead + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = self.project();
+        let mut buf = ReadBuf::new(*me.buf);
+        ready!(Pin::new(me.reader).poll_read(cx, &mut buf))?;
+        Poll::Ready(Ok(buf.filled().len()))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/read_buf.rs b/third_party/rust/tokio/src/io/util/read_buf.rs
new file mode 100644
index 0000000000..8ec57c0d6f
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_buf.rs
@@ -0,0 +1,72 @@
+use crate::io::AsyncRead;
+
+use bytes::BufMut;
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pub(crate) fn read_buf<'a, R, B>(reader: &'a mut R, buf: &'a mut B) -> ReadBuf<'a, R, B>
+where
+    R: AsyncRead + Unpin,
+    B: BufMut,
+{
+    ReadBuf {
+        reader,
+        buf,
+        _pin: PhantomPinned,
+    }
+}
+
+pin_project! {
+    /// Future returned by [`read_buf`](crate::io::AsyncReadExt::read_buf).
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct ReadBuf<'a, R, B> {
+        reader: &'a mut R,
+        buf: &'a mut B,
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+impl<R, B> Future for ReadBuf<'_, R, B>
+where
+    R: AsyncRead + Unpin,
+    B: BufMut,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        use crate::io::ReadBuf;
+        use std::mem::MaybeUninit;
+
+        let me = self.project();
+
+        if !me.buf.has_remaining_mut() {
+            return Poll::Ready(Ok(0));
+        }
+
+        let n = {
+            let dst = me.buf.chunk_mut();
+            let dst = unsafe { &mut *(dst as *mut _ as *mut [MaybeUninit<u8>]) };
+            let mut buf = ReadBuf::uninit(dst);
+            let ptr = buf.filled().as_ptr();
+            ready!(Pin::new(me.reader).poll_read(cx, &mut buf)?);
+
+            // Ensure the pointer does not change from under us
+            assert_eq!(ptr, buf.filled().as_ptr());
+            buf.filled().len()
+        };
+
+        // Safety: This is guaranteed to be the number of initialized (and read)
+        // bytes due to the invariants provided by `ReadBuf::filled`.
+        unsafe {
+            me.buf.advance_mut(n);
+        }
+
+        Poll::Ready(Ok(n))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/read_exact.rs b/third_party/rust/tokio/src/io/util/read_exact.rs
new file mode 100644
index 0000000000..dbdd58bae9
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_exact.rs
@@ -0,0 +1,69 @@
+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,
+    }
+}
+
+fn eof() -> io::Error {
+    io::Error::new(io::ErrorKind::UnexpectedEof, "early eof")
+}
+
+impl<A> Future for ReadExact<'_, A>
+where
+    A: AsyncRead + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = self.project();
+
+        loop {
+            // if our buffer is empty, then we need to read some data to continue.
+            let rem = me.buf.remaining();
+            if rem != 0 {
+                ready!(Pin::new(&mut *me.reader).poll_read(cx, me.buf))?;
+                if me.buf.remaining() == rem {
+                    return Err(eof()).into();
+                }
+            } else {
+                return Poll::Ready(Ok(me.buf.capacity()));
+            }
+        }
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/read_int.rs b/third_party/rust/tokio/src/io/util/read_int.rs
new file mode 100644
index 0000000000..164dcf5963
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_int.rs
@@ -0,0 +1,159 @@
+use crate::io::{AsyncRead, ReadBuf};
+
+use bytes::Buf;
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::io::ErrorKind::UnexpectedEof;
+use std::marker::PhantomPinned;
+use std::mem::size_of;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+macro_rules! reader {
+    ($name:ident, $ty:ty, $reader:ident) => {
+        reader!($name, $ty, $reader, size_of::<$ty>());
+    };
+    ($name:ident, $ty:ty, $reader:ident, $bytes:expr) => {
+        pin_project! {
+            #[doc(hidden)]
+            #[must_use = "futures do nothing unless you `.await` or poll them"]
+            pub struct $name<R> {
+                #[pin]
+                src: R,
+                buf: [u8; $bytes],
+                read: u8,
+                // Make this future `!Unpin` for compatibility with async trait methods.
+                #[pin]
+                _pin: PhantomPinned,
+            }
+        }
+
+        impl<R> $name<R> {
+            pub(crate) fn new(src: R) -> Self {
+                $name {
+                    src,
+                    buf: [0; $bytes],
+                    read: 0,
+                    _pin: PhantomPinned,
+                }
+            }
+        }
+
+        impl<R> Future for $name<R>
+        where
+            R: AsyncRead,
+        {
+            type Output = io::Result<$ty>;
+
+            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+                let mut me = self.project();
+
+                if *me.read == $bytes as u8 {
+                    return Poll::Ready(Ok(Buf::$reader(&mut &me.buf[..])));
+                }
+
+                while *me.read < $bytes as u8 {
+                    let mut buf = ReadBuf::new(&mut me.buf[*me.read as usize..]);
+
+                    *me.read += match me.src.as_mut().poll_read(cx, &mut buf) {
+                        Poll::Pending => return Poll::Pending,
+                        Poll::Ready(Err(e)) => return Poll::Ready(Err(e.into())),
+                        Poll::Ready(Ok(())) => {
+                            let n = buf.filled().len();
+                            if n == 0 {
+                                return Poll::Ready(Err(UnexpectedEof.into()));
+                            }
+
+                            n as u8
+                        }
+                    };
+                }
+
+                let num = Buf::$reader(&mut &me.buf[..]);
+
+                Poll::Ready(Ok(num))
+            }
+        }
+    };
+}
+
+macro_rules! reader8 {
+    ($name:ident, $ty:ty) => {
+        pin_project! {
+            /// Future returned from `read_u8`
+            #[doc(hidden)]
+            #[must_use = "futures do nothing unless you `.await` or poll them"]
+            pub struct $name<R> {
+                #[pin]
+                reader: R,
+                // Make this future `!Unpin` for compatibility with async trait methods.
+                #[pin]
+                _pin: PhantomPinned,
+            }
+        }
+
+        impl<R> $name<R> {
+            pub(crate) fn new(reader: R) -> $name<R> {
+                $name {
+                    reader,
+                    _pin: PhantomPinned,
+                }
+            }
+        }
+
+        impl<R> Future for $name<R>
+        where
+            R: AsyncRead,
+        {
+            type Output = io::Result<$ty>;
+
+            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+                let me = self.project();
+
+                let mut buf = [0; 1];
+                let mut buf = ReadBuf::new(&mut buf);
+                match me.reader.poll_read(cx, &mut buf) {
+                    Poll::Pending => Poll::Pending,
+                    Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
+                    Poll::Ready(Ok(())) => {
+                        if buf.filled().len() == 0 {
+                            return Poll::Ready(Err(UnexpectedEof.into()));
+                        }
+
+                        Poll::Ready(Ok(buf.filled()[0] as $ty))
+                    }
+                }
+            }
+        }
+    };
+}
+
+reader8!(ReadU8, u8);
+reader8!(ReadI8, i8);
+
+reader!(ReadU16, u16, get_u16);
+reader!(ReadU32, u32, get_u32);
+reader!(ReadU64, u64, get_u64);
+reader!(ReadU128, u128, get_u128);
+
+reader!(ReadI16, i16, get_i16);
+reader!(ReadI32, i32, get_i32);
+reader!(ReadI64, i64, get_i64);
+reader!(ReadI128, i128, get_i128);
+
+reader!(ReadF32, f32, get_f32);
+reader!(ReadF64, f64, get_f64);
+
+reader!(ReadU16Le, u16, get_u16_le);
+reader!(ReadU32Le, u32, get_u32_le);
+reader!(ReadU64Le, u64, get_u64_le);
+reader!(ReadU128Le, u128, get_u128_le);
+
+reader!(ReadI16Le, i16, get_i16_le);
+reader!(ReadI32Le, i32, get_i32_le);
+reader!(ReadI64Le, i64, get_i64_le);
+reader!(ReadI128Le, i128, get_i128_le);
+
+reader!(ReadF32Le, f32, get_f32_le);
+reader!(ReadF64Le, f64, get_f64_le);
diff --git a/third_party/rust/tokio/src/io/util/read_line.rs b/third_party/rust/tokio/src/io/util/read_line.rs
new file mode 100644
index 0000000000..e641f51532
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_line.rs
@@ -0,0 +1,119 @@
+use crate::io::util::read_until::read_until_internal;
+use crate::io::AsyncBufRead;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::mem;
+use std::pin::Pin;
+use std::string::FromUtf8Error;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Future for the [`read_line`](crate::io::AsyncBufReadExt::read_line) method.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct ReadLine<'a, R: ?Sized> {
+        reader: &'a mut R,
+        // This is the buffer we were provided. It will be replaced with an empty string
+        // while reading to postpone utf-8 handling until after reading.
+        output: &'a mut String,
+        // The actual allocation of the string is moved into this vector instead.
+        buf: Vec<u8>,
+        // The number of bytes appended to buf. This can be less than buf.len() if
+        // the buffer was not empty when the operation was started.
+        read: usize,
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+pub(crate) fn read_line<'a, R>(reader: &'a mut R, string: &'a mut String) -> ReadLine<'a, R>
+where
+    R: AsyncBufRead + ?Sized + Unpin,
+{
+    ReadLine {
+        reader,
+        buf: mem::take(string).into_bytes(),
+        output: string,
+        read: 0,
+        _pin: PhantomPinned,
+    }
+}
+
+fn put_back_original_data(output: &mut String, mut vector: Vec<u8>, num_bytes_read: usize) {
+    let original_len = vector.len() - num_bytes_read;
+    vector.truncate(original_len);
+    *output = String::from_utf8(vector).expect("The original data must be valid utf-8.");
+}
+
+/// This handles the various failure cases and puts the string back into `output`.
+///
+/// The `truncate_on_io_error` bool is necessary because `read_to_string` and `read_line`
+/// disagree on what should happen when an IO error occurs.
+pub(super) fn finish_string_read(
+    io_res: io::Result<usize>,
+    utf8_res: Result<String, FromUtf8Error>,
+    read: usize,
+    output: &mut String,
+    truncate_on_io_error: bool,
+) -> Poll<io::Result<usize>> {
+    match (io_res, utf8_res) {
+        (Ok(num_bytes), Ok(string)) => {
+            debug_assert_eq!(read, 0);
+            *output = string;
+            Poll::Ready(Ok(num_bytes))
+        }
+        (Err(io_err), Ok(string)) => {
+            *output = string;
+            if truncate_on_io_error {
+                let original_len = output.len() - read;
+                output.truncate(original_len);
+            }
+            Poll::Ready(Err(io_err))
+        }
+        (Ok(num_bytes), Err(utf8_err)) => {
+            debug_assert_eq!(read, 0);
+            put_back_original_data(output, utf8_err.into_bytes(), num_bytes);
+
+            Poll::Ready(Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stream did not contain valid UTF-8",
+            )))
+        }
+        (Err(io_err), Err(utf8_err)) => {
+            put_back_original_data(output, utf8_err.into_bytes(), read);
+
+            Poll::Ready(Err(io_err))
+        }
+    }
+}
+
+pub(super) fn read_line_internal<R: AsyncBufRead + ?Sized>(
+    reader: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    output: &mut String,
+    buf: &mut Vec<u8>,
+    read: &mut usize,
+) -> Poll<io::Result<usize>> {
+    let io_res = ready!(read_until_internal(reader, cx, b'\n', buf, read));
+    let utf8_res = String::from_utf8(mem::take(buf));
+
+    // At this point both buf and output are empty. The allocation is in utf8_res.
+
+    debug_assert!(buf.is_empty());
+    debug_assert!(output.is_empty());
+    finish_string_read(io_res, utf8_res, *read, output, false)
+}
+
+impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadLine<'_, R> {
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+
+        read_line_internal(Pin::new(*me.reader), cx, me.output, me.buf, me.read)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/read_to_end.rs b/third_party/rust/tokio/src/io/util/read_to_end.rs
new file mode 100644
index 0000000000..f4a564d7dd
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_to_end.rs
@@ -0,0 +1,112 @@
+use crate::io::util::vec_with_initialized::{into_read_buf_parts, VecU8, VecWithInitialized};
+use crate::io::AsyncRead;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct ReadToEnd<'a, R: ?Sized> {
+        reader: &'a mut R,
+        buf: VecWithInitialized<&'a mut Vec<u8>>,
+        // The number of bytes appended to buf. This can be less than buf.len() if
+        // the buffer was not empty when the operation was started.
+        read: usize,
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+pub(crate) fn read_to_end<'a, R>(reader: &'a mut R, buffer: &'a mut Vec<u8>) -> ReadToEnd<'a, R>
+where
+    R: AsyncRead + Unpin + ?Sized,
+{
+    ReadToEnd {
+        reader,
+        buf: VecWithInitialized::new(buffer),
+        read: 0,
+        _pin: PhantomPinned,
+    }
+}
+
+pub(super) fn read_to_end_internal<V: VecU8, R: AsyncRead + ?Sized>(
+    buf: &mut VecWithInitialized<V>,
+    mut reader: Pin<&mut R>,
+    num_read: &mut usize,
+    cx: &mut Context<'_>,
+) -> Poll<io::Result<usize>> {
+    loop {
+        let ret = ready!(poll_read_to_end(buf, reader.as_mut(), cx));
+        match ret {
+            Err(err) => return Poll::Ready(Err(err)),
+            Ok(0) => return Poll::Ready(Ok(mem::replace(num_read, 0))),
+            Ok(num) => {
+                *num_read += num;
+            }
+        }
+    }
+}
+
+/// Tries to read from the provided AsyncRead.
+///
+/// The length of the buffer is increased by the number of bytes read.
+fn poll_read_to_end<V: VecU8, R: AsyncRead + ?Sized>(
+    buf: &mut VecWithInitialized<V>,
+    read: Pin<&mut R>,
+    cx: &mut Context<'_>,
+) -> Poll<io::Result<usize>> {
+    // This uses an adaptive system to extend the vector when it fills. We want to
+    // avoid paying to allocate and zero a huge chunk of memory if the reader only
+    // has 4 bytes while still making large reads if the reader does have a ton
+    // of data to return. Simply tacking on an extra DEFAULT_BUF_SIZE space every
+    // time is 4,500 times (!) slower than this if the reader has a very small
+    // amount of data to return.
+    buf.reserve(32);
+
+    // Get a ReadBuf into the vector.
+    let mut read_buf = buf.get_read_buf();
+
+    let filled_before = read_buf.filled().len();
+    let poll_result = read.poll_read(cx, &mut read_buf);
+    let filled_after = read_buf.filled().len();
+    let n = filled_after - filled_before;
+
+    // Update the length of the vector using the result of poll_read.
+    let read_buf_parts = into_read_buf_parts(read_buf);
+    buf.apply_read_buf(read_buf_parts);
+
+    match poll_result {
+        Poll::Pending => {
+            // In this case, nothing should have been read. However we still
+            // update the vector in case the poll_read call initialized parts of
+            // the vector's unused capacity.
+            debug_assert_eq!(filled_before, filled_after);
+            Poll::Pending
+        }
+        Poll::Ready(Err(err)) => {
+            debug_assert_eq!(filled_before, filled_after);
+            Poll::Ready(Err(err))
+        }
+        Poll::Ready(Ok(())) => Poll::Ready(Ok(n)),
+    }
+}
+
+impl<A> Future for ReadToEnd<'_, A>
+where
+    A: AsyncRead + ?Sized + Unpin,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+
+        read_to_end_internal(me.buf, Pin::new(*me.reader), me.read, cx)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/read_to_string.rs b/third_party/rust/tokio/src/io/util/read_to_string.rs
new file mode 100644
index 0000000000..b3d82a26ba
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_to_string.rs
@@ -0,0 +1,78 @@
+use crate::io::util::read_line::finish_string_read;
+use crate::io::util::read_to_end::read_to_end_internal;
+use crate::io::util::vec_with_initialized::VecWithInitialized;
+use crate::io::AsyncRead;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::{io, mem};
+
+pin_project! {
+    /// Future for the [`read_to_string`](super::AsyncReadExt::read_to_string) method.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct ReadToString<'a, R: ?Sized> {
+        reader: &'a mut R,
+        // This is the buffer we were provided. It will be replaced with an empty string
+        // while reading to postpone utf-8 handling until after reading.
+        output: &'a mut String,
+        // The actual allocation of the string is moved into this vector instead.
+        buf: VecWithInitialized<Vec<u8>>,
+        // The number of bytes appended to buf. This can be less than buf.len() if
+        // the buffer was not empty when the operation was started.
+        read: usize,
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+pub(crate) fn read_to_string<'a, R>(
+    reader: &'a mut R,
+    string: &'a mut String,
+) -> ReadToString<'a, R>
+where
+    R: AsyncRead + ?Sized + Unpin,
+{
+    let buf = mem::take(string).into_bytes();
+    ReadToString {
+        reader,
+        buf: VecWithInitialized::new(buf),
+        output: string,
+        read: 0,
+        _pin: PhantomPinned,
+    }
+}
+
+fn read_to_string_internal<R: AsyncRead + ?Sized>(
+    reader: Pin<&mut R>,
+    output: &mut String,
+    buf: &mut VecWithInitialized<Vec<u8>>,
+    read: &mut usize,
+    cx: &mut Context<'_>,
+) -> Poll<io::Result<usize>> {
+    let io_res = ready!(read_to_end_internal(buf, reader, read, cx));
+    let utf8_res = String::from_utf8(buf.take());
+
+    // At this point both buf and output are empty. The allocation is in utf8_res.
+
+    debug_assert!(buf.is_empty());
+    debug_assert!(output.is_empty());
+    finish_string_read(io_res, utf8_res, *read, output, true)
+}
+
+impl<A> Future for ReadToString<'_, A>
+where
+    A: AsyncRead + ?Sized + Unpin,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+
+        read_to_string_internal(Pin::new(*me.reader), me.output, me.buf, me.read, cx)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/read_until.rs b/third_party/rust/tokio/src/io/util/read_until.rs
new file mode 100644
index 0000000000..90a0e8a18d
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_until.rs
@@ -0,0 +1,79 @@
+use crate::io::AsyncBufRead;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Future for the [`read_until`](crate::io::AsyncBufReadExt::read_until) method.
+    /// The delimiter is included in the resulting vector.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct ReadUntil<'a, R: ?Sized> {
+        reader: &'a mut R,
+        delimiter: u8,
+        buf: &'a mut Vec<u8>,
+        // The number of bytes appended to buf. This can be less than buf.len() if
+        // the buffer was not empty when the operation was started.
+        read: usize,
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+pub(crate) fn read_until<'a, R>(
+    reader: &'a mut R,
+    delimiter: u8,
+    buf: &'a mut Vec<u8>,
+) -> ReadUntil<'a, R>
+where
+    R: AsyncBufRead + ?Sized + Unpin,
+{
+    ReadUntil {
+        reader,
+        delimiter,
+        buf,
+        read: 0,
+        _pin: PhantomPinned,
+    }
+}
+
+pub(super) fn read_until_internal<R: AsyncBufRead + ?Sized>(
+    mut reader: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    delimiter: u8,
+    buf: &mut Vec<u8>,
+    read: &mut usize,
+) -> Poll<io::Result<usize>> {
+    loop {
+        let (done, used) = {
+            let available = ready!(reader.as_mut().poll_fill_buf(cx))?;
+            if let Some(i) = memchr::memchr(delimiter, available) {
+                buf.extend_from_slice(&available[..=i]);
+                (true, i + 1)
+            } else {
+                buf.extend_from_slice(available);
+                (false, available.len())
+            }
+        };
+        reader.as_mut().consume(used);
+        *read += used;
+        if done || used == 0 {
+            return Poll::Ready(Ok(mem::replace(read, 0)));
+        }
+    }
+}
+
+impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadUntil<'_, R> {
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+        read_until_internal(Pin::new(*me.reader), cx, *me.delimiter, me.buf, me.read)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/repeat.rs b/third_party/rust/tokio/src/io/util/repeat.rs
new file mode 100644
index 0000000000..1142765df5
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/repeat.rs
@@ -0,0 +1,72 @@
+use crate::io::{AsyncRead, ReadBuf};
+
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// An async reader which yields one byte over and over and over and over and
+    /// over and...
+    ///
+    /// This struct is generally created by calling [`repeat`][repeat]. Please
+    /// see the documentation of `repeat()` for more details.
+    ///
+    /// This is an asynchronous version of [`std::io::Repeat`][std].
+    ///
+    /// [repeat]: fn@repeat
+    /// [std]: std::io::Repeat
+    #[derive(Debug)]
+    pub struct Repeat {
+        byte: u8,
+    }
+
+    /// Creates an instance of an async reader that infinitely repeats one byte.
+    ///
+    /// All reads from this reader will succeed by filling the specified buffer with
+    /// the given byte.
+    ///
+    /// This is an asynchronous version of [`std::io::repeat`][std].
+    ///
+    /// [std]: std::io::repeat
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::{self, AsyncReadExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let mut buffer = [0; 3];
+    ///     io::repeat(0b101).read_exact(&mut buffer).await.unwrap();
+    ///     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>,
+        _: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        // 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>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/shutdown.rs b/third_party/rust/tokio/src/io/util/shutdown.rs
new file mode 100644
index 0000000000..6d30b004b1
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/shutdown.rs
@@ -0,0 +1,46 @@
+use crate::io::AsyncWrite;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// A future used to shutdown an I/O object.
+    ///
+    /// Created by the [`AsyncWriteExt::shutdown`][shutdown] function.
+    /// [shutdown]: crate::io::AsyncWriteExt::shutdown
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    #[derive(Debug)]
+    pub struct Shutdown<'a, A: ?Sized> {
+        a: &'a mut A,
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+/// Creates a future which will shutdown an I/O object.
+pub(super) fn shutdown<A>(a: &mut A) -> Shutdown<'_, A>
+where
+    A: AsyncWrite + Unpin + ?Sized,
+{
+    Shutdown {
+        a,
+        _pin: PhantomPinned,
+    }
+}
+
+impl<A> Future for Shutdown<'_, A>
+where
+    A: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+        Pin::new(me.a).poll_shutdown(cx)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/sink.rs b/third_party/rust/tokio/src/io/util/sink.rs
new file mode 100644
index 0000000000..05ee773fa3
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/sink.rs
@@ -0,0 +1,87 @@
+use crate::io::AsyncWrite;
+
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// An async writer which will move data into the void.
+    ///
+    /// This struct is generally created by calling [`sink`][sink]. Please
+    /// see the documentation of `sink()` for more details.
+    ///
+    /// This is an asynchronous version of [`std::io::Sink`][std].
+    ///
+    /// [sink]: sink()
+    /// [std]: std::io::Sink
+    pub struct Sink {
+        _p: (),
+    }
+
+    /// Creates an instance of an async writer which will successfully consume all
+    /// data.
+    ///
+    /// All calls to [`poll_write`] on the returned instance will return
+    /// `Poll::Ready(Ok(buf.len()))` and the contents of the buffer will not be
+    /// inspected.
+    ///
+    /// This is an asynchronous version of [`std::io::sink`][std].
+    ///
+    /// [`poll_write`]: crate::io::AsyncWrite::poll_write()
+    /// [std]: std::io::sink
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::{self, AsyncWriteExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> io::Result<()> {
+    ///     let buffer = vec![1, 2, 3, 5, 8];
+    ///     let num_bytes = io::sink().write(&buffer).await?;
+    ///     assert_eq!(num_bytes, 5);
+    ///     Ok(())
+    /// }
+    /// ```
+    pub fn sink() -> Sink {
+        Sink { _p: () }
+    }
+}
+
+impl AsyncWrite for Sink {
+    #[inline]
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        Poll::Ready(Ok(buf.len()))
+    }
+
+    #[inline]
+    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    #[inline]
+    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl fmt::Debug for Sink {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("Sink { .. }")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Sink>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/split.rs b/third_party/rust/tokio/src/io/util/split.rs
new file mode 100644
index 0000000000..7489c24281
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/split.rs
@@ -0,0 +1,121 @@
+use crate::io::util::read_until::read_until_internal;
+use crate::io::AsyncBufRead;
+
+use pin_project_lite::pin_project;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Splitter for the [`split`](crate::io::AsyncBufReadExt::split) method.
+    ///
+    /// A `Split` can be turned into a `Stream` with [`SplitStream`].
+    ///
+    /// [`SplitStream`]: https://docs.rs/tokio-stream/0.1/tokio_stream/wrappers/struct.SplitStream.html
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+    pub struct Split<R> {
+        #[pin]
+        reader: R,
+        buf: Vec<u8>,
+        delim: u8,
+        read: usize,
+    }
+}
+
+pub(crate) fn split<R>(reader: R, delim: u8) -> Split<R>
+where
+    R: AsyncBufRead,
+{
+    Split {
+        reader,
+        buf: Vec::new(),
+        delim,
+        read: 0,
+    }
+}
+
+impl<R> Split<R>
+where
+    R: AsyncBufRead + Unpin,
+{
+    /// Returns the next segment in the stream.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use tokio::io::AsyncBufRead;
+    /// use tokio::io::AsyncBufReadExt;
+    ///
+    /// # async fn dox(my_buf_read: impl AsyncBufRead + Unpin) -> std::io::Result<()> {
+    /// let mut segments = my_buf_read.split(b'f');
+    ///
+    /// while let Some(segment) = segments.next_segment().await? {
+    ///     println!("length = {}", segment.len())
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn next_segment(&mut self) -> io::Result<Option<Vec<u8>>> {
+        use crate::future::poll_fn;
+
+        poll_fn(|cx| Pin::new(&mut *self).poll_next_segment(cx)).await
+    }
+}
+
+impl<R> Split<R>
+where
+    R: AsyncBufRead,
+{
+    /// Polls for the next segment in the stream.
+    ///
+    /// This method returns:
+    ///
+    ///  * `Poll::Pending` if the next segment is not yet available.
+    ///  * `Poll::Ready(Ok(Some(segment)))` if the next segment is available.
+    ///  * `Poll::Ready(Ok(None))` if there are no more segments in this stream.
+    ///  * `Poll::Ready(Err(err))` if an IO error occurred while reading the
+    ///    next segment.
+    ///
+    /// When the method returns `Poll::Pending`, the `Waker` in the provided
+    /// `Context` is scheduled to receive a wakeup when more bytes become
+    /// available on the underlying IO resource.
+    ///
+    /// Note that on multiple calls to `poll_next_segment`, only the `Waker`
+    /// from the `Context` passed to the most recent call is scheduled to
+    /// receive a wakeup.
+    pub fn poll_next_segment(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<Option<Vec<u8>>>> {
+        let me = self.project();
+
+        let n = ready!(read_until_internal(
+            me.reader, cx, *me.delim, me.buf, me.read,
+        ))?;
+        // read_until_internal resets me.read to zero once it finds the delimiter
+        debug_assert_eq!(*me.read, 0);
+
+        if n == 0 && me.buf.is_empty() {
+            return Poll::Ready(Ok(None));
+        }
+
+        if me.buf.last() == Some(me.delim) {
+            me.buf.pop();
+        }
+
+        Poll::Ready(Ok(Some(mem::take(me.buf))))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Split<()>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/take.rs b/third_party/rust/tokio/src/io/util/take.rs
new file mode 100644
index 0000000000..df2f61b9e6
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/take.rs
@@ -0,0 +1,137 @@
+use crate::io::{AsyncBufRead, AsyncRead, ReadBuf};
+
+use pin_project_lite::pin_project;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::{cmp, io};
+
+pin_project! {
+    /// Stream for the [`take`](super::AsyncReadExt::take) method.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless you `.await` or poll them"]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+    pub struct Take<R> {
+        #[pin]
+        inner: R,
+        // Add '_' to avoid conflicts with `limit` method.
+        limit_: u64,
+    }
+}
+
+pub(super) fn take<R: AsyncRead>(inner: R, limit: u64) -> Take<R> {
+    Take {
+        inner,
+        limit_: limit,
+    }
+}
+
+impl<R: AsyncRead> Take<R> {
+    /// Returns the remaining number of bytes that can be
+    /// read before this instance will return EOF.
+    ///
+    /// # Note
+    ///
+    /// This instance may reach `EOF` after reading fewer bytes than indicated by
+    /// this method if the underlying [`AsyncRead`] instance reaches EOF.
+    pub fn limit(&self) -> u64 {
+        self.limit_
+    }
+
+    /// Sets the number of bytes that can be read before this instance will
+    /// return EOF. This is the same as constructing a new `Take` instance, so
+    /// the amount of bytes read and the previous limit value don't matter when
+    /// calling this method.
+    pub fn set_limit(&mut self, limit: u64) {
+        self.limit_ = limit
+    }
+
+    /// Gets a reference to the underlying reader.
+    pub fn get_ref(&self) -> &R {
+        &self.inner
+    }
+
+    /// Gets a mutable reference to the underlying reader.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying reader as doing so may corrupt the internal limit of this
+    /// `Take`.
+    pub fn get_mut(&mut self) -> &mut R {
+        &mut self.inner
+    }
+
+    /// Gets a pinned mutable reference to the underlying reader.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying reader as doing so may corrupt the internal limit of this
+    /// `Take`.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R> {
+        self.project().inner
+    }
+
+    /// Consumes the `Take`, returning the wrapped reader.
+    pub fn into_inner(self) -> R {
+        self.inner
+    }
+}
+
+impl<R: AsyncRead> AsyncRead for Take<R> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<Result<(), io::Error>> {
+        if self.limit_ == 0 {
+            return Poll::Ready(Ok(()));
+        }
+
+        let me = self.project();
+        let mut b = buf.take(*me.limit_ as usize);
+
+        let buf_ptr = b.filled().as_ptr();
+        ready!(me.inner.poll_read(cx, &mut b))?;
+        assert_eq!(b.filled().as_ptr(), buf_ptr);
+
+        let n = b.filled().len();
+
+        // We need to update the original ReadBuf
+        unsafe {
+            buf.assume_init(n);
+        }
+        buf.advance(n);
+        *me.limit_ -= n as u64;
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<R: AsyncBufRead> AsyncBufRead for Take<R> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        let me = self.project();
+
+        // Don't call into inner reader at all at EOF because it may still block
+        if *me.limit_ == 0 {
+            return Poll::Ready(Ok(&[]));
+        }
+
+        let buf = ready!(me.inner.poll_fill_buf(cx)?);
+        let cap = cmp::min(buf.len() as u64, *me.limit_) as usize;
+        Poll::Ready(Ok(&buf[..cap]))
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        let me = self.project();
+        // Don't let callers reset the limit by passing an overlarge value
+        let amt = cmp::min(amt as u64, *me.limit_) as usize;
+        *me.limit_ -= amt as u64;
+        me.inner.consume(amt);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Take<()>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/vec_with_initialized.rs b/third_party/rust/tokio/src/io/util/vec_with_initialized.rs
new file mode 100644
index 0000000000..208cc939c1
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/vec_with_initialized.rs
@@ -0,0 +1,132 @@
+use crate::io::ReadBuf;
+use std::mem::MaybeUninit;
+
+mod private {
+    pub trait Sealed {}
+
+    impl Sealed for Vec<u8> {}
+    impl Sealed for &mut Vec<u8> {}
+}
+
+/// A sealed trait that constrains the generic type parameter in `VecWithInitialized<V>`.  That struct's safety relies
+/// on certain invariants upheld by `Vec<u8>`.
+pub(crate) trait VecU8: AsMut<Vec<u8>> + private::Sealed {}
+
+impl VecU8 for Vec<u8> {}
+impl VecU8 for &mut Vec<u8> {}
+/// This struct wraps a `Vec<u8>` or `&mut Vec<u8>`, combining it with a
+/// `num_initialized`, which keeps track of the number of initialized bytes
+/// in the unused capacity.
+///
+/// The purpose of this struct is to remember how many bytes were initialized
+/// through a `ReadBuf` from call to call.
+///
+/// This struct has the safety invariant that the first `num_initialized` of the
+/// vector's allocation must be initialized at any time.
+#[derive(Debug)]
+pub(crate) struct VecWithInitialized<V> {
+    vec: V,
+    // The number of initialized bytes in the vector.
+    // Always between `vec.len()` and `vec.capacity()`.
+    num_initialized: usize,
+}
+
+impl VecWithInitialized<Vec<u8>> {
+    #[cfg(feature = "io-util")]
+    pub(crate) fn take(&mut self) -> Vec<u8> {
+        self.num_initialized = 0;
+        std::mem::take(&mut self.vec)
+    }
+}
+
+impl<V> VecWithInitialized<V>
+where
+    V: VecU8,
+{
+    pub(crate) fn new(mut vec: V) -> Self {
+        // SAFETY: The safety invariants of vector guarantee that the bytes up
+        // to its length are initialized.
+        Self {
+            num_initialized: vec.as_mut().len(),
+            vec,
+        }
+    }
+
+    pub(crate) fn reserve(&mut self, num_bytes: usize) {
+        let vec = self.vec.as_mut();
+        if vec.capacity() - vec.len() >= num_bytes {
+            return;
+        }
+        // SAFETY: Setting num_initialized to `vec.len()` is correct as
+        // `reserve` does not change the length of the vector.
+        self.num_initialized = vec.len();
+        vec.reserve(num_bytes);
+    }
+
+    #[cfg(feature = "io-util")]
+    pub(crate) fn is_empty(&mut self) -> bool {
+        self.vec.as_mut().is_empty()
+    }
+
+    pub(crate) fn get_read_buf<'a>(&'a mut self) -> ReadBuf<'a> {
+        let num_initialized = self.num_initialized;
+
+        // SAFETY: Creating the slice is safe because of the safety invariants
+        // on Vec<u8>. The safety invariants of `ReadBuf` will further guarantee
+        // that no bytes in the slice are de-initialized.
+        let vec = self.vec.as_mut();
+        let len = vec.len();
+        let cap = vec.capacity();
+        let ptr = vec.as_mut_ptr().cast::<MaybeUninit<u8>>();
+        let slice = unsafe { std::slice::from_raw_parts_mut::<'a, MaybeUninit<u8>>(ptr, cap) };
+
+        // SAFETY: This is safe because the safety invariants of
+        // VecWithInitialized say that the first num_initialized bytes must be
+        // initialized.
+        let mut read_buf = ReadBuf::uninit(slice);
+        unsafe {
+            read_buf.assume_init(num_initialized);
+        }
+        read_buf.set_filled(len);
+
+        read_buf
+    }
+
+    pub(crate) fn apply_read_buf(&mut self, parts: ReadBufParts) {
+        let vec = self.vec.as_mut();
+        assert_eq!(vec.as_ptr(), parts.ptr);
+
+        // SAFETY:
+        // The ReadBufParts really does point inside `self.vec` due to the above
+        // check, and the safety invariants of `ReadBuf` guarantee that the
+        // first `parts.initialized` bytes of `self.vec` really have been
+        // initialized. Additionally, `ReadBuf` guarantees that `parts.len` is
+        // at most `parts.initialized`, so the first `parts.len` bytes are also
+        // initialized.
+        //
+        // Note that this relies on the fact that `V` is either `Vec<u8>` or
+        // `&mut Vec<u8>`, so the vector returned by `self.vec.as_mut()` cannot
+        // change from call to call.
+        unsafe {
+            self.num_initialized = parts.initialized;
+            vec.set_len(parts.len);
+        }
+    }
+}
+
+pub(crate) struct ReadBufParts {
+    // Pointer is only used to check that the ReadBuf actually came from the
+    // right VecWithInitialized.
+    ptr: *const u8,
+    len: usize,
+    initialized: usize,
+}
+
+// This is needed to release the borrow on `VecWithInitialized<V>`.
+pub(crate) fn into_read_buf_parts(rb: ReadBuf<'_>) -> ReadBufParts {
+    ReadBufParts {
+        ptr: rb.filled().as_ptr(),
+        len: rb.filled().len(),
+        initialized: rb.initialized().len(),
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/write.rs b/third_party/rust/tokio/src/io/util/write.rs
new file mode 100644
index 0000000000..92169ebc1d
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write.rs
@@ -0,0 +1,46 @@
+use crate::io::AsyncWrite;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// A future to write some of the buffer to an `AsyncWrite`.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct Write<'a, W: ?Sized> {
+        writer: &'a mut W,
+        buf: &'a [u8],
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+/// Tries to write some bytes from the given `buf` to the writer in an
+/// asynchronous manner, returning a future.
+pub(crate) fn write<'a, W>(writer: &'a mut W, buf: &'a [u8]) -> Write<'a, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    Write {
+        writer,
+        buf,
+        _pin: PhantomPinned,
+    }
+}
+
+impl<W> Future for Write<'_, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = self.project();
+        Pin::new(&mut *me.writer).poll_write(cx, me.buf)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/write_all.rs b/third_party/rust/tokio/src/io/util/write_all.rs
new file mode 100644
index 0000000000..abd3e39d31
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write_all.rs
@@ -0,0 +1,55 @@
+use crate::io::AsyncWrite;
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct WriteAll<'a, W: ?Sized> {
+        writer: &'a mut W,
+        buf: &'a [u8],
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+pub(crate) fn write_all<'a, W>(writer: &'a mut W, buf: &'a [u8]) -> WriteAll<'a, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    WriteAll {
+        writer,
+        buf,
+        _pin: PhantomPinned,
+    }
+}
+
+impl<W> Future for WriteAll<'_, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let me = self.project();
+        while !me.buf.is_empty() {
+            let n = ready!(Pin::new(&mut *me.writer).poll_write(cx, me.buf))?;
+            {
+                let (_, rest) = mem::take(&mut *me.buf).split_at(n);
+                *me.buf = rest;
+            }
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/write_all_buf.rs b/third_party/rust/tokio/src/io/util/write_all_buf.rs
new file mode 100644
index 0000000000..05af7fe99b
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write_all_buf.rs
@@ -0,0 +1,56 @@
+use crate::io::AsyncWrite;
+
+use bytes::Buf;
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// A future to write some of the buffer to an `AsyncWrite`.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct WriteAllBuf<'a, W, B> {
+        writer: &'a mut W,
+        buf: &'a mut B,
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+/// Tries to write some bytes from the given `buf` to the writer in an
+/// asynchronous manner, returning a future.
+pub(crate) fn write_all_buf<'a, W, B>(writer: &'a mut W, buf: &'a mut B) -> WriteAllBuf<'a, W, B>
+where
+    W: AsyncWrite + Unpin,
+    B: Buf,
+{
+    WriteAllBuf {
+        writer,
+        buf,
+        _pin: PhantomPinned,
+    }
+}
+
+impl<W, B> Future for WriteAllBuf<'_, W, B>
+where
+    W: AsyncWrite + Unpin,
+    B: Buf,
+{
+    type Output = io::Result<()>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let me = self.project();
+        while me.buf.has_remaining() {
+            let n = ready!(Pin::new(&mut *me.writer).poll_write(cx, me.buf.chunk())?);
+            me.buf.advance(n);
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/write_buf.rs b/third_party/rust/tokio/src/io/util/write_buf.rs
new file mode 100644
index 0000000000..82fd7a759f
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write_buf.rs
@@ -0,0 +1,55 @@
+use crate::io::AsyncWrite;
+
+use bytes::Buf;
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// A future to write some of the buffer to an `AsyncWrite`.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct WriteBuf<'a, W, B> {
+        writer: &'a mut W,
+        buf: &'a mut B,
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+/// Tries to write some bytes from the given `buf` to the writer in an
+/// asynchronous manner, returning a future.
+pub(crate) fn write_buf<'a, W, B>(writer: &'a mut W, buf: &'a mut B) -> WriteBuf<'a, W, B>
+where
+    W: AsyncWrite + Unpin,
+    B: Buf,
+{
+    WriteBuf {
+        writer,
+        buf,
+        _pin: PhantomPinned,
+    }
+}
+
+impl<W, B> Future for WriteBuf<'_, W, B>
+where
+    W: AsyncWrite + Unpin,
+    B: Buf,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = self.project();
+
+        if !me.buf.has_remaining() {
+            return Poll::Ready(Ok(0));
+        }
+
+        let n = ready!(Pin::new(me.writer).poll_write(cx, me.buf.chunk()))?;
+        me.buf.advance(n);
+        Poll::Ready(Ok(n))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/write_int.rs b/third_party/rust/tokio/src/io/util/write_int.rs
new file mode 100644
index 0000000000..63cd49126f
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write_int.rs
@@ -0,0 +1,152 @@
+use crate::io::AsyncWrite;
+
+use bytes::BufMut;
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomPinned;
+use std::mem::size_of;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+macro_rules! writer {
+    ($name:ident, $ty:ty, $writer:ident) => {
+        writer!($name, $ty, $writer, size_of::<$ty>());
+    };
+    ($name:ident, $ty:ty, $writer:ident, $bytes:expr) => {
+        pin_project! {
+            #[doc(hidden)]
+            #[must_use = "futures do nothing unless you `.await` or poll them"]
+            pub struct $name<W> {
+                #[pin]
+                dst: W,
+                buf: [u8; $bytes],
+                written: u8,
+                // Make this future `!Unpin` for compatibility with async trait methods.
+                #[pin]
+                _pin: PhantomPinned,
+            }
+        }
+
+        impl<W> $name<W> {
+            pub(crate) fn new(w: W, value: $ty) -> Self {
+                let mut writer = Self {
+                    buf: [0; $bytes],
+                    written: 0,
+                    dst: w,
+                    _pin: PhantomPinned,
+                };
+                BufMut::$writer(&mut &mut writer.buf[..], value);
+                writer
+            }
+        }
+
+        impl<W> Future for $name<W>
+        where
+            W: AsyncWrite,
+        {
+            type Output = io::Result<()>;
+
+            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+                let mut me = self.project();
+
+                if *me.written == $bytes as u8 {
+                    return Poll::Ready(Ok(()));
+                }
+
+                while *me.written < $bytes as u8 {
+                    *me.written += match me
+                        .dst
+                        .as_mut()
+                        .poll_write(cx, &me.buf[*me.written as usize..])
+                    {
+                        Poll::Pending => return Poll::Pending,
+                        Poll::Ready(Err(e)) => return Poll::Ready(Err(e.into())),
+                        Poll::Ready(Ok(0)) => {
+                            return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+                        }
+                        Poll::Ready(Ok(n)) => n as u8,
+                    };
+                }
+                Poll::Ready(Ok(()))
+            }
+        }
+    };
+}
+
+macro_rules! writer8 {
+    ($name:ident, $ty:ty) => {
+        pin_project! {
+            #[doc(hidden)]
+            #[must_use = "futures do nothing unless you `.await` or poll them"]
+            pub struct $name<W> {
+                #[pin]
+                dst: W,
+                byte: $ty,
+                // Make this future `!Unpin` for compatibility with async trait methods.
+                #[pin]
+                _pin: PhantomPinned,
+            }
+        }
+
+        impl<W> $name<W> {
+            pub(crate) fn new(dst: W, byte: $ty) -> Self {
+                Self {
+                    dst,
+                    byte,
+                    _pin: PhantomPinned,
+                }
+            }
+        }
+
+        impl<W> Future for $name<W>
+        where
+            W: AsyncWrite,
+        {
+            type Output = io::Result<()>;
+
+            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+                let me = self.project();
+
+                let buf = [*me.byte as u8];
+
+                match me.dst.poll_write(cx, &buf[..]) {
+                    Poll::Pending => Poll::Pending,
+                    Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
+                    Poll::Ready(Ok(0)) => Poll::Ready(Err(io::ErrorKind::WriteZero.into())),
+                    Poll::Ready(Ok(1)) => Poll::Ready(Ok(())),
+                    Poll::Ready(Ok(_)) => unreachable!(),
+                }
+            }
+        }
+    };
+}
+
+writer8!(WriteU8, u8);
+writer8!(WriteI8, i8);
+
+writer!(WriteU16, u16, put_u16);
+writer!(WriteU32, u32, put_u32);
+writer!(WriteU64, u64, put_u64);
+writer!(WriteU128, u128, put_u128);
+
+writer!(WriteI16, i16, put_i16);
+writer!(WriteI32, i32, put_i32);
+writer!(WriteI64, i64, put_i64);
+writer!(WriteI128, i128, put_i128);
+
+writer!(WriteF32, f32, put_f32);
+writer!(WriteF64, f64, put_f64);
+
+writer!(WriteU16Le, u16, put_u16_le);
+writer!(WriteU32Le, u32, put_u32_le);
+writer!(WriteU64Le, u64, put_u64_le);
+writer!(WriteU128Le, u128, put_u128_le);
+
+writer!(WriteI16Le, i16, put_i16_le);
+writer!(WriteI32Le, i32, put_i32_le);
+writer!(WriteI64Le, i64, put_i64_le);
+writer!(WriteI128Le, i128, put_i128_le);
+
+writer!(WriteF32Le, f32, put_f32_le);
+writer!(WriteF64Le, f64, put_f64_le);
diff --git a/third_party/rust/tokio/src/io/util/write_vectored.rs b/third_party/rust/tokio/src/io/util/write_vectored.rs
new file mode 100644
index 0000000000..be40322943
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write_vectored.rs
@@ -0,0 +1,47 @@
+use crate::io::AsyncWrite;
+
+use pin_project_lite::pin_project;
+use std::io;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::{future::Future, io::IoSlice};
+
+pin_project! {
+    /// A future to write a slice of buffers to an `AsyncWrite`.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct WriteVectored<'a, 'b, W: ?Sized> {
+        writer: &'a mut W,
+        bufs: &'a [IoSlice<'b>],
+        // Make this future `!Unpin` for compatibility with async trait methods.
+        #[pin]
+        _pin: PhantomPinned,
+    }
+}
+
+pub(crate) fn write_vectored<'a, 'b, W>(
+    writer: &'a mut W,
+    bufs: &'a [IoSlice<'b>],
+) -> WriteVectored<'a, 'b, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    WriteVectored {
+        writer,
+        bufs,
+        _pin: PhantomPinned,
+    }
+}
+
+impl<W> Future for WriteVectored<'_, '_, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = self.project();
+        Pin::new(&mut *me.writer).poll_write_vectored(cx, me.bufs)
+    }
+}