Adding upstream version 110.0.1.upstream/110.0.1upstream

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

33 files changed, 3795 insertions, 0 deletions

diff --git a/third_party/rust/futures-util/src/io/allow_std.rs b/third_party/rust/futures-util/src/io/allow_std.rs
new file mode 100644
index 0000000000..ec30ee31e5
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/allow_std.rs
@@ -0,0 +1,200 @@
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSlice, IoSliceMut, SeekFrom};
+use std::pin::Pin;
+use std::{fmt, io};
+
+/// A simple wrapper type which allows types which implement only
+/// implement `std::io::Read` or `std::io::Write`
+/// to be used in contexts which expect an `AsyncRead` or `AsyncWrite`.
+///
+/// If these types issue an error with the kind `io::ErrorKind::WouldBlock`,
+/// it is expected that they will notify the current task on readiness.
+/// Synchronous `std` types should not issue errors of this kind and
+/// are safe to use in this context. However, using these types with
+/// `AllowStdIo` will cause the event loop to block, so they should be used
+/// with care.
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
+pub struct AllowStdIo<T>(T);
+
+impl<T> Unpin for AllowStdIo<T> {}
+
+macro_rules! try_with_interrupt {
+    ($e:expr) => {
+        loop {
+            match $e {
+                Ok(e) => {
+                    break e;
+                }
+                Err(ref e) if e.kind() == ::std::io::ErrorKind::Interrupted => {
+                    continue;
+                }
+                Err(e) => {
+                    return Poll::Ready(Err(e));
+                }
+            }
+        }
+    };
+}
+
+impl<T> AllowStdIo<T> {
+    /// Creates a new `AllowStdIo` from an existing IO object.
+    pub fn new(io: T) -> Self {
+        Self(io)
+    }
+
+    /// Returns a reference to the contained IO object.
+    pub fn get_ref(&self) -> &T {
+        &self.0
+    }
+
+    /// Returns a mutable reference to the contained IO object.
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.0
+    }
+
+    /// Consumes self and returns the contained IO object.
+    pub fn into_inner(self) -> T {
+        self.0
+    }
+}
+
+impl<T> io::Write for AllowStdIo<T>
+where
+    T: io::Write,
+{
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.0.write(buf)
+    }
+    fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> io::Result<usize> {
+        self.0.write_vectored(bufs)
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        self.0.flush()
+    }
+    fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
+        self.0.write_all(buf)
+    }
+    fn write_fmt(&mut self, fmt: fmt::Arguments<'_>) -> io::Result<()> {
+        self.0.write_fmt(fmt)
+    }
+}
+
+impl<T> AsyncWrite for AllowStdIo<T>
+where
+    T: io::Write,
+{
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(try_with_interrupt!(self.0.write(buf))))
+    }
+
+    fn poll_write_vectored(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(try_with_interrupt!(self.0.write_vectored(bufs))))
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        try_with_interrupt!(self.0.flush());
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.poll_flush(cx)
+    }
+}
+
+impl<T> io::Read for AllowStdIo<T>
+where
+    T: io::Read,
+{
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.0.read(buf)
+    }
+    fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> io::Result<usize> {
+        self.0.read_vectored(bufs)
+    }
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.0.read_to_end(buf)
+    }
+    fn read_to_string(&mut self, buf: &mut String) -> io::Result<usize> {
+        self.0.read_to_string(buf)
+    }
+    fn read_exact(&mut self, buf: &mut [u8]) -> io::Result<()> {
+        self.0.read_exact(buf)
+    }
+}
+
+impl<T> AsyncRead for AllowStdIo<T>
+where
+    T: io::Read,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(try_with_interrupt!(self.0.read(buf))))
+    }
+
+    fn poll_read_vectored(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(try_with_interrupt!(self.0.read_vectored(bufs))))
+    }
+}
+
+impl<T> io::Seek for AllowStdIo<T>
+where
+    T: io::Seek,
+{
+    fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
+        self.0.seek(pos)
+    }
+}
+
+impl<T> AsyncSeek for AllowStdIo<T>
+where
+    T: io::Seek,
+{
+    fn poll_seek(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        pos: SeekFrom,
+    ) -> Poll<io::Result<u64>> {
+        Poll::Ready(Ok(try_with_interrupt!(self.0.seek(pos))))
+    }
+}
+
+impl<T> io::BufRead for AllowStdIo<T>
+where
+    T: io::BufRead,
+{
+    fn fill_buf(&mut self) -> io::Result<&[u8]> {
+        self.0.fill_buf()
+    }
+    fn consume(&mut self, amt: usize) {
+        self.0.consume(amt)
+    }
+}
+
+impl<T> AsyncBufRead for AllowStdIo<T>
+where
+    T: io::BufRead,
+{
+    fn poll_fill_buf(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        let this: *mut Self = &mut *self as *mut _;
+        Poll::Ready(Ok(try_with_interrupt!(unsafe { &mut *this }.0.fill_buf())))
+    }
+
+    fn consume(mut self: Pin<&mut Self>, amt: usize) {
+        self.0.consume(amt)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/buf_reader.rs b/third_party/rust/futures-util/src/io/buf_reader.rs
new file mode 100644
index 0000000000..0334a9f081
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/buf_reader.rs
@@ -0,0 +1,263 @@
+use super::DEFAULT_BUF_SIZE;
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSliceMut, SeekFrom};
+use pin_project_lite::pin_project;
+use std::io::{self, Read};
+use std::pin::Pin;
+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.
+    ///
+    /// [`AsyncRead`]: futures_io::AsyncRead
+    ///
+    // TODO: Examples
+    pub struct BufReader<R> {
+        #[pin]
+        inner: R,
+        buffer: Box<[u8]>,
+        pos: usize,
+        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);
+            super::initialize(&inner, &mut buffer);
+            Self { inner, buffer: buffer.into_boxed_slice(), pos: 0, cap: 0 }
+        }
+    }
+
+    delegate_access_inner!(inner, R, ());
+
+    /// 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.buffer[self.pos..self.cap]
+    }
+
+    /// Invalidates all data in the internal buffer.
+    #[inline]
+    fn discard_buffer(self: Pin<&mut Self>) {
+        let this = self.project();
+        *this.pos = 0;
+        *this.cap = 0;
+    }
+}
+
+impl<R: AsyncRead + AsyncSeek> BufReader<R> {
+    /// Seeks relative to the current position. If the new position lies within the buffer,
+    /// the buffer will not be flushed, allowing for more efficient seeks.
+    /// This method does not return the location of the underlying reader, so the caller
+    /// must track this information themselves if it is required.
+    pub fn seek_relative(self: Pin<&mut Self>, offset: i64) -> SeeKRelative<'_, R> {
+        SeeKRelative { inner: self, offset, first: true }
+    }
+
+    /// Attempts to seek relative to the current position. If the new position lies within the buffer,
+    /// the buffer will not be flushed, allowing for more efficient seeks.
+    /// This method does not return the location of the underlying reader, so the caller
+    /// must track this information themselves if it is required.
+    pub fn poll_seek_relative(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        offset: i64,
+    ) -> Poll<io::Result<()>> {
+        let pos = self.pos as u64;
+        if offset < 0 {
+            if let Some(new_pos) = pos.checked_sub((-offset) as u64) {
+                *self.project().pos = new_pos as usize;
+                return Poll::Ready(Ok(()));
+            }
+        } else if let Some(new_pos) = pos.checked_add(offset as u64) {
+            if new_pos <= self.cap as u64 {
+                *self.project().pos = new_pos as usize;
+                return Poll::Ready(Ok(()));
+            }
+        }
+        self.poll_seek(cx, SeekFrom::Current(offset)).map(|res| res.map(|_| ()))
+    }
+}
+
+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.buffer.len() {
+            let res = ready!(self.as_mut().project().inner.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))
+    }
+
+    fn poll_read_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+    ) -> Poll<io::Result<usize>> {
+        let total_len = bufs.iter().map(|b| b.len()).sum::<usize>();
+        if self.pos == self.cap && total_len >= self.buffer.len() {
+            let res = ready!(self.as_mut().project().inner.poll_read_vectored(cx, bufs));
+            self.discard_buffer();
+            return Poll::Ready(res);
+        }
+        let mut rem = ready!(self.as_mut().poll_fill_buf(cx))?;
+        let nread = rem.read_vectored(bufs)?;
+        self.consume(nread);
+        Poll::Ready(Ok(nread))
+    }
+}
+
+impl<R: AsyncRead> AsyncBufRead for BufReader<R> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        let this = 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 *this.pos >= *this.cap {
+            debug_assert!(*this.pos == *this.cap);
+            *this.cap = ready!(this.inner.poll_read(cx, this.buffer))?;
+            *this.pos = 0;
+        }
+        Poll::Ready(Ok(&this.buffer[*this.pos..*this.cap]))
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        *self.project().pos = cmp::min(self.pos + amt, self.cap);
+    }
+}
+
+impl<R: AsyncWrite> AsyncWrite for BufReader<R> {
+    delegate_async_write!(inner);
+}
+
+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.buffer.len()))
+            .finish()
+    }
+}
+
+impl<R: AsyncRead + AsyncSeek> AsyncSeek for BufReader<R> {
+    /// Seek to an offset, in bytes, in the underlying reader.
+    ///
+    /// The position used for seeking with `SeekFrom::Current(_)` is the
+    /// position the underlying reader would be at if the `BufReader` had no
+    /// internal buffer.
+    ///
+    /// Seeking always discards the internal buffer, even if the seek position
+    /// would otherwise fall within it. This guarantees that calling
+    /// `.into_inner()` immediately after a seek yields the underlying reader
+    /// at the same position.
+    ///
+    /// To seek without discarding the internal buffer, use
+    /// [`BufReader::seek_relative`](BufReader::seek_relative) or
+    /// [`BufReader::poll_seek_relative`](BufReader::poll_seek_relative).
+    ///
+    /// See [`AsyncSeek`](futures_io::AsyncSeek) for more details.
+    ///
+    /// Note: In the edge case where you're seeking with `SeekFrom::Current(n)`
+    /// where `n` minus the internal buffer length overflows an `i64`, two
+    /// seeks will be performed instead of one. If the second seek returns
+    /// `Err`, the underlying reader will be left at the same position it would
+    /// have if you called `seek` with `SeekFrom::Current(0)`.
+    fn poll_seek(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        pos: SeekFrom,
+    ) -> Poll<io::Result<u64>> {
+        let result: u64;
+        if let SeekFrom::Current(n) = pos {
+            let remainder = (self.cap - self.pos) as i64;
+            // it should be safe to assume that remainder fits within an i64 as the alternative
+            // means we managed to allocate 8 exbibytes and that's absurd.
+            // But it's not out of the realm of possibility for some weird underlying reader to
+            // support seeking by i64::min_value() so we need to handle underflow when subtracting
+            // remainder.
+            if let Some(offset) = n.checked_sub(remainder) {
+                result =
+                    ready!(self.as_mut().project().inner.poll_seek(cx, SeekFrom::Current(offset)))?;
+            } else {
+                // seek backwards by our remainder, and then by the offset
+                ready!(self.as_mut().project().inner.poll_seek(cx, SeekFrom::Current(-remainder)))?;
+                self.as_mut().discard_buffer();
+                result = ready!(self.as_mut().project().inner.poll_seek(cx, SeekFrom::Current(n)))?;
+            }
+        } else {
+            // Seeking with Start/End doesn't care about our buffer length.
+            result = ready!(self.as_mut().project().inner.poll_seek(cx, pos))?;
+        }
+        self.discard_buffer();
+        Poll::Ready(Ok(result))
+    }
+}
+
+/// Future for the [`BufReader::seek_relative`](self::BufReader::seek_relative) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless polled"]
+pub struct SeeKRelative<'a, R> {
+    inner: Pin<&'a mut BufReader<R>>,
+    offset: i64,
+    first: bool,
+}
+
+impl<R> Future for SeeKRelative<'_, R>
+where
+    R: AsyncRead + AsyncSeek,
+{
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let offset = self.offset;
+        if self.first {
+            self.first = false;
+            self.inner.as_mut().poll_seek_relative(cx, offset)
+        } else {
+            self.inner
+                .as_mut()
+                .as_mut()
+                .poll_seek(cx, SeekFrom::Current(offset))
+                .map(|res| res.map(|_| ()))
+        }
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/buf_writer.rs b/third_party/rust/futures-util/src/io/buf_writer.rs
new file mode 100644
index 0000000000..cb74863ad0
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/buf_writer.rs
@@ -0,0 +1,224 @@
+use super::DEFAULT_BUF_SIZE;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSlice, SeekFrom};
+use pin_project_lite::pin_project;
+use std::fmt;
+use std::io::{self, Write};
+use std::pin::Pin;
+use std::ptr;
+
+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`]: futures_io::AsyncWrite
+    /// [`flush`]: super::AsyncWriteExt::flush
+    ///
+    // TODO: Examples
+    pub struct BufWriter<W> {
+        #[pin]
+        inner: W,
+        buf: Vec<u8>,
+        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 }
+    }
+
+    pub(super) fn flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let mut this = self.project();
+
+        let len = this.buf.len();
+        let mut ret = Ok(());
+        while *this.written < len {
+            match ready!(this.inner.as_mut().poll_write(cx, &this.buf[*this.written..])) {
+                Ok(0) => {
+                    ret = Err(io::Error::new(
+                        io::ErrorKind::WriteZero,
+                        "failed to write the buffered data",
+                    ));
+                    break;
+                }
+                Ok(n) => *this.written += n,
+                Err(e) => {
+                    ret = Err(e);
+                    break;
+                }
+            }
+        }
+        if *this.written > 0 {
+            this.buf.drain(..*this.written);
+        }
+        *this.written = 0;
+        Poll::Ready(ret)
+    }
+
+    delegate_access_inner!(inner, W, ());
+
+    /// Returns a reference to the internally buffered data.
+    pub fn buffer(&self) -> &[u8] {
+        &self.buf
+    }
+
+    /// Capacity of `buf`. how many chars can be held in buffer
+    pub(super) fn capacity(&self) -> usize {
+        self.buf.capacity()
+    }
+
+    /// Remaining number of bytes to reach `buf` 's capacity
+    #[inline]
+    pub(super) fn spare_capacity(&self) -> usize {
+        self.buf.capacity() - self.buf.len()
+    }
+
+    /// Write a byte slice directly into buffer
+    ///
+    /// Will truncate the number of bytes written to `spare_capacity()` so you want to
+    /// calculate the size of your slice to avoid losing bytes
+    ///
+    /// Based on `std::io::BufWriter`
+    pub(super) fn write_to_buf(self: Pin<&mut Self>, buf: &[u8]) -> usize {
+        let available = self.spare_capacity();
+        let amt_to_buffer = available.min(buf.len());
+
+        // SAFETY: `amt_to_buffer` is <= buffer's spare capacity by construction.
+        unsafe {
+            self.write_to_buffer_unchecked(&buf[..amt_to_buffer]);
+        }
+
+        amt_to_buffer
+    }
+
+    /// Write byte slice directly into `self.buf`
+    ///
+    /// Based on `std::io::BufWriter`
+    #[inline]
+    unsafe fn write_to_buffer_unchecked(self: Pin<&mut Self>, buf: &[u8]) {
+        debug_assert!(buf.len() <= self.spare_capacity());
+        let this = self.project();
+        let old_len = this.buf.len();
+        let buf_len = buf.len();
+        let src = buf.as_ptr();
+        let dst = this.buf.as_mut_ptr().add(old_len);
+        ptr::copy_nonoverlapping(src, dst, buf_len);
+        this.buf.set_len(old_len + buf_len);
+    }
+
+    /// Write directly using `inner`, bypassing buffering
+    pub(super) fn inner_poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        self.project().inner.poll_write(cx, buf)
+    }
+
+    /// Write directly using `inner`, bypassing buffering
+    pub(super) fn inner_poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        self.project().inner.poll_write_vectored(cx, bufs)
+    }
+}
+
+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))?;
+        }
+        if buf.len() >= self.buf.capacity() {
+            self.project().inner.poll_write(cx, buf)
+        } else {
+            Poll::Ready(self.project().buf.write(buf))
+        }
+    }
+
+    fn poll_write_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        let total_len = bufs.iter().map(|b| b.len()).sum::<usize>();
+        if self.buf.len() + total_len > self.buf.capacity() {
+            ready!(self.as_mut().flush_buf(cx))?;
+        }
+        if total_len >= self.buf.capacity() {
+            self.project().inner.poll_write_vectored(cx, bufs)
+        } else {
+            Poll::Ready(self.project().buf.write_vectored(bufs))
+        }
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        ready!(self.as_mut().flush_buf(cx))?;
+        self.project().inner.poll_flush(cx)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        ready!(self.as_mut().flush_buf(cx))?;
+        self.project().inner.poll_close(cx)
+    }
+}
+
+impl<W: AsyncRead> AsyncRead for BufWriter<W> {
+    delegate_async_read!(inner);
+}
+
+impl<W: AsyncBufRead> AsyncBufRead for BufWriter<W> {
+    delegate_async_buf_read!(inner);
+}
+
+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()
+    }
+}
+
+impl<W: AsyncWrite + AsyncSeek> AsyncSeek for BufWriter<W> {
+    /// Seek to the offset, in bytes, in the underlying writer.
+    ///
+    /// Seeking always writes out the internal buffer before seeking.
+    fn poll_seek(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        pos: SeekFrom,
+    ) -> Poll<io::Result<u64>> {
+        ready!(self.as_mut().flush_buf(cx))?;
+        self.project().inner.poll_seek(cx, pos)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/chain.rs b/third_party/rust/futures-util/src/io/chain.rs
new file mode 100644
index 0000000000..728a3d2dc0
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/chain.rs
@@ -0,0 +1,142 @@
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncBufRead, AsyncRead, IoSliceMut};
+use pin_project_lite::pin_project;
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+
+pin_project! {
+    /// Reader for the [`chain`](super::AsyncReadExt::chain) method.
+    #[must_use = "readers do nothing unless polled"]
+    pub struct Chain<T, U> {
+        #[pin]
+        first: T,
+        #[pin]
+        second: U,
+        done_first: bool,
+    }
+}
+
+impl<T, U> Chain<T, U>
+where
+    T: AsyncRead,
+    U: AsyncRead,
+{
+    pub(super) fn new(first: T, second: U) -> Self {
+        Self { first, second, done_first: false }
+    }
+
+    /// 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 this = self.project();
+        (this.first, this.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)
+            .field("done_first", &self.done_first)
+            .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 this = self.project();
+
+        if !*this.done_first {
+            match ready!(this.first.poll_read(cx, buf)?) {
+                0 if !buf.is_empty() => *this.done_first = true,
+                n => return Poll::Ready(Ok(n)),
+            }
+        }
+        this.second.poll_read(cx, buf)
+    }
+
+    fn poll_read_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+    ) -> Poll<io::Result<usize>> {
+        let this = self.project();
+
+        if !*this.done_first {
+            let n = ready!(this.first.poll_read_vectored(cx, bufs)?);
+            if n == 0 && bufs.iter().any(|b| !b.is_empty()) {
+                *this.done_first = true
+            } else {
+                return Poll::Ready(Ok(n));
+            }
+        }
+        this.second.poll_read_vectored(cx, bufs)
+    }
+}
+
+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 this = self.project();
+
+        if !*this.done_first {
+            match ready!(this.first.poll_fill_buf(cx)?) {
+                buf if buf.is_empty() => {
+                    *this.done_first = true;
+                }
+                buf => return Poll::Ready(Ok(buf)),
+            }
+        }
+        this.second.poll_fill_buf(cx)
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        let this = self.project();
+
+        if !*this.done_first {
+            this.first.consume(amt)
+        } else {
+            this.second.consume(amt)
+        }
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/close.rs b/third_party/rust/futures-util/src/io/close.rs
new file mode 100644
index 0000000000..b94459279a
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/close.rs
@@ -0,0 +1,28 @@
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncWrite;
+use std::io;
+use std::pin::Pin;
+
+/// Future for the [`close`](super::AsyncWriteExt::close) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Close<'a, W: ?Sized> {
+    writer: &'a mut W,
+}
+
+impl<W: ?Sized + Unpin> Unpin for Close<'_, W> {}
+
+impl<'a, W: AsyncWrite + ?Sized + Unpin> Close<'a, W> {
+    pub(super) fn new(writer: &'a mut W) -> Self {
+        Self { writer }
+    }
+}
+
+impl<W: AsyncWrite + ?Sized + Unpin> Future for Close<'_, W> {
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut *self.writer).poll_close(cx)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/copy.rs b/third_party/rust/futures-util/src/io/copy.rs
new file mode 100644
index 0000000000..c80add271b
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/copy.rs
@@ -0,0 +1,58 @@
+use super::{copy_buf, BufReader, CopyBuf};
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncRead, AsyncWrite};
+use pin_project_lite::pin_project;
+use std::io;
+use std::pin::Pin;
+
+/// Creates a future which copies all the bytes from one object to another.
+///
+/// The returned future will copy all the bytes read from this `AsyncRead` into the
+/// `writer` specified. This future will only complete once the `reader` has hit
+/// EOF and all bytes have been written to and flushed from the `writer`
+/// provided.
+///
+/// On success the number of bytes is returned.
+///
+/// # Examples
+///
+/// ```
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncWriteExt, Cursor};
+///
+/// let reader = Cursor::new([1, 2, 3, 4]);
+/// let mut writer = Cursor::new(vec![0u8; 5]);
+///
+/// let bytes = io::copy(reader, &mut writer).await?;
+/// writer.close().await?;
+///
+/// assert_eq!(bytes, 4);
+/// assert_eq!(writer.into_inner(), [1, 2, 3, 4, 0]);
+/// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+/// ```
+pub fn copy<R, W>(reader: R, writer: &mut W) -> Copy<'_, R, W>
+where
+    R: AsyncRead,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    Copy { inner: copy_buf(BufReader::new(reader), writer) }
+}
+
+pin_project! {
+    /// Future for the [`copy()`] function.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct Copy<'a, R, W: ?Sized> {
+        #[pin]
+        inner: CopyBuf<'a, BufReader<R>, W>,
+    }
+}
+
+impl<R: AsyncRead, W: AsyncWrite + Unpin + ?Sized> Future for Copy<'_, R, W> {
+    type Output = io::Result<u64>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().inner.poll(cx)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/copy_buf.rs b/third_party/rust/futures-util/src/io/copy_buf.rs
new file mode 100644
index 0000000000..50f7abdca9
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/copy_buf.rs
@@ -0,0 +1,78 @@
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncBufRead, AsyncWrite};
+use pin_project_lite::pin_project;
+use std::io;
+use std::pin::Pin;
+
+/// Creates a future which copies all the bytes from one object to another.
+///
+/// The returned future will copy all the bytes read from this `AsyncBufRead` into the
+/// `writer` specified. This future will only complete once the `reader` has hit
+/// EOF and all bytes have been written to and flushed from the `writer`
+/// provided.
+///
+/// On success the number of bytes is returned.
+///
+/// # Examples
+///
+/// ```
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncWriteExt, Cursor};
+///
+/// let reader = Cursor::new([1, 2, 3, 4]);
+/// let mut writer = Cursor::new(vec![0u8; 5]);
+///
+/// let bytes = io::copy_buf(reader, &mut writer).await?;
+/// writer.close().await?;
+///
+/// assert_eq!(bytes, 4);
+/// assert_eq!(writer.into_inner(), [1, 2, 3, 4, 0]);
+/// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+/// ```
+pub fn copy_buf<R, W>(reader: R, writer: &mut W) -> CopyBuf<'_, R, W>
+where
+    R: AsyncBufRead,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    CopyBuf { reader, writer, amt: 0 }
+}
+
+pin_project! {
+    /// Future for the [`copy_buf()`] function.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct CopyBuf<'a, R, W: ?Sized> {
+        #[pin]
+        reader: R,
+        writer: &'a mut W,
+        amt: u64,
+    }
+}
+
+impl<R, W> Future for CopyBuf<'_, R, W>
+where
+    R: AsyncBufRead,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<u64>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.project();
+        loop {
+            let buffer = ready!(this.reader.as_mut().poll_fill_buf(cx))?;
+            if buffer.is_empty() {
+                ready!(Pin::new(&mut this.writer).poll_flush(cx))?;
+                return Poll::Ready(Ok(*this.amt));
+            }
+
+            let i = ready!(Pin::new(&mut this.writer).poll_write(cx, buffer))?;
+            if i == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+            *this.amt += i as u64;
+            this.reader.as_mut().consume(i);
+        }
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/copy_buf_abortable.rs b/third_party/rust/futures-util/src/io/copy_buf_abortable.rs
new file mode 100644
index 0000000000..fdbc4a5f00
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/copy_buf_abortable.rs
@@ -0,0 +1,124 @@
+use crate::abortable::{AbortHandle, AbortInner, Aborted};
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncBufRead, AsyncWrite};
+use pin_project_lite::pin_project;
+use std::io;
+use std::pin::Pin;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+
+/// Creates a future which copies all the bytes from one object to another, with its `AbortHandle`.
+///
+/// The returned future will copy all the bytes read from this `AsyncBufRead` into the
+/// `writer` specified. This future will only complete once abort has been requested or the `reader` has hit
+/// EOF and all bytes have been written to and flushed from the `writer`
+/// provided.
+///
+/// On success the number of bytes is returned. If aborted, `Aborted` is returned. Otherwise, the underlying error is returned.
+///
+/// # Examples
+///
+/// ```
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncWriteExt, Cursor};
+/// use futures::future::Aborted;
+///
+/// let reader = Cursor::new([1, 2, 3, 4]);
+/// let mut writer = Cursor::new(vec![0u8; 5]);
+///
+/// let (fut, abort_handle) = io::copy_buf_abortable(reader, &mut writer);
+/// let bytes = fut.await;
+/// abort_handle.abort();
+/// writer.close().await.unwrap();
+/// match bytes {
+///     Ok(Ok(n)) => {
+///         assert_eq!(n, 4);
+///         assert_eq!(writer.into_inner(), [1, 2, 3, 4, 0]);
+///         Ok(n)
+///     },
+///     Ok(Err(a)) => {
+///         Err::<u64, Aborted>(a)
+///     }
+///     Err(e) => panic!("{}", e)
+/// }
+/// #  }).unwrap();
+/// ```
+pub fn copy_buf_abortable<R, W>(
+    reader: R,
+    writer: &mut W,
+) -> (CopyBufAbortable<'_, R, W>, AbortHandle)
+where
+    R: AsyncBufRead,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    let (handle, reg) = AbortHandle::new_pair();
+    (CopyBufAbortable { reader, writer, amt: 0, inner: reg.inner }, handle)
+}
+
+pin_project! {
+    /// Future for the [`copy_buf()`] function.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct CopyBufAbortable<'a, R, W: ?Sized> {
+        #[pin]
+        reader: R,
+        writer: &'a mut W,
+        amt: u64,
+        inner: Arc<AbortInner>
+    }
+}
+
+macro_rules! ready_or_break {
+    ($e:expr $(,)?) => {
+        match $e {
+            $crate::task::Poll::Ready(t) => t,
+            $crate::task::Poll::Pending => break,
+        }
+    };
+}
+
+impl<R, W> Future for CopyBufAbortable<'_, R, W>
+where
+    R: AsyncBufRead,
+    W: AsyncWrite + Unpin + Sized,
+{
+    type Output = Result<Result<u64, Aborted>, io::Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.project();
+        loop {
+            // Check if the task has been aborted
+            if this.inner.aborted.load(Ordering::Relaxed) {
+                return Poll::Ready(Ok(Err(Aborted)));
+            }
+
+            // Read some bytes from the reader, and if we have reached EOF, return total bytes read
+            let buffer = ready_or_break!(this.reader.as_mut().poll_fill_buf(cx))?;
+            if buffer.is_empty() {
+                ready_or_break!(Pin::new(&mut this.writer).poll_flush(cx))?;
+                return Poll::Ready(Ok(Ok(*this.amt)));
+            }
+
+            // Pass the buffer to the writer, and update the amount written
+            let i = ready_or_break!(Pin::new(&mut this.writer).poll_write(cx, buffer))?;
+            if i == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+            *this.amt += i as u64;
+            this.reader.as_mut().consume(i);
+        }
+        // Schedule the task to be woken up again.
+        // Never called unless Poll::Pending is returned from io objects.
+        this.inner.waker.register(cx.waker());
+
+        // Check to see if the task was aborted between the first check and
+        // registration.
+        // Checking with  `Relaxed` is sufficient because
+        // `register` introduces an `AcqRel` barrier.
+        if this.inner.aborted.load(Ordering::Relaxed) {
+            return Poll::Ready(Ok(Err(Aborted)));
+        }
+        Poll::Pending
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/cursor.rs b/third_party/rust/futures-util/src/io/cursor.rs
new file mode 100644
index 0000000000..b6fb3724c7
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/cursor.rs
@@ -0,0 +1,240 @@
+use futures_core::task::{Context, Poll};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
+use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSlice, IoSliceMut, SeekFrom};
+use std::io;
+use std::pin::Pin;
+
+/// A `Cursor` wraps an in-memory buffer and provides it with a
+/// [`AsyncSeek`] implementation.
+///
+/// `Cursor`s are used with in-memory buffers, anything implementing
+/// `AsRef<[u8]>`, to allow them to implement [`AsyncRead`] and/or [`AsyncWrite`],
+/// allowing these buffers to be used anywhere you might use a reader or writer
+/// that does actual I/O.
+///
+/// This library implements some I/O traits on various types which
+/// are commonly used as a buffer, like `Cursor<`[`Vec`]`<u8>>` and
+/// `Cursor<`[`&[u8]`][bytes]`>`.
+///
+/// [`AsyncSeek`]: trait.AsyncSeek.html
+/// [`AsyncRead`]: trait.AsyncRead.html
+/// [`AsyncWrite`]: trait.AsyncWrite.html
+/// [bytes]: https://doc.rust-lang.org/std/primitive.slice.html
+#[derive(Clone, Debug, Default)]
+pub struct Cursor<T> {
+    inner: io::Cursor<T>,
+}
+
+impl<T> Cursor<T> {
+    /// Creates a new cursor wrapping the provided underlying in-memory buffer.
+    ///
+    /// Cursor initial position is `0` even if underlying buffer (e.g., `Vec`)
+    /// is not empty. So writing to cursor starts with overwriting `Vec`
+    /// content, not with appending to it.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use futures::io::Cursor;
+    ///
+    /// let buff = Cursor::new(Vec::new());
+    /// # fn force_inference(_: &Cursor<Vec<u8>>) {}
+    /// # force_inference(&buff);
+    /// ```
+    pub fn new(inner: T) -> Self {
+        Self { inner: io::Cursor::new(inner) }
+    }
+
+    /// Consumes this cursor, returning the underlying value.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use futures::io::Cursor;
+    ///
+    /// let buff = Cursor::new(Vec::new());
+    /// # fn force_inference(_: &Cursor<Vec<u8>>) {}
+    /// # force_inference(&buff);
+    ///
+    /// let vec = buff.into_inner();
+    /// ```
+    pub fn into_inner(self) -> T {
+        self.inner.into_inner()
+    }
+
+    /// Gets a reference to the underlying value in this cursor.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use futures::io::Cursor;
+    ///
+    /// let buff = Cursor::new(Vec::new());
+    /// # fn force_inference(_: &Cursor<Vec<u8>>) {}
+    /// # force_inference(&buff);
+    ///
+    /// let reference = buff.get_ref();
+    /// ```
+    pub fn get_ref(&self) -> &T {
+        self.inner.get_ref()
+    }
+
+    /// Gets a mutable reference to the underlying value in this cursor.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying value as it may corrupt this cursor's position.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use futures::io::Cursor;
+    ///
+    /// let mut buff = Cursor::new(Vec::new());
+    /// # fn force_inference(_: &Cursor<Vec<u8>>) {}
+    /// # force_inference(&buff);
+    ///
+    /// let reference = buff.get_mut();
+    /// ```
+    pub fn get_mut(&mut self) -> &mut T {
+        self.inner.get_mut()
+    }
+
+    /// Returns the current position of this cursor.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncSeekExt, Cursor, SeekFrom};
+    ///
+    /// let mut buff = Cursor::new(vec![1, 2, 3, 4, 5]);
+    ///
+    /// assert_eq!(buff.position(), 0);
+    ///
+    /// buff.seek(SeekFrom::Current(2)).await?;
+    /// assert_eq!(buff.position(), 2);
+    ///
+    /// buff.seek(SeekFrom::Current(-1)).await?;
+    /// assert_eq!(buff.position(), 1);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    pub fn position(&self) -> u64 {
+        self.inner.position()
+    }
+
+    /// Sets the position of this cursor.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use futures::io::Cursor;
+    ///
+    /// let mut buff = Cursor::new(vec![1, 2, 3, 4, 5]);
+    ///
+    /// assert_eq!(buff.position(), 0);
+    ///
+    /// buff.set_position(2);
+    /// assert_eq!(buff.position(), 2);
+    ///
+    /// buff.set_position(4);
+    /// assert_eq!(buff.position(), 4);
+    /// ```
+    pub fn set_position(&mut self, pos: u64) {
+        self.inner.set_position(pos)
+    }
+}
+
+impl<T> AsyncSeek for Cursor<T>
+where
+    T: AsRef<[u8]> + Unpin,
+{
+    fn poll_seek(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        pos: SeekFrom,
+    ) -> Poll<io::Result<u64>> {
+        Poll::Ready(io::Seek::seek(&mut self.inner, pos))
+    }
+}
+
+impl<T: AsRef<[u8]> + Unpin> AsyncRead for Cursor<T> {
+    #[cfg(feature = "read_initializer")]
+    #[inline]
+    unsafe fn initializer(&self) -> Initializer {
+        io::Read::initializer(&self.inner)
+    }
+
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(io::Read::read(&mut self.inner, buf))
+    }
+
+    fn poll_read_vectored(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(io::Read::read_vectored(&mut self.inner, bufs))
+    }
+}
+
+impl<T> AsyncBufRead for Cursor<T>
+where
+    T: AsRef<[u8]> + Unpin,
+{
+    fn poll_fill_buf(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        Poll::Ready(io::BufRead::fill_buf(&mut self.get_mut().inner))
+    }
+
+    fn consume(mut self: Pin<&mut Self>, amt: usize) {
+        io::BufRead::consume(&mut self.inner, amt)
+    }
+}
+
+macro_rules! delegate_async_write_to_stdio {
+    () => {
+        fn poll_write(
+            mut self: Pin<&mut Self>,
+            _: &mut Context<'_>,
+            buf: &[u8],
+        ) -> Poll<io::Result<usize>> {
+            Poll::Ready(io::Write::write(&mut self.inner, buf))
+        }
+
+        fn poll_write_vectored(
+            mut self: Pin<&mut Self>,
+            _: &mut Context<'_>,
+            bufs: &[IoSlice<'_>],
+        ) -> Poll<io::Result<usize>> {
+            Poll::Ready(io::Write::write_vectored(&mut self.inner, bufs))
+        }
+
+        fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(io::Write::flush(&mut self.inner))
+        }
+
+        fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            self.poll_flush(cx)
+        }
+    };
+}
+
+impl AsyncWrite for Cursor<&mut [u8]> {
+    delegate_async_write_to_stdio!();
+}
+
+impl AsyncWrite for Cursor<&mut Vec<u8>> {
+    delegate_async_write_to_stdio!();
+}
+
+impl AsyncWrite for Cursor<Vec<u8>> {
+    delegate_async_write_to_stdio!();
+}
+
+impl AsyncWrite for Cursor<Box<[u8]>> {
+    delegate_async_write_to_stdio!();
+}
diff --git a/third_party/rust/futures-util/src/io/empty.rs b/third_party/rust/futures-util/src/io/empty.rs
new file mode 100644
index 0000000000..02f6103f54
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/empty.rs
@@ -0,0 +1,59 @@
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncBufRead, AsyncRead};
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+
+/// Reader for the [`empty()`] function.
+#[must_use = "readers do nothing unless polled"]
+pub struct Empty {
+    _priv: (),
+}
+
+/// Constructs a new handle to an empty reader.
+///
+/// All reads from the returned reader will return `Poll::Ready(Ok(0))`.
+///
+/// # Examples
+///
+/// A slightly sad example of not reading anything into a buffer:
+///
+/// ```
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncReadExt};
+///
+/// let mut buffer = String::new();
+/// let mut reader = io::empty();
+/// reader.read_to_string(&mut buffer).await?;
+/// assert!(buffer.is_empty());
+/// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+/// ```
+pub fn empty() -> Empty {
+    Empty { _priv: () }
+}
+
+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 { .. }")
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/fill_buf.rs b/third_party/rust/futures-util/src/io/fill_buf.rs
new file mode 100644
index 0000000000..a1484c0322
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/fill_buf.rs
@@ -0,0 +1,51 @@
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncBufRead;
+use std::io;
+use std::pin::Pin;
+
+/// Future for the [`fill_buf`](super::AsyncBufReadExt::fill_buf) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct FillBuf<'a, R: ?Sized> {
+    reader: Option<&'a mut R>,
+}
+
+impl<R: ?Sized> Unpin for FillBuf<'_, R> {}
+
+impl<'a, R: AsyncBufRead + ?Sized + Unpin> FillBuf<'a, R> {
+    pub(super) fn new(reader: &'a mut R) -> Self {
+        Self { reader: Some(reader) }
+    }
+}
+
+impl<'a, R> Future for FillBuf<'a, R>
+where
+    R: AsyncBufRead + ?Sized + Unpin,
+{
+    type Output = io::Result<&'a [u8]>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        let reader = this.reader.take().expect("Polled FillBuf after completion");
+
+        match Pin::new(&mut *reader).poll_fill_buf(cx) {
+            // With polonius it is possible to remove this inner match and just have the correct
+            // lifetime of the reference inferred based on which branch is taken
+            Poll::Ready(Ok(_)) => match Pin::new(reader).poll_fill_buf(cx) {
+                Poll::Ready(Ok(slice)) => Poll::Ready(Ok(slice)),
+                Poll::Ready(Err(err)) => {
+                    unreachable!("reader indicated readiness but then returned an error: {:?}", err)
+                }
+                Poll::Pending => {
+                    unreachable!("reader indicated readiness but then returned pending")
+                }
+            },
+            Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
+            Poll::Pending => {
+                this.reader = Some(reader);
+                Poll::Pending
+            }
+        }
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/flush.rs b/third_party/rust/futures-util/src/io/flush.rs
new file mode 100644
index 0000000000..b75d14c5d3
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/flush.rs
@@ -0,0 +1,31 @@
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncWrite;
+use std::io;
+use std::pin::Pin;
+
+/// Future for the [`flush`](super::AsyncWriteExt::flush) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Flush<'a, W: ?Sized> {
+    writer: &'a mut W,
+}
+
+impl<W: ?Sized + Unpin> Unpin for Flush<'_, W> {}
+
+impl<'a, W: AsyncWrite + ?Sized + Unpin> Flush<'a, W> {
+    pub(super) fn new(writer: &'a mut W) -> Self {
+        Self { writer }
+    }
+}
+
+impl<W> Future for Flush<'_, W>
+where
+    W: AsyncWrite + ?Sized + Unpin,
+{
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut *self.writer).poll_flush(cx)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/into_sink.rs b/third_party/rust/futures-util/src/io/into_sink.rs
new file mode 100644
index 0000000000..6a41ee2269
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/into_sink.rs
@@ -0,0 +1,82 @@
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncWrite;
+use futures_sink::Sink;
+use pin_project_lite::pin_project;
+use std::io;
+use std::pin::Pin;
+
+#[derive(Debug)]
+struct Block<Item> {
+    offset: usize,
+    bytes: Item,
+}
+
+pin_project! {
+    /// Sink for the [`into_sink`](super::AsyncWriteExt::into_sink) method.
+    #[must_use = "sinks do nothing unless polled"]
+    #[derive(Debug)]
+    #[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
+    pub struct IntoSink<W, Item> {
+        #[pin]
+        writer: W,
+        // An outstanding block for us to push into the underlying writer, along with an offset of how
+        // far into this block we have written already.
+        buffer: Option<Block<Item>>,
+    }
+}
+
+impl<W: AsyncWrite, Item: AsRef<[u8]>> IntoSink<W, Item> {
+    pub(super) fn new(writer: W) -> Self {
+        Self { writer, buffer: None }
+    }
+
+    /// If we have an outstanding block in `buffer` attempt to push it into the writer, does _not_
+    /// flush the writer after it succeeds in pushing the block into it.
+    fn poll_flush_buffer(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), io::Error>> {
+        let mut this = self.project();
+
+        if let Some(buffer) = this.buffer {
+            loop {
+                let bytes = buffer.bytes.as_ref();
+                let written = ready!(this.writer.as_mut().poll_write(cx, &bytes[buffer.offset..]))?;
+                buffer.offset += written;
+                if buffer.offset == bytes.len() {
+                    break;
+                }
+            }
+        }
+        *this.buffer = None;
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<W: AsyncWrite, Item: AsRef<[u8]>> Sink<Item> for IntoSink<W, Item> {
+    type Error = io::Error;
+
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        ready!(self.poll_flush_buffer(cx))?;
+        Poll::Ready(Ok(()))
+    }
+
+    fn start_send(self: Pin<&mut Self>, item: Item) -> Result<(), Self::Error> {
+        debug_assert!(self.buffer.is_none());
+        *self.project().buffer = Some(Block { offset: 0, bytes: item });
+        Ok(())
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        ready!(self.as_mut().poll_flush_buffer(cx))?;
+        ready!(self.project().writer.poll_flush(cx))?;
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        ready!(self.as_mut().poll_flush_buffer(cx))?;
+        ready!(self.project().writer.poll_close(cx))?;
+        Poll::Ready(Ok(()))
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/line_writer.rs b/third_party/rust/futures-util/src/io/line_writer.rs
new file mode 100644
index 0000000000..71cd668325
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/line_writer.rs
@@ -0,0 +1,155 @@
+use super::buf_writer::BufWriter;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncWrite;
+use futures_io::IoSlice;
+use pin_project_lite::pin_project;
+use std::io;
+use std::pin::Pin;
+
+pin_project! {
+/// Wrap a writer, like [`BufWriter`] does, but prioritizes buffering lines
+///
+/// This was written based on `std::io::LineWriter` which goes into further details
+/// explaining the code.
+///
+/// Buffering is actually done using `BufWriter`. This class will leverage `BufWriter`
+/// to write on-each-line.
+#[derive(Debug)]
+pub struct LineWriter<W: AsyncWrite> {
+    #[pin]
+    buf_writer: BufWriter<W>,
+}
+}
+
+impl<W: AsyncWrite> LineWriter<W> {
+    /// Create a new `LineWriter` with default buffer capacity. The default is currently 1KB
+    /// which was taken from `std::io::LineWriter`
+    pub fn new(inner: W) -> LineWriter<W> {
+        LineWriter::with_capacity(1024, inner)
+    }
+
+    /// Creates a new `LineWriter` with the specified buffer capacity.
+    pub fn with_capacity(capacity: usize, inner: W) -> LineWriter<W> {
+        LineWriter { buf_writer: BufWriter::with_capacity(capacity, inner) }
+    }
+
+    /// Flush `buf_writer` if last char is "new line"
+    fn flush_if_completed_line(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = self.project();
+        match this.buf_writer.buffer().last().copied() {
+            Some(b'\n') => this.buf_writer.flush_buf(cx),
+            _ => Poll::Ready(Ok(())),
+        }
+    }
+
+    /// Returns a reference to `buf_writer`'s internally buffered data.
+    pub fn buffer(&self) -> &[u8] {
+        self.buf_writer.buffer()
+    }
+
+    /// Acquires a reference to the underlying sink or stream that this combinator is
+    /// pulling from.
+    pub fn get_ref(&self) -> &W {
+        self.buf_writer.get_ref()
+    }
+}
+
+impl<W: AsyncWrite> AsyncWrite for LineWriter<W> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        let mut this = self.as_mut().project();
+        let newline_index = match memchr::memrchr(b'\n', buf) {
+            None => {
+                ready!(self.as_mut().flush_if_completed_line(cx)?);
+                return self.project().buf_writer.poll_write(cx, buf);
+            }
+            Some(newline_index) => newline_index + 1,
+        };
+
+        ready!(this.buf_writer.as_mut().poll_flush(cx)?);
+
+        let lines = &buf[..newline_index];
+
+        let flushed = { ready!(this.buf_writer.as_mut().inner_poll_write(cx, lines))? };
+
+        if flushed == 0 {
+            return Poll::Ready(Ok(0));
+        }
+
+        let tail = if flushed >= newline_index {
+            &buf[flushed..]
+        } else if newline_index - flushed <= this.buf_writer.capacity() {
+            &buf[flushed..newline_index]
+        } else {
+            let scan_area = &buf[flushed..];
+            let scan_area = &scan_area[..this.buf_writer.capacity()];
+            match memchr::memrchr(b'\n', scan_area) {
+                Some(newline_index) => &scan_area[..newline_index + 1],
+                None => scan_area,
+            }
+        };
+
+        let buffered = this.buf_writer.as_mut().write_to_buf(tail);
+        Poll::Ready(Ok(flushed + buffered))
+    }
+
+    fn poll_write_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        let mut this = self.as_mut().project();
+        // `is_write_vectored()` is handled in original code, but not in this crate
+        // see https://github.com/rust-lang/rust/issues/70436
+
+        let last_newline_buf_idx = bufs
+            .iter()
+            .enumerate()
+            .rev()
+            .find_map(|(i, buf)| memchr::memchr(b'\n', buf).map(|_| i));
+        let last_newline_buf_idx = match last_newline_buf_idx {
+            None => {
+                ready!(self.as_mut().flush_if_completed_line(cx)?);
+                return self.project().buf_writer.poll_write_vectored(cx, bufs);
+            }
+            Some(i) => i,
+        };
+
+        ready!(this.buf_writer.as_mut().poll_flush(cx)?);
+
+        let (lines, tail) = bufs.split_at(last_newline_buf_idx + 1);
+
+        let flushed = { ready!(this.buf_writer.as_mut().inner_poll_write_vectored(cx, lines))? };
+        if flushed == 0 {
+            return Poll::Ready(Ok(0));
+        }
+
+        let lines_len = lines.iter().map(|buf| buf.len()).sum();
+        if flushed < lines_len {
+            return Poll::Ready(Ok(flushed));
+        }
+
+        let buffered: usize = tail
+            .iter()
+            .filter(|buf| !buf.is_empty())
+            .map(|buf| this.buf_writer.as_mut().write_to_buf(buf))
+            .take_while(|&n| n > 0)
+            .sum();
+
+        Poll::Ready(Ok(flushed + buffered))
+    }
+
+    /// Forward to `buf_writer` 's `BufWriter::poll_flush()`
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.as_mut().project().buf_writer.poll_flush(cx)
+    }
+
+    /// Forward to `buf_writer` 's `BufWriter::poll_close()`
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.as_mut().project().buf_writer.poll_close(cx)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/lines.rs b/third_party/rust/futures-util/src/io/lines.rs
new file mode 100644
index 0000000000..b5561bfa7d
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/lines.rs
@@ -0,0 +1,47 @@
+use super::read_line::read_line_internal;
+use futures_core::ready;
+use futures_core::stream::Stream;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncBufRead;
+use pin_project_lite::pin_project;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+
+pin_project! {
+    /// Stream for the [`lines`](super::AsyncBufReadExt::lines) method.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    pub struct Lines<R> {
+        #[pin]
+        reader: R,
+        buf: String,
+        bytes: Vec<u8>,
+        read: usize,
+    }
+}
+
+impl<R: AsyncBufRead> Lines<R> {
+    pub(super) fn new(reader: R) -> Self {
+        Self { reader, buf: String::new(), bytes: Vec::new(), read: 0 }
+    }
+}
+
+impl<R: AsyncBufRead> Stream for Lines<R> {
+    type Item = io::Result<String>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.project();
+        let n = ready!(read_line_internal(this.reader, cx, this.buf, this.bytes, this.read))?;
+        if n == 0 && this.buf.is_empty() {
+            return Poll::Ready(None);
+        }
+        if this.buf.ends_with('\n') {
+            this.buf.pop();
+            if this.buf.ends_with('\r') {
+                this.buf.pop();
+            }
+        }
+        Poll::Ready(Some(Ok(mem::take(this.buf))))
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/mod.rs b/third_party/rust/futures-util/src/io/mod.rs
new file mode 100644
index 0000000000..8ce3ad644b
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/mod.rs
@@ -0,0 +1,841 @@
+//! Asynchronous I/O.
+//!
+//! This module is the asynchronous version of `std::io`. It defines four
+//! traits, [`AsyncRead`], [`AsyncWrite`], [`AsyncSeek`], and [`AsyncBufRead`],
+//! which mirror the `Read`, `Write`, `Seek`, and `BufRead` traits of the
+//! standard library. However, these traits integrate with the asynchronous
+//! task system, so that if an I/O object isn't ready for reading (or writing),
+//! the thread is not blocked, and instead the current task is queued to be
+//! woken when I/O is ready.
+//!
+//! In addition, the [`AsyncReadExt`], [`AsyncWriteExt`], [`AsyncSeekExt`], and
+//! [`AsyncBufReadExt`] extension traits offer a variety of useful combinators
+//! for operating with asynchronous I/O objects, including ways to work with
+//! them using futures, streams and sinks.
+//!
+//! This module is only available when the `std` feature of this
+//! library is activated, and it is activated by default.
+
+#[cfg(feature = "io-compat")]
+#[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
+use crate::compat::Compat;
+use crate::future::assert_future;
+use crate::stream::assert_stream;
+use std::{pin::Pin, ptr};
+
+// Re-export some types from `std::io` so that users don't have to deal
+// with conflicts when `use`ing `futures::io` and `std::io`.
+#[doc(no_inline)]
+pub use std::io::{Error, ErrorKind, IoSlice, IoSliceMut, Result, SeekFrom};
+
+pub use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite};
+
+// 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;
+
+/// Initializes a buffer if necessary.
+///
+/// A buffer is currently always initialized.
+#[inline]
+unsafe fn initialize<R: AsyncRead>(_reader: &R, buf: &mut [u8]) {
+    ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len())
+}
+
+mod allow_std;
+pub use self::allow_std::AllowStdIo;
+
+mod buf_reader;
+pub use self::buf_reader::{BufReader, SeeKRelative};
+
+mod buf_writer;
+pub use self::buf_writer::BufWriter;
+
+mod line_writer;
+pub use self::line_writer::LineWriter;
+
+mod chain;
+pub use self::chain::Chain;
+
+mod close;
+pub use self::close::Close;
+
+mod copy;
+pub use self::copy::{copy, Copy};
+
+mod copy_buf;
+pub use self::copy_buf::{copy_buf, CopyBuf};
+
+mod copy_buf_abortable;
+pub use self::copy_buf_abortable::{copy_buf_abortable, CopyBufAbortable};
+
+mod cursor;
+pub use self::cursor::Cursor;
+
+mod empty;
+pub use self::empty::{empty, Empty};
+
+mod fill_buf;
+pub use self::fill_buf::FillBuf;
+
+mod flush;
+pub use self::flush::Flush;
+
+#[cfg(feature = "sink")]
+#[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
+mod into_sink;
+#[cfg(feature = "sink")]
+#[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
+pub use self::into_sink::IntoSink;
+
+mod lines;
+pub use self::lines::Lines;
+
+mod read;
+pub use self::read::Read;
+
+mod read_vectored;
+pub use self::read_vectored::ReadVectored;
+
+mod read_exact;
+pub use self::read_exact::ReadExact;
+
+mod read_line;
+pub use self::read_line::ReadLine;
+
+mod read_to_end;
+pub use self::read_to_end::ReadToEnd;
+
+mod read_to_string;
+pub use self::read_to_string::ReadToString;
+
+mod read_until;
+pub use self::read_until::ReadUntil;
+
+mod repeat;
+pub use self::repeat::{repeat, Repeat};
+
+mod seek;
+pub use self::seek::Seek;
+
+mod sink;
+pub use self::sink::{sink, Sink};
+
+mod split;
+pub use self::split::{ReadHalf, ReuniteError, WriteHalf};
+
+mod take;
+pub use self::take::Take;
+
+mod window;
+pub use self::window::Window;
+
+mod write;
+pub use self::write::Write;
+
+mod write_vectored;
+pub use self::write_vectored::WriteVectored;
+
+mod write_all;
+pub use self::write_all::WriteAll;
+
+#[cfg(feature = "write-all-vectored")]
+mod write_all_vectored;
+#[cfg(feature = "write-all-vectored")]
+pub use self::write_all_vectored::WriteAllVectored;
+
+/// An extension trait which adds utility methods to `AsyncRead` types.
+pub trait AsyncReadExt: AsyncRead {
+    /// Creates an adaptor which will chain this stream with another.
+    ///
+    /// 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
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncReadExt, Cursor};
+    ///
+    /// let reader1 = Cursor::new([1, 2, 3, 4]);
+    /// let reader2 = Cursor::new([5, 6, 7, 8]);
+    ///
+    /// let mut reader = reader1.chain(reader2);
+    /// let mut buffer = Vec::new();
+    ///
+    /// // read the value into a Vec.
+    /// reader.read_to_end(&mut buffer).await?;
+    /// assert_eq!(buffer, [1, 2, 3, 4, 5, 6, 7, 8]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn chain<R>(self, next: R) -> Chain<Self, R>
+    where
+        Self: Sized,
+        R: AsyncRead,
+    {
+        assert_read(Chain::new(self, next))
+    }
+
+    /// Tries to read some bytes directly into the given `buf` in asynchronous
+    /// manner, returning a future type.
+    ///
+    /// The returned future will resolve to the number of bytes read once the read
+    /// operation is completed.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncReadExt, Cursor};
+    ///
+    /// let mut reader = Cursor::new([1, 2, 3, 4]);
+    /// let mut output = [0u8; 5];
+    ///
+    /// let bytes = reader.read(&mut output[..]).await?;
+    ///
+    /// // This is only guaranteed to be 4 because `&[u8]` is a synchronous
+    /// // reader. In a real system you could get anywhere from 1 to
+    /// // `output.len()` bytes in a single read.
+    /// assert_eq!(bytes, 4);
+    /// assert_eq!(output, [1, 2, 3, 4, 0]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Read<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<usize>, _>(Read::new(self, buf))
+    }
+
+    /// Creates a future which will read from the `AsyncRead` into `bufs` using vectored
+    /// IO operations.
+    ///
+    /// The returned future will resolve to the number of bytes read once the read
+    /// operation is completed.
+    fn read_vectored<'a>(&'a mut self, bufs: &'a mut [IoSliceMut<'a>]) -> ReadVectored<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<usize>, _>(ReadVectored::new(self, bufs))
+    }
+
+    /// Creates a future which will read exactly enough bytes to fill `buf`,
+    /// returning an error if end of file (EOF) is hit sooner.
+    ///
+    /// The returned future will resolve once the read operation is completed.
+    ///
+    /// In the case of an error the buffer and the object will be discarded, with
+    /// the error yielded.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncReadExt, Cursor};
+    ///
+    /// let mut reader = Cursor::new([1, 2, 3, 4]);
+    /// let mut output = [0u8; 4];
+    ///
+    /// reader.read_exact(&mut output).await?;
+    ///
+    /// assert_eq!(output, [1, 2, 3, 4]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    ///
+    /// ## EOF is hit before `buf` is filled
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{self, AsyncReadExt, Cursor};
+    ///
+    /// let mut reader = Cursor::new([1, 2, 3, 4]);
+    /// let mut output = [0u8; 5];
+    ///
+    /// let result = reader.read_exact(&mut output).await;
+    ///
+    /// assert_eq!(result.unwrap_err().kind(), io::ErrorKind::UnexpectedEof);
+    /// # });
+    /// ```
+    fn read_exact<'a>(&'a mut self, buf: &'a mut [u8]) -> ReadExact<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<()>, _>(ReadExact::new(self, buf))
+    }
+
+    /// Creates a future which will read all the bytes from this `AsyncRead`.
+    ///
+    /// On success the total number of bytes read is returned.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncReadExt, Cursor};
+    ///
+    /// let mut reader = Cursor::new([1, 2, 3, 4]);
+    /// let mut output = Vec::with_capacity(4);
+    ///
+    /// let bytes = reader.read_to_end(&mut output).await?;
+    ///
+    /// assert_eq!(bytes, 4);
+    /// assert_eq!(output, vec![1, 2, 3, 4]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn read_to_end<'a>(&'a mut self, buf: &'a mut Vec<u8>) -> ReadToEnd<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<usize>, _>(ReadToEnd::new(self, buf))
+    }
+
+    /// Creates a future which will read all the bytes from this `AsyncRead`.
+    ///
+    /// On success the total number of bytes read is returned.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncReadExt, Cursor};
+    ///
+    /// let mut reader = Cursor::new(&b"1234"[..]);
+    /// let mut buffer = String::with_capacity(4);
+    ///
+    /// let bytes = reader.read_to_string(&mut buffer).await?;
+    ///
+    /// assert_eq!(bytes, 4);
+    /// assert_eq!(buffer, String::from("1234"));
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn read_to_string<'a>(&'a mut self, buf: &'a mut String) -> ReadToString<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<usize>, _>(ReadToString::new(self, buf))
+    }
+
+    /// Helper method for splitting this read/write object into two halves.
+    ///
+    /// The two halves returned implement the `AsyncRead` and `AsyncWrite`
+    /// traits, respectively.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{self, AsyncReadExt, Cursor};
+    ///
+    /// // Note that for `Cursor` the read and write halves share a single
+    /// // seek position. This may or may not be true for other types that
+    /// // implement both `AsyncRead` and `AsyncWrite`.
+    ///
+    /// let reader = Cursor::new([1, 2, 3, 4]);
+    /// let mut buffer = Cursor::new(vec![0, 0, 0, 0, 5, 6, 7, 8]);
+    /// let mut writer = Cursor::new(vec![0u8; 5]);
+    ///
+    /// {
+    ///     let (buffer_reader, mut buffer_writer) = (&mut buffer).split();
+    ///     io::copy(reader, &mut buffer_writer).await?;
+    ///     io::copy(buffer_reader, &mut writer).await?;
+    /// }
+    ///
+    /// assert_eq!(buffer.into_inner(), [1, 2, 3, 4, 5, 6, 7, 8]);
+    /// assert_eq!(writer.into_inner(), [5, 6, 7, 8, 0]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn split(self) -> (ReadHalf<Self>, WriteHalf<Self>)
+    where
+        Self: AsyncWrite + Sized,
+    {
+        let (r, w) = split::split(self);
+        (assert_read(r), assert_write(w))
+    }
+
+    /// Creates an AsyncRead adapter which will read at most `limit` bytes
+    /// from the underlying reader.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncReadExt, Cursor};
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 5];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// assert_eq!(n, 4);
+    /// assert_eq!(&buffer, b"1234\0");
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn take(self, limit: u64) -> Take<Self>
+    where
+        Self: Sized,
+    {
+        assert_read(Take::new(self, limit))
+    }
+
+    /// Wraps an [`AsyncRead`] in a compatibility wrapper that allows it to be
+    /// used as a futures 0.1 / tokio-io 0.1 `AsyncRead`. If the wrapped type
+    /// implements [`AsyncWrite`] as well, the result will also implement the
+    /// futures 0.1 / tokio 0.1 `AsyncWrite` trait.
+    ///
+    /// Requires the `io-compat` feature to enable.
+    #[cfg(feature = "io-compat")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
+    fn compat(self) -> Compat<Self>
+    where
+        Self: Sized + Unpin,
+    {
+        Compat::new(self)
+    }
+}
+
+impl<R: AsyncRead + ?Sized> AsyncReadExt for R {}
+
+/// An extension trait which adds utility methods to `AsyncWrite` types.
+pub trait AsyncWriteExt: AsyncWrite {
+    /// Creates a future which will entirely flush this `AsyncWrite`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AllowStdIo, AsyncWriteExt};
+    /// use std::io::{BufWriter, Cursor};
+    ///
+    /// let mut output = vec![0u8; 5];
+    ///
+    /// {
+    ///     let writer = Cursor::new(&mut output);
+    ///     let mut buffered = AllowStdIo::new(BufWriter::new(writer));
+    ///     buffered.write_all(&[1, 2]).await?;
+    ///     buffered.write_all(&[3, 4]).await?;
+    ///     buffered.flush().await?;
+    /// }
+    ///
+    /// assert_eq!(output, [1, 2, 3, 4, 0]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn flush(&mut self) -> Flush<'_, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<()>, _>(Flush::new(self))
+    }
+
+    /// Creates a future which will entirely close this `AsyncWrite`.
+    fn close(&mut self) -> Close<'_, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<()>, _>(Close::new(self))
+    }
+
+    /// Creates a future which will write bytes from `buf` into the object.
+    ///
+    /// The returned future will resolve to the number of bytes written once the write
+    /// operation is completed.
+    fn write<'a>(&'a mut self, buf: &'a [u8]) -> Write<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<usize>, _>(Write::new(self, buf))
+    }
+
+    /// Creates a future which will write bytes from `bufs` into the object using vectored
+    /// IO operations.
+    ///
+    /// The returned future will resolve to the number of bytes written once the write
+    /// operation is completed.
+    fn write_vectored<'a>(&'a mut self, bufs: &'a [IoSlice<'a>]) -> WriteVectored<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<usize>, _>(WriteVectored::new(self, bufs))
+    }
+
+    /// Write data into this object.
+    ///
+    /// Creates a future that will write the entire contents of the buffer `buf` into
+    /// this `AsyncWrite`.
+    ///
+    /// The returned future will not complete until all the data has been written.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncWriteExt, Cursor};
+    ///
+    /// let mut writer = Cursor::new(vec![0u8; 5]);
+    ///
+    /// writer.write_all(&[1, 2, 3, 4]).await?;
+    ///
+    /// assert_eq!(writer.into_inner(), [1, 2, 3, 4, 0]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn write_all<'a>(&'a mut self, buf: &'a [u8]) -> WriteAll<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<()>, _>(WriteAll::new(self, buf))
+    }
+
+    /// Attempts to write multiple buffers into this writer.
+    ///
+    /// Creates a future that will write the entire contents of `bufs` into this
+    /// `AsyncWrite` using [vectored writes].
+    ///
+    /// The returned future will not complete until all the data has been
+    /// written.
+    ///
+    /// [vectored writes]: std::io::Write::write_vectored
+    ///
+    /// # Notes
+    ///
+    /// Unlike `io::Write::write_vectored`, this takes a *mutable* reference to
+    /// a slice of `IoSlice`s, not an immutable one. That's because we need to
+    /// modify the slice to keep track of the bytes already written.
+    ///
+    /// Once this futures returns, the contents of `bufs` are unspecified, as
+    /// this depends on how many calls to `write_vectored` were necessary. It is
+    /// best to understand this function as taking ownership of `bufs` and to
+    /// not use `bufs` afterwards. The underlying buffers, to which the
+    /// `IoSlice`s point (but not the `IoSlice`s themselves), are unchanged and
+    /// can be reused.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncWriteExt;
+    /// use futures_util::io::Cursor;
+    /// use std::io::IoSlice;
+    ///
+    /// let mut writer = Cursor::new(Vec::new());
+    /// let bufs = &mut [
+    ///     IoSlice::new(&[1]),
+    ///     IoSlice::new(&[2, 3]),
+    ///     IoSlice::new(&[4, 5, 6]),
+    /// ];
+    ///
+    /// writer.write_all_vectored(bufs).await?;
+    /// // Note: the contents of `bufs` is now unspecified, see the Notes section.
+    ///
+    /// assert_eq!(writer.into_inner(), &[1, 2, 3, 4, 5, 6]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    #[cfg(feature = "write-all-vectored")]
+    fn write_all_vectored<'a>(
+        &'a mut self,
+        bufs: &'a mut [IoSlice<'a>],
+    ) -> WriteAllVectored<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<()>, _>(WriteAllVectored::new(self, bufs))
+    }
+
+    /// Wraps an [`AsyncWrite`] in a compatibility wrapper that allows it to be
+    /// used as a futures 0.1 / tokio-io 0.1 `AsyncWrite`.
+    /// Requires the `io-compat` feature to enable.
+    #[cfg(feature = "io-compat")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
+    fn compat_write(self) -> Compat<Self>
+    where
+        Self: Sized + Unpin,
+    {
+        Compat::new(self)
+    }
+
+    /// Allow using an [`AsyncWrite`] as a [`Sink`](futures_sink::Sink)`<Item: AsRef<[u8]>>`.
+    ///
+    /// This adapter produces a sink that will write each value passed to it
+    /// into the underlying writer.
+    ///
+    /// Note that this function consumes the given writer, returning a wrapped
+    /// version.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncWriteExt;
+    /// use futures::stream::{self, StreamExt};
+    ///
+    /// let stream = stream::iter(vec![Ok([1, 2, 3]), Ok([4, 5, 6])]);
+    ///
+    /// let mut writer = vec![];
+    ///
+    /// stream.forward((&mut writer).into_sink()).await?;
+    ///
+    /// assert_eq!(writer, vec![1, 2, 3, 4, 5, 6]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(())
+    /// # })?;
+    /// # Ok::<(), Box<dyn std::error::Error>>(())
+    /// ```
+    #[cfg(feature = "sink")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
+    fn into_sink<Item: AsRef<[u8]>>(self) -> IntoSink<Self, Item>
+    where
+        Self: Sized,
+    {
+        crate::sink::assert_sink::<Item, Error, _>(IntoSink::new(self))
+    }
+}
+
+impl<W: AsyncWrite + ?Sized> AsyncWriteExt for W {}
+
+/// An extension trait which adds utility methods to `AsyncSeek` types.
+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.
+    fn seek(&mut self, pos: SeekFrom) -> Seek<'_, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<u64>, _>(Seek::new(self, pos))
+    }
+
+    /// Creates a future which will return the current seek position from the
+    /// start of the stream.
+    ///
+    /// This is equivalent to `self.seek(SeekFrom::Current(0))`.
+    fn stream_position(&mut self) -> Seek<'_, Self>
+    where
+        Self: Unpin,
+    {
+        self.seek(SeekFrom::Current(0))
+    }
+}
+
+impl<S: AsyncSeek + ?Sized> AsyncSeekExt for S {}
+
+/// An extension trait which adds utility methods to `AsyncBufRead` types.
+pub trait AsyncBufReadExt: AsyncBufRead {
+    /// Creates a future which will wait for a non-empty buffer to be available from this I/O
+    /// object or EOF to be reached.
+    ///
+    /// This method is the async equivalent to [`BufRead::fill_buf`](std::io::BufRead::fill_buf).
+    ///
+    /// ```rust
+    /// # futures::executor::block_on(async {
+    /// use futures::{io::AsyncBufReadExt as _, stream::{iter, TryStreamExt as _}};
+    ///
+    /// let mut stream = iter(vec![Ok(vec![1, 2, 3]), Ok(vec![4, 5, 6])]).into_async_read();
+    ///
+    /// assert_eq!(stream.fill_buf().await?, vec![1, 2, 3]);
+    /// stream.consume_unpin(2);
+    ///
+    /// assert_eq!(stream.fill_buf().await?, vec![3]);
+    /// stream.consume_unpin(1);
+    ///
+    /// assert_eq!(stream.fill_buf().await?, vec![4, 5, 6]);
+    /// stream.consume_unpin(3);
+    ///
+    /// assert_eq!(stream.fill_buf().await?, vec![]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn fill_buf(&mut self) -> FillBuf<'_, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<&[u8]>, _>(FillBuf::new(self))
+    }
+
+    /// A convenience for calling [`AsyncBufRead::consume`] on [`Unpin`] IO types.
+    ///
+    /// ```rust
+    /// # futures::executor::block_on(async {
+    /// use futures::{io::AsyncBufReadExt as _, stream::{iter, TryStreamExt as _}};
+    ///
+    /// let mut stream = iter(vec![Ok(vec![1, 2, 3])]).into_async_read();
+    ///
+    /// assert_eq!(stream.fill_buf().await?, vec![1, 2, 3]);
+    /// stream.consume_unpin(2);
+    ///
+    /// assert_eq!(stream.fill_buf().await?, vec![3]);
+    /// stream.consume_unpin(1);
+    ///
+    /// assert_eq!(stream.fill_buf().await?, vec![]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn consume_unpin(&mut self, amt: usize)
+    where
+        Self: Unpin,
+    {
+        Pin::new(self).consume(amt)
+    }
+
+    /// Creates a future which will read all the bytes associated with this I/O
+    /// object into `buf` until the delimiter `byte` or EOF is reached.
+    /// This method is the async equivalent to [`BufRead::read_until`](std::io::BufRead::read_until).
+    ///
+    /// 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`.
+    ///
+    /// The returned future will resolve to the number of bytes read once the read
+    /// operation is completed.
+    ///
+    /// In the case of an error the buffer and the object will be discarded, with
+    /// the error yielded.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncBufReadExt, Cursor};
+    ///
+    /// 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?;
+    /// 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?;
+    /// 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?;
+    /// assert_eq!(num_bytes, 0);
+    /// assert_eq!(buf, b"");
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn read_until<'a>(&'a mut self, byte: u8, buf: &'a mut Vec<u8>) -> ReadUntil<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<usize>, _>(ReadUntil::new(self, byte, buf))
+    }
+
+    /// Creates a future which will read all the bytes associated with this I/O
+    /// object into `buf` until a newline (the 0xA byte) or EOF is reached,
+    /// This method is the async equivalent to [`BufRead::read_line`](std::io::BufRead::read_line).
+    ///
+    /// 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`.
+    ///
+    /// The returned future will resolve to the number of bytes read once the read
+    /// operation is completed.
+    ///
+    /// In the case of an error the buffer and the object will be discarded, with
+    /// the error yielded.
+    ///
+    /// # 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
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncBufReadExt, Cursor};
+    ///
+    /// 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?;
+    /// 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?;
+    /// assert_eq!(num_bytes, 3);
+    /// assert_eq!(buf, "bar");
+    /// buf.clear();
+    ///
+    /// // cursor is at EOF
+    /// let num_bytes = cursor.read_line(&mut buf).await?;
+    /// assert_eq!(num_bytes, 0);
+    /// assert_eq!(buf, "");
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn read_line<'a>(&'a mut self, buf: &'a mut String) -> ReadLine<'a, Self>
+    where
+        Self: Unpin,
+    {
+        assert_future::<Result<usize>, _>(ReadLine::new(self, buf))
+    }
+
+    /// 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`].
+    ///
+    /// [`AsyncBufReadExt::read_line`]: AsyncBufReadExt::read_line
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncBufReadExt, Cursor};
+    /// use futures::stream::StreamExt;
+    ///
+    /// let cursor = Cursor::new(b"lorem\nipsum\r\ndolor");
+    ///
+    /// let mut lines_stream = cursor.lines().map(|l| l.unwrap());
+    /// assert_eq!(lines_stream.next().await, Some(String::from("lorem")));
+    /// assert_eq!(lines_stream.next().await, Some(String::from("ipsum")));
+    /// assert_eq!(lines_stream.next().await, Some(String::from("dolor")));
+    /// assert_eq!(lines_stream.next().await, None);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn lines(self) -> Lines<Self>
+    where
+        Self: Sized,
+    {
+        assert_stream::<Result<String>, _>(Lines::new(self))
+    }
+}
+
+impl<R: AsyncBufRead + ?Sized> AsyncBufReadExt for R {}
+
+// Just a helper function to ensure the reader we're returning all have the
+// right implementations.
+pub(crate) fn assert_read<R>(reader: R) -> R
+where
+    R: AsyncRead,
+{
+    reader
+}
+// Just a helper function to ensure the writer we're returning all have the
+// right implementations.
+pub(crate) fn assert_write<W>(writer: W) -> W
+where
+    W: AsyncWrite,
+{
+    writer
+}
diff --git a/third_party/rust/futures-util/src/io/read.rs b/third_party/rust/futures-util/src/io/read.rs
new file mode 100644
index 0000000000..677ba818d9
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/read.rs
@@ -0,0 +1,30 @@
+use crate::io::AsyncRead;
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use std::io;
+use std::pin::Pin;
+
+/// Future for the [`read`](super::AsyncReadExt::read) method.
+#[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: ?Sized + Unpin> Unpin for Read<'_, R> {}
+
+impl<'a, R: AsyncRead + ?Sized + Unpin> Read<'a, R> {
+    pub(super) fn new(reader: &'a mut R, buf: &'a mut [u8]) -> Self {
+        Self { reader, buf }
+    }
+}
+
+impl<R: AsyncRead + ?Sized + Unpin> Future for Read<'_, R> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        Pin::new(&mut this.reader).poll_read(cx, this.buf)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/read_exact.rs b/third_party/rust/futures-util/src/io/read_exact.rs
new file mode 100644
index 0000000000..cd0b20e597
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/read_exact.rs
@@ -0,0 +1,42 @@
+use crate::io::AsyncRead;
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use std::io;
+use std::mem;
+use std::pin::Pin;
+
+/// Future for the [`read_exact`](super::AsyncReadExt::read_exact) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadExact<'a, R: ?Sized> {
+    reader: &'a mut R,
+    buf: &'a mut [u8],
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadExact<'_, R> {}
+
+impl<'a, R: AsyncRead + ?Sized + Unpin> ReadExact<'a, R> {
+    pub(super) fn new(reader: &'a mut R, buf: &'a mut [u8]) -> Self {
+        Self { reader, buf }
+    }
+}
+
+impl<R: AsyncRead + ?Sized + Unpin> Future for ReadExact<'_, R> {
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        while !this.buf.is_empty() {
+            let n = ready!(Pin::new(&mut this.reader).poll_read(cx, this.buf))?;
+            {
+                let (_, rest) = mem::take(&mut this.buf).split_at_mut(n);
+                this.buf = rest;
+            }
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::UnexpectedEof.into()));
+            }
+        }
+        Poll::Ready(Ok(()))
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/read_line.rs b/third_party/rust/futures-util/src/io/read_line.rs
new file mode 100644
index 0000000000..e1b8fc9455
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/read_line.rs
@@ -0,0 +1,57 @@
+use super::read_until::read_until_internal;
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncBufRead;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::str;
+
+/// Future for the [`read_line`](super::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,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadLine<'_, R> {}
+
+impl<'a, R: AsyncBufRead + ?Sized + Unpin> ReadLine<'a, R> {
+    pub(super) fn new(reader: &'a mut R, buf: &'a mut String) -> Self {
+        Self { reader, bytes: mem::take(buf).into_bytes(), 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)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/read_to_end.rs b/third_party/rust/futures-util/src/io/read_to_end.rs
new file mode 100644
index 0000000000..919d7d13c7
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/read_to_end.rs
@@ -0,0 +1,91 @@
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncRead;
+use std::io;
+use std::pin::Pin;
+use std::vec::Vec;
+
+/// Future for the [`read_to_end`](super::AsyncReadExt::read_to_end) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadToEnd<'a, R: ?Sized> {
+    reader: &'a mut R,
+    buf: &'a mut Vec<u8>,
+    start_len: usize,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadToEnd<'_, R> {}
+
+impl<'a, R: AsyncRead + ?Sized + Unpin> ReadToEnd<'a, R> {
+    pub(super) fn new(reader: &'a mut R, buf: &'a mut Vec<u8>) -> Self {
+        let start_len = buf.len();
+        Self { 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 };
+    loop {
+        if g.len == g.buf.len() {
+            unsafe {
+                g.buf.reserve(32);
+                let capacity = g.buf.capacity();
+                g.buf.set_len(capacity);
+                super::initialize(&rd, &mut g.buf[g.len..]);
+            }
+        }
+
+        let buf = &mut g.buf[g.len..];
+        match ready!(rd.as_mut().poll_read(cx, buf)) {
+            Ok(0) => return Poll::Ready(Ok(g.len - start_len)),
+            Ok(n) => {
+                // We can't allow bogus values from read. If it is too large, the returned vec could have its length
+                // set past its capacity, or if it overflows the vec could be shortened which could create an invalid
+                // string if this is called via read_to_string.
+                assert!(n <= buf.len());
+                g.len += n;
+            }
+            Err(e) => return Poll::Ready(Err(e)),
+        }
+    }
+}
+
+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)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/read_to_string.rs b/third_party/rust/futures-util/src/io/read_to_string.rs
new file mode 100644
index 0000000000..c175396d81
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/read_to_string.rs
@@ -0,0 +1,59 @@
+use super::read_to_end::read_to_end_internal;
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncRead;
+use std::pin::Pin;
+use std::vec::Vec;
+use std::{io, mem, str};
+
+/// 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,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadToString<'_, R> {}
+
+impl<'a, R: AsyncRead + ?Sized + Unpin> ReadToString<'a, R> {
+    pub(super) fn new(reader: &'a mut R, buf: &'a mut String) -> Self {
+        let start_len = buf.len();
+        Self { reader, bytes: mem::take(buf).into_bytes(), 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)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/read_until.rs b/third_party/rust/futures-util/src/io/read_until.rs
new file mode 100644
index 0000000000..72b59eab13
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/read_until.rs
@@ -0,0 +1,60 @@
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncBufRead;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+
+/// Future for the [`read_until`](super::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,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadUntil<'_, R> {}
+
+impl<'a, R: AsyncBufRead + ?Sized + Unpin> ReadUntil<'a, R> {
+    pub(super) fn new(reader: &'a mut R, byte: u8, buf: &'a mut Vec<u8>) -> Self {
+        Self { 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)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/read_vectored.rs b/third_party/rust/futures-util/src/io/read_vectored.rs
new file mode 100644
index 0000000000..4e22df57e9
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/read_vectored.rs
@@ -0,0 +1,30 @@
+use crate::io::AsyncRead;
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use std::io::{self, IoSliceMut};
+use std::pin::Pin;
+
+/// Future for the [`read_vectored`](super::AsyncReadExt::read_vectored) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadVectored<'a, R: ?Sized> {
+    reader: &'a mut R,
+    bufs: &'a mut [IoSliceMut<'a>],
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadVectored<'_, R> {}
+
+impl<'a, R: AsyncRead + ?Sized + Unpin> ReadVectored<'a, R> {
+    pub(super) fn new(reader: &'a mut R, bufs: &'a mut [IoSliceMut<'a>]) -> Self {
+        Self { reader, bufs }
+    }
+}
+
+impl<R: AsyncRead + ?Sized + Unpin> Future for ReadVectored<'_, R> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        Pin::new(&mut this.reader).poll_read_vectored(cx, this.bufs)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/repeat.rs b/third_party/rust/futures-util/src/io/repeat.rs
new file mode 100644
index 0000000000..2828bf0114
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/repeat.rs
@@ -0,0 +1,66 @@
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncRead, IoSliceMut};
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+
+/// Reader for the [`repeat()`] function.
+#[must_use = "readers do nothing unless polled"]
+pub struct Repeat {
+    byte: u8,
+}
+
+/// Creates an instance of a reader that infinitely repeats one byte.
+///
+/// All reads from this reader will succeed by filling the specified buffer with
+/// the given byte.
+///
+/// # Examples
+///
+/// ```
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncReadExt};
+///
+/// let mut buffer = [0; 3];
+/// let mut reader = io::repeat(0b101);
+/// reader.read_exact(&mut buffer).await.unwrap();
+/// assert_eq!(buffer, [0b101, 0b101, 0b101]);
+/// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+/// ```
+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 slot in &mut *buf {
+            *slot = self.byte;
+        }
+        Poll::Ready(Ok(buf.len()))
+    }
+
+    #[inline]
+    fn poll_read_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+    ) -> Poll<io::Result<usize>> {
+        let mut nwritten = 0;
+        for buf in bufs {
+            nwritten += ready!(self.as_mut().poll_read(cx, buf))?;
+        }
+        Poll::Ready(Ok(nwritten))
+    }
+}
+
+impl fmt::Debug for Repeat {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("Repeat { .. }")
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/seek.rs b/third_party/rust/futures-util/src/io/seek.rs
new file mode 100644
index 0000000000..0aa2371393
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/seek.rs
@@ -0,0 +1,30 @@
+use crate::io::{AsyncSeek, SeekFrom};
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use std::io;
+use std::pin::Pin;
+
+/// 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: SeekFrom,
+}
+
+impl<S: ?Sized + Unpin> Unpin for Seek<'_, S> {}
+
+impl<'a, S: AsyncSeek + ?Sized + Unpin> Seek<'a, S> {
+    pub(super) fn new(seek: &'a mut S, pos: SeekFrom) -> Self {
+        Self { seek, pos }
+    }
+}
+
+impl<S: AsyncSeek + ?Sized + Unpin> Future for Seek<'_, S> {
+    type Output = io::Result<u64>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        Pin::new(&mut this.seek).poll_seek(cx, this.pos)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/sink.rs b/third_party/rust/futures-util/src/io/sink.rs
new file mode 100644
index 0000000000..4a32ca7041
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/sink.rs
@@ -0,0 +1,67 @@
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncWrite, IoSlice};
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+
+/// Writer for the [`sink()`] function.
+#[must_use = "writers do nothing unless polled"]
+pub struct Sink {
+    _priv: (),
+}
+
+/// Creates an instance of a 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.
+///
+/// # Examples
+///
+/// ```rust
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncWriteExt};
+///
+/// let buffer = vec![1, 2, 3, 5, 8];
+/// let mut writer = io::sink();
+/// let num_bytes = writer.write(&buffer).await?;
+/// assert_eq!(num_bytes, 5);
+/// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+/// ```
+pub fn sink() -> Sink {
+    Sink { _priv: () }
+}
+
+impl AsyncWrite for Sink {
+    #[inline]
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(buf.len()))
+    }
+
+    #[inline]
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(bufs.iter().map(|b| b.len()).sum()))
+    }
+
+    #[inline]
+    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+    #[inline]
+    fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl fmt::Debug for Sink {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("Sink { .. }")
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/split.rs b/third_party/rust/futures-util/src/io/split.rs
new file mode 100644
index 0000000000..3f1b9af456
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/split.rs
@@ -0,0 +1,115 @@
+use crate::lock::BiLock;
+use core::fmt;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncRead, AsyncWrite, IoSlice, IoSliceMut};
+use std::io;
+use std::pin::Pin;
+
+/// The readable half of an object returned from `AsyncRead::split`.
+#[derive(Debug)]
+pub struct ReadHalf<T> {
+    handle: BiLock<T>,
+}
+
+/// The writable half of an object returned from `AsyncRead::split`.
+#[derive(Debug)]
+pub struct WriteHalf<T> {
+    handle: BiLock<T>,
+}
+
+fn lock_and_then<T, U, E, F>(lock: &BiLock<T>, cx: &mut Context<'_>, f: F) -> Poll<Result<U, E>>
+where
+    F: FnOnce(Pin<&mut T>, &mut Context<'_>) -> Poll<Result<U, E>>,
+{
+    let mut l = ready!(lock.poll_lock(cx));
+    f(l.as_pin_mut(), cx)
+}
+
+pub(super) fn split<T: AsyncRead + AsyncWrite>(t: T) -> (ReadHalf<T>, WriteHalf<T>) {
+    let (a, b) = BiLock::new(t);
+    (ReadHalf { handle: a }, WriteHalf { handle: b })
+}
+
+impl<T: Unpin> ReadHalf<T> {
+    /// Attempts to put the two "halves" of a split `AsyncRead + AsyncWrite` back
+    /// together. Succeeds only if the `ReadHalf<T>` and `WriteHalf<T>` are
+    /// a matching pair originating from the same call to `AsyncReadExt::split`.
+    pub fn reunite(self, other: WriteHalf<T>) -> Result<T, ReuniteError<T>> {
+        self.handle
+            .reunite(other.handle)
+            .map_err(|err| ReuniteError(ReadHalf { handle: err.0 }, WriteHalf { handle: err.1 }))
+    }
+}
+
+impl<T: Unpin> WriteHalf<T> {
+    /// Attempts to put the two "halves" of a split `AsyncRead + AsyncWrite` back
+    /// together. Succeeds only if the `ReadHalf<T>` and `WriteHalf<T>` are
+    /// a matching pair originating from the same call to `AsyncReadExt::split`.
+    pub fn reunite(self, other: ReadHalf<T>) -> Result<T, ReuniteError<T>> {
+        other.reunite(self)
+    }
+}
+
+impl<R: AsyncRead> AsyncRead for ReadHalf<R> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        lock_and_then(&self.handle, cx, |l, cx| l.poll_read(cx, buf))
+    }
+
+    fn poll_read_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+    ) -> Poll<io::Result<usize>> {
+        lock_and_then(&self.handle, cx, |l, cx| l.poll_read_vectored(cx, bufs))
+    }
+}
+
+impl<W: AsyncWrite> AsyncWrite for WriteHalf<W> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        lock_and_then(&self.handle, cx, |l, cx| l.poll_write(cx, buf))
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        lock_and_then(&self.handle, cx, |l, cx| l.poll_write_vectored(cx, bufs))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        lock_and_then(&self.handle, cx, |l, cx| l.poll_flush(cx))
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        lock_and_then(&self.handle, cx, |l, cx| l.poll_close(cx))
+    }
+}
+
+/// Error indicating a `ReadHalf<T>` and `WriteHalf<T>` were not two halves
+/// of a `AsyncRead + AsyncWrite`, and thus could not be `reunite`d.
+pub struct ReuniteError<T>(pub ReadHalf<T>, pub WriteHalf<T>);
+
+impl<T> fmt::Debug for ReuniteError<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_tuple("ReuniteError").field(&"...").finish()
+    }
+}
+
+impl<T> fmt::Display for ReuniteError<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "tried to reunite a ReadHalf and WriteHalf that don't form a pair")
+    }
+}
+
+#[cfg(feature = "std")]
+impl<T: core::any::Any> std::error::Error for ReuniteError<T> {}
diff --git a/third_party/rust/futures-util/src/io/take.rs b/third_party/rust/futures-util/src/io/take.rs
new file mode 100644
index 0000000000..2c494804d9
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/take.rs
@@ -0,0 +1,125 @@
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncBufRead, AsyncRead};
+use pin_project_lite::pin_project;
+use std::pin::Pin;
+use std::{cmp, io};
+
+pin_project! {
+    /// Reader for the [`take`](super::AsyncReadExt::take) method.
+    #[derive(Debug)]
+    #[must_use = "readers do nothing unless you `.await` or poll them"]
+    pub struct Take<R> {
+        #[pin]
+        inner: R,
+        limit: u64,
+    }
+}
+
+impl<R: AsyncRead> Take<R> {
+    pub(super) fn new(inner: R, limit: u64) -> Self {
+        Self { inner, limit }
+    }
+
+    /// 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.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncReadExt, Cursor};
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 2];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// assert_eq!(take.limit(), 2);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    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.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::{AsyncReadExt, Cursor};
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 4];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// assert_eq!(n, 4);
+    /// assert_eq!(take.limit(), 0);
+    ///
+    /// take.set_limit(10);
+    /// let n = take.read(&mut buffer).await?;
+    /// assert_eq!(n, 4);
+    ///
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    pub fn set_limit(&mut self, limit: u64) {
+        self.limit = limit
+    }
+
+    delegate_access_inner!(inner, R, ());
+}
+
+impl<R: AsyncRead> AsyncRead for Take<R> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        let this = self.project();
+
+        if *this.limit == 0 {
+            return Poll::Ready(Ok(0));
+        }
+
+        let max = cmp::min(buf.len() as u64, *this.limit) as usize;
+        let n = ready!(this.inner.poll_read(cx, &mut buf[..max]))?;
+        *this.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 this = self.project();
+
+        // Don't call into inner reader at all at EOF because it may still block
+        if *this.limit == 0 {
+            return Poll::Ready(Ok(&[]));
+        }
+
+        let buf = ready!(this.inner.poll_fill_buf(cx)?);
+        let cap = cmp::min(buf.len() as u64, *this.limit) as usize;
+        Poll::Ready(Ok(&buf[..cap]))
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        let this = self.project();
+
+        // Don't let callers reset the limit by passing an overlarge value
+        let amt = cmp::min(amt as u64, *this.limit) as usize;
+        *this.limit -= amt as u64;
+        this.inner.consume(amt);
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/window.rs b/third_party/rust/futures-util/src/io/window.rs
new file mode 100644
index 0000000000..77b7267c69
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/window.rs
@@ -0,0 +1,104 @@
+use std::ops::{Bound, Range, RangeBounds};
+
+/// A owned window around an underlying buffer.
+///
+/// Normally slices work great for considering sub-portions of a buffer, but
+/// unfortunately a slice is a *borrowed* type in Rust which has an associated
+/// lifetime. When working with future and async I/O these lifetimes are not
+/// always appropriate, and are sometimes difficult to store in tasks. This
+/// type strives to fill this gap by providing an "owned slice" around an
+/// underlying buffer of bytes.
+///
+/// A `Window<T>` wraps an underlying buffer, `T`, and has configurable
+/// start/end indexes to alter the behavior of the `AsRef<[u8]>` implementation
+/// that this type carries.
+///
+/// This type can be particularly useful when working with the `write_all`
+/// combinator in this crate. Data can be sliced via `Window`, consumed by
+/// `write_all`, and then earned back once the write operation finishes through
+/// the `into_inner` method on this type.
+#[derive(Debug)]
+pub struct Window<T> {
+    inner: T,
+    range: Range<usize>,
+}
+
+impl<T: AsRef<[u8]>> Window<T> {
+    /// Creates a new window around the buffer `t` defaulting to the entire
+    /// slice.
+    ///
+    /// Further methods can be called on the returned `Window<T>` to alter the
+    /// window into the data provided.
+    pub fn new(t: T) -> Self {
+        Self { range: 0..t.as_ref().len(), inner: t }
+    }
+
+    /// Gets a shared reference to the underlying buffer inside of this
+    /// `Window`.
+    pub fn get_ref(&self) -> &T {
+        &self.inner
+    }
+
+    /// Gets a mutable reference to the underlying buffer inside of this
+    /// `Window`.
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.inner
+    }
+
+    /// Consumes this `Window`, returning the underlying buffer.
+    pub fn into_inner(self) -> T {
+        self.inner
+    }
+
+    /// Returns the starting index of this window into the underlying buffer
+    /// `T`.
+    pub fn start(&self) -> usize {
+        self.range.start
+    }
+
+    /// Returns the end index of this window into the underlying buffer
+    /// `T`.
+    pub fn end(&self) -> usize {
+        self.range.end
+    }
+
+    /// Changes the range of this window to the range specified.
+    ///
+    /// # Panics
+    ///
+    /// This method will panic if `range` is out of bounds for the underlying
+    /// slice or if [`start_bound()`] of `range` comes after the [`end_bound()`].
+    ///
+    /// [`start_bound()`]: std::ops::RangeBounds::start_bound
+    /// [`end_bound()`]: std::ops::RangeBounds::end_bound
+    pub fn set<R: RangeBounds<usize>>(&mut self, range: R) {
+        let start = match range.start_bound() {
+            Bound::Included(n) => *n,
+            Bound::Excluded(n) => *n + 1,
+            Bound::Unbounded => 0,
+        };
+        let end = match range.end_bound() {
+            Bound::Included(n) => *n + 1,
+            Bound::Excluded(n) => *n,
+            Bound::Unbounded => self.inner.as_ref().len(),
+        };
+
+        assert!(end <= self.inner.as_ref().len());
+        assert!(start <= end);
+
+        self.range.start = start;
+        self.range.end = end;
+    }
+}
+
+impl<T: AsRef<[u8]>> AsRef<[u8]> for Window<T> {
+    fn as_ref(&self) -> &[u8] {
+        &self.inner.as_ref()[self.range.start..self.range.end]
+    }
+}
+
+impl<T: AsMut<[u8]>> AsMut<[u8]> for Window<T> {
+    fn as_mut(&mut self) -> &mut [u8] {
+        &mut self.inner.as_mut()[self.range.start..self.range.end]
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/write.rs b/third_party/rust/futures-util/src/io/write.rs
new file mode 100644
index 0000000000..c47ef9e2eb
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/write.rs
@@ -0,0 +1,30 @@
+use crate::io::AsyncWrite;
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use std::io;
+use std::pin::Pin;
+
+/// Future for the [`write`](super::AsyncWriteExt::write) method.
+#[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],
+}
+
+impl<W: ?Sized + Unpin> Unpin for Write<'_, W> {}
+
+impl<'a, W: AsyncWrite + ?Sized + Unpin> Write<'a, W> {
+    pub(super) fn new(writer: &'a mut W, buf: &'a [u8]) -> Self {
+        Self { writer, buf }
+    }
+}
+
+impl<W: AsyncWrite + ?Sized + Unpin> Future for Write<'_, W> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        Pin::new(&mut this.writer).poll_write(cx, this.buf)
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/write_all.rs b/third_party/rust/futures-util/src/io/write_all.rs
new file mode 100644
index 0000000000..08c025f94d
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/write_all.rs
@@ -0,0 +1,43 @@
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncWrite;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+
+/// Future for the [`write_all`](super::AsyncWriteExt::write_all) method.
+#[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],
+}
+
+impl<W: ?Sized + Unpin> Unpin for WriteAll<'_, W> {}
+
+impl<'a, W: AsyncWrite + ?Sized + Unpin> WriteAll<'a, W> {
+    pub(super) fn new(writer: &'a mut W, buf: &'a [u8]) -> Self {
+        Self { writer, buf }
+    }
+}
+
+impl<W: AsyncWrite + ?Sized + Unpin> Future for WriteAll<'_, W> {
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = &mut *self;
+        while !this.buf.is_empty() {
+            let n = ready!(Pin::new(&mut this.writer).poll_write(cx, this.buf))?;
+            {
+                let (_, rest) = mem::take(&mut this.buf).split_at(n);
+                this.buf = rest;
+            }
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/write_all_vectored.rs b/third_party/rust/futures-util/src/io/write_all_vectored.rs
new file mode 100644
index 0000000000..a8fc4c641c
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/write_all_vectored.rs
@@ -0,0 +1,193 @@
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncWrite;
+use futures_io::IoSlice;
+use std::io;
+use std::pin::Pin;
+
+/// Future for the
+/// [`write_all_vectored`](super::AsyncWriteExt::write_all_vectored) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct WriteAllVectored<'a, W: ?Sized + Unpin> {
+    writer: &'a mut W,
+    bufs: &'a mut [IoSlice<'a>],
+}
+
+impl<W: ?Sized + Unpin> Unpin for WriteAllVectored<'_, W> {}
+
+impl<'a, W: AsyncWrite + ?Sized + Unpin> WriteAllVectored<'a, W> {
+    pub(super) fn new(writer: &'a mut W, mut bufs: &'a mut [IoSlice<'a>]) -> Self {
+        IoSlice::advance_slices(&mut bufs, 0);
+        Self { writer, bufs }
+    }
+}
+
+impl<W: AsyncWrite + ?Sized + Unpin> Future for WriteAllVectored<'_, W> {
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = &mut *self;
+        while !this.bufs.is_empty() {
+            let n = ready!(Pin::new(&mut this.writer).poll_write_vectored(cx, this.bufs))?;
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            } else {
+                IoSlice::advance_slices(&mut this.bufs, n);
+            }
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::cmp::min;
+    use std::future::Future;
+    use std::io;
+    use std::pin::Pin;
+    use std::task::{Context, Poll};
+
+    use crate::io::{AsyncWrite, AsyncWriteExt, IoSlice};
+    use crate::task::noop_waker;
+
+    /// Create a new writer that reads from at most `n_bufs` and reads
+    /// `per_call` bytes (in total) per call to write.
+    fn test_writer(n_bufs: usize, per_call: usize) -> TestWriter {
+        TestWriter { n_bufs, per_call, written: Vec::new() }
+    }
+
+    // TODO: maybe move this the future-test crate?
+    struct TestWriter {
+        n_bufs: usize,
+        per_call: usize,
+        written: Vec<u8>,
+    }
+
+    impl AsyncWrite for TestWriter {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            buf: &[u8],
+        ) -> Poll<io::Result<usize>> {
+            self.poll_write_vectored(cx, &[IoSlice::new(buf)])
+        }
+
+        fn poll_write_vectored(
+            mut self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            bufs: &[IoSlice<'_>],
+        ) -> Poll<io::Result<usize>> {
+            let mut left = self.per_call;
+            let mut written = 0;
+            for buf in bufs.iter().take(self.n_bufs) {
+                let n = min(left, buf.len());
+                self.written.extend_from_slice(&buf[0..n]);
+                left -= n;
+                written += n;
+            }
+            Poll::Ready(Ok(written))
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    // TODO: maybe move this the future-test crate?
+    macro_rules! assert_poll_ok {
+        ($e:expr, $expected:expr) => {
+            let expected = $expected;
+            match $e {
+                Poll::Ready(Ok(ok)) if ok == expected => {}
+                got => {
+                    panic!("unexpected result, got: {:?}, wanted: Ready(Ok({:?}))", got, expected)
+                }
+            }
+        };
+    }
+
+    #[test]
+    fn test_writer_read_from_one_buf() {
+        let waker = noop_waker();
+        let mut cx = Context::from_waker(&waker);
+
+        let mut dst = test_writer(1, 2);
+        let mut dst = Pin::new(&mut dst);
+
+        assert_poll_ok!(dst.as_mut().poll_write(&mut cx, &[]), 0);
+        assert_poll_ok!(dst.as_mut().poll_write_vectored(&mut cx, &[]), 0);
+
+        // Read at most 2 bytes.
+        assert_poll_ok!(dst.as_mut().poll_write(&mut cx, &[1, 1, 1]), 2);
+        let bufs = &[IoSlice::new(&[2, 2, 2])];
+        assert_poll_ok!(dst.as_mut().poll_write_vectored(&mut cx, bufs), 2);
+
+        // Only read from first buf.
+        let bufs = &[IoSlice::new(&[3]), IoSlice::new(&[4, 4])];
+        assert_poll_ok!(dst.as_mut().poll_write_vectored(&mut cx, bufs), 1);
+
+        assert_eq!(dst.written, &[1, 1, 2, 2, 3]);
+    }
+
+    #[test]
+    fn test_writer_read_from_multiple_bufs() {
+        let waker = noop_waker();
+        let mut cx = Context::from_waker(&waker);
+
+        let mut dst = test_writer(3, 3);
+        let mut dst = Pin::new(&mut dst);
+
+        // Read at most 3 bytes from two buffers.
+        let bufs = &[IoSlice::new(&[1]), IoSlice::new(&[2, 2, 2])];
+        assert_poll_ok!(dst.as_mut().poll_write_vectored(&mut cx, bufs), 3);
+
+        // Read at most 3 bytes from three buffers.
+        let bufs = &[IoSlice::new(&[3]), IoSlice::new(&[4]), IoSlice::new(&[5, 5])];
+        assert_poll_ok!(dst.as_mut().poll_write_vectored(&mut cx, bufs), 3);
+
+        assert_eq!(dst.written, &[1, 2, 2, 3, 4, 5]);
+    }
+
+    #[test]
+    fn test_write_all_vectored() {
+        let waker = noop_waker();
+        let mut cx = Context::from_waker(&waker);
+
+        #[rustfmt::skip] // Becomes unreadable otherwise.
+        let tests: Vec<(_, &'static [u8])> = vec![
+            (vec![], &[]),
+            (vec![IoSlice::new(&[]), IoSlice::new(&[])], &[]),
+            (vec![IoSlice::new(&[1])], &[1]),
+            (vec![IoSlice::new(&[1, 2])], &[1, 2]),
+            (vec![IoSlice::new(&[1, 2, 3])], &[1, 2, 3]),
+            (vec![IoSlice::new(&[1, 2, 3, 4])], &[1, 2, 3, 4]),
+            (vec![IoSlice::new(&[1, 2, 3, 4, 5])], &[1, 2, 3, 4, 5]),
+            (vec![IoSlice::new(&[1]), IoSlice::new(&[2])], &[1, 2]),
+            (vec![IoSlice::new(&[1, 1]), IoSlice::new(&[2, 2])], &[1, 1, 2, 2]),
+            (vec![IoSlice::new(&[1, 1, 1]), IoSlice::new(&[2, 2, 2])], &[1, 1, 1, 2, 2, 2]),
+            (vec![IoSlice::new(&[1, 1, 1, 1]), IoSlice::new(&[2, 2, 2, 2])], &[1, 1, 1, 1, 2, 2, 2, 2]),
+            (vec![IoSlice::new(&[1]), IoSlice::new(&[2]), IoSlice::new(&[3])], &[1, 2, 3]),
+            (vec![IoSlice::new(&[1, 1]), IoSlice::new(&[2, 2]), IoSlice::new(&[3, 3])], &[1, 1, 2, 2, 3, 3]),
+            (vec![IoSlice::new(&[1, 1, 1]), IoSlice::new(&[2, 2, 2]), IoSlice::new(&[3, 3, 3])], &[1, 1, 1, 2, 2, 2, 3, 3, 3]),
+        ];
+
+        for (mut input, wanted) in tests {
+            let mut dst = test_writer(2, 2);
+            {
+                let mut future = dst.write_all_vectored(&mut *input);
+                match Pin::new(&mut future).poll(&mut cx) {
+                    Poll::Ready(Ok(())) => {}
+                    other => panic!("unexpected result polling future: {:?}", other),
+                }
+            }
+            assert_eq!(&*dst.written, &*wanted);
+        }
+    }
+}
diff --git a/third_party/rust/futures-util/src/io/write_vectored.rs b/third_party/rust/futures-util/src/io/write_vectored.rs
new file mode 100644
index 0000000000..14a01d7302
--- /dev/null
+++ b/third_party/rust/futures-util/src/io/write_vectored.rs
@@ -0,0 +1,30 @@
+use crate::io::AsyncWrite;
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use std::io::{self, IoSlice};
+use std::pin::Pin;
+
+/// Future for the [`write_vectored`](super::AsyncWriteExt::write_vectored) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct WriteVectored<'a, W: ?Sized> {
+    writer: &'a mut W,
+    bufs: &'a [IoSlice<'a>],
+}
+
+impl<W: ?Sized + Unpin> Unpin for WriteVectored<'_, W> {}
+
+impl<'a, W: AsyncWrite + ?Sized + Unpin> WriteVectored<'a, W> {
+    pub(super) fn new(writer: &'a mut W, bufs: &'a [IoSlice<'a>]) -> Self {
+        Self { writer, bufs }
+    }
+}
+
+impl<W: AsyncWrite + ?Sized + Unpin> Future for WriteVectored<'_, W> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        Pin::new(&mut this.writer).poll_write_vectored(cx, this.bufs)
+    }
+}