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