Adding upstream version 86.0.1.upstream/86.0.1upstream

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

47 files changed, 7589 insertions, 0 deletions

diff --git a/third_party/rust/tokio/src/io/async_buf_read.rs b/third_party/rust/tokio/src/io/async_buf_read.rs
new file mode 100644
index 0000000000..1ab73cd9b7
--- /dev/null
+++ b/third_party/rust/tokio/src/io/async_buf_read.rs
@@ -0,0 +1,115 @@
+use crate::io::AsyncRead;
+
+use std::io;
+use std::ops::DerefMut;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Reads bytes asynchronously.
+///
+/// This trait inherits from [`std::io::BufRead`] and indicates that an I/O object is
+/// **non-blocking**. All non-blocking I/O objects must return an error when
+/// bytes are unavailable instead of blocking the current thread.
+///
+/// Utilities for working with `AsyncBufRead` values are provided by
+/// [`AsyncBufReadExt`].
+///
+/// [`std::io::BufRead`]: std::io::BufRead
+/// [`AsyncBufReadExt`]: crate::io::AsyncBufReadExt
+pub trait AsyncBufRead: AsyncRead {
+    /// Attempts to return the contents of the internal buffer, filling it with more data
+    /// from the inner reader if it is empty.
+    ///
+    /// On success, returns `Poll::Ready(Ok(buf))`.
+    ///
+    /// If no data is available for reading, the method returns
+    /// `Poll::Pending` and arranges for the current task (via
+    /// `cx.waker().wake_by_ref()`) to receive a notification when the object becomes
+    /// readable or is closed.
+    ///
+    /// 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 [`poll_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.
+    ///
+    /// [`poll_read`]: AsyncRead::poll_read
+    /// [`consume`]: AsyncBufRead::consume
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>>;
+
+    /// Tells this buffer that `amt` bytes have been consumed from the buffer,
+    /// so they should no longer be returned in calls to [`poll_read`].
+    ///
+    /// This function is a lower-level call. It needs to be paired with the
+    /// [`poll_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 [`poll_fill_buf`], has been consumed and should
+    /// no longer be returned. As such, this function may do odd things if
+    /// [`poll_fill_buf`] isn't called before calling it.
+    ///
+    /// The `amt` must be `<=` the number of bytes in the buffer returned by
+    /// [`poll_fill_buf`].
+    ///
+    /// [`poll_read`]: AsyncRead::poll_read
+    /// [`poll_fill_buf`]: AsyncBufRead::poll_fill_buf
+    fn consume(self: Pin<&mut Self>, amt: usize);
+}
+
+macro_rules! deref_async_buf_read {
+    () => {
+        fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>)
+            -> Poll<io::Result<&[u8]>>
+        {
+            Pin::new(&mut **self.get_mut()).poll_fill_buf(cx)
+        }
+
+        fn consume(mut self: Pin<&mut Self>, amt: usize) {
+            Pin::new(&mut **self).consume(amt)
+        }
+    }
+}
+
+impl<T: ?Sized + AsyncBufRead + Unpin> AsyncBufRead for Box<T> {
+    deref_async_buf_read!();
+}
+
+impl<T: ?Sized + AsyncBufRead + Unpin> AsyncBufRead for &mut T {
+    deref_async_buf_read!();
+}
+
+impl<P> AsyncBufRead for Pin<P>
+where
+    P: DerefMut + Unpin,
+    P::Target: AsyncBufRead,
+{
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        self.get_mut().as_mut().poll_fill_buf(cx)
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        self.get_mut().as_mut().consume(amt)
+    }
+}
+
+impl AsyncBufRead for &[u8] {
+    fn poll_fill_buf(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        Poll::Ready(Ok(*self))
+    }
+
+    fn consume(mut self: Pin<&mut Self>, amt: usize) {
+        *self = &self[amt..];
+    }
+}
+
+impl<T: AsRef<[u8]> + Unpin> AsyncBufRead for io::Cursor<T> {
+    fn poll_fill_buf(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        Poll::Ready(io::BufRead::fill_buf(self.get_mut()))
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        io::BufRead::consume(self.get_mut(), amt)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/async_read.rs b/third_party/rust/tokio/src/io/async_read.rs
new file mode 100644
index 0000000000..de08d65810
--- /dev/null
+++ b/third_party/rust/tokio/src/io/async_read.rs
@@ -0,0 +1,203 @@
+use bytes::BufMut;
+use std::io;
+use std::mem::MaybeUninit;
+use std::ops::DerefMut;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Reads bytes from a source.
+///
+/// This trait is analogous to the [`std::io::Read`] trait, but integrates with
+/// the asynchronous task system. In particular, the [`poll_read`] method,
+/// unlike [`Read::read`], will automatically queue the current task for wakeup
+/// and return if data is not yet available, rather than blocking the calling
+/// thread.
+///
+/// Specifically, this means that the `poll_read` function will return one of
+/// the following:
+///
+/// * `Poll::Ready(Ok(n))` means that `n` bytes of data was immediately read
+///   and placed into the output buffer, where `n` == 0 implies that EOF has
+///   been reached.
+///
+/// * `Poll::Pending` means that no data was read into the buffer
+///   provided. The I/O object is not currently readable but may become readable
+///   in the future. Most importantly, **the current future's task is scheduled
+///   to get unparked when the object is readable**. This means that like
+///   `Future::poll` you'll receive a notification when the I/O object is
+///   readable again.
+///
+/// * `Poll::Ready(Err(e))` for other errors are standard I/O errors coming from the
+///   underlying object.
+///
+/// This trait importantly means that the `read` method only works in the
+/// context of a future's task. The object may panic if used outside of a task.
+///
+/// Utilities for working with `AsyncRead` values are provided by
+/// [`AsyncReadExt`].
+///
+/// [`poll_read`]: AsyncRead::poll_read
+/// [`std::io::Read`]: std::io::Read
+/// [`Read::read`]: std::io::Read::read
+/// [`AsyncReadExt`]: crate::io::AsyncReadExt
+pub trait AsyncRead {
+    /// Prepares an uninitialized buffer to be safe to pass to `read`. Returns
+    /// `true` if the supplied buffer was zeroed out.
+    ///
+    /// While it would be highly unusual, implementations of [`io::Read`] are
+    /// able to read data from the buffer passed as an argument. Because of
+    /// this, the buffer passed to [`io::Read`] must be initialized memory. In
+    /// situations where large numbers of buffers are used, constantly having to
+    /// zero out buffers can be expensive.
+    ///
+    /// This function does any necessary work to prepare an uninitialized buffer
+    /// to be safe to pass to `read`. If `read` guarantees to never attempt to
+    /// read data out of the supplied buffer, then `prepare_uninitialized_buffer`
+    /// doesn't need to do any work.
+    ///
+    /// If this function returns `true`, then the memory has been zeroed out.
+    /// This allows implementations of `AsyncRead` which are composed of
+    /// multiple subimplementations to efficiently implement
+    /// `prepare_uninitialized_buffer`.
+    ///
+    /// This function isn't actually `unsafe` to call but `unsafe` to implement.
+    /// The implementer must ensure that either the whole `buf` has been zeroed
+    /// or `poll_read_buf()` overwrites the buffer without reading it and returns
+    /// correct value.
+    ///
+    /// This function is called from [`poll_read_buf`].
+    ///
+    /// # Safety
+    ///
+    /// Implementations that return `false` must never read from data slices
+    /// that they did not write to.
+    ///
+    /// [`io::Read`]: std::io::Read
+    /// [`poll_read_buf`]: #method.poll_read_buf
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
+        for x in buf {
+            *x.as_mut_ptr() = 0;
+        }
+
+        true
+    }
+
+    /// Attempts to read from the `AsyncRead` into `buf`.
+    ///
+    /// On success, returns `Poll::Ready(Ok(num_bytes_read))`.
+    ///
+    /// If no data is available for reading, the method returns
+    /// `Poll::Pending` and arranges for the current task (via
+    /// `cx.waker()`) to receive a notification when the object becomes
+    /// readable or is closed.
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>>;
+
+    /// Pulls some bytes from this source into the specified `BufMut`, returning
+    /// how many bytes were read.
+    ///
+    /// The `buf` provided will have bytes read into it and the internal cursor
+    /// will be advanced if any bytes were read. Note that this method typically
+    /// will not reallocate the buffer provided.
+    fn poll_read_buf<B: BufMut>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>>
+    where
+        Self: Sized,
+    {
+        if !buf.has_remaining_mut() {
+            return Poll::Ready(Ok(0));
+        }
+
+        unsafe {
+            let n = {
+                let b = buf.bytes_mut();
+
+                self.prepare_uninitialized_buffer(b);
+
+                // Convert to `&mut [u8]`
+                let b = &mut *(b as *mut [MaybeUninit<u8>] as *mut [u8]);
+
+                let n = ready!(self.poll_read(cx, b))?;
+                assert!(n <= b.len(), "Bad AsyncRead implementation, more bytes were reported as read than the buffer can hold");
+                n
+            };
+
+            buf.advance_mut(n);
+            Poll::Ready(Ok(n))
+        }
+    }
+}
+
+macro_rules! deref_async_read {
+    () => {
+        unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
+            (**self).prepare_uninitialized_buffer(buf)
+        }
+
+        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8])
+            -> Poll<io::Result<usize>>
+        {
+            Pin::new(&mut **self).poll_read(cx, buf)
+        }
+    }
+}
+
+impl<T: ?Sized + AsyncRead + Unpin> AsyncRead for Box<T> {
+    deref_async_read!();
+}
+
+impl<T: ?Sized + AsyncRead + Unpin> AsyncRead for &mut T {
+    deref_async_read!();
+}
+
+impl<P> AsyncRead for Pin<P>
+where
+    P: DerefMut + Unpin,
+    P::Target: AsyncRead,
+{
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
+        (**self).prepare_uninitialized_buffer(buf)
+    }
+
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.get_mut().as_mut().poll_read(cx, buf)
+    }
+}
+
+impl AsyncRead for &[u8] {
+    unsafe fn prepare_uninitialized_buffer(&self, _buf: &mut [MaybeUninit<u8>]) -> bool {
+        false
+    }
+
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(io::Read::read(self.get_mut(), buf))
+    }
+}
+
+impl<T: AsRef<[u8]> + Unpin> AsyncRead for io::Cursor<T> {
+    unsafe fn prepare_uninitialized_buffer(&self, _buf: &mut [MaybeUninit<u8>]) -> bool {
+        false
+    }
+
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(io::Read::read(self.get_mut(), buf))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/async_seek.rs b/third_party/rust/tokio/src/io/async_seek.rs
new file mode 100644
index 0000000000..0be9c90d56
--- /dev/null
+++ b/third_party/rust/tokio/src/io/async_seek.rs
@@ -0,0 +1,104 @@
+use std::io::{self, SeekFrom};
+use std::ops::DerefMut;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Seek bytes asynchronously.
+///
+/// This trait is analogous to the [`std::io::Seek`] trait, but integrates
+/// with the asynchronous task system. In particular, the `start_seek`
+/// method, unlike [`Seek::seek`], will not block the calling thread.
+///
+/// Utilities for working with `AsyncSeek` values are provided by
+/// [`AsyncSeekExt`].
+///
+/// [`std::io::Seek`]: std::io::Seek
+/// [`Seek::seek`]: std::io::Seek::seek()
+/// [`AsyncSeekExt`]: crate::io::AsyncSeekExt
+pub trait AsyncSeek {
+    /// Attempts to seek to an offset, in bytes, in a stream.
+    ///
+    /// A seek beyond the end of a stream is allowed, but behavior is defined
+    /// by the implementation.
+    ///
+    /// If this function returns successfully, then the job has been submitted.
+    /// To find out when it completes, call `poll_complete`.
+    fn start_seek(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        position: SeekFrom,
+    ) -> Poll<io::Result<()>>;
+
+    /// Waits for a seek operation to complete.
+    ///
+    /// If the seek operation completed successfully,
+    /// this method returns the new position from the start of the stream.
+    /// That position can be used later with [`SeekFrom::Start`].
+    ///
+    /// # Errors
+    ///
+    /// Seeking to a negative offset is considered an error.
+    ///
+    /// # Panics
+    ///
+    /// Calling this method without calling `start_seek` first is an error.
+    fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>>;
+}
+
+macro_rules! deref_async_seek {
+    () => {
+        fn start_seek(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            pos: SeekFrom,
+        ) -> Poll<io::Result<()>> {
+            Pin::new(&mut **self).start_seek(cx, pos)
+        }
+
+        fn poll_complete(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<io::Result<u64>> {
+            Pin::new(&mut **self).poll_complete(cx)
+        }
+    }
+}
+
+impl<T: ?Sized + AsyncSeek + Unpin> AsyncSeek for Box<T> {
+    deref_async_seek!();
+}
+
+impl<T: ?Sized + AsyncSeek + Unpin> AsyncSeek for &mut T {
+    deref_async_seek!();
+}
+
+impl<P> AsyncSeek for Pin<P>
+where
+    P: DerefMut + Unpin,
+    P::Target: AsyncSeek,
+{
+    fn start_seek(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        pos: SeekFrom,
+    ) -> Poll<io::Result<()>> {
+        self.get_mut().as_mut().start_seek(cx, pos)
+    }
+
+    fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
+        self.get_mut().as_mut().poll_complete(cx)
+    }
+}
+
+impl<T: AsRef<[u8]> + Unpin> AsyncSeek for io::Cursor<T> {
+    fn start_seek(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        pos: SeekFrom,
+    ) -> Poll<io::Result<()>> {
+        Poll::Ready(io::Seek::seek(&mut *self, pos).map(drop))
+    }
+    fn poll_complete(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<u64>> {
+        Poll::Ready(Ok(self.get_mut().position()))
+    }
+}
diff --git a/third_party/rust/tokio/src/io/async_write.rs b/third_party/rust/tokio/src/io/async_write.rs
new file mode 100644
index 0000000000..0bfed056ef
--- /dev/null
+++ b/third_party/rust/tokio/src/io/async_write.rs
@@ -0,0 +1,291 @@
+use bytes::Buf;
+use std::io;
+use std::ops::DerefMut;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Writes bytes asynchronously.
+///
+/// The trait inherits from [`std::io::Write`] and indicates that an I/O object is
+/// **nonblocking**. All non-blocking I/O objects must return an error when
+/// bytes cannot be written instead of blocking the current thread.
+///
+/// Specifically, this means that the [`poll_write`] function will return one of
+/// the following:
+///
+/// * `Poll::Ready(Ok(n))` means that `n` bytes of data was immediately
+///   written.
+///
+/// * `Poll::Pending` means that no data was written from the buffer
+///   provided. The I/O object is not currently writable but may become writable
+///   in the future. Most importantly, **the current future's task is scheduled
+///   to get unparked when the object is writable**. This means that like
+///   `Future::poll` you'll receive a notification when the I/O object is
+///   writable again.
+///
+/// * `Poll::Ready(Err(e))` for other errors are standard I/O errors coming from the
+///   underlying object.
+///
+/// This trait importantly means that the [`write`][stdwrite] method only works in
+/// the context of a future's task. The object may panic if used outside of a task.
+///
+/// Note that this trait also represents that the  [`Write::flush`][stdflush] method
+/// works very similarly to the `write` method, notably that `Ok(())` means that the
+/// writer has successfully been flushed, a "would block" error means that the
+/// current task is ready to receive a notification when flushing can make more
+/// progress, and otherwise normal errors can happen as well.
+///
+/// Utilities for working with `AsyncWrite` values are provided by
+/// [`AsyncWriteExt`].
+///
+/// [`std::io::Write`]: std::io::Write
+/// [`poll_write`]: AsyncWrite::poll_write()
+/// [stdwrite]: std::io::Write::write()
+/// [stdflush]: std::io::Write::flush()
+/// [`AsyncWriteExt`]: crate::io::AsyncWriteExt
+pub trait AsyncWrite {
+    /// Attempt to write bytes from `buf` into the object.
+    ///
+    /// On success, returns `Poll::Ready(Ok(num_bytes_written))`.
+    ///
+    /// If the object is not ready for writing, the method returns
+    /// `Poll::Pending` and arranges for the current task (via
+    /// `cx.waker()`) to receive a notification when the object becomes
+    /// readable or is closed.
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, io::Error>>;
+
+    /// Attempts to flush the object, ensuring that any buffered data reach
+    /// their destination.
+    ///
+    /// On success, returns `Poll::Ready(Ok(()))`.
+    ///
+    /// If flushing cannot immediately complete, this method returns
+    /// `Poll::Pending` and arranges for the current task (via
+    /// `cx.waker()`) to receive a notification when the object can make
+    /// progress towards flushing.
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>>;
+
+    /// Initiates or attempts to shut down this writer, returning success when
+    /// the I/O connection has completely shut down.
+    ///
+    /// This method is intended to be used for asynchronous shutdown of I/O
+    /// connections. For example this is suitable for implementing shutdown of a
+    /// TLS connection or calling `TcpStream::shutdown` on a proxied connection.
+    /// Protocols sometimes need to flush out final pieces of data or otherwise
+    /// perform a graceful shutdown handshake, reading/writing more data as
+    /// appropriate. This method is the hook for such protocols to implement the
+    /// graceful shutdown logic.
+    ///
+    /// This `shutdown` method is required by implementers of the
+    /// `AsyncWrite` trait. Wrappers typically just want to proxy this call
+    /// through to the wrapped type, and base types will typically implement
+    /// shutdown logic here or just return `Ok(().into())`. Note that if you're
+    /// wrapping an underlying `AsyncWrite` a call to `shutdown` implies that
+    /// transitively the entire stream has been shut down. After your wrapper's
+    /// shutdown logic has been executed you should shut down the underlying
+    /// stream.
+    ///
+    /// Invocation of a `shutdown` implies an invocation of `flush`. Once this
+    /// method returns `Ready` it implies that a flush successfully happened
+    /// before the shutdown happened. That is, callers don't need to call
+    /// `flush` before calling `shutdown`. They can rely that by calling
+    /// `shutdown` any pending buffered data will be written out.
+    ///
+    /// # Return value
+    ///
+    /// This function returns a `Poll<io::Result<()>>` classified as such:
+    ///
+    /// * `Poll::Ready(Ok(()))` - indicates that the connection was
+    ///   successfully shut down and is now safe to deallocate/drop/close
+    ///   resources associated with it. This method means that the current task
+    ///   will no longer receive any notifications due to this method and the
+    ///   I/O object itself is likely no longer usable.
+    ///
+    /// * `Poll::Pending` - indicates that shutdown is initiated but could
+    ///   not complete just yet. This may mean that more I/O needs to happen to
+    ///   continue this shutdown operation. The current task is scheduled to
+    ///   receive a notification when it's otherwise ready to continue the
+    ///   shutdown operation. When woken up this method should be called again.
+    ///
+    /// * `Poll::Ready(Err(e))` - indicates a fatal error has happened with shutdown,
+    ///   indicating that the shutdown operation did not complete successfully.
+    ///   This typically means that the I/O object is no longer usable.
+    ///
+    /// # Errors
+    ///
+    /// This function can return normal I/O errors through `Err`, described
+    /// above. Additionally this method may also render the underlying
+    /// `Write::write` method no longer usable (e.g. will return errors in the
+    /// future). It's recommended that once `shutdown` is called the
+    /// `write` method is no longer called.
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if not called within the context of a future's
+    /// task.
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>>;
+
+    /// Writes a `Buf` into this value, returning how many bytes were written.
+    ///
+    /// Note that this method will advance the `buf` provided automatically by
+    /// the number of bytes written.
+    fn poll_write_buf<B: Buf>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<Result<usize, io::Error>>
+    where
+        Self: Sized,
+    {
+        if !buf.has_remaining() {
+            return Poll::Ready(Ok(0));
+        }
+
+        let n = ready!(self.poll_write(cx, buf.bytes()))?;
+        buf.advance(n);
+        Poll::Ready(Ok(n))
+    }
+}
+
+macro_rules! deref_async_write {
+    () => {
+        fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8])
+            -> Poll<io::Result<usize>>
+        {
+            Pin::new(&mut **self).poll_write(cx, buf)
+        }
+
+        fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Pin::new(&mut **self).poll_flush(cx)
+        }
+
+        fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Pin::new(&mut **self).poll_shutdown(cx)
+        }
+    }
+}
+
+impl<T: ?Sized + AsyncWrite + Unpin> AsyncWrite for Box<T> {
+    deref_async_write!();
+}
+
+impl<T: ?Sized + AsyncWrite + Unpin> AsyncWrite for &mut T {
+    deref_async_write!();
+}
+
+impl<P> AsyncWrite for Pin<P>
+where
+    P: DerefMut + Unpin,
+    P::Target: AsyncWrite,
+{
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        self.get_mut().as_mut().poll_write(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.get_mut().as_mut().poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.get_mut().as_mut().poll_shutdown(cx)
+    }
+}
+
+impl AsyncWrite for Vec<u8> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        self.get_mut().extend_from_slice(buf);
+        Poll::Ready(Ok(buf.len()))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl AsyncWrite for io::Cursor<&mut [u8]> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(io::Write::write(&mut *self, buf))
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(io::Write::flush(&mut *self))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.poll_flush(cx)
+    }
+}
+
+impl AsyncWrite for io::Cursor<&mut Vec<u8>> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(io::Write::write(&mut *self, buf))
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(io::Write::flush(&mut *self))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.poll_flush(cx)
+    }
+}
+
+impl AsyncWrite for io::Cursor<Vec<u8>> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(io::Write::write(&mut *self, buf))
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(io::Write::flush(&mut *self))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.poll_flush(cx)
+    }
+}
+
+impl AsyncWrite for io::Cursor<Box<[u8]>> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(io::Write::write(&mut *self, buf))
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(io::Write::flush(&mut *self))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.poll_flush(cx)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/blocking.rs b/third_party/rust/tokio/src/io/blocking.rs
new file mode 100644
index 0000000000..2491039a3f
--- /dev/null
+++ b/third_party/rust/tokio/src/io/blocking.rs
@@ -0,0 +1,279 @@
+use crate::io::sys;
+use crate::io::{AsyncRead, AsyncWrite};
+
+use std::cmp;
+use std::future::Future;
+use std::io;
+use std::io::prelude::*;
+use std::pin::Pin;
+use std::task::Poll::*;
+use std::task::{Context, Poll};
+
+use self::State::*;
+
+/// `T` should not implement _both_ Read and Write.
+#[derive(Debug)]
+pub(crate) struct Blocking<T> {
+    inner: Option<T>,
+    state: State<T>,
+    /// `true` if the lower IO layer needs flushing
+    need_flush: bool,
+}
+
+#[derive(Debug)]
+pub(crate) struct Buf {
+    buf: Vec<u8>,
+    pos: usize,
+}
+
+pub(crate) const MAX_BUF: usize = 16 * 1024;
+
+#[derive(Debug)]
+enum State<T> {
+    Idle(Option<Buf>),
+    Busy(sys::Blocking<(io::Result<usize>, Buf, T)>),
+}
+
+cfg_io_std! {
+    impl<T> Blocking<T> {
+        pub(crate) fn new(inner: T) -> Blocking<T> {
+            Blocking {
+                inner: Some(inner),
+                state: State::Idle(Some(Buf::with_capacity(0))),
+                need_flush: false,
+            }
+        }
+    }
+}
+
+impl<T> AsyncRead for Blocking<T>
+where
+    T: Read + Unpin + Send + 'static,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        dst: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        loop {
+            match self.state {
+                Idle(ref mut buf_cell) => {
+                    let mut buf = buf_cell.take().unwrap();
+
+                    if !buf.is_empty() {
+                        let n = buf.copy_to(dst);
+                        *buf_cell = Some(buf);
+                        return Ready(Ok(n));
+                    }
+
+                    buf.ensure_capacity_for(dst);
+                    let mut inner = self.inner.take().unwrap();
+
+                    self.state = Busy(sys::run(move || {
+                        let res = buf.read_from(&mut inner);
+                        (res, buf, inner)
+                    }));
+                }
+                Busy(ref mut rx) => {
+                    let (res, mut buf, inner) = ready!(Pin::new(rx).poll(cx))?;
+                    self.inner = Some(inner);
+
+                    match res {
+                        Ok(_) => {
+                            let n = buf.copy_to(dst);
+                            self.state = Idle(Some(buf));
+                            return Ready(Ok(n));
+                        }
+                        Err(e) => {
+                            assert!(buf.is_empty());
+
+                            self.state = Idle(Some(buf));
+                            return Ready(Err(e));
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl<T> AsyncWrite for Blocking<T>
+where
+    T: Write + Unpin + Send + 'static,
+{
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        src: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        loop {
+            match self.state {
+                Idle(ref mut buf_cell) => {
+                    let mut buf = buf_cell.take().unwrap();
+
+                    assert!(buf.is_empty());
+
+                    let n = buf.copy_from(src);
+                    let mut inner = self.inner.take().unwrap();
+
+                    self.state = Busy(sys::run(move || {
+                        let n = buf.len();
+                        let res = buf.write_to(&mut inner).map(|_| n);
+
+                        (res, buf, inner)
+                    }));
+                    self.need_flush = true;
+
+                    return Ready(Ok(n));
+                }
+                Busy(ref mut rx) => {
+                    let (res, buf, inner) = ready!(Pin::new(rx).poll(cx))?;
+                    self.state = Idle(Some(buf));
+                    self.inner = Some(inner);
+
+                    // If error, return
+                    res?;
+                }
+            }
+        }
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        loop {
+            let need_flush = self.need_flush;
+            match self.state {
+                // The buffer is not used here
+                Idle(ref mut buf_cell) => {
+                    if need_flush {
+                        let buf = buf_cell.take().unwrap();
+                        let mut inner = self.inner.take().unwrap();
+
+                        self.state = Busy(sys::run(move || {
+                            let res = inner.flush().map(|_| 0);
+                            (res, buf, inner)
+                        }));
+
+                        self.need_flush = false;
+                    } else {
+                        return Ready(Ok(()));
+                    }
+                }
+                Busy(ref mut rx) => {
+                    let (res, buf, inner) = ready!(Pin::new(rx).poll(cx))?;
+                    self.state = Idle(Some(buf));
+                    self.inner = Some(inner);
+
+                    // If error, return
+                    res?;
+                }
+            }
+        }
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+/// Repeates operations that are interrupted
+macro_rules! uninterruptibly {
+    ($e:expr) => {{
+        loop {
+            match $e {
+                Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
+                res => break res,
+            }
+        }
+    }};
+}
+
+impl Buf {
+    pub(crate) fn with_capacity(n: usize) -> Buf {
+        Buf {
+            buf: Vec::with_capacity(n),
+            pos: 0,
+        }
+    }
+
+    pub(crate) fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    pub(crate) fn len(&self) -> usize {
+        self.buf.len() - self.pos
+    }
+
+    pub(crate) fn copy_to(&mut self, dst: &mut [u8]) -> usize {
+        let n = cmp::min(self.len(), dst.len());
+        dst[..n].copy_from_slice(&self.bytes()[..n]);
+        self.pos += n;
+
+        if self.pos == self.buf.len() {
+            self.buf.truncate(0);
+            self.pos = 0;
+        }
+
+        n
+    }
+
+    pub(crate) fn copy_from(&mut self, src: &[u8]) -> usize {
+        assert!(self.is_empty());
+
+        let n = cmp::min(src.len(), MAX_BUF);
+
+        self.buf.extend_from_slice(&src[..n]);
+        n
+    }
+
+    pub(crate) fn bytes(&self) -> &[u8] {
+        &self.buf[self.pos..]
+    }
+
+    pub(crate) fn ensure_capacity_for(&mut self, bytes: &[u8]) {
+        assert!(self.is_empty());
+
+        let len = cmp::min(bytes.len(), MAX_BUF);
+
+        if self.buf.len() < len {
+            self.buf.reserve(len - self.buf.len());
+        }
+
+        unsafe {
+            self.buf.set_len(len);
+        }
+    }
+
+    pub(crate) fn read_from<T: Read>(&mut self, rd: &mut T) -> io::Result<usize> {
+        let res = uninterruptibly!(rd.read(&mut self.buf));
+
+        if let Ok(n) = res {
+            self.buf.truncate(n);
+        } else {
+            self.buf.clear();
+        }
+
+        assert_eq!(self.pos, 0);
+
+        res
+    }
+
+    pub(crate) fn write_to<T: Write>(&mut self, wr: &mut T) -> io::Result<()> {
+        assert_eq!(self.pos, 0);
+
+        // `write_all` already ignores interrupts
+        let res = wr.write_all(&self.buf);
+        self.buf.clear();
+        res
+    }
+}
+
+cfg_fs! {
+    impl Buf {
+        pub(crate) fn discard_read(&mut self) -> i64 {
+            let ret = -(self.bytes().len() as i64);
+            self.pos = 0;
+            self.buf.truncate(0);
+            ret
+        }
+    }
+}
diff --git a/third_party/rust/tokio/src/io/driver/mod.rs b/third_party/rust/tokio/src/io/driver/mod.rs
new file mode 100644
index 0000000000..d8535d9ab2
--- /dev/null
+++ b/third_party/rust/tokio/src/io/driver/mod.rs
@@ -0,0 +1,396 @@
+pub(crate) mod platform;
+
+mod scheduled_io;
+pub(crate) use scheduled_io::ScheduledIo; // pub(crate) for tests
+
+use crate::loom::sync::atomic::AtomicUsize;
+use crate::park::{Park, Unpark};
+use crate::runtime::context;
+use crate::util::slab::{Address, Slab};
+
+use mio::event::Evented;
+use std::fmt;
+use std::io;
+use std::sync::atomic::Ordering::SeqCst;
+use std::sync::{Arc, Weak};
+use std::task::Waker;
+use std::time::Duration;
+
+/// I/O driver, backed by Mio
+pub(crate) struct Driver {
+    /// Reuse the `mio::Events` value across calls to poll.
+    events: mio::Events,
+
+    /// State shared between the reactor and the handles.
+    inner: Arc<Inner>,
+
+    _wakeup_registration: mio::Registration,
+}
+
+/// A reference to an I/O driver
+#[derive(Clone)]
+pub(crate) struct Handle {
+    inner: Weak<Inner>,
+}
+
+pub(super) struct Inner {
+    /// The underlying system event queue.
+    io: mio::Poll,
+
+    /// Dispatch slabs for I/O and futures events
+    pub(super) io_dispatch: Slab<ScheduledIo>,
+
+    /// The number of sources in `io_dispatch`.
+    n_sources: AtomicUsize,
+
+    /// Used to wake up the reactor from a call to `turn`
+    wakeup: mio::SetReadiness,
+}
+
+#[derive(Debug, Eq, PartialEq, Clone, Copy)]
+pub(super) enum Direction {
+    Read,
+    Write,
+}
+
+const TOKEN_WAKEUP: mio::Token = mio::Token(Address::NULL);
+
+fn _assert_kinds() {
+    fn _assert<T: Send + Sync>() {}
+
+    _assert::<Handle>();
+}
+
+// ===== impl Driver =====
+
+impl Driver {
+    /// Creates a new event loop, returning any error that happened during the
+    /// creation.
+    pub(crate) fn new() -> io::Result<Driver> {
+        let io = mio::Poll::new()?;
+        let wakeup_pair = mio::Registration::new2();
+
+        io.register(
+            &wakeup_pair.0,
+            TOKEN_WAKEUP,
+            mio::Ready::readable(),
+            mio::PollOpt::level(),
+        )?;
+
+        Ok(Driver {
+            events: mio::Events::with_capacity(1024),
+            _wakeup_registration: wakeup_pair.0,
+            inner: Arc::new(Inner {
+                io,
+                io_dispatch: Slab::new(),
+                n_sources: AtomicUsize::new(0),
+                wakeup: wakeup_pair.1,
+            }),
+        })
+    }
+
+    /// Returns a handle to this event loop which can be sent across threads
+    /// and can be used as a proxy to the event loop itself.
+    ///
+    /// Handles are cloneable and clones always refer to the same event loop.
+    /// This handle is typically passed into functions that create I/O objects
+    /// to bind them to this event loop.
+    pub(crate) fn handle(&self) -> Handle {
+        Handle {
+            inner: Arc::downgrade(&self.inner),
+        }
+    }
+
+    fn turn(&mut self, max_wait: Option<Duration>) -> io::Result<()> {
+        // Block waiting for an event to happen, peeling out how many events
+        // happened.
+        match self.inner.io.poll(&mut self.events, max_wait) {
+            Ok(_) => {}
+            Err(e) => return Err(e),
+        }
+
+        // Process all the events that came in, dispatching appropriately
+
+        for event in self.events.iter() {
+            let token = event.token();
+
+            if token == TOKEN_WAKEUP {
+                self.inner
+                    .wakeup
+                    .set_readiness(mio::Ready::empty())
+                    .unwrap();
+            } else {
+                self.dispatch(token, event.readiness());
+            }
+        }
+
+        Ok(())
+    }
+
+    fn dispatch(&self, token: mio::Token, ready: mio::Ready) {
+        let mut rd = None;
+        let mut wr = None;
+
+        let address = Address::from_usize(token.0);
+
+        let io = match self.inner.io_dispatch.get(address) {
+            Some(io) => io,
+            None => return,
+        };
+
+        if io
+            .set_readiness(address, |curr| curr | ready.as_usize())
+            .is_err()
+        {
+            // token no longer valid!
+            return;
+        }
+
+        if ready.is_writable() || platform::is_hup(ready) || platform::is_error(ready) {
+            wr = io.writer.take_waker();
+        }
+
+        if !(ready & (!mio::Ready::writable())).is_empty() {
+            rd = io.reader.take_waker();
+        }
+
+        if let Some(w) = rd {
+            w.wake();
+        }
+
+        if let Some(w) = wr {
+            w.wake();
+        }
+    }
+}
+
+impl Park for Driver {
+    type Unpark = Handle;
+    type Error = io::Error;
+
+    fn unpark(&self) -> Self::Unpark {
+        self.handle()
+    }
+
+    fn park(&mut self) -> io::Result<()> {
+        self.turn(None)?;
+        Ok(())
+    }
+
+    fn park_timeout(&mut self, duration: Duration) -> io::Result<()> {
+        self.turn(Some(duration))?;
+        Ok(())
+    }
+}
+
+impl fmt::Debug for Driver {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Driver")
+    }
+}
+
+// ===== impl Handle =====
+
+impl Handle {
+    /// Returns a handle to the current reactor
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is no current reactor set.
+    pub(super) fn current() -> Self {
+        context::io_handle()
+            .expect("there is no reactor running, must be called from the context of Tokio runtime")
+    }
+
+    /// Forces a reactor blocked in a call to `turn` to wakeup, or otherwise
+    /// makes the next call to `turn` return immediately.
+    ///
+    /// This method is intended to be used in situations where a notification
+    /// needs to otherwise be sent to the main reactor. If the reactor is
+    /// currently blocked inside of `turn` then it will wake up and soon return
+    /// after this method has been called. If the reactor is not currently
+    /// blocked in `turn`, then the next call to `turn` will not block and
+    /// return immediately.
+    fn wakeup(&self) {
+        if let Some(inner) = self.inner() {
+            inner.wakeup.set_readiness(mio::Ready::readable()).unwrap();
+        }
+    }
+
+    pub(super) fn inner(&self) -> Option<Arc<Inner>> {
+        self.inner.upgrade()
+    }
+}
+
+impl Unpark for Handle {
+    fn unpark(&self) {
+        self.wakeup();
+    }
+}
+
+impl fmt::Debug for Handle {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Handle")
+    }
+}
+
+// ===== impl Inner =====
+
+impl Inner {
+    /// Registers an I/O resource with the reactor.
+    ///
+    /// The registration token is returned.
+    pub(super) fn add_source(&self, source: &dyn Evented) -> io::Result<Address> {
+        let address = self.io_dispatch.alloc().ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::Other,
+                "reactor at max registered I/O resources",
+            )
+        })?;
+
+        self.n_sources.fetch_add(1, SeqCst);
+
+        self.io.register(
+            source,
+            mio::Token(address.to_usize()),
+            mio::Ready::all(),
+            mio::PollOpt::edge(),
+        )?;
+
+        Ok(address)
+    }
+
+    /// Deregisters an I/O resource from the reactor.
+    pub(super) fn deregister_source(&self, source: &dyn Evented) -> io::Result<()> {
+        self.io.deregister(source)
+    }
+
+    pub(super) fn drop_source(&self, address: Address) {
+        self.io_dispatch.remove(address);
+        self.n_sources.fetch_sub(1, SeqCst);
+    }
+
+    /// Registers interest in the I/O resource associated with `token`.
+    pub(super) fn register(&self, token: Address, dir: Direction, w: Waker) {
+        let sched = self
+            .io_dispatch
+            .get(token)
+            .unwrap_or_else(|| panic!("IO resource for token {:?} does not exist!", token));
+
+        let waker = match dir {
+            Direction::Read => &sched.reader,
+            Direction::Write => &sched.writer,
+        };
+
+        waker.register(w);
+    }
+}
+
+impl Direction {
+    pub(super) fn mask(self) -> mio::Ready {
+        match self {
+            Direction::Read => {
+                // Everything except writable is signaled through read.
+                mio::Ready::all() - mio::Ready::writable()
+            }
+            Direction::Write => mio::Ready::writable() | platform::hup() | platform::error(),
+        }
+    }
+}
+
+#[cfg(all(test, loom))]
+mod tests {
+    use super::*;
+    use loom::thread;
+
+    // No-op `Evented` impl just so we can have something to pass to `add_source`.
+    struct NotEvented;
+
+    impl Evented for NotEvented {
+        fn register(
+            &self,
+            _: &mio::Poll,
+            _: mio::Token,
+            _: mio::Ready,
+            _: mio::PollOpt,
+        ) -> io::Result<()> {
+            Ok(())
+        }
+
+        fn reregister(
+            &self,
+            _: &mio::Poll,
+            _: mio::Token,
+            _: mio::Ready,
+            _: mio::PollOpt,
+        ) -> io::Result<()> {
+            Ok(())
+        }
+
+        fn deregister(&self, _: &mio::Poll) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn tokens_unique_when_dropped() {
+        loom::model(|| {
+            let reactor = Driver::new().unwrap();
+            let inner = reactor.inner;
+            let inner2 = inner.clone();
+
+            let token_1 = inner.add_source(&NotEvented).unwrap();
+            let thread = thread::spawn(move || {
+                inner2.drop_source(token_1);
+            });
+
+            let token_2 = inner.add_source(&NotEvented).unwrap();
+            thread.join().unwrap();
+
+            assert!(token_1 != token_2);
+        })
+    }
+
+    #[test]
+    fn tokens_unique_when_dropped_on_full_page() {
+        loom::model(|| {
+            let reactor = Driver::new().unwrap();
+            let inner = reactor.inner;
+            let inner2 = inner.clone();
+            // add sources to fill up the first page so that the dropped index
+            // may be reused.
+            for _ in 0..31 {
+                inner.add_source(&NotEvented).unwrap();
+            }
+
+            let token_1 = inner.add_source(&NotEvented).unwrap();
+            let thread = thread::spawn(move || {
+                inner2.drop_source(token_1);
+            });
+
+            let token_2 = inner.add_source(&NotEvented).unwrap();
+            thread.join().unwrap();
+
+            assert!(token_1 != token_2);
+        })
+    }
+
+    #[test]
+    fn tokens_unique_concurrent_add() {
+        loom::model(|| {
+            let reactor = Driver::new().unwrap();
+            let inner = reactor.inner;
+            let inner2 = inner.clone();
+
+            let thread = thread::spawn(move || {
+                let token_2 = inner2.add_source(&NotEvented).unwrap();
+                token_2
+            });
+
+            let token_1 = inner.add_source(&NotEvented).unwrap();
+            let token_2 = thread.join().unwrap();
+
+            assert!(token_1 != token_2);
+        })
+    }
+}
diff --git a/third_party/rust/tokio/src/io/driver/platform.rs b/third_party/rust/tokio/src/io/driver/platform.rs
new file mode 100644
index 0000000000..6b27988ce6
--- /dev/null
+++ b/third_party/rust/tokio/src/io/driver/platform.rs
@@ -0,0 +1,44 @@
+pub(crate) use self::sys::*;
+
+#[cfg(unix)]
+mod sys {
+    use mio::unix::UnixReady;
+    use mio::Ready;
+
+    pub(crate) fn hup() -> Ready {
+        UnixReady::hup().into()
+    }
+
+    pub(crate) fn is_hup(ready: Ready) -> bool {
+        UnixReady::from(ready).is_hup()
+    }
+
+    pub(crate) fn error() -> Ready {
+        UnixReady::error().into()
+    }
+
+    pub(crate) fn is_error(ready: Ready) -> bool {
+        UnixReady::from(ready).is_error()
+    }
+}
+
+#[cfg(windows)]
+mod sys {
+    use mio::Ready;
+
+    pub(crate) fn hup() -> Ready {
+        Ready::empty()
+    }
+
+    pub(crate) fn is_hup(_: Ready) -> bool {
+        false
+    }
+
+    pub(crate) fn error() -> Ready {
+        Ready::empty()
+    }
+
+    pub(crate) fn is_error(_: Ready) -> bool {
+        false
+    }
+}
diff --git a/third_party/rust/tokio/src/io/driver/scheduled_io.rs b/third_party/rust/tokio/src/io/driver/scheduled_io.rs
new file mode 100644
index 0000000000..7f6446e3f5
--- /dev/null
+++ b/third_party/rust/tokio/src/io/driver/scheduled_io.rs
@@ -0,0 +1,141 @@
+use crate::loom::future::AtomicWaker;
+use crate::loom::sync::atomic::AtomicUsize;
+use crate::util::bit;
+use crate::util::slab::{Address, Entry, Generation};
+
+use std::sync::atomic::Ordering::{AcqRel, Acquire, SeqCst};
+
+#[derive(Debug)]
+pub(crate) struct ScheduledIo {
+    readiness: AtomicUsize,
+    pub(crate) reader: AtomicWaker,
+    pub(crate) writer: AtomicWaker,
+}
+
+const PACK: bit::Pack = bit::Pack::most_significant(Generation::WIDTH);
+
+impl Entry for ScheduledIo {
+    fn generation(&self) -> Generation {
+        unpack_generation(self.readiness.load(SeqCst))
+    }
+
+    fn reset(&self, generation: Generation) -> bool {
+        let mut current = self.readiness.load(Acquire);
+
+        loop {
+            if unpack_generation(current) != generation {
+                return false;
+            }
+
+            let next = PACK.pack(generation.next().to_usize(), 0);
+
+            match self
+                .readiness
+                .compare_exchange(current, next, AcqRel, Acquire)
+            {
+                Ok(_) => break,
+                Err(actual) => current = actual,
+            }
+        }
+
+        drop(self.reader.take_waker());
+        drop(self.writer.take_waker());
+
+        true
+    }
+}
+
+impl Default for ScheduledIo {
+    fn default() -> ScheduledIo {
+        ScheduledIo {
+            readiness: AtomicUsize::new(0),
+            reader: AtomicWaker::new(),
+            writer: AtomicWaker::new(),
+        }
+    }
+}
+
+impl ScheduledIo {
+    #[cfg(all(test, loom))]
+    /// Returns the current readiness value of this `ScheduledIo`, if the
+    /// provided `token` is still a valid access.
+    ///
+    /// # Returns
+    ///
+    /// If the given token's generation no longer matches the `ScheduledIo`'s
+    /// generation, then the corresponding IO resource has been removed and
+    /// replaced with a new resource. In that case, this method returns `None`.
+    /// Otherwise, this returns the current readiness.
+    pub(crate) fn get_readiness(&self, address: Address) -> Option<usize> {
+        let ready = self.readiness.load(Acquire);
+
+        if unpack_generation(ready) != address.generation() {
+            return None;
+        }
+
+        Some(ready & !PACK.mask())
+    }
+
+    /// Sets the readiness on this `ScheduledIo` by invoking the given closure on
+    /// the current value, returning the previous readiness value.
+    ///
+    /// # Arguments
+    /// - `token`: the token for this `ScheduledIo`.
+    /// - `f`: a closure returning a new readiness value given the previous
+    ///   readiness.
+    ///
+    /// # Returns
+    ///
+    /// If the given token's generation no longer matches the `ScheduledIo`'s
+    /// generation, then the corresponding IO resource has been removed and
+    /// replaced with a new resource. In that case, this method returns `Err`.
+    /// Otherwise, this returns the previous readiness.
+    pub(crate) fn set_readiness(
+        &self,
+        address: Address,
+        f: impl Fn(usize) -> usize,
+    ) -> Result<usize, ()> {
+        let generation = address.generation();
+
+        let mut current = self.readiness.load(Acquire);
+
+        loop {
+            // Check that the generation for this access is still the current
+            // one.
+            if unpack_generation(current) != generation {
+                return Err(());
+            }
+            // Mask out the generation bits so that the modifying function
+            // doesn't see them.
+            let current_readiness = current & mio::Ready::all().as_usize();
+            let new = f(current_readiness);
+
+            debug_assert!(
+                new <= !PACK.max_value(),
+                "new readiness value would overwrite generation bits!"
+            );
+
+            match self.readiness.compare_exchange(
+                current,
+                PACK.pack(generation.to_usize(), new),
+                AcqRel,
+                Acquire,
+            ) {
+                Ok(_) => return Ok(current),
+                // we lost the race, retry!
+                Err(actual) => current = actual,
+            }
+        }
+    }
+}
+
+impl Drop for ScheduledIo {
+    fn drop(&mut self) {
+        self.writer.wake();
+        self.reader.wake();
+    }
+}
+
+fn unpack_generation(src: usize) -> Generation {
+    Generation::new(PACK.unpack(src))
+}
diff --git a/third_party/rust/tokio/src/io/mod.rs b/third_party/rust/tokio/src/io/mod.rs
new file mode 100644
index 0000000000..29d8bc5554
--- /dev/null
+++ b/third_party/rust/tokio/src/io/mod.rs
@@ -0,0 +1,229 @@
+#![cfg_attr(loom, allow(dead_code, unreachable_pub))]
+
+//! Traits, helpers, and type definitions for asynchronous I/O functionality.
+//!
+//! This module is the asynchronous version of `std::io`. Primarily, it
+//! defines two traits, [`AsyncRead`] and [`AsyncWrite`], which are asynchronous
+//! versions of the [`Read`] and [`Write`] traits in the standard library.
+//!
+//! # AsyncRead and AsyncWrite
+//!
+//! Like the standard library's [`Read`] and [`Write`] traits, [`AsyncRead`] and
+//! [`AsyncWrite`] provide the most general interface for reading and writing
+//! input and output. Unlike the standard library's traits, however, they are
+//! _asynchronous_ &mdash; meaning that reading from or writing to a `tokio::io`
+//! type will _yield_ to the Tokio scheduler when IO is not ready, rather than
+//! blocking. This allows other tasks to run while waiting on IO.
+//!
+//! Another difference is that [`AsyncRead`] and [`AsyncWrite`] only contain
+//! core methods needed to provide asynchronous reading and writing
+//! functionality. Instead, utility methods are defined in the [`AsyncReadExt`]
+//! and [`AsyncWriteExt`] extension traits. These traits are automatically
+//! implemented for all values that implement [`AsyncRead`] and [`AsyncWrite`]
+//! respectively.
+//!
+//! End users will rarely interact directly with [`AsyncRead`] and
+//! [`AsyncWrite`]. Instead, they will use the async functions defined in the
+//! extension traits. Library authors are expected to implement [`AsyncRead`]
+//! and [`AsyncWrite`] in order to provide types that behave like byte streams.
+//!
+//! Even with these differences, Tokio's [`AsyncRead`] and [`AsyncWrite`] traits
+//! can be used in almost exactly the same manner as the standard library's
+//! `Read` and `Write`. Most types in the standard library that implement `Read`
+//! and `Write` have asynchronous equivalents in `tokio` that implement
+//! `AsyncRead` and `AsyncWrite`, such as [`File`] and [`TcpStream`].
+//!
+//! For example, the standard library documentation introduces `Read` by
+//! [demonstrating][std_example] reading some bytes from a [`std::fs::File`]. We
+//! can do the same with [`tokio::fs::File`][`File`]:
+//!
+//! ```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 = [0; 10];
+//!
+//!     // read up to 10 bytes
+//!     let n = f.read(&mut buffer).await?;
+//!
+//!     println!("The bytes: {:?}", &buffer[..n]);
+//!     Ok(())
+//! }
+//! ```
+//!
+//! [`File`]: crate::fs::File
+//! [`TcpStream`]: crate::net::TcpStream
+//! [`std::fs::File`]: std::fs::File
+//! [std_example]: https://doc.rust-lang.org/std/io/index.html#read-and-write
+//!
+//! ## Buffered Readers and Writers
+//!
+//! Byte-based interfaces are unwieldy and can be inefficient, as we'd need to be
+//! making near-constant calls to the operating system. To help with this,
+//! `std::io` comes with [support for _buffered_ readers and writers][stdbuf],
+//! and therefore, `tokio::io` does as well.
+//!
+//! Tokio provides an async version of the [`std::io::BufRead`] trait,
+//! [`AsyncBufRead`]; and async [`BufReader`] and [`BufWriter`] structs, which
+//! wrap readers and writers. These wrappers use a buffer, reducing the number
+//! of calls and providing nicer methods for accessing exactly what you want.
+//!
+//! For example, [`BufReader`] works with the [`AsyncBufRead`] trait to add
+//! extra methods to any async reader:
+//!
+//! ```no_run
+//! use tokio::io::{self, BufReader, AsyncBufReadExt};
+//! use tokio::fs::File;
+//!
+//! #[tokio::main]
+//! async fn main() -> io::Result<()> {
+//!     let f = File::open("foo.txt").await?;
+//!     let mut reader = BufReader::new(f);
+//!     let mut buffer = String::new();
+//!
+//!     // read a line into buffer
+//!     reader.read_line(&mut buffer).await?;
+//!
+//!     println!("{}", buffer);
+//!     Ok(())
+//! }
+//! ```
+//!
+//! [`BufWriter`] doesn't add any new ways of writing; it just buffers every call
+//! to [`write`](crate::io::AsyncWriteExt::write):
+//!
+//! ```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 writer = BufWriter::new(f);
+//!
+//!         // write a byte to the buffer
+//!         writer.write(&[42u8]).await?;
+//!
+//!     } // the buffer is flushed once writer goes out of scope
+//!
+//!     Ok(())
+//! }
+//! ```
+//!
+//! [stdbuf]: https://doc.rust-lang.org/std/io/index.html#bufreader-and-bufwriter
+//! [`std::io::BufRead`]: std::io::BufRead
+//! [`AsyncBufRead`]: crate::io::AsyncBufRead
+//! [`BufReader`]: crate::io::BufReader
+//! [`BufWriter`]: crate::io::BufWriter
+//!
+//! ## Implementing AsyncRead and AsyncWrite
+//!
+//! Because they are traits, we can implement `AsyncRead` and `AsyncWrite` for
+//! our own types, as well. Note that these traits must only be implemented for
+//! non-blocking I/O types that integrate with the futures type system. In
+//! other words, these types must never block the thread, and instead the
+//! current task is notified when the I/O resource is ready.
+//!
+//! # Standard input and output
+//!
+//! Tokio provides asynchronous APIs to standard [input], [output], and [error].
+//! These APIs are very similar to the ones provided by `std`, but they also
+//! implement [`AsyncRead`] and [`AsyncWrite`].
+//!
+//! Note that the standard input / output APIs  **must** be used from the
+//! context of the Tokio runtime, as they require Tokio-specific features to
+//! function. Calling these functions outside of a Tokio runtime will panic.
+//!
+//! [input]: fn@stdin
+//! [output]: fn@stdout
+//! [error]: fn@stderr
+//!
+//! # `std` re-exports
+//!
+//! Additionally, [`Error`], [`ErrorKind`], and [`Result`] are re-exported
+//! from `std::io` for ease of use.
+//!
+//! [`AsyncRead`]: trait@AsyncRead
+//! [`AsyncWrite`]: trait@AsyncWrite
+//! [`Error`]: struct@Error
+//! [`ErrorKind`]: enum@ErrorKind
+//! [`Result`]: type@Result
+//! [`Read`]: std::io::Read
+//! [`Write`]: std::io::Write
+cfg_io_blocking! {
+    pub(crate) mod blocking;
+}
+
+mod async_buf_read;
+pub use self::async_buf_read::AsyncBufRead;
+
+mod async_read;
+pub use self::async_read::AsyncRead;
+
+mod async_seek;
+pub use self::async_seek::AsyncSeek;
+
+mod async_write;
+pub use self::async_write::AsyncWrite;
+
+cfg_io_driver! {
+    pub(crate) mod driver;
+
+    mod poll_evented;
+    pub use poll_evented::PollEvented;
+
+    mod registration;
+    pub use registration::Registration;
+}
+
+cfg_io_std! {
+    mod stderr;
+    pub use stderr::{stderr, Stderr};
+
+    mod stdin;
+    pub use stdin::{stdin, Stdin};
+
+    mod stdout;
+    pub use stdout::{stdout, Stdout};
+}
+
+cfg_io_util! {
+    mod split;
+    pub use split::{split, ReadHalf, WriteHalf};
+
+    pub(crate) mod seek;
+    pub use self::seek::Seek;
+
+    pub(crate) mod util;
+    pub use util::{
+        copy, empty, repeat, sink, AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt,
+        BufReader, BufStream, BufWriter, Copy, Empty, Lines, Repeat, Sink, Split, Take,
+    };
+
+    cfg_stream! {
+        pub use util::{stream_reader, StreamReader};
+    }
+
+    // Re-export io::Error so that users don't have to deal with conflicts when
+    // `use`ing `tokio::io` and `std::io`.
+    pub use std::io::{Error, ErrorKind, Result};
+}
+
+cfg_not_io_util! {
+    cfg_process! {
+        pub(crate) mod util;
+    }
+}
+
+cfg_io_blocking! {
+    /// Types in this module can be mocked out in tests.
+    mod sys {
+        // TODO: don't rename
+        pub(crate) use crate::runtime::spawn_blocking as run;
+        pub(crate) use crate::task::JoinHandle as Blocking;
+    }
+}
diff --git a/third_party/rust/tokio/src/io/poll_evented.rs b/third_party/rust/tokio/src/io/poll_evented.rs
new file mode 100644
index 0000000000..298e6e58cf
--- /dev/null
+++ b/third_party/rust/tokio/src/io/poll_evented.rs
@@ -0,0 +1,423 @@
+use crate::io::driver::platform;
+use crate::io::{AsyncRead, AsyncWrite, Registration};
+
+use mio::event::Evented;
+use std::fmt;
+use std::io::{self, Read, Write};
+use std::marker::Unpin;
+use std::pin::Pin;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering::Relaxed;
+use std::task::{Context, Poll};
+
+cfg_io_driver! {
+    /// Associates an I/O resource that implements the [`std::io::Read`] and/or
+    /// [`std::io::Write`] traits with the reactor that drives it.
+    ///
+    /// `PollEvented` uses [`Registration`] internally to take a type that
+    /// implements [`mio::Evented`] as well as [`std::io::Read`] and or
+    /// [`std::io::Write`] and associate it with a reactor that will drive it.
+    ///
+    /// Once the [`mio::Evented`] type is wrapped by `PollEvented`, it can be
+    /// used from within the future's execution model. As such, the
+    /// `PollEvented` type provides [`AsyncRead`] and [`AsyncWrite`]
+    /// implementations using the underlying I/O resource as well as readiness
+    /// events provided by the reactor.
+    ///
+    /// **Note**: While `PollEvented` is `Sync` (if the underlying I/O type is
+    /// `Sync`), the caller must ensure that there are at most two tasks that
+    /// use a `PollEvented` instance concurrently. One for reading and one for
+    /// writing. While violating this requirement is "safe" from a Rust memory
+    /// model point of view, it will result in unexpected behavior in the form
+    /// of lost notifications and tasks hanging.
+    ///
+    /// ## Readiness events
+    ///
+    /// Besides just providing [`AsyncRead`] and [`AsyncWrite`] implementations,
+    /// this type also supports access to the underlying readiness event stream.
+    /// While similar in function to what [`Registration`] provides, the
+    /// semantics are a bit different.
+    ///
+    /// Two functions are provided to access the readiness events:
+    /// [`poll_read_ready`] and [`poll_write_ready`]. These functions return the
+    /// current readiness state of the `PollEvented` instance. If
+    /// [`poll_read_ready`] indicates read readiness, immediately calling
+    /// [`poll_read_ready`] again will also indicate read readiness.
+    ///
+    /// When the operation is attempted and is unable to succeed due to the I/O
+    /// resource not being ready, the caller must call [`clear_read_ready`] or
+    /// [`clear_write_ready`]. This clears the readiness state until a new
+    /// readiness event is received.
+    ///
+    /// This allows the caller to implement additional functions. For example,
+    /// [`TcpListener`] implements poll_accept by using [`poll_read_ready`] and
+    /// [`clear_read_ready`].
+    ///
+    /// ```rust
+    /// use tokio::io::PollEvented;
+    ///
+    /// use futures::ready;
+    /// use mio::Ready;
+    /// use mio::net::{TcpStream, TcpListener};
+    /// use std::io;
+    /// use std::task::{Context, Poll};
+    ///
+    /// struct MyListener {
+    ///     poll_evented: PollEvented<TcpListener>,
+    /// }
+    ///
+    /// impl MyListener {
+    ///     pub fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<TcpStream, io::Error>> {
+    ///         let ready = Ready::readable();
+    ///
+    ///         ready!(self.poll_evented.poll_read_ready(cx, ready))?;
+    ///
+    ///         match self.poll_evented.get_ref().accept() {
+    ///             Ok((socket, _)) => Poll::Ready(Ok(socket)),
+    ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+    ///                 self.poll_evented.clear_read_ready(cx, ready)?;
+    ///                 Poll::Pending
+    ///             }
+    ///             Err(e) => Poll::Ready(Err(e)),
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    ///
+    /// ## Platform-specific events
+    ///
+    /// `PollEvented` also allows receiving platform-specific `mio::Ready` events.
+    /// These events are included as part of the read readiness event stream. The
+    /// write readiness event stream is only for `Ready::writable()` events.
+    ///
+    /// [`std::io::Read`]: https://doc.rust-lang.org/std/io/trait.Read.html
+    /// [`std::io::Write`]: https://doc.rust-lang.org/std/io/trait.Write.html
+    /// [`AsyncRead`]: ../io/trait.AsyncRead.html
+    /// [`AsyncWrite`]: ../io/trait.AsyncWrite.html
+    /// [`mio::Evented`]: https://docs.rs/mio/0.6/mio/trait.Evented.html
+    /// [`Registration`]: struct@Registration
+    /// [`TcpListener`]: ../net/struct.TcpListener.html
+    /// [`clear_read_ready`]: #method.clear_read_ready
+    /// [`clear_write_ready`]: #method.clear_write_ready
+    /// [`poll_read_ready`]: #method.poll_read_ready
+    /// [`poll_write_ready`]: #method.poll_write_ready
+    pub struct PollEvented<E: Evented> {
+        io: Option<E>,
+        inner: Inner,
+    }
+}
+
+struct Inner {
+    registration: Registration,
+
+    /// Currently visible read readiness
+    read_readiness: AtomicUsize,
+
+    /// Currently visible write readiness
+    write_readiness: AtomicUsize,
+}
+
+// ===== impl PollEvented =====
+
+macro_rules! poll_ready {
+    ($me:expr, $mask:expr, $cache:ident, $take:ident, $poll:expr) => {{
+        // Load cached & encoded readiness.
+        let mut cached = $me.inner.$cache.load(Relaxed);
+        let mask = $mask | platform::hup() | platform::error();
+
+        // See if the current readiness matches any bits.
+        let mut ret = mio::Ready::from_usize(cached) & $mask;
+
+        if ret.is_empty() {
+            // Readiness does not match, consume the registration's readiness
+            // stream. This happens in a loop to ensure that the stream gets
+            // drained.
+            loop {
+                let ready = match $poll? {
+                    Poll::Ready(v) => v,
+                    Poll::Pending => return Poll::Pending,
+                };
+                cached |= ready.as_usize();
+
+                // Update the cache store
+                $me.inner.$cache.store(cached, Relaxed);
+
+                ret |= ready & mask;
+
+                if !ret.is_empty() {
+                    return Poll::Ready(Ok(ret));
+                }
+            }
+        } else {
+            // Check what's new with the registration stream. This will not
+            // request to be notified
+            if let Some(ready) = $me.inner.registration.$take()? {
+                cached |= ready.as_usize();
+                $me.inner.$cache.store(cached, Relaxed);
+            }
+
+            Poll::Ready(Ok(mio::Ready::from_usize(cached)))
+        }
+    }};
+}
+
+impl<E> PollEvented<E>
+where
+    E: Evented,
+{
+    /// Creates a new `PollEvented` associated with the default reactor.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if thread-local runtime is not set.
+    ///
+    /// The runtime is usually set implicitly when this function is called
+    /// from a future driven by a tokio runtime, otherwise runtime can be set
+    /// explicitly with [`Handle::enter`](crate::runtime::Handle::enter) function.
+    pub fn new(io: E) -> io::Result<Self> {
+        let registration = Registration::new(&io)?;
+        Ok(Self {
+            io: Some(io),
+            inner: Inner {
+                registration,
+                read_readiness: AtomicUsize::new(0),
+                write_readiness: AtomicUsize::new(0),
+            },
+        })
+    }
+
+    /// Returns a shared reference to the underlying I/O object this readiness
+    /// stream is wrapping.
+    pub fn get_ref(&self) -> &E {
+        self.io.as_ref().unwrap()
+    }
+
+    /// Returns a mutable reference to the underlying I/O object this readiness
+    /// stream is wrapping.
+    pub fn get_mut(&mut self) -> &mut E {
+        self.io.as_mut().unwrap()
+    }
+
+    /// Consumes self, returning the inner I/O object
+    ///
+    /// This function will deregister the I/O resource from the reactor before
+    /// returning. If the deregistration operation fails, an error is returned.
+    ///
+    /// Note that deregistering does not guarantee that the I/O resource can be
+    /// registered with a different reactor. Some I/O resource types can only be
+    /// associated with a single reactor instance for their lifetime.
+    pub fn into_inner(mut self) -> io::Result<E> {
+        let io = self.io.take().unwrap();
+        self.inner.registration.deregister(&io)?;
+        Ok(io)
+    }
+
+    /// Checks the I/O resource's read readiness state.
+    ///
+    /// The mask argument allows specifying what readiness to notify on. This
+    /// can be any value, including platform specific readiness, **except**
+    /// `writable`. HUP is always implicitly included on platforms that support
+    /// it.
+    ///
+    /// If the resource is not ready for a read then `Poll::Pending` is returned
+    /// and the current task is notified once a new event is received.
+    ///
+    /// The I/O resource will remain in a read-ready state until readiness is
+    /// cleared by calling [`clear_read_ready`].
+    ///
+    /// [`clear_read_ready`]: #method.clear_read_ready
+    ///
+    /// # Panics
+    ///
+    /// This function panics if:
+    ///
+    /// * `ready` includes writable.
+    /// * called from outside of a task context.
+    pub fn poll_read_ready(
+        &self,
+        cx: &mut Context<'_>,
+        mask: mio::Ready,
+    ) -> Poll<io::Result<mio::Ready>> {
+        assert!(!mask.is_writable(), "cannot poll for write readiness");
+        poll_ready!(
+            self,
+            mask,
+            read_readiness,
+            take_read_ready,
+            self.inner.registration.poll_read_ready(cx)
+        )
+    }
+
+    /// Clears the I/O resource's read readiness state and registers the current
+    /// task to be notified once a read readiness event is received.
+    ///
+    /// After calling this function, `poll_read_ready` will return
+    /// `Poll::Pending` until a new read readiness event has been received.
+    ///
+    /// The `mask` argument specifies the readiness bits to clear. This may not
+    /// include `writable` or `hup`.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if:
+    ///
+    /// * `ready` includes writable or HUP
+    /// * called from outside of a task context.
+    pub fn clear_read_ready(&self, cx: &mut Context<'_>, ready: mio::Ready) -> io::Result<()> {
+        // Cannot clear write readiness
+        assert!(!ready.is_writable(), "cannot clear write readiness");
+        assert!(!platform::is_hup(ready), "cannot clear HUP readiness");
+
+        self.inner
+            .read_readiness
+            .fetch_and(!ready.as_usize(), Relaxed);
+
+        if self.poll_read_ready(cx, ready)?.is_ready() {
+            // Notify the current task
+            cx.waker().wake_by_ref();
+        }
+
+        Ok(())
+    }
+
+    /// Checks the I/O resource's write readiness state.
+    ///
+    /// This always checks for writable readiness and also checks for HUP
+    /// readiness on platforms that support it.
+    ///
+    /// If the resource is not ready for a write then `Poll::Pending` is
+    /// returned and the current task is notified once a new event is received.
+    ///
+    /// The I/O resource will remain in a write-ready state until readiness is
+    /// cleared by calling [`clear_write_ready`].
+    ///
+    /// [`clear_write_ready`]: #method.clear_write_ready
+    ///
+    /// # Panics
+    ///
+    /// This function panics if:
+    ///
+    /// * `ready` contains bits besides `writable` and `hup`.
+    /// * called from outside of a task context.
+    pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
+        poll_ready!(
+            self,
+            mio::Ready::writable(),
+            write_readiness,
+            take_write_ready,
+            self.inner.registration.poll_write_ready(cx)
+        )
+    }
+
+    /// Resets the I/O resource's write readiness state and registers the current
+    /// task to be notified once a write readiness event is received.
+    ///
+    /// This only clears writable readiness. HUP (on platforms that support HUP)
+    /// cannot be cleared as it is a final state.
+    ///
+    /// After calling this function, `poll_write_ready(Ready::writable())` will
+    /// return `NotReady` until a new write readiness event has been received.
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if called from outside of a task context.
+    pub fn clear_write_ready(&self, cx: &mut Context<'_>) -> io::Result<()> {
+        let ready = mio::Ready::writable();
+
+        self.inner
+            .write_readiness
+            .fetch_and(!ready.as_usize(), Relaxed);
+
+        if self.poll_write_ready(cx)?.is_ready() {
+            // Notify the current task
+            cx.waker().wake_by_ref();
+        }
+
+        Ok(())
+    }
+}
+
+// ===== Read / Write impls =====
+
+impl<E> AsyncRead for PollEvented<E>
+where
+    E: Evented + Read + Unpin,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.poll_read_ready(cx, mio::Ready::readable()))?;
+
+        let r = (*self).get_mut().read(buf);
+
+        if is_wouldblock(&r) {
+            self.clear_read_ready(cx, mio::Ready::readable())?;
+            return Poll::Pending;
+        }
+
+        Poll::Ready(r)
+    }
+}
+
+impl<E> AsyncWrite for PollEvented<E>
+where
+    E: Evented + Write + Unpin,
+{
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        ready!(self.poll_write_ready(cx))?;
+
+        let r = (*self).get_mut().write(buf);
+
+        if is_wouldblock(&r) {
+            self.clear_write_ready(cx)?;
+            return Poll::Pending;
+        }
+
+        Poll::Ready(r)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        ready!(self.poll_write_ready(cx))?;
+
+        let r = (*self).get_mut().flush();
+
+        if is_wouldblock(&r) {
+            self.clear_write_ready(cx)?;
+            return Poll::Pending;
+        }
+
+        Poll::Ready(r)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+fn is_wouldblock<T>(r: &io::Result<T>) -> bool {
+    match *r {
+        Ok(_) => false,
+        Err(ref e) => e.kind() == io::ErrorKind::WouldBlock,
+    }
+}
+
+impl<E: Evented + fmt::Debug> fmt::Debug for PollEvented<E> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("PollEvented").field("io", &self.io).finish()
+    }
+}
+
+impl<E: Evented> Drop for PollEvented<E> {
+    fn drop(&mut self) {
+        if let Some(io) = self.io.take() {
+            // Ignore errors
+            let _ = self.inner.registration.deregister(&io);
+        }
+    }
+}
diff --git a/third_party/rust/tokio/src/io/registration.rs b/third_party/rust/tokio/src/io/registration.rs
new file mode 100644
index 0000000000..4df11999f5
--- /dev/null
+++ b/third_party/rust/tokio/src/io/registration.rs
@@ -0,0 +1,299 @@
+use crate::io::driver::{platform, Direction, Handle};
+use crate::util::slab::Address;
+
+use mio::{self, Evented};
+use std::io;
+use std::task::{Context, Poll};
+
+cfg_io_driver! {
+    /// Associates an I/O resource with the reactor instance that drives it.
+    ///
+    /// A registration represents an I/O resource registered with a Reactor such
+    /// that it will receive task notifications on readiness. This is the lowest
+    /// level API for integrating with a reactor.
+    ///
+    /// The association between an I/O resource is made by calling [`new`]. Once
+    /// the association is established, it remains established until the
+    /// registration instance is dropped.
+    ///
+    /// A registration instance represents two separate readiness streams. One
+    /// for the read readiness and one for write readiness. These streams are
+    /// independent and can be consumed from separate tasks.
+    ///
+    /// **Note**: while `Registration` is `Sync`, the caller must ensure that
+    /// there are at most two tasks that use a registration instance
+    /// concurrently. One task for [`poll_read_ready`] and one task for
+    /// [`poll_write_ready`]. While violating this requirement is "safe" from a
+    /// Rust memory safety point of view, it will result in unexpected behavior
+    /// in the form of lost notifications and tasks hanging.
+    ///
+    /// ## Platform-specific events
+    ///
+    /// `Registration` also allows receiving platform-specific `mio::Ready`
+    /// events. These events are included as part of the read readiness event
+    /// stream. The write readiness event stream is only for `Ready::writable()`
+    /// events.
+    ///
+    /// [`new`]: #method.new
+    /// [`poll_read_ready`]: #method.poll_read_ready`]
+    /// [`poll_write_ready`]: #method.poll_write_ready`]
+    #[derive(Debug)]
+    pub struct Registration {
+        handle: Handle,
+        address: Address,
+    }
+}
+
+// ===== impl Registration =====
+
+impl Registration {
+    /// Registers the I/O resource with the default reactor.
+    ///
+    /// # Return
+    ///
+    /// - `Ok` if the registration happened successfully
+    /// - `Err` if an error was encountered during registration
+    ///
+    ///
+    /// # Panics
+    ///
+    /// This function panics if thread-local runtime is not set.
+    ///
+    /// The runtime is usually set implicitly when this function is called
+    /// from a future driven by a tokio runtime, otherwise runtime can be set
+    /// explicitly with [`Handle::enter`](crate::runtime::Handle::enter) function.
+    pub fn new<T>(io: &T) -> io::Result<Registration>
+    where
+        T: Evented,
+    {
+        let handle = Handle::current();
+        let address = if let Some(inner) = handle.inner() {
+            inner.add_source(io)?
+        } else {
+            return Err(io::Error::new(
+                io::ErrorKind::Other,
+                "failed to find event loop",
+            ));
+        };
+
+        Ok(Registration { handle, address })
+    }
+
+    /// Deregisters the I/O resource from the reactor it is associated with.
+    ///
+    /// This function must be called before the I/O resource associated with the
+    /// registration is dropped.
+    ///
+    /// Note that deregistering does not guarantee that the I/O resource can be
+    /// registered with a different reactor. Some I/O resource types can only be
+    /// associated with a single reactor instance for their lifetime.
+    ///
+    /// # Return
+    ///
+    /// If the deregistration was successful, `Ok` is returned. Any calls to
+    /// `Reactor::turn` that happen after a successful call to `deregister` will
+    /// no longer result in notifications getting sent for this registration.
+    ///
+    /// `Err` is returned if an error is encountered.
+    pub fn deregister<T>(&mut self, io: &T) -> io::Result<()>
+    where
+        T: Evented,
+    {
+        let inner = match self.handle.inner() {
+            Some(inner) => inner,
+            None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
+        };
+        inner.deregister_source(io)
+    }
+
+    /// Polls for events on the I/O resource's read readiness stream.
+    ///
+    /// If the I/O resource receives a new read readiness event since the last
+    /// call to `poll_read_ready`, it is returned. If it has not, the current
+    /// task is notified once a new event is received.
+    ///
+    /// All events except `HUP` are [edge-triggered]. Once `HUP` is returned,
+    /// the function will always return `Ready(HUP)`. This should be treated as
+    /// the end of the readiness stream.
+    ///
+    /// Ensure that [`register`] has been called first.
+    ///
+    /// # Return value
+    ///
+    /// There are several possible return values:
+    ///
+    /// * `Poll::Ready(Ok(readiness))` means that the I/O resource has received
+    ///   a new readiness event. The readiness value is included.
+    ///
+    /// * `Poll::Pending` means that no new readiness events have been received
+    ///   since the last call to `poll_read_ready`.
+    ///
+    /// * `Poll::Ready(Err(err))` means that the registration has encountered an
+    ///   error. This error either represents a permanent internal error **or**
+    ///   the fact that [`register`] was not called first.
+    ///
+    /// [`register`]: #method.register
+    /// [edge-triggered]: https://docs.rs/mio/0.6/mio/struct.Poll.html#edge-triggered-and-level-triggered
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if called from outside of a task context.
+    pub fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
+        // Keep track of task budget
+        ready!(crate::coop::poll_proceed(cx));
+
+        let v = self.poll_ready(Direction::Read, Some(cx))?;
+        match v {
+            Some(v) => Poll::Ready(Ok(v)),
+            None => Poll::Pending,
+        }
+    }
+
+    /// Consume any pending read readiness event.
+    ///
+    /// This function is identical to [`poll_read_ready`] **except** that it
+    /// will not notify the current task when a new event is received. As such,
+    /// it is safe to call this function from outside of a task context.
+    ///
+    /// [`poll_read_ready`]: #method.poll_read_ready
+    pub fn take_read_ready(&self) -> io::Result<Option<mio::Ready>> {
+        self.poll_ready(Direction::Read, None)
+    }
+
+    /// Polls for events on the I/O resource's write readiness stream.
+    ///
+    /// If the I/O resource receives a new write readiness event since the last
+    /// call to `poll_write_ready`, it is returned. If it has not, the current
+    /// task is notified once a new event is received.
+    ///
+    /// All events except `HUP` are [edge-triggered]. Once `HUP` is returned,
+    /// the function will always return `Ready(HUP)`. This should be treated as
+    /// the end of the readiness stream.
+    ///
+    /// Ensure that [`register`] has been called first.
+    ///
+    /// # Return value
+    ///
+    /// There are several possible return values:
+    ///
+    /// * `Poll::Ready(Ok(readiness))` means that the I/O resource has received
+    ///   a new readiness event. The readiness value is included.
+    ///
+    /// * `Poll::Pending` means that no new readiness events have been received
+    ///   since the last call to `poll_write_ready`.
+    ///
+    /// * `Poll::Ready(Err(err))` means that the registration has encountered an
+    ///   error. This error either represents a permanent internal error **or**
+    ///   the fact that [`register`] was not called first.
+    ///
+    /// [`register`]: #method.register
+    /// [edge-triggered]: https://docs.rs/mio/0.6/mio/struct.Poll.html#edge-triggered-and-level-triggered
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if called from outside of a task context.
+    pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
+        // Keep track of task budget
+        ready!(crate::coop::poll_proceed(cx));
+
+        let v = self.poll_ready(Direction::Write, Some(cx))?;
+        match v {
+            Some(v) => Poll::Ready(Ok(v)),
+            None => Poll::Pending,
+        }
+    }
+
+    /// Consumes any pending write readiness event.
+    ///
+    /// This function is identical to [`poll_write_ready`] **except** that it
+    /// will not notify the current task when a new event is received. As such,
+    /// it is safe to call this function from outside of a task context.
+    ///
+    /// [`poll_write_ready`]: #method.poll_write_ready
+    pub fn take_write_ready(&self) -> io::Result<Option<mio::Ready>> {
+        self.poll_ready(Direction::Write, None)
+    }
+
+    /// Polls for events on the I/O resource's `direction` readiness stream.
+    ///
+    /// If called with a task context, notify the task when a new event is
+    /// received.
+    fn poll_ready(
+        &self,
+        direction: Direction,
+        cx: Option<&mut Context<'_>>,
+    ) -> io::Result<Option<mio::Ready>> {
+        let inner = match self.handle.inner() {
+            Some(inner) => inner,
+            None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
+        };
+
+        // If the task should be notified about new events, ensure that it has
+        // been registered
+        if let Some(ref cx) = cx {
+            inner.register(self.address, direction, cx.waker().clone())
+        }
+
+        let mask = direction.mask();
+        let mask_no_hup = (mask - platform::hup() - platform::error()).as_usize();
+
+        let sched = inner.io_dispatch.get(self.address).unwrap();
+
+        // This consumes the current readiness state **except** for HUP and
+        // error. HUP and error are excluded because a) they are final states
+        // and never transitition out and b) both the read AND the write
+        // directions need to be able to obvserve these states.
+        //
+        // # Platform-specific behavior
+        //
+        // HUP and error readiness are platform-specific. On epoll platforms,
+        // HUP has specific conditions that must be met by both peers of a
+        // connection in order to be triggered.
+        //
+        // On epoll platforms, `EPOLLERR` is signaled through
+        // `UnixReady::error()` and is important to be observable by both read
+        // AND write. A specific case that `EPOLLERR` occurs is when the read
+        // end of a pipe is closed. When this occurs, a peer blocked by
+        // writing to the pipe should be notified.
+        let curr_ready = sched
+            .set_readiness(self.address, |curr| curr & (!mask_no_hup))
+            .unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address));
+
+        let mut ready = mask & mio::Ready::from_usize(curr_ready);
+
+        if ready.is_empty() {
+            if let Some(cx) = cx {
+                // Update the task info
+                match direction {
+                    Direction::Read => sched.reader.register_by_ref(cx.waker()),
+                    Direction::Write => sched.writer.register_by_ref(cx.waker()),
+                }
+
+                // Try again
+                let curr_ready = sched
+                    .set_readiness(self.address, |curr| curr & (!mask_no_hup))
+                    .unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address));
+                ready = mask & mio::Ready::from_usize(curr_ready);
+            }
+        }
+
+        if ready.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(ready))
+        }
+    }
+}
+
+unsafe impl Send for Registration {}
+unsafe impl Sync for Registration {}
+
+impl Drop for Registration {
+    fn drop(&mut self) {
+        let inner = match self.handle.inner() {
+            Some(inner) => inner,
+            None => return,
+        };
+        inner.drop_source(self.address);
+    }
+}
diff --git a/third_party/rust/tokio/src/io/seek.rs b/third_party/rust/tokio/src/io/seek.rs
new file mode 100644
index 0000000000..e3b5bf6b6f
--- /dev/null
+++ b/third_party/rust/tokio/src/io/seek.rs
@@ -0,0 +1,56 @@
+use crate::io::AsyncSeek;
+use std::future::Future;
+use std::io::{self, SeekFrom};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Future for the [`seek`](crate::io::AsyncSeekExt::seek) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Seek<'a, S: ?Sized> {
+    seek: &'a mut S,
+    pos: Option<SeekFrom>,
+}
+
+pub(crate) fn seek<S>(seek: &mut S, pos: SeekFrom) -> Seek<'_, S>
+where
+    S: AsyncSeek + ?Sized + Unpin,
+{
+    Seek {
+        seek,
+        pos: Some(pos),
+    }
+}
+
+impl<S> Future for Seek<'_, S>
+where
+    S: AsyncSeek + ?Sized + Unpin,
+{
+    type Output = io::Result<u64>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = &mut *self;
+        match me.pos {
+            Some(pos) => match Pin::new(&mut me.seek).start_seek(cx, pos) {
+                Poll::Ready(Ok(())) => {
+                    me.pos = None;
+                    Pin::new(&mut me.seek).poll_complete(cx)
+                }
+                Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
+                Poll::Pending => Poll::Pending,
+            },
+            None => Pin::new(&mut me.seek).poll_complete(cx),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Seek<'_, PhantomPinned>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/split.rs b/third_party/rust/tokio/src/io/split.rs
new file mode 100644
index 0000000000..134b937a5f
--- /dev/null
+++ b/third_party/rust/tokio/src/io/split.rs
@@ -0,0 +1,195 @@
+//! Split a single value implementing `AsyncRead + AsyncWrite` into separate
+//! `AsyncRead` and `AsyncWrite` handles.
+//!
+//! To restore this read/write object from its `split::ReadHalf` and
+//! `split::WriteHalf` use `unsplit`.
+
+use crate::io::{AsyncRead, AsyncWrite};
+
+use bytes::{Buf, BufMut};
+use std::cell::UnsafeCell;
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering::{Acquire, Release};
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// The readable half of a value returned from [`split`](split()).
+    pub struct ReadHalf<T> {
+        inner: Arc<Inner<T>>,
+    }
+
+    /// The writable half of a value returned from [`split`](split()).
+    pub struct WriteHalf<T> {
+        inner: Arc<Inner<T>>,
+    }
+
+    /// Splits a single value implementing `AsyncRead + AsyncWrite` into separate
+    /// `AsyncRead` and `AsyncWrite` handles.
+    ///
+    /// To restore this read/write object from its `ReadHalf` and
+    /// `WriteHalf` use [`unsplit`](ReadHalf::unsplit()).
+    pub fn split<T>(stream: T) -> (ReadHalf<T>, WriteHalf<T>)
+    where
+        T: AsyncRead + AsyncWrite,
+    {
+        let inner = Arc::new(Inner {
+            locked: AtomicBool::new(false),
+            stream: UnsafeCell::new(stream),
+        });
+
+        let rd = ReadHalf {
+            inner: inner.clone(),
+        };
+
+        let wr = WriteHalf { inner };
+
+        (rd, wr)
+    }
+}
+
+struct Inner<T> {
+    locked: AtomicBool,
+    stream: UnsafeCell<T>,
+}
+
+struct Guard<'a, T> {
+    inner: &'a Inner<T>,
+}
+
+impl<T> ReadHalf<T> {
+    /// Checks if this `ReadHalf` and some `WriteHalf` were split from the same
+    /// stream.
+    pub fn is_pair_of(&self, other: &WriteHalf<T>) -> bool {
+        other.is_pair_of(&self)
+    }
+
+    /// Reunites with a previously split `WriteHalf`.
+    ///
+    /// # Panics
+    ///
+    /// If this `ReadHalf` and the given `WriteHalf` do not originate from the
+    /// same `split` operation this method will panic.
+    /// This can be checked ahead of time by comparing the stream ID
+    /// of the two halves.
+    pub fn unsplit(self, wr: WriteHalf<T>) -> T {
+        if self.is_pair_of(&wr) {
+            drop(wr);
+
+            let inner = Arc::try_unwrap(self.inner)
+                .ok()
+                .expect("`Arc::try_unwrap` failed");
+
+            inner.stream.into_inner()
+        } else {
+            panic!("Unrelated `split::Write` passed to `split::Read::unsplit`.")
+        }
+    }
+}
+
+impl<T> WriteHalf<T> {
+    /// Check if this `WriteHalf` and some `ReadHalf` were split from the same
+    /// stream.
+    pub fn is_pair_of(&self, other: &ReadHalf<T>) -> bool {
+        Arc::ptr_eq(&self.inner, &other.inner)
+    }
+}
+
+impl<T: AsyncRead> AsyncRead for ReadHalf<T> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        let mut inner = ready!(self.inner.poll_lock(cx));
+        inner.stream_pin().poll_read(cx, buf)
+    }
+
+    fn poll_read_buf<B: BufMut>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
+        let mut inner = ready!(self.inner.poll_lock(cx));
+        inner.stream_pin().poll_read_buf(cx, buf)
+    }
+}
+
+impl<T: AsyncWrite> AsyncWrite for WriteHalf<T> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        let mut inner = ready!(self.inner.poll_lock(cx));
+        inner.stream_pin().poll_write(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        let mut inner = ready!(self.inner.poll_lock(cx));
+        inner.stream_pin().poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        let mut inner = ready!(self.inner.poll_lock(cx));
+        inner.stream_pin().poll_shutdown(cx)
+    }
+
+    fn poll_write_buf<B: Buf>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<Result<usize, io::Error>> {
+        let mut inner = ready!(self.inner.poll_lock(cx));
+        inner.stream_pin().poll_write_buf(cx, buf)
+    }
+}
+
+impl<T> Inner<T> {
+    fn poll_lock(&self, cx: &mut Context<'_>) -> Poll<Guard<'_, T>> {
+        if !self.locked.compare_and_swap(false, true, Acquire) {
+            Poll::Ready(Guard { inner: self })
+        } else {
+            // Spin... but investigate a better strategy
+
+            std::thread::yield_now();
+            cx.waker().wake_by_ref();
+
+            Poll::Pending
+        }
+    }
+}
+
+impl<T> Guard<'_, T> {
+    fn stream_pin(&mut self) -> Pin<&mut T> {
+        // safety: the stream is pinned in `Arc` and the `Guard` ensures mutual
+        // exclusion.
+        unsafe { Pin::new_unchecked(&mut *self.inner.stream.get()) }
+    }
+}
+
+impl<T> Drop for Guard<'_, T> {
+    fn drop(&mut self) {
+        self.inner.locked.store(false, Release);
+    }
+}
+
+unsafe impl<T: Send> Send for ReadHalf<T> {}
+unsafe impl<T: Send> Send for WriteHalf<T> {}
+unsafe impl<T: Sync> Sync for ReadHalf<T> {}
+unsafe impl<T: Sync> Sync for WriteHalf<T> {}
+
+impl<T: fmt::Debug> fmt::Debug for ReadHalf<T> {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt.debug_struct("split::ReadHalf").finish()
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for WriteHalf<T> {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt.debug_struct("split::WriteHalf").finish()
+    }
+}
diff --git a/third_party/rust/tokio/src/io/stderr.rs b/third_party/rust/tokio/src/io/stderr.rs
new file mode 100644
index 0000000000..99607dc604
--- /dev/null
+++ b/third_party/rust/tokio/src/io/stderr.rs
@@ -0,0 +1,108 @@
+use crate::io::blocking::Blocking;
+use crate::io::AsyncWrite;
+
+use std::io;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+cfg_io_std! {
+    /// A handle to the standard error stream of a process.
+    ///
+    /// Concurrent writes to stderr must be executed with care: Only individual
+    /// writes to this [`AsyncWrite`] are guaranteed to be intact. In particular
+    /// you should be aware that writes using [`write_all`] are not guaranteed
+    /// to occur as a single write, so multiple threads writing data with
+    /// [`write_all`] may result in interleaved output.
+    ///
+    /// Created by the [`stderr`] function.
+    ///
+    /// [`stderr`]: stderr()
+    /// [`AsyncWrite`]: AsyncWrite
+    /// [`write_all`]: crate::io::AsyncWriteExt::write_all()
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::{self, AsyncWriteExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> io::Result<()> {
+    ///     let mut stderr = io::stdout();
+    ///     stderr.write_all(b"Print some error here.").await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    #[derive(Debug)]
+    pub struct Stderr {
+        std: Blocking<std::io::Stderr>,
+    }
+
+    /// Constructs a new handle to the standard error of the current process.
+    ///
+    /// The returned handle allows writing to standard error from the within the
+    /// Tokio runtime.
+    ///
+    /// Concurrent writes to stderr must be executed with care: Only individual
+    /// writes to this [`AsyncWrite`] are guaranteed to be intact. In particular
+    /// you should be aware that writes using [`write_all`] are not guaranteed
+    /// to occur as a single write, so multiple threads writing data with
+    /// [`write_all`] may result in interleaved output.
+    ///
+    /// [`AsyncWrite`]: AsyncWrite
+    /// [`write_all`]: crate::io::AsyncWriteExt::write_all()
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::{self, AsyncWriteExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> io::Result<()> {
+    ///     let mut stderr = io::stdout();
+    ///     stderr.write_all(b"Print some error here.").await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    pub fn stderr() -> Stderr {
+        let std = io::stderr();
+        Stderr {
+            std: Blocking::new(std),
+        }
+    }
+}
+
+#[cfg(unix)]
+impl std::os::unix::io::AsRawFd for Stderr {
+    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
+        std::io::stderr().as_raw_fd()
+    }
+}
+
+#[cfg(windows)]
+impl std::os::windows::io::AsRawHandle for Stderr {
+    fn as_raw_handle(&self) -> std::os::windows::io::RawHandle {
+        std::io::stderr().as_raw_handle()
+    }
+}
+
+impl AsyncWrite for Stderr {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.std).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        Pin::new(&mut self.std).poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), io::Error>> {
+        Pin::new(&mut self.std).poll_shutdown(cx)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/stdin.rs b/third_party/rust/tokio/src/io/stdin.rs
new file mode 100644
index 0000000000..214c4d0564
--- /dev/null
+++ b/third_party/rust/tokio/src/io/stdin.rs
@@ -0,0 +1,70 @@
+use crate::io::blocking::Blocking;
+use crate::io::AsyncRead;
+
+use std::io;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+cfg_io_std! {
+    /// A handle to the standard input stream of a process.
+    ///
+    /// The handle implements the [`AsyncRead`] trait, but beware that concurrent
+    /// reads of `Stdin` must be executed with care.
+    ///
+    /// As an additional caveat, reading from the handle may block the calling
+    /// future indefinitely if there is not enough data available. This makes this
+    /// handle unsuitable for use in any circumstance where immediate reaction to
+    /// available data is required, e.g. interactive use or when implementing a
+    /// subprocess driven by requests on the standard input.
+    ///
+    /// Created by the [`stdin`] function.
+    ///
+    /// [`stdin`]: fn@stdin
+    /// [`AsyncRead`]: trait@AsyncRead
+    #[derive(Debug)]
+    pub struct Stdin {
+        std: Blocking<std::io::Stdin>,
+    }
+
+    /// Constructs a new handle to the standard input of the current process.
+    ///
+    /// The returned handle allows reading from standard input from the within the
+    /// Tokio runtime.
+    ///
+    /// As an additional caveat, reading from the handle may block the calling
+    /// future indefinitely if there is not enough data available. This makes this
+    /// handle unsuitable for use in any circumstance where immediate reaction to
+    /// available data is required, e.g. interactive use or when implementing a
+    /// subprocess driven by requests on the standard input.
+    pub fn stdin() -> Stdin {
+        let std = io::stdin();
+        Stdin {
+            std: Blocking::new(std),
+        }
+    }
+}
+
+#[cfg(unix)]
+impl std::os::unix::io::AsRawFd for Stdin {
+    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
+        std::io::stdin().as_raw_fd()
+    }
+}
+
+#[cfg(windows)]
+impl std::os::windows::io::AsRawHandle for Stdin {
+    fn as_raw_handle(&self) -> std::os::windows::io::RawHandle {
+        std::io::stdin().as_raw_handle()
+    }
+}
+
+impl AsyncRead for Stdin {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.std).poll_read(cx, buf)
+    }
+}
diff --git a/third_party/rust/tokio/src/io/stdout.rs b/third_party/rust/tokio/src/io/stdout.rs
new file mode 100644
index 0000000000..5377993a46
--- /dev/null
+++ b/third_party/rust/tokio/src/io/stdout.rs
@@ -0,0 +1,108 @@
+use crate::io::blocking::Blocking;
+use crate::io::AsyncWrite;
+
+use std::io;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+cfg_io_std! {
+    /// A handle to the standard output stream of a process.
+    ///
+    /// Concurrent writes to stdout must be executed with care: Only individual
+    /// writes to this [`AsyncWrite`] are guaranteed to be intact. In particular
+    /// you should be aware that writes using [`write_all`] are not guaranteed
+    /// to occur as a single write, so multiple threads writing data with
+    /// [`write_all`] may result in interleaved output.
+    ///
+    /// Created by the [`stdout`] function.
+    ///
+    /// [`stdout`]: stdout()
+    /// [`AsyncWrite`]: AsyncWrite
+    /// [`write_all`]: crate::io::AsyncWriteExt::write_all()
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::{self, AsyncWriteExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> io::Result<()> {
+    ///     let mut stdout = io::stdout();
+    ///     stdout.write_all(b"Hello world!").await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    #[derive(Debug)]
+    pub struct Stdout {
+        std: Blocking<std::io::Stdout>,
+    }
+
+    /// Constructs a new handle to the standard output of the current process.
+    ///
+    /// The returned handle allows writing to standard out from the within the
+    /// Tokio runtime.
+    ///
+    /// Concurrent writes to stdout must be executed with care: Only individual
+    /// writes to this [`AsyncWrite`] are guaranteed to be intact. In particular
+    /// you should be aware that writes using [`write_all`] are not guaranteed
+    /// to occur as a single write, so multiple threads writing data with
+    /// [`write_all`] may result in interleaved output.
+    ///
+    /// [`AsyncWrite`]: AsyncWrite
+    /// [`write_all`]: crate::io::AsyncWriteExt::write_all()
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::io::{self, AsyncWriteExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> io::Result<()> {
+    ///     let mut stdout = io::stdout();
+    ///     stdout.write_all(b"Hello world!").await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    pub fn stdout() -> Stdout {
+        let std = io::stdout();
+        Stdout {
+            std: Blocking::new(std),
+        }
+    }
+}
+
+#[cfg(unix)]
+impl std::os::unix::io::AsRawFd for Stdout {
+    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
+        std::io::stdout().as_raw_fd()
+    }
+}
+
+#[cfg(windows)]
+impl std::os::windows::io::AsRawHandle for Stdout {
+    fn as_raw_handle(&self) -> std::os::windows::io::RawHandle {
+        std::io::stdout().as_raw_handle()
+    }
+}
+
+impl AsyncWrite for Stdout {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.std).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        Pin::new(&mut self.std).poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), io::Error>> {
+        Pin::new(&mut self.std).poll_shutdown(cx)
+    }
+}
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..1bfab90220
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/async_buf_read_ext.rs
@@ -0,0 +1,258 @@
+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, 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.
+        ///
+        /// # 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
+        ///
+        /// # 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
+        ///
+        /// # 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`]`<`[`Vec<u8>`]`>`. Each vector returned will *not* have
+        /// the delimiter byte at the end.
+        ///
+        /// [`io::Result`]: std::io::Result
+        /// [`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 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`]`<`[`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
+        /// [`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 tokio::stream::StreamExt;
+        ///
+        /// use std::io::Cursor;
+        ///
+        /// #[tokio::main]
+        /// async fn main() {
+        ///     let cursor = Cursor::new(b"lorem\nipsum\r\ndolor");
+        ///
+        ///     let mut lines = cursor.lines().map(|res| res.unwrap());
+        ///
+        ///     assert_eq!(lines.next().await, Some(String::from("lorem")));
+        ///     assert_eq!(lines.next().await, Some(String::from("ipsum")));
+        ///     assert_eq!(lines.next().await, Some(String::from("dolor")));
+        ///     assert_eq!(lines.next().await, 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..d4402db621
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/async_read_ext.rs
@@ -0,0 +1,807 @@
+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::{ReadI128, ReadI16, ReadI32, ReadI64, ReadI8};
+use crate::io::util::read_int::{ReadU128, ReadU16, ReadU32, ReadU64, 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 extention trait, adding utility methods to all
+    /// [`AsyncRead`] types. Callers will tend to import this trait instead of
+    /// [`AsyncRead`].
+    ///
+    /// As a convenience, this trait may be imported using the [`prelude`]:
+    ///
+    /// ```no_run
+    /// use tokio::fs::File;
+    /// use tokio::prelude::*;
+    ///
+    /// #[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
+    /// [`prelude`]: crate::prelude
+    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 function does not provide any guarantees about whether it
+        /// completes immediately or asynchronously
+        ///
+        /// 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.
+        ///
+        /// # 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 function does not provide any guarantees about whether it
+        /// completes immediately or asynchronously
+        ///
+        /// # Return
+        ///
+        /// On a successful read, the number of read bytes is returned. If the
+        /// supplied buffer is not empty and the function returns `Ok(0)` then
+        /// the source as reached an "end-of-file" event.
+        ///
+        /// # 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.
+        ///
+        /// # 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,
+            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`.
+        ///
+        /// 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.
+        ///
+        /// # 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.
+        ///
+        /// # 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_value(), 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_value(), reader.read_i128().await?);
+            ///     Ok(())
+            /// }
+            /// ```
+            fn read_i128(&mut self) -> ReadI128;
+        }
+
+        /// 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()`][read] returns `Ok(0)`.
+        ///
+        /// If successful, the total number of bytes read is returned.
+        ///
+        /// # 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`]: crate::fs::read::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`]: crate::fs::read_to_string::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()`][read] may succeed.
+        ///
+        /// # 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..c7243c7f3e
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/async_seek_ext.rs
@@ -0,0 +1,60 @@
+use crate::io::seek::{seek, Seek};
+use crate::io::AsyncSeek;
+use std::io::SeekFrom;
+
+/// An extension trait which adds utility methods to `AsyncSeek` types.
+///
+/// # Examples
+///
+/// ```
+/// use std::io::{Cursor, SeekFrom};
+/// use tokio::prelude::*;
+///
+/// #[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(())
+/// }
+/// ```
+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.
+    ///
+    /// 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::prelude::*;
+    ///
+    /// 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)
+    }
+}
+
+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..377f4ecaf8
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/async_write_ext.rs
@@ -0,0 +1,689 @@
+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_buf::{write_buf, WriteBuf};
+use crate::io::util::write_int::{WriteI128, WriteI16, WriteI32, WriteI64, WriteI8};
+use crate::io::util::write_int::{WriteU128, WriteU16, WriteU32, WriteU64, WriteU8};
+use crate::io::AsyncWrite;
+
+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 extention trait, adding utility methods to all
+    /// [`AsyncWrite`] types. Callers will tend to import this trait instead of
+    /// [`AsyncWrite`].
+    ///
+    /// As a convenience, this trait may be imported using the [`prelude`]:
+    ///
+    /// ```no_run
+    /// use tokio::prelude::*;
+    /// 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
+    /// [`prelude`]: crate::prelude
+    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.
+        ///
+        /// # 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)
+        }
+
+        /// 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` 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.
+        ///
+        /// # Examples
+        ///
+        /// [`File`] implements `Read` and [`Cursor<&[u8]>`] implements [`Buf`]:
+        ///
+        /// [`File`]: crate::fs::File
+        /// [`Buf`]: bytes::Buf
+        ///
+        /// ```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,
+            B: Buf,
+        {
+            write_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.
+        ///
+        /// # 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 buffer = File::create("foo.txt").await?;
+        ///
+        ///     buffer.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_value()).await?;
+            ///     writer.write_i64(i64::max_value()).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_value()).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;
+        }
+
+        /// 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.
+        ///
+        /// # 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..0177c0e344
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/buf_reader.rs
@@ -0,0 +1,194 @@
+use crate::io::util::DEFAULT_BUF_SIZE;
+use crate::io::{AsyncBufRead, AsyncRead, AsyncWrite};
+
+use pin_project_lite::pin_project;
+use std::io::{self, Read};
+use std::mem::MaybeUninit;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::{cmp, fmt};
+
+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,
+    }
+}
+
+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 {
+        unsafe {
+            let mut buffer = Vec::with_capacity(capacity);
+            buffer.set_len(capacity);
+
+            {
+                // Convert to MaybeUninit
+                let b = &mut *(&mut buffer[..] as *mut [u8] as *mut [MaybeUninit<u8>]);
+                inner.prepare_uninitialized_buffer(b);
+            }
+            Self {
+                inner,
+                buf: buffer.into_boxed_slice(),
+                pos: 0,
+                cap: 0,
+            }
+        }
+    }
+
+    /// 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 `BufWriter`, 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 [u8],
+    ) -> Poll<io::Result<usize>> {
+        // 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.len() >= self.buf.len() {
+            let res = ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
+            self.discard_buffer();
+            return Poll::Ready(res);
+        }
+        let mut rem = ready!(self.as_mut().poll_fill_buf(cx))?;
+        let nread = rem.read(buf)?;
+        self.consume(nread);
+        Poll::Ready(Ok(nread))
+    }
+
+    // we can't skip unconditionally because of the large buffer case in read.
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
+        self.inner.prepare_uninitialized_buffer(buf)
+    }
+}
+
+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);
+            *me.cap = ready!(me.inner.poll_read(cx, me.buf))?;
+            *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);
+    }
+}
+
+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_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..a56a4517fa
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/buf_stream.rs
@@ -0,0 +1,169 @@
+use crate::io::util::{BufReader, BufWriter};
+use crate::io::{AsyncBufRead, AsyncRead, AsyncWrite};
+
+use pin_project_lite::pin_project;
+use std::io;
+use std::mem::MaybeUninit;
+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,
+                },
+            buf: wbuf,
+            written,
+        } = b;
+
+        BufStream {
+            inner: BufReader {
+                inner: BufWriter {
+                    inner,
+                    buf: wbuf,
+                    written,
+                },
+                buf: rbuf,
+                pos,
+                cap,
+            },
+        }
+    }
+}
+
+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_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 [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.project().inner.poll_read(cx, buf)
+    }
+
+    // we can't skip unconditionally because of the large buffer case in read.
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
+        self.inner.prepare_uninitialized_buffer(buf)
+    }
+}
+
+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..efd053ebac
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/buf_writer.rs
@@ -0,0 +1,192 @@
+use crate::io::util::DEFAULT_BUF_SIZE;
+use crate::io::{AsyncBufRead, AsyncRead, AsyncWrite};
+
+use pin_project_lite::pin_project;
+use std::fmt;
+use std::io::{self, Write};
+use std::mem::MaybeUninit;
+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,
+    }
+}
+
+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,
+        }
+    }
+
+    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_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)
+    }
+}
+
+impl<W: AsyncWrite + AsyncRead> AsyncRead for BufWriter<W> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.get_pin_mut().poll_read(cx, buf)
+    }
+
+    // we can't skip unconditionally because of the large buffer case in read.
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
+        self.get_ref().prepare_uninitialized_buffer(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..bc76af341d
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/chain.rs
@@ -0,0 +1,141 @@
+use crate::io::{AsyncBufRead, AsyncRead};
+
+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 [u8],
+    ) -> Poll<io::Result<usize>> {
+        let me = self.project();
+
+        if !*me.done_first {
+            match ready!(me.first.poll_read(cx, buf)?) {
+                0 if !buf.is_empty() => *me.done_first = true,
+                n => return Poll::Ready(Ok(n)),
+            }
+        }
+        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..8e0058c1c2
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/copy.rs
@@ -0,0 +1,135 @@
+use crate::io::{AsyncRead, 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`][copy]. Please
+    /// see the documentation of `copy()` for more details.
+    ///
+    /// [copy]: copy()
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct Copy<'a, R: ?Sized, W: ?Sized> {
+        reader: &'a mut R,
+        read_done: bool,
+        writer: &'a mut W,
+        pos: usize,
+        cap: usize,
+        amt: u64,
+        buf: Box<[u8]>,
+    }
+
+    /// 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 finish with an error 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 fn copy<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> Copy<'a, R, W>
+    where
+        R: AsyncRead + Unpin + ?Sized,
+        W: AsyncWrite + Unpin + ?Sized,
+    {
+        Copy {
+            reader,
+            read_done: false,
+            writer,
+            amt: 0,
+            pos: 0,
+            cap: 0,
+            buf: Box::new([0; 2048]),
+        }
+    }
+}
+
+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>> {
+        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 n = ready!(Pin::new(&mut *me.reader).poll_read(cx, &mut me.buf))?;
+                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!(Pin::new(&mut *me.writer).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;
+                }
+            }
+
+            // 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 {
+                let me = &mut *self;
+                ready!(Pin::new(&mut *me.writer).poll_flush(cx))?;
+                return Poll::Ready(Ok(self.amt));
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Copy<'_, 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..121102c78f
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/empty.rs
@@ -0,0 +1,84 @@
+use crate::io::{AsyncBufRead, AsyncRead};
+
+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>,
+        _: &mut Context<'_>,
+        _: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(0))
+    }
+}
+
+impl AsyncBufRead for Empty {
+    #[inline]
+    fn poll_fill_buf(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        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(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Empty>();
+    }
+}
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..1465f30448
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/flush.rs
@@ -0,0 +1,47 @@
+use crate::io::AsyncWrite;
+
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// A future used to fully flush an I/O object.
+    ///
+    /// Created by the [`AsyncWriteExt::flush`] function.
+    #[derive(Debug)]
+    pub struct Flush<'a, A: ?Sized> {
+        a: &'a mut A,
+    }
+}
+
+/// 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 }
+}
+
+impl<A> Future for Flush<'_, A>
+where
+    A: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = &mut *self;
+        Pin::new(&mut *me.a).poll_flush(cx)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Flush<'_, PhantomPinned>>();
+    }
+}
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..f0e75de4b1
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/lines.rs
@@ -0,0 +1,114 @@
+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! {
+    /// Stream for the [`lines`](crate::io::AsyncBufReadExt::lines) method.
+    #[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.
+    ///
+    /// # 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
+    }
+}
+
+impl<R> Lines<R>
+where
+    R: AsyncBufRead,
+{
+    #[doc(hidden)]
+    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))?;
+
+        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::replace(me.buf, String::new()))))
+    }
+}
+
+#[cfg(feature = "stream")]
+impl<R: AsyncBufRead> crate::stream::Stream for Lines<R> {
+    type Item = io::Result<String>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Poll::Ready(match ready!(self.poll_next_line(cx)) {
+            Ok(Some(line)) => Some(Ok(line)),
+            Ok(None) => None,
+            Err(err) => Some(Err(err)),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Lines<()>>();
+    }
+}
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..c4754abf05
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/mod.rs
@@ -0,0 +1,88 @@
+#![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, Copy};
+
+    mod empty;
+    pub use empty::{empty, Empty};
+
+    mod flush;
+
+    mod lines;
+    pub use lines::Lines;
+
+    mod read;
+    mod read_buf;
+    mod read_exact;
+    mod read_int;
+    mod read_line;
+
+    mod read_to_end;
+    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;
+
+    cfg_stream! {
+        mod stream_reader;
+        pub use stream_reader::{stream_reader, StreamReader};
+    }
+
+    mod take;
+    pub use take::Take;
+
+    mod write;
+    mod write_all;
+    mod write_buf;
+    mod write_int;
+
+
+    // used by `BufReader` and `BufWriter`
+    // https://github.com/rust-lang/rust/blob/master/src/libstd/sys_common/io.rs#L1
+    const DEFAULT_BUF_SIZE: usize = 8 * 1024;
+}
+
+cfg_not_io_util! {
+    cfg_process! {
+        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..a8ca370ea8
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read.rs
@@ -0,0 +1,55 @@
+use crate::io::AsyncRead;
+
+use std::future::Future;
+use std::io;
+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 }
+}
+
+cfg_io_util! {
+    /// 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],
+    }
+}
+
+impl<R> Future for Read<'_, R>
+where
+    R: AsyncRead + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = &mut *self;
+        Pin::new(&mut *me.reader).poll_read(cx, me.buf)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Read<'_, PhantomPinned>>();
+    }
+}
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..550499b933
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_buf.rs
@@ -0,0 +1,41 @@
+use crate::io::AsyncRead;
+
+use bytes::BufMut;
+use std::future::Future;
+use std::io;
+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,
+    B: BufMut,
+{
+    ReadBuf { reader, buf }
+}
+
+cfg_io_util! {
+    /// Future returned by [`read_buf`](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,
+    }
+}
+
+impl<R, B> Future for ReadBuf<'_, R, B>
+where
+    R: AsyncRead,
+    B: BufMut,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        // safety: no data is moved from self
+        unsafe {
+            let me = self.get_unchecked_mut();
+            Pin::new_unchecked(&mut *me.reader).poll_read_buf(cx, &mut me.buf)
+        }
+    }
+}
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..d6983c9953
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_exact.rs
@@ -0,0 +1,76 @@
+use crate::io::AsyncRead;
+
+use std::future::Future;
+use std::io;
+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 [`AsyncRead::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,
+        pos: 0,
+    }
+}
+
+cfg_io_util! {
+    /// 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: &'a mut [u8],
+        pos: usize,
+    }
+}
+
+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(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        loop {
+            // if our buffer is empty, then we need to read some data to continue.
+            if self.pos < self.buf.len() {
+                let me = &mut *self;
+                let n = ready!(Pin::new(&mut *me.reader).poll_read(cx, &mut me.buf[me.pos..]))?;
+                me.pos += n;
+                if n == 0 {
+                    return Err(eof()).into();
+                }
+            }
+
+            if self.pos >= self.buf.len() {
+                return Poll::Ready(Ok(self.pos));
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadExact<'_, PhantomPinned>>();
+    }
+}
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..9dc4402f88
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_int.rs
@@ -0,0 +1,123 @@
+use crate::io::AsyncRead;
+
+use bytes::Buf;
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+use std::io::ErrorKind::UnexpectedEof;
+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)]
+            pub struct $name<R> {
+                #[pin]
+                src: R,
+                buf: [u8; $bytes],
+                read: u8,
+            }
+        }
+
+        impl<R> $name<R> {
+            pub(crate) fn new(src: R) -> Self {
+                $name {
+                    src,
+                    buf: [0; $bytes],
+                    read: 0,
+                }
+            }
+        }
+
+        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 {
+                    *me.read += match me
+                        .src
+                        .as_mut()
+                        .poll_read(cx, &mut me.buf[*me.read 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(UnexpectedEof.into()));
+                        }
+                        Poll::Ready(Ok(n)) => 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)]
+            pub struct $name<R> {
+                #[pin]
+                reader: R,
+            }
+        }
+
+        impl<R> $name<R> {
+            pub(crate) fn new(reader: R) -> $name<R> {
+                $name { reader }
+            }
+        }
+
+        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];
+                match me.reader.poll_read(cx, &mut buf[..]) {
+                    Poll::Pending => Poll::Pending,
+                    Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
+                    Poll::Ready(Ok(0)) => Poll::Ready(Err(UnexpectedEof.into())),
+                    Poll::Ready(Ok(1)) => Poll::Ready(Ok(buf[0] as $ty)),
+                    Poll::Ready(Ok(_)) => unreachable!(),
+                }
+            }
+        }
+    };
+}
+
+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);
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..c5ee597486
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_line.rs
@@ -0,0 +1,82 @@
+use crate::io::util::read_until::read_until_internal;
+use crate::io::AsyncBufRead;
+
+use std::future::Future;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::str;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// 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,
+        buf: &'a mut String,
+        bytes: Vec<u8>,
+        read: usize,
+    }
+}
+
+pub(crate) fn read_line<'a, R>(reader: &'a mut R, buf: &'a mut String) -> ReadLine<'a, R>
+where
+    R: AsyncBufRead + ?Sized + Unpin,
+{
+    ReadLine {
+        reader,
+        bytes: unsafe { mem::replace(buf.as_mut_vec(), Vec::new()) },
+        buf,
+        read: 0,
+    }
+}
+
+pub(super) fn read_line_internal<R: AsyncBufRead + ?Sized>(
+    reader: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    buf: &mut String,
+    bytes: &mut Vec<u8>,
+    read: &mut usize,
+) -> Poll<io::Result<usize>> {
+    let ret = ready!(read_until_internal(reader, cx, b'\n', bytes, read));
+    if str::from_utf8(&bytes).is_err() {
+        Poll::Ready(ret.and_then(|_| {
+            Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stream did not contain valid UTF-8",
+            ))
+        }))
+    } else {
+        debug_assert!(buf.is_empty());
+        debug_assert_eq!(*read, 0);
+        // Safety: `bytes` is a valid UTF-8 because `str::from_utf8` returned `Ok`.
+        mem::swap(unsafe { buf.as_mut_vec() }, bytes);
+        Poll::Ready(ret)
+    }
+}
+
+impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadLine<'_, R> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self {
+            reader,
+            buf,
+            bytes,
+            read,
+        } = &mut *self;
+        read_line_internal(Pin::new(reader), cx, buf, bytes, read)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadLine<'_, PhantomPinned>>();
+    }
+}
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..a2cd99bed0
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_to_end.rs
@@ -0,0 +1,113 @@
+use crate::io::AsyncRead;
+
+use std::future::Future;
+use std::io;
+use std::mem::MaybeUninit;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+#[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+pub struct ReadToEnd<'a, R: ?Sized> {
+    reader: &'a mut R,
+    buf: &'a mut Vec<u8>,
+    start_len: usize,
+}
+
+pub(crate) fn read_to_end<'a, R>(reader: &'a mut R, buf: &'a mut Vec<u8>) -> ReadToEnd<'a, R>
+where
+    R: AsyncRead + Unpin + ?Sized,
+{
+    let start_len = buf.len();
+    ReadToEnd {
+        reader,
+        buf,
+        start_len,
+    }
+}
+
+struct Guard<'a> {
+    buf: &'a mut Vec<u8>,
+    len: usize,
+}
+
+impl Drop for Guard<'_> {
+    fn drop(&mut self) {
+        unsafe {
+            self.buf.set_len(self.len);
+        }
+    }
+}
+
+// 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.
+//
+// Because we're extending the buffer with uninitialized data for trusted
+// readers, we need to make sure to truncate that if any of this panics.
+pub(super) fn read_to_end_internal<R: AsyncRead + ?Sized>(
+    mut rd: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    buf: &mut Vec<u8>,
+    start_len: usize,
+) -> Poll<io::Result<usize>> {
+    let mut g = Guard {
+        len: buf.len(),
+        buf,
+    };
+    let ret;
+    loop {
+        if g.len == g.buf.len() {
+            unsafe {
+                g.buf.reserve(32);
+                let capacity = g.buf.capacity();
+                g.buf.set_len(capacity);
+
+                let b = &mut *(&mut g.buf[g.len..] as *mut [u8] as *mut [MaybeUninit<u8>]);
+
+                rd.prepare_uninitialized_buffer(b);
+            }
+        }
+
+        match ready!(rd.as_mut().poll_read(cx, &mut g.buf[g.len..])) {
+            Ok(0) => {
+                ret = Poll::Ready(Ok(g.len - start_len));
+                break;
+            }
+            Ok(n) => g.len += n,
+            Err(e) => {
+                ret = Poll::Ready(Err(e));
+                break;
+            }
+        }
+    }
+
+    ret
+}
+
+impl<A> Future for ReadToEnd<'_, A>
+where
+    A: AsyncRead + ?Sized + Unpin,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        read_to_end_internal(Pin::new(&mut this.reader), cx, this.buf, this.start_len)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadToEnd<'_, PhantomPinned>>();
+    }
+}
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..e77d836dee
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_to_string.rs
@@ -0,0 +1,83 @@
+use crate::io::util::read_to_end::read_to_end_internal;
+use crate::io::AsyncRead;
+
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::{io, mem, str};
+
+cfg_io_util! {
+    /// 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,
+        buf: &'a mut String,
+        bytes: Vec<u8>,
+        start_len: usize,
+    }
+}
+
+pub(crate) fn read_to_string<'a, R>(reader: &'a mut R, buf: &'a mut String) -> ReadToString<'a, R>
+where
+    R: AsyncRead + ?Sized + Unpin,
+{
+    let start_len = buf.len();
+    ReadToString {
+        reader,
+        bytes: unsafe { mem::replace(buf.as_mut_vec(), Vec::new()) },
+        buf,
+        start_len,
+    }
+}
+
+fn read_to_string_internal<R: AsyncRead + ?Sized>(
+    reader: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    buf: &mut String,
+    bytes: &mut Vec<u8>,
+    start_len: usize,
+) -> Poll<io::Result<usize>> {
+    let ret = ready!(read_to_end_internal(reader, cx, bytes, start_len));
+    if str::from_utf8(&bytes).is_err() {
+        Poll::Ready(ret.and_then(|_| {
+            Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stream did not contain valid UTF-8",
+            ))
+        }))
+    } else {
+        debug_assert!(buf.is_empty());
+        // Safety: `bytes` is a valid UTF-8 because `str::from_utf8` returned `Ok`.
+        mem::swap(unsafe { buf.as_mut_vec() }, bytes);
+        Poll::Ready(ret)
+    }
+}
+
+impl<A> Future for ReadToString<'_, A>
+where
+    A: AsyncRead + ?Sized + Unpin,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self {
+            reader,
+            buf,
+            bytes,
+            start_len,
+        } = &mut *self;
+        read_to_string_internal(Pin::new(reader), cx, buf, bytes, *start_len)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadToString<'_, PhantomPinned>>();
+    }
+}
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..1adeda66f0
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/read_until.rs
@@ -0,0 +1,86 @@
+use crate::io::AsyncBufRead;
+
+use std::future::Future;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// Future for the [`read_until`](crate::io::AsyncBufReadExt::read_until) method.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct ReadUntil<'a, R: ?Sized> {
+        reader: &'a mut R,
+        byte: u8,
+        buf: &'a mut Vec<u8>,
+        read: usize,
+    }
+}
+
+pub(crate) fn read_until<'a, R>(
+    reader: &'a mut R,
+    byte: u8,
+    buf: &'a mut Vec<u8>,
+) -> ReadUntil<'a, R>
+where
+    R: AsyncBufRead + ?Sized + Unpin,
+{
+    ReadUntil {
+        reader,
+        byte,
+        buf,
+        read: 0,
+    }
+}
+
+pub(super) fn read_until_internal<R: AsyncBufRead + ?Sized>(
+    mut reader: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    byte: 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(byte, 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(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self {
+            reader,
+            byte,
+            buf,
+            read,
+        } = &mut *self;
+        read_until_internal(Pin::new(reader), cx, *byte, buf, read)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadUntil<'_, PhantomPinned>>();
+    }
+}
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..6b9067e853
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/repeat.rs
@@ -0,0 +1,71 @@
+use crate::io::AsyncRead;
+
+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 [u8],
+    ) -> Poll<io::Result<usize>> {
+        for byte in &mut *buf {
+            *byte = self.byte;
+        }
+        Poll::Ready(Ok(buf.len()))
+    }
+}
+
+#[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..f24e288541
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/shutdown.rs
@@ -0,0 +1,47 @@
+use crate::io::AsyncWrite;
+
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// A future used to shutdown an I/O object.
+    ///
+    /// Created by the [`AsyncWriteExt::shutdown`] function.
+    #[derive(Debug)]
+    pub struct Shutdown<'a, A: ?Sized> {
+        a: &'a mut A,
+    }
+}
+
+/// 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 }
+}
+
+impl<A> Future for Shutdown<'_, A>
+where
+    A: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = &mut *self;
+        Pin::new(&mut *me.a).poll_shutdown(cx)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Shutdown<'_, PhantomPinned>>();
+    }
+}
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..f1ed2fd89d
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/split.rs
@@ -0,0 +1,112 @@
+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! {
+    /// Stream for the [`split`](crate::io::AsyncBufReadExt::split) method.
+    #[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,
+{
+    #[doc(hidden)]
+    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,
+        ))?;
+
+        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::replace(me.buf, Vec::new()))))
+    }
+}
+
+#[cfg(feature = "stream")]
+impl<R: AsyncBufRead> crate::stream::Stream for Split<R> {
+    type Item = io::Result<Vec<u8>>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Poll::Ready(match ready!(self.poll_next_segment(cx)) {
+            Ok(Some(segment)) => Some(Ok(segment)),
+            Ok(None) => None,
+            Err(err) => Some(Err(err)),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Split<()>>();
+    }
+}
diff --git a/third_party/rust/tokio/src/io/util/stream_reader.rs b/third_party/rust/tokio/src/io/util/stream_reader.rs
new file mode 100644
index 0000000000..b98f8bdfc2
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/stream_reader.rs
@@ -0,0 +1,184 @@
+use crate::io::{AsyncBufRead, AsyncRead};
+use crate::stream::Stream;
+use bytes::{Buf, BufMut};
+use pin_project_lite::pin_project;
+use std::io;
+use std::mem::MaybeUninit;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pin_project! {
+    /// Convert a stream of byte chunks into an [`AsyncRead`].
+    ///
+    /// This type is usually created using the [`stream_reader`] function.
+    ///
+    /// [`AsyncRead`]: crate::io::AsyncRead
+    /// [`stream_reader`]: crate::io::stream_reader
+    #[derive(Debug)]
+    #[cfg_attr(docsrs, doc(cfg(feature = "stream")))]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+    pub struct StreamReader<S, B> {
+        #[pin]
+        inner: S,
+        chunk: Option<B>,
+    }
+}
+
+/// Convert a stream of byte chunks into an [`AsyncRead`](crate::io::AsyncRead).
+///
+/// # Example
+///
+/// ```
+/// use bytes::Bytes;
+/// use tokio::io::{stream_reader, AsyncReadExt};
+/// # #[tokio::main]
+/// # async fn main() -> std::io::Result<()> {
+///
+/// // Create a stream from an iterator.
+/// let stream = tokio::stream::iter(vec![
+///     Ok(Bytes::from_static(&[0, 1, 2, 3])),
+///     Ok(Bytes::from_static(&[4, 5, 6, 7])),
+///     Ok(Bytes::from_static(&[8, 9, 10, 11])),
+/// ]);
+///
+/// // Convert it to an AsyncRead.
+/// let mut read = stream_reader(stream);
+///
+/// // Read five bytes from the stream.
+/// let mut buf = [0; 5];
+/// read.read_exact(&mut buf).await?;
+/// assert_eq!(buf, [0, 1, 2, 3, 4]);
+///
+/// // Read the rest of the current chunk.
+/// assert_eq!(read.read(&mut buf).await?, 3);
+/// assert_eq!(&buf[..3], [5, 6, 7]);
+///
+/// // Read the next chunk.
+/// assert_eq!(read.read(&mut buf).await?, 4);
+/// assert_eq!(&buf[..4], [8, 9, 10, 11]);
+///
+/// // We have now reached the end.
+/// assert_eq!(read.read(&mut buf).await?, 0);
+///
+/// # Ok(())
+/// # }
+/// ```
+#[cfg_attr(docsrs, doc(cfg(feature = "stream")))]
+#[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
+pub fn stream_reader<S, B>(stream: S) -> StreamReader<S, B>
+where
+    S: Stream<Item = Result<B, io::Error>>,
+    B: Buf,
+{
+    StreamReader::new(stream)
+}
+
+impl<S, B> StreamReader<S, B>
+where
+    S: Stream<Item = Result<B, io::Error>>,
+    B: Buf,
+{
+    /// Convert the provided stream into an `AsyncRead`.
+    fn new(stream: S) -> Self {
+        Self {
+            inner: stream,
+            chunk: None,
+        }
+    }
+    /// Do we have a chunk and is it non-empty?
+    fn has_chunk(self: Pin<&mut Self>) -> bool {
+        if let Some(chunk) = self.project().chunk {
+            chunk.remaining() > 0
+        } else {
+            false
+        }
+    }
+}
+
+impl<S, B> AsyncRead for StreamReader<S, B>
+where
+    S: Stream<Item = Result<B, io::Error>>,
+    B: Buf,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        if buf.is_empty() {
+            return Poll::Ready(Ok(0));
+        }
+
+        let inner_buf = match self.as_mut().poll_fill_buf(cx) {
+            Poll::Ready(Ok(buf)) => buf,
+            Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+            Poll::Pending => return Poll::Pending,
+        };
+        let len = std::cmp::min(inner_buf.len(), buf.len());
+        (&mut buf[..len]).copy_from_slice(&inner_buf[..len]);
+
+        self.consume(len);
+        Poll::Ready(Ok(len))
+    }
+    fn poll_read_buf<BM: BufMut>(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut BM,
+    ) -> Poll<io::Result<usize>>
+    where
+        Self: Sized,
+    {
+        if !buf.has_remaining_mut() {
+            return Poll::Ready(Ok(0));
+        }
+
+        let inner_buf = match self.as_mut().poll_fill_buf(cx) {
+            Poll::Ready(Ok(buf)) => buf,
+            Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+            Poll::Pending => return Poll::Pending,
+        };
+        let len = std::cmp::min(inner_buf.len(), buf.remaining_mut());
+        buf.put_slice(&inner_buf[..len]);
+
+        self.consume(len);
+        Poll::Ready(Ok(len))
+    }
+    unsafe fn prepare_uninitialized_buffer(&self, _buf: &mut [MaybeUninit<u8>]) -> bool {
+        false
+    }
+}
+
+impl<S, B> AsyncBufRead for StreamReader<S, B>
+where
+    S: Stream<Item = Result<B, io::Error>>,
+    B: Buf,
+{
+    fn poll_fill_buf(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        loop {
+            if self.as_mut().has_chunk() {
+                // This unwrap is very sad, but it can't be avoided.
+                let buf = self.project().chunk.as_ref().unwrap().bytes();
+                return Poll::Ready(Ok(buf));
+            } else {
+                match self.as_mut().project().inner.poll_next(cx) {
+                    Poll::Ready(Some(Ok(chunk))) => {
+                        // Go around the loop in case the chunk is empty.
+                        *self.as_mut().project().chunk = Some(chunk);
+                    }
+                    Poll::Ready(Some(Err(err))) => return Poll::Ready(Err(err)),
+                    Poll::Ready(None) => return Poll::Ready(Ok(&[])),
+                    Poll::Pending => return Poll::Pending,
+                }
+            }
+        }
+    }
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        if amt > 0 {
+            self.project()
+                .chunk
+                .as_mut()
+                .expect("No chunk present")
+                .advance(amt);
+        }
+    }
+}
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..5d6bd90aa3
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/take.rs
@@ -0,0 +1,131 @@
+use crate::io::{AsyncBufRead, AsyncRead};
+
+use pin_project_lite::pin_project;
+use std::mem::MaybeUninit;
+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> {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
+        self.inner.prepare_uninitialized_buffer(buf)
+    }
+
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        if self.limit_ == 0 {
+            return Poll::Ready(Ok(0));
+        }
+
+        let me = self.project();
+        let max = std::cmp::min(buf.len() as u64, *me.limit_) as usize;
+        let n = ready!(me.inner.poll_read(cx, &mut buf[..max]))?;
+        *me.limit_ -= n as u64;
+        Poll::Ready(Ok(n))
+    }
+}
+
+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/write.rs b/third_party/rust/tokio/src/io/util/write.rs
new file mode 100644
index 0000000000..433a421d34
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write.rs
@@ -0,0 +1,37 @@
+use crate::io::AsyncWrite;
+
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// 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],
+    }
+}
+
+/// 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 }
+}
+
+impl<W> Future for Write<'_, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = &mut *self;
+        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..898006c56c
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write_all.rs
@@ -0,0 +1,57 @@
+use crate::io::AsyncWrite;
+
+use std::future::Future;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    #[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],
+    }
+}
+
+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 }
+}
+
+impl<W> Future for WriteAll<'_, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let me = &mut *self;
+        while !me.buf.is_empty() {
+            let n = ready!(Pin::new(&mut me.writer).poll_write(cx, me.buf))?;
+            {
+                let (_, rest) = mem::replace(&mut me.buf, &[]).split_at(n);
+                me.buf = rest;
+            }
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<WriteAll<'_, PhantomPinned>>();
+    }
+}
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..e49282fe0c
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write_buf.rs
@@ -0,0 +1,43 @@
+use crate::io::AsyncWrite;
+
+use bytes::Buf;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_io_util! {
+    /// 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,
+    }
+}
+
+/// 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,
+    B: Buf,
+{
+    WriteBuf { writer, buf }
+}
+
+impl<W, B> Future for WriteBuf<'_, W, B>
+where
+    W: AsyncWrite,
+    B: Buf,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        // safety: no data is moved from self
+        unsafe {
+            let me = self.get_unchecked_mut();
+            Pin::new_unchecked(&mut *me.writer).poll_write_buf(cx, &mut me.buf)
+        }
+    }
+}
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..672c35f076
--- /dev/null
+++ b/third_party/rust/tokio/src/io/util/write_int.rs
@@ -0,0 +1,122 @@
+use crate::io::AsyncWrite;
+
+use bytes::BufMut;
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::io;
+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)]
+            pub struct $name<W> {
+                #[pin]
+                dst: W,
+                buf: [u8; $bytes],
+                written: u8,
+            }
+        }
+
+        impl<W> $name<W> {
+            pub(crate) fn new(w: W, value: $ty) -> Self {
+                let mut writer = $name {
+                    buf: [0; $bytes],
+                    written: 0,
+                    dst: w,
+                };
+                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)]
+            pub struct $name<W> {
+                #[pin]
+                dst: W,
+                byte: $ty,
+            }
+        }
+
+        impl<W> $name<W> {
+            pub(crate) fn new(dst: W, byte: $ty) -> Self {
+                Self { dst, byte }
+            }
+        }
+
+        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);