io: move io helpers back into tokio-io (#1377)

Utilities are made optional with a feature flag.

tokio-io/src/io/async_buf_read_ext.rs

@@ -0,0 +1,81 @@
+use crate::io::lines::{lines, Lines};
+use crate::io::read_line::{read_line, ReadLine};
+use crate::io::read_until::{read_until, ReadUntil};
+use crate::AsyncBufRead;
+
+/// An extension trait which adds utility methods to `AsyncBufRead` types.
+pub trait AsyncBufReadExt: AsyncBufRead {
+    /// Creates a future which will read all the bytes associated with this I/O
+    /// object into `buf` until the delimiter `byte` or EOF is reached.
+    /// This method is the async equivalent to [`BufRead::read_until`](std::io::BufRead::read_until).
+    ///
+    /// This function will read bytes from the underlying stream until the
+    /// delimiter or EOF is found. Once found, all bytes up to, and including,
+    /// the delimiter (if found) will be appended to `buf`.
+    ///
+    /// The returned future will resolve to the number of bytes read once the read
+    /// operation is completed.
+    ///
+    /// In the case of an error the buffer and the object will be discarded, with
+    /// the error yielded.
+    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)
+    }
+
+    /// Creates a future which will read all the bytes associated with this I/O
+    /// object into `buf` until a newline (the 0xA byte) or EOF is reached,
+    /// This method is the async equivalent to [`BufRead::read_line`](std::io::BufRead::read_line).
+    ///
+    /// This function will read bytes from the underlying stream until the
+    /// newline delimiter (the 0xA byte) or EOF is found. Once found, all bytes
+    /// up to, and including, the delimiter (if found) will be appended to
+    /// `buf`.
+    ///
+    /// The returned future will resolve to the number of bytes read once the read
+    /// operation is completed.
+    ///
+    /// In the case of an error the buffer and the object will be discarded, with
+    /// the error yielded.
+    ///
+    /// # Errors
+    ///
+    /// This function has the same error semantics as [`read_until`] and will
+    /// also return an error if the read bytes are not valid UTF-8. If an I/O
+    /// error is encountered then `buf` may contain some bytes already read in
+    /// the event that all data read so far was valid UTF-8.
+    ///
+    /// [`read_until`]: AsyncBufReadExt::read_until
+    fn read_line<'a>(&'a mut self, buf: &'a mut String) -> ReadLine<'a, Self>
+    where
+        Self: Unpin,
+    {
+        read_line(self, buf)
+    }
+
+    /// Returns a stream over the lines of this reader.
+    /// This method is the async equivalent to [`BufRead::lines`](std::io::BufRead::lines).
+    ///
+    /// The stream returned from this function will yield instances of
+    /// [`io::Result`]`<`[`String`]`>`. Each string returned will *not* have a newline
+    /// byte (the 0xA byte) or CRLF (0xD, 0xA bytes) at the end.
+    ///
+    /// [`io::Result`]: std::io::Result
+    /// [`String`]: String
+    ///
+    /// # Errors
+    ///
+    /// Each line of the stream has the same error semantics as [`AsyncBufReadExt::read_line`].
+    ///
+    /// [`AsyncBufReadExt::read_line`]: AsyncBufReadExt::read_line
+    fn lines(self) -> Lines<Self>
+    where
+        Self: Sized,
+    {
+        lines(self)
+    }
+}
+
+impl<R: AsyncBufRead + ?Sized> AsyncBufReadExt for R {}

tokio-io/src/io/async_read_ext.rs

@@ -0,0 +1,68 @@
+use crate::io::copy::{copy, Copy};
+use crate::io::read::{read, Read};
+use crate::io::read_exact::{read_exact, ReadExact};
+use crate::io::read_to_end::{read_to_end, ReadToEnd};
+use crate::io::read_to_string::{read_to_string, ReadToString};
+use crate::{AsyncRead, AsyncWrite};
+
+/// An extension trait which adds utility methods to `AsyncRead` types.
+pub trait AsyncReadExt: AsyncRead {
+    /// Copy all data from `self` into the provided `AsyncWrite`.
+    ///
+    /// The returned future will copy all the bytes read from `reader` into the
+    /// `writer` specified.  This future will only complete once the `reader`
+    /// has hit EOF and all bytes have been written to and flushed from the
+    /// `writer` provided.
+    ///
+    /// On success the number of bytes is returned and the `reader` and `writer`
+    /// are consumed. On error the error is returned and the I/O objects are
+    /// consumed as well.
+    fn copy<'a, W>(&'a mut self, dst: &'a mut W) -> Copy<'a, Self, W>
+    where
+        Self: Unpin,
+        W: AsyncWrite + Unpin + ?Sized,
+    {
+        copy(self, dst)
+    }
+
+    /// Read data into the provided buffer.
+    ///
+    /// The returned future will resolve to the number of bytes read once the
+    /// read operation is completed.
+    fn read<'a>(&'a mut self, dst: &'a mut [u8]) -> Read<'a, Self>
+    where
+        Self: Unpin,
+    {
+        read(self, dst)
+    }
+
+    /// Read exactly the amount of data needed to fill the provided buffer.
+    fn read_exact<'a>(&'a mut self, dst: &'a mut [u8]) -> ReadExact<'a, Self>
+    where
+        Self: Unpin,
+    {
+        read_exact(self, dst)
+    }
+
+    /// Read all bytes until EOF in this source, placing them into `dst`.
+    ///
+    /// On success the total number of bytes read is returned.
+    fn read_to_end<'a>(&'a mut self, dst: &'a mut Vec<u8>) -> ReadToEnd<'a, Self>
+    where
+        Self: Unpin,
+    {
+        read_to_end(self, dst)
+    }
+
+    /// Read all bytes until EOF in this source, placing them into `dst`.
+    ///
+    /// On success the total number of bytes read is returned.
+    fn read_to_string<'a>(&'a mut self, dst: &'a mut String) -> ReadToString<'a, Self>
+    where
+        Self: Unpin,
+    {
+        read_to_string(self, dst)
+    }
+}
+
+impl<R: AsyncRead + ?Sized> AsyncReadExt for R {}

tokio-io/src/io/async_write_ext.rs

@@ -0,0 +1,33 @@
+use crate::io::flush::{flush, Flush};
+use crate::io::write::{write, Write};
+use crate::io::write_all::{write_all, WriteAll};
+use crate::AsyncWrite;
+
+/// An extension trait which adds utility methods to `AsyncWrite` types.
+pub trait AsyncWriteExt: AsyncWrite {
+    /// Write a buffer into this writter, returning how many bytes were written.
+    fn write<'a>(&'a mut self, src: &'a [u8]) -> Write<'a, Self>
+    where
+        Self: Unpin,
+    {
+        write(self, src)
+    }
+
+    /// Attempt to write an entire buffer into this writter.
+    fn write_all<'a>(&'a mut self, src: &'a [u8]) -> WriteAll<'a, Self>
+    where
+        Self: Unpin,
+    {
+        write_all(self, src)
+    }
+
+    /// Flush the contents of this writer.
+    fn flush(&mut self) -> Flush<'_, Self>
+    where
+        Self: Unpin,
+    {
+        flush(self)
+    }
+}
+
+impl<W: AsyncWrite + ?Sized> AsyncWriteExt for W {}

tokio-io/src/io/copy.rs

@@ -0,0 +1,83 @@
+use crate::{AsyncRead, AsyncWrite};
+use futures_core::ready;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Copy<'a, R: ?Sized, W: ?Sized> {
+    reader: &'a mut R,
+    read_done: bool,
+    writer: &'a mut W,
+    pos: usize,
+    cap: usize,
+    amt: u64,
+    buf: Box<[u8]>,
+}
+
+pub(crate) fn copy<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> Copy<'a, R, W>
+where
+    R: AsyncRead + Unpin + ?Sized,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    Copy {
+        reader,
+        read_done: false,
+        writer,
+        amt: 0,
+        pos: 0,
+        cap: 0,
+        buf: Box::new([0; 2048]),
+    }
+}
+
+impl<'a, R, W> Future for Copy<'a, R, W>
+where
+    R: AsyncRead + Unpin + ?Sized,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<u64>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
+        loop {
+            // If our buffer is empty, then we need to read some data to
+            // continue.
+            if self.pos == self.cap && !self.read_done {
+                let me = &mut *self;
+                let n = ready!(Pin::new(&mut *me.reader).poll_read(cx, &mut me.buf))?;
+                if n == 0 {
+                    self.read_done = true;
+                } else {
+                    self.pos = 0;
+                    self.cap = n;
+                }
+            }
+
+            // If our buffer has some data, let's write it out!
+            while self.pos < self.cap {
+                let me = &mut *self;
+                let i = ready!(Pin::new(&mut *me.writer).poll_write(cx, &me.buf[me.pos..me.cap]))?;
+                if i == 0 {
+                    return Poll::Ready(Err(io::Error::new(
+                        io::ErrorKind::WriteZero,
+                        "write zero byte into writer",
+                    )));
+                } else {
+                    self.pos += i;
+                    self.amt += i as u64;
+                }
+            }
+
+            // If we've written al the data and we've seen EOF, flush out the
+            // data and finish the transfer.
+            // done with the entire transfer.
+            if self.pos == self.cap && self.read_done {
+                let me = &mut *self;
+                ready!(Pin::new(&mut *me.writer).poll_flush(cx))?;
+                return Poll::Ready(Ok(self.amt));
+            }
+        }
+    }
+}

tokio-io/src/io/flush.rs

@@ -0,0 +1,37 @@
+use crate::AsyncWrite;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A future used to fully flush an I/O object.
+///
+/// Created by the [`AsyncWriteExt::flush`] function.
+///
+/// [`flush`]: fn.flush.html
+#[derive(Debug)]
+pub struct Flush<'a, A: ?Sized> {
+    a: &'a mut A,
+}
+
+/// Creates a future which will entirely flush an I/O object.
+pub(super) fn flush<A>(a: &mut A) -> Flush<'_, A>
+where
+    A: AsyncWrite + Unpin + ?Sized,
+{
+    Flush { a }
+}
+
+impl<'a, A> Unpin for Flush<'a, A> where A: Unpin + ?Sized {}
+
+impl<A> Future for Flush<'_, A>
+where
+    A: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = &mut *self;
+        Pin::new(&mut *me.a).poll_flush(cx)
+    }
+}

tokio-io/src/io/lines.rs

@@ -0,0 +1,57 @@
+use super::read_line::read_line_internal;
+use crate::AsyncBufRead;
+
+use futures_core::{ready, Stream};
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Stream for the [`lines`](crate::io::AsyncBufReadExt::lines) method.
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Lines<R> {
+    reader: R,
+    buf: String,
+    bytes: Vec<u8>,
+    read: usize,
+}
+
+impl<R: Unpin> Unpin for Lines<R> {}
+
+pub(crate) fn lines<R>(reader: R) -> Lines<R>
+where
+    R: AsyncBufRead,
+{
+    Lines {
+        reader,
+        buf: String::new(),
+        bytes: Vec::new(),
+        read: 0,
+    }
+}
+
+impl<R: AsyncBufRead> Stream for Lines<R> {
+    type Item = io::Result<String>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let Self {
+            reader,
+            buf,
+            bytes,
+            read,
+        } = unsafe { self.get_unchecked_mut() };
+        let reader = unsafe { Pin::new_unchecked(reader) };
+        let n = ready!(read_line_internal(reader, cx, buf, bytes, read))?;
+        if n == 0 && buf.is_empty() {
+            return Poll::Ready(None);
+        }
+        if buf.ends_with('\n') {
+            buf.pop();
+            if buf.ends_with('\r') {
+                buf.pop();
+            }
+        }
+        Poll::Ready(Some(Ok(mem::replace(buf, String::new()))))
+    }
+}

tokio-io/src/io/mod.rs

@@ -0,0 +1,56 @@
+//! Asynchronous I/O.
+//!
+//! This module is the asynchronous version of `std::io`. Primarily, it
+//! defines two traits, [`AsyncRead`] and [`AsyncWrite`], which extend the
+//! `Read` and `Write` traits of the standard library.
+//!
+//! # AsyncRead and AsyncWrite
+//!
+//! [`AsyncRead`] and [`AsyncWrite`] must only be implemented for
+//! non-blocking I/O types that integrate with the futures type system. In
+//! other words, these types must never block the thread, and instead the
+//! current task is notified when the I/O resource is ready.
+//!
+//! # Standard input and output
+//!
+//! Tokio provides asynchronous APIs to standard [input], [output], and [error].
+//! These APIs are very similar to the ones provided by `std`, but they also
+//! implement [`AsyncRead`] and [`AsyncWrite`].
+//!
+//! Unlike *most* other Tokio APIs, the standard input / output APIs
+//! **must** be used from the context of the Tokio runtime as they require
+//! Tokio specific features to function.
+//!
+//! [input]: fn.stdin.html
+//! [output]: fn.stdout.html
+//! [error]: fn.stderr.html
+//!
+//! # `std` re-exports
+//!
+//! Additionally, [`Error`], [`ErrorKind`], and [`Result`] are re-exported
+//! from `std::io` for ease of use.
+//!
+//! [`AsyncRead`]: trait.AsyncRead.html
+//! [`AsyncWrite`]: trait.AsyncWrite.html
+//! [`Error`]: struct.Error.html
+//! [`ErrorKind`]: enum.ErrorKind.html
+//! [`Result`]: type.Result.html
+
+mod async_buf_read_ext;
+mod async_read_ext;
+mod async_write_ext;
+mod copy;
+mod flush;
+mod lines;
+mod read;
+mod read_exact;
+mod read_line;
+mod read_to_end;
+mod read_to_string;
+mod read_until;
+mod write;
+mod write_all;
+
+pub use self::async_buf_read_ext::AsyncBufReadExt;
+pub use self::async_read_ext::AsyncReadExt;
+pub use self::async_write_ext::AsyncWriteExt;

tokio-io/src/io/read.rs

@@ -0,0 +1,44 @@
+use crate::AsyncRead;
+use std::future::Future;
+use std::io;
+use std::marker::Unpin;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Tries to read some bytes directly into the given `buf` in asynchronous
+/// manner, returning a future type.
+///
+/// The returned future will resolve to both the I/O stream and the buffer
+/// as well as the number of bytes read once the read operation is completed.
+pub(crate) fn read<'a, R>(reader: &'a mut R, buf: &'a mut [u8]) -> Read<'a, R>
+where
+    R: AsyncRead + Unpin + ?Sized,
+{
+    Read { reader, buf }
+}
+
+/// 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],
+}
+
+// forward Unpin
+impl<'a, R: Unpin + ?Sized> Unpin for Read<'_, R> {}
+
+impl<R> Future for Read<'_, R>
+where
+    R: AsyncRead + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = &mut *self;
+        Pin::new(&mut *me.reader).poll_read(cx, me.buf)
+    }
+}

tokio-io/src/io/read_exact.rs

@@ -0,0 +1,68 @@
+use crate::AsyncRead;
+use futures_core::ready;
+use std::future::Future;
+use std::io;
+use std::marker::Unpin;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A future which can be used to easily read exactly enough bytes to fill
+/// a buffer.
+///
+/// Created by the [`read_exact`] function.
+///
+/// [`read_exact`]: fn.read_exact.html
+pub(crate) fn read_exact<'a, A>(reader: &'a mut A, buf: &'a mut [u8]) -> ReadExact<'a, A>
+where
+    A: AsyncRead + Unpin + ?Sized,
+{
+    ReadExact {
+        reader,
+        buf,
+        pos: 0,
+    }
+}
+
+/// Creates a future which will read exactly enough bytes to fill `buf`,
+/// returning an error if EOF is hit sooner.
+///
+/// On success the number of bytes is returned
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadExact<'a, A: ?Sized> {
+    reader: &'a mut A,
+    buf: &'a mut [u8],
+    pos: usize,
+}
+
+fn eof() -> io::Error {
+    io::Error::new(io::ErrorKind::UnexpectedEof, "early eof")
+}
+
+// forward Unpin
+impl<'a, A: Unpin + ?Sized> Unpin for ReadExact<'_, A> {}
+
+impl<A> Future for ReadExact<'_, A>
+where
+    A: AsyncRead + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        loop {
+            // if our buffer is empty, then we need to read some data to continue.
+            if self.pos < self.buf.len() {
+                let me = &mut *self;
+                let n = ready!(Pin::new(&mut *me.reader).poll_read(cx, &mut me.buf[me.pos..]))?;
+                me.pos += n;
+                if n == 0 {
+                    return Err(eof()).into();
+                }
+            }
+
+            if self.pos >= self.buf.len() {
+                return Poll::Ready(Ok(self.pos));
+            }
+        }
+    }
+}

tokio-io/src/io/read_line.rs

@@ -0,0 +1,71 @@
+use super::read_until::read_until_internal;
+use crate::AsyncBufRead;
+use futures_core::ready;
+use std::future::Future;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::str;
+use std::task::{Context, Poll};
+
+/// 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 + Unpin> {
+    reader: &'a mut R,
+    buf: &'a mut String,
+    bytes: Vec<u8>,
+    read: usize,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadLine<'_, R> {}
+
+pub(crate) fn read_line<'a, R>(reader: &'a mut R, buf: &'a mut String) -> ReadLine<'a, R>
+where
+    R: AsyncBufRead + ?Sized + Unpin,
+{
+    ReadLine {
+        reader,
+        bytes: unsafe { mem::replace(buf.as_mut_vec(), Vec::new()) },
+        buf,
+        read: 0,
+    }
+}
+
+pub(super) fn read_line_internal<R: AsyncBufRead + ?Sized>(
+    reader: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    buf: &mut String,
+    bytes: &mut Vec<u8>,
+    read: &mut usize,
+) -> Poll<io::Result<usize>> {
+    let ret = ready!(read_until_internal(reader, cx, b'\n', bytes, read));
+    if str::from_utf8(&bytes).is_err() {
+        Poll::Ready(ret.and_then(|_| {
+            Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stream did not contain valid UTF-8",
+            ))
+        }))
+    } else {
+        debug_assert!(buf.is_empty());
+        debug_assert_eq!(*read, 0);
+        // Safety: `bytes` is a valid UTF-8 because `str::from_utf8` returned `Ok`.
+        mem::swap(unsafe { buf.as_mut_vec() }, bytes);
+        Poll::Ready(ret)
+    }
+}
+
+impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadLine<'_, R> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self {
+            reader,
+            buf,
+            bytes,
+            read,
+        } = &mut *self;
+        read_line_internal(Pin::new(reader), cx, buf, bytes, read)
+    }
+}

tokio-io/src/io/read_to_end.rs

@@ -0,0 +1,99 @@
+use crate::AsyncRead;
+use futures_core::ready;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadToEnd<'a, R: ?Sized> {
+    reader: &'a mut R,
+    buf: &'a mut Vec<u8>,
+    start_len: usize,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadToEnd<'_, R> {}
+
+pub(crate) fn read_to_end<'a, R>(reader: &'a mut R, buf: &'a mut Vec<u8>) -> ReadToEnd<'a, R>
+where
+    R: AsyncRead + Unpin + ?Sized,
+{
+    let start_len = buf.len();
+    ReadToEnd {
+        reader,
+        buf,
+        start_len,
+    }
+}
+
+struct Guard<'a> {
+    buf: &'a mut Vec<u8>,
+    len: usize,
+}
+
+impl Drop for Guard<'_> {
+    fn drop(&mut self) {
+        unsafe {
+            self.buf.set_len(self.len);
+        }
+    }
+}
+
+// This uses an adaptive system to extend the vector when it fills. We want to
+// avoid paying to allocate and zero a huge chunk of memory if the reader only
+// has 4 bytes while still making large reads if the reader does have a ton
+// of data to return. Simply tacking on an extra DEFAULT_BUF_SIZE space every
+// time is 4,500 times (!) slower than this if the reader has a very small
+// amount of data to return.
+//
+// Because we're extending the buffer with uninitialized data for trusted
+// readers, we need to make sure to truncate that if any of this panics.
+pub(super) fn read_to_end_internal<R: AsyncRead + ?Sized>(
+    mut rd: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    buf: &mut Vec<u8>,
+    start_len: usize,
+) -> Poll<io::Result<usize>> {
+    let mut g = Guard {
+        len: buf.len(),
+        buf,
+    };
+    let ret;
+    loop {
+        if g.len == g.buf.len() {
+            unsafe {
+                g.buf.reserve(32);
+                let capacity = g.buf.capacity();
+                g.buf.set_len(capacity);
+                rd.prepare_uninitialized_buffer(&mut g.buf[g.len..]);
+            }
+        }
+
+        match ready!(rd.as_mut().poll_read(cx, &mut g.buf[g.len..])) {
+            Ok(0) => {
+                ret = Poll::Ready(Ok(g.len - start_len));
+                break;
+            }
+            Ok(n) => g.len += n,
+            Err(e) => {
+                ret = Poll::Ready(Err(e));
+                break;
+            }
+        }
+    }
+
+    ret
+}
+
+impl<A> Future for ReadToEnd<'_, A>
+where
+    A: AsyncRead + ?Sized + Unpin,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        read_to_end_internal(Pin::new(&mut this.reader), cx, this.buf, this.start_len)
+    }
+}

tokio-io/src/io/read_to_string.rs

@@ -0,0 +1,72 @@
+use super::read_to_end::read_to_end_internal;
+use crate::AsyncRead;
+use futures_core::ready;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::{io, mem, str};
+
+/// Future for the [`read_to_string`](super::AsyncReadExt::read_to_string) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadToString<'a, R: ?Sized + Unpin> {
+    reader: &'a mut R,
+    buf: &'a mut String,
+    bytes: Vec<u8>,
+    start_len: usize,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadToString<'_, R> {}
+
+pub(crate) fn read_to_string<'a, R>(reader: &'a mut R, buf: &'a mut String) -> ReadToString<'a, R>
+where
+    R: AsyncRead + ?Sized + Unpin,
+{
+    let start_len = buf.len();
+    ReadToString {
+        reader,
+        bytes: unsafe { mem::replace(buf.as_mut_vec(), Vec::new()) },
+        buf,
+        start_len,
+    }
+}
+
+fn read_to_string_internal<R: AsyncRead + ?Sized>(
+    reader: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    buf: &mut String,
+    bytes: &mut Vec<u8>,
+    start_len: usize,
+) -> Poll<io::Result<usize>> {
+    let ret = ready!(read_to_end_internal(reader, cx, bytes, start_len));
+    if str::from_utf8(&bytes).is_err() {
+        Poll::Ready(ret.and_then(|_| {
+            Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stream did not contain valid UTF-8",
+            ))
+        }))
+    } else {
+        debug_assert!(buf.is_empty());
+        // Safety: `bytes` is a valid UTF-8 because `str::from_utf8` returned `Ok`.
+        mem::swap(unsafe { buf.as_mut_vec() }, bytes);
+        Poll::Ready(ret)
+    }
+}
+
+impl<A> Future for ReadToString<'_, A>
+where
+    A: AsyncRead + ?Sized + Unpin,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self {
+            reader,
+            buf,
+            bytes,
+            start_len,
+        } = &mut *self;
+        read_to_string_internal(Pin::new(reader), cx, buf, bytes, *start_len)
+    }
+}

tokio-io/src/io/read_until.rs

@@ -0,0 +1,75 @@
+use crate::AsyncBufRead;
+use futures_core::ready;
+use std::future::Future;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Future for the [`read_until`](crate::io::AsyncBufReadExt::read_until) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadUntil<'a, R: ?Sized + Unpin> {
+    reader: &'a mut R,
+    byte: u8,
+    buf: &'a mut Vec<u8>,
+    read: usize,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadUntil<'_, R> {}
+
+pub(crate) fn read_until<'a, R>(
+    reader: &'a mut R,
+    byte: u8,
+    buf: &'a mut Vec<u8>,
+) -> ReadUntil<'a, R>
+where
+    R: AsyncBufRead + ?Sized + Unpin,
+{
+    ReadUntil {
+        reader,
+        byte,
+        buf,
+        read: 0,
+    }
+}
+
+pub(super) fn read_until_internal<R: AsyncBufRead + ?Sized>(
+    mut reader: Pin<&mut R>,
+    cx: &mut Context<'_>,
+    byte: u8,
+    buf: &mut Vec<u8>,
+    read: &mut usize,
+) -> Poll<io::Result<usize>> {
+    loop {
+        let (done, used) = {
+            let available = ready!(reader.as_mut().poll_fill_buf(cx))?;
+            if let Some(i) = memchr::memchr(byte, available) {
+                buf.extend_from_slice(&available[..=i]);
+                (true, i + 1)
+            } else {
+                buf.extend_from_slice(available);
+                (false, available.len())
+            }
+        };
+        reader.as_mut().consume(used);
+        *read += used;
+        if done || used == 0 {
+            return Poll::Ready(Ok(mem::replace(read, 0)));
+        }
+    }
+}
+
+impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadUntil<'_, R> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self {
+            reader,
+            byte,
+            buf,
+            read,
+        } = &mut *self;
+        read_until_internal(Pin::new(reader), cx, *byte, buf, read)
+    }
+}

tokio-io/src/io/write.rs

@@ -0,0 +1,37 @@
+use crate::AsyncWrite;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A future to write some of the buffer to an `AsyncWrite`.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Write<'a, W: ?Sized> {
+    writer: &'a mut W,
+    buf: &'a [u8],
+}
+
+/// Tries to write some bytes from the given `buf` to the writer in an
+/// asynchronous manner, returning a future.
+pub(crate) fn write<'a, W>(writer: &'a mut W, buf: &'a [u8]) -> Write<'a, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    Write { writer, buf }
+}
+
+// forward Unpin
+impl<'a, W: Unpin + ?Sized> Unpin for Write<'a, W> {}
+
+impl<W> Future for Write<'_, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
+        let me = &mut *self;
+        Pin::new(&mut *me.writer).poll_write(cx, me.buf)
+    }
+}

tokio-io/src/io/write_all.rs

@@ -0,0 +1,46 @@
+use crate::AsyncWrite;
+use futures_core::ready;
+use std::future::Future;
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct WriteAll<'a, W: ?Sized> {
+    writer: &'a mut W,
+    buf: &'a [u8],
+}
+
+pub(crate) fn write_all<'a, W>(writer: &'a mut W, buf: &'a [u8]) -> WriteAll<'a, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    WriteAll { writer, buf }
+}
+
+impl<W: ?Sized + Unpin> Unpin for WriteAll<'_, W> {}
+
+impl<W> Future for WriteAll<'_, W>
+where
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let me = &mut *self;
+        while !me.buf.is_empty() {
+            let n = ready!(Pin::new(&mut me.writer).poll_write(cx, me.buf))?;
+            {
+                let (_, rest) = mem::replace(&mut me.buf, &[]).split_at(n);
+                me.buf = rest;
+            }
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}

tokio-io/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio-io/0.1.12")]
+#![doc(html_root_url = "https://docs.rs/tokio-io/0.2.0")]
 #![deny(missing_debug_implementations, missing_docs, rust_2018_idioms)]
 #![cfg_attr(test, deny(warnings))]
 #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))]
@@ -15,9 +15,15 @@ mod async_buf_read;
 mod async_read;
 mod async_write;
 
+#[cfg(feature = "util")]
+mod io;
+
 pub use self::async_buf_read::AsyncBufRead;
 pub use self::async_read::AsyncRead;
 pub use self::async_write::AsyncWrite;
 
+#[cfg(feature = "util")]
+pub use self::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt};
+
 // Re-export `Buf` and `BufMut` since they are part of the API
 pub use bytes::{Buf, BufMut};