Merge #151

151: Split io into multiple files r=stjepang a=yoshuawuyts

Counterpart to #150, splits `io::read` and `io::write` into multiple files. This is useful to prevent a single file from becoming hard to navigate as we add more combinators. No other changes were made. Ref #131. Thanks!

Co-authored-by: Yoshua Wuyts <yoshuawuyts@gmail.com>

src/io/read/mod.rs

@@ -1,15 +1,21 @@
-use std::io::IoSliceMut;
+mod read;
+mod read_exact;
+mod read_to_end;
+mod read_to_string;
+mod read_vectored;
+
+use read::ReadFuture;
+use read_exact::ReadExactFuture;
+use read_to_end::{read_to_end_internal, ReadToEndFuture};
+use read_to_string::ReadToStringFuture;
+use read_vectored::ReadVectoredFuture;
+
+use std::io;
 use std::mem;
-use std::pin::Pin;
-use std::str;
 
 use cfg_if::cfg_if;
 use futures_io::AsyncRead;
 
-use crate::future::Future;
-use crate::io;
-use crate::task::{Context, Poll};
-
 cfg_if! {
     if #[cfg(feature = "docs")] {
         #[doc(hidden)]
@@ -80,7 +86,7 @@ pub trait Read {
     /// [`read`]: #tymethod.read
     fn read_vectored<'a>(
         &'a mut self,
-        bufs: &'a mut [IoSliceMut<'a>],
+        bufs: &'a mut [io::IoSliceMut<'a>],
     ) -> ret!('a, ReadVectoredFuture, io::Result<usize>)
     where
         Self: Unpin,
@@ -215,180 +221,3 @@ impl<T: AsyncRead + Unpin + ?Sized> Read for T {
         ReadFuture { reader: self, buf }
     }
 }
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct ReadFuture<'a, T: Unpin + ?Sized> {
-    reader: &'a mut T,
-    buf: &'a mut [u8],
-}
-
-impl<T: AsyncRead + Unpin + ?Sized> Future for ReadFuture<'_, T> {
-    type Output = io::Result<usize>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let Self { reader, buf } = &mut *self;
-        Pin::new(reader).poll_read(cx, buf)
-    }
-}
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct ReadVectoredFuture<'a, T: Unpin + ?Sized> {
-    reader: &'a mut T,
-    bufs: &'a mut [IoSliceMut<'a>],
-}
-
-impl<T: AsyncRead + Unpin + ?Sized> Future for ReadVectoredFuture<'_, T> {
-    type Output = io::Result<usize>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let Self { reader, bufs } = &mut *self;
-        Pin::new(reader).poll_read_vectored(cx, bufs)
-    }
-}
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct ReadToEndFuture<'a, T: Unpin + ?Sized> {
-    reader: &'a mut T,
-    buf: &'a mut Vec<u8>,
-    start_len: usize,
-}
-
-impl<T: AsyncRead + Unpin + ?Sized> Future for ReadToEndFuture<'_, T> {
-    type Output = io::Result<usize>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let Self {
-            reader,
-            buf,
-            start_len,
-        } = &mut *self;
-        read_to_end_internal(Pin::new(reader), cx, buf, *start_len)
-    }
-}
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct ReadToStringFuture<'a, T: Unpin + ?Sized> {
-    reader: &'a mut T,
-    buf: &'a mut String,
-    bytes: Vec<u8>,
-    start_len: usize,
-}
-
-impl<T: AsyncRead + Unpin + ?Sized> Future for ReadToStringFuture<'_, T> {
-    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;
-        let reader = Pin::new(reader);
-
-        let ret = futures_core::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)
-        }
-    }
-}
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct ReadExactFuture<'a, T: Unpin + ?Sized> {
-    reader: &'a mut T,
-    buf: &'a mut [u8],
-}
-
-impl<T: AsyncRead + Unpin + ?Sized> Future for ReadExactFuture<'_, T> {
-    type Output = io::Result<()>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let Self { reader, buf } = &mut *self;
-
-        while !buf.is_empty() {
-            let n = futures_core::ready!(Pin::new(&mut *reader).poll_read(cx, buf))?;
-            let (_, rest) = mem::replace(buf, &mut []).split_at_mut(n);
-            *buf = rest;
-
-            if n == 0 {
-                return Poll::Ready(Err(io::ErrorKind::UnexpectedEof.into()));
-            }
-        }
-
-        Poll::Ready(Ok(()))
-    }
-}
-
-// 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 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>> {
-    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);
-            }
-        }
-    }
-
-    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.initializer().initialize(&mut g.buf[g.len..]);
-            }
-        }
-
-        match futures_core::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
-}

src/io/read/read.rs

@@ -0,0 +1,23 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io;
+use std::pin::Pin;
+
+use futures_io::AsyncRead;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct ReadFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) reader: &'a mut T,
+    pub(crate) buf: &'a mut [u8],
+}
+
+impl<T: AsyncRead + Unpin + ?Sized> Future for ReadFuture<'_, T> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self { reader, buf } = &mut *self;
+        Pin::new(reader).poll_read(cx, buf)
+    }
+}

src/io/read/read_exact.rs

@@ -0,0 +1,35 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io;
+use std::mem;
+use std::pin::Pin;
+
+use futures_io::AsyncRead;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct ReadExactFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) reader: &'a mut T,
+    pub(crate) buf: &'a mut [u8],
+}
+
+impl<T: AsyncRead + Unpin + ?Sized> Future for ReadExactFuture<'_, T> {
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self { reader, buf } = &mut *self;
+
+        while !buf.is_empty() {
+            let n = futures_core::ready!(Pin::new(&mut *reader).poll_read(cx, buf))?;
+            let (_, rest) = mem::replace(buf, &mut []).split_at_mut(n);
+            *buf = rest;
+
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::UnexpectedEof.into()));
+            }
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}

src/io/read/read_to_end.rs

@@ -0,0 +1,87 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io;
+use std::pin::Pin;
+
+use futures_io::AsyncRead;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct ReadToEndFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) reader: &'a mut T,
+    pub(crate) buf: &'a mut Vec<u8>,
+    pub(crate) start_len: usize,
+}
+
+impl<T: AsyncRead + Unpin + ?Sized> Future for ReadToEndFuture<'_, T> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self {
+            reader,
+            buf,
+            start_len,
+        } = &mut *self;
+        read_to_end_internal(Pin::new(reader), cx, buf, *start_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 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>> {
+    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);
+            }
+        }
+    }
+
+    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.initializer().initialize(&mut g.buf[g.len..]);
+            }
+        }
+
+        match futures_core::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
+}

src/io/read/read_to_string.rs

@@ -0,0 +1,48 @@
+use super::read_to_end_internal;
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::str;
+
+use futures_io::AsyncRead;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct ReadToStringFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) reader: &'a mut T,
+    pub(crate) buf: &'a mut String,
+    pub(crate) bytes: Vec<u8>,
+    pub(crate) start_len: usize,
+}
+
+impl<T: AsyncRead + Unpin + ?Sized> Future for ReadToStringFuture<'_, T> {
+    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;
+        let reader = Pin::new(reader);
+
+        let ret = futures_core::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)
+        }
+    }
+}

src/io/read/read_vectored.rs

@@ -0,0 +1,25 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io::IoSliceMut;
+use std::pin::Pin;
+
+use futures_io::AsyncRead;
+
+use crate::io;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct ReadVectoredFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) reader: &'a mut T,
+    pub(crate) bufs: &'a mut [IoSliceMut<'a>],
+}
+
+impl<T: AsyncRead + Unpin + ?Sized> Future for ReadVectoredFuture<'_, T> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self { reader, bufs } = &mut *self;
+        Pin::new(reader).poll_read_vectored(cx, bufs)
+    }
+}

src/io/write/flush.rs

@@ -0,0 +1,21 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io;
+use std::pin::Pin;
+
+use futures_io::AsyncWrite;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct FlushFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) writer: &'a mut T,
+}
+
+impl<T: AsyncWrite + Unpin + ?Sized> Future for FlushFuture<'_, T> {
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut *self.writer).poll_flush(cx)
+    }
+}

src/io/write/mod.rs

@@ -1,14 +1,18 @@
-use std::io::IoSlice;
-use std::mem;
-use std::pin::Pin;
+mod flush;
+mod write;
+mod write_all;
+mod write_vectored;
+
+use flush::FlushFuture;
+use write::WriteFuture;
+use write_all::WriteAllFuture;
+use write_vectored::WriteVectoredFuture;
+
+use std::io;
 
 use cfg_if::cfg_if;
 use futures_io::AsyncWrite;
 
-use crate::future::Future;
-use crate::io;
-use crate::task::{Context, Poll};
-
 cfg_if! {
     if #[cfg(feature = "docs")] {
         #[doc(hidden)]
@@ -95,7 +99,7 @@ pub trait Write {
     /// [`write`]: #tymethod.write
     fn write_vectored<'a>(
         &'a mut self,
-        bufs: &'a [IoSlice<'a>],
+        bufs: &'a [io::IoSlice<'a>],
     ) -> ret!('a, WriteVectoredFuture, io::Result<usize>)
     where
         Self: Unpin,
@@ -140,76 +144,3 @@ impl<T: AsyncWrite + Unpin + ?Sized> Write for T {
         FlushFuture { writer: self }
     }
 }
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct WriteFuture<'a, T: Unpin + ?Sized> {
-    writer: &'a mut T,
-    buf: &'a [u8],
-}
-
-impl<T: AsyncWrite + Unpin + ?Sized> Future for WriteFuture<'_, T> {
-    type Output = io::Result<usize>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let buf = self.buf;
-        Pin::new(&mut *self.writer).poll_write(cx, buf)
-    }
-}
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct FlushFuture<'a, T: Unpin + ?Sized> {
-    writer: &'a mut T,
-}
-
-impl<T: AsyncWrite + Unpin + ?Sized> Future for FlushFuture<'_, T> {
-    type Output = io::Result<()>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        Pin::new(&mut *self.writer).poll_flush(cx)
-    }
-}
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct WriteVectoredFuture<'a, T: Unpin + ?Sized> {
-    writer: &'a mut T,
-    bufs: &'a [IoSlice<'a>],
-}
-
-impl<T: AsyncWrite + Unpin + ?Sized> Future for WriteVectoredFuture<'_, T> {
-    type Output = io::Result<usize>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let bufs = self.bufs;
-        Pin::new(&mut *self.writer).poll_write_vectored(cx, bufs)
-    }
-}
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct WriteAllFuture<'a, T: Unpin + ?Sized> {
-    writer: &'a mut T,
-    buf: &'a [u8],
-}
-
-impl<T: AsyncWrite + Unpin + ?Sized> Future for WriteAllFuture<'_, T> {
-    type Output = io::Result<()>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let Self { writer, buf } = &mut *self;
-
-        while !buf.is_empty() {
-            let n = futures_core::ready!(Pin::new(&mut **writer).poll_write(cx, buf))?;
-            let (_, rest) = mem::replace(buf, &[]).split_at(n);
-            *buf = rest;
-
-            if n == 0 {
-                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
-            }
-        }
-
-        Poll::Ready(Ok(()))
-    }
-}

src/io/write/write.rs

@@ -0,0 +1,23 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io;
+use std::pin::Pin;
+
+use futures_io::AsyncWrite;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct WriteFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) writer: &'a mut T,
+    pub(crate) buf: &'a [u8],
+}
+
+impl<T: AsyncWrite + Unpin + ?Sized> Future for WriteFuture<'_, T> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let buf = self.buf;
+        Pin::new(&mut *self.writer).poll_write(cx, buf)
+    }
+}

src/io/write/write_all.rs

@@ -0,0 +1,35 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io;
+use std::mem;
+use std::pin::Pin;
+
+use futures_io::AsyncWrite;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct WriteAllFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) writer: &'a mut T,
+    pub(crate) buf: &'a [u8],
+}
+
+impl<T: AsyncWrite + Unpin + ?Sized> Future for WriteAllFuture<'_, T> {
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self { writer, buf } = &mut *self;
+
+        while !buf.is_empty() {
+            let n = futures_core::ready!(Pin::new(&mut **writer).poll_write(cx, buf))?;
+            let (_, rest) = mem::replace(buf, &[]).split_at(n);
+            *buf = rest;
+
+            if n == 0 {
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+        }
+
+        Poll::Ready(Ok(()))
+    }
+}

src/io/write/write_vectored.rs

@@ -0,0 +1,24 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::io;
+use std::io::IoSlice;
+use std::pin::Pin;
+
+use futures_io::AsyncWrite;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct WriteVectoredFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) writer: &'a mut T,
+    pub(crate) bufs: &'a [IoSlice<'a>],
+}
+
+impl<T: AsyncWrite + Unpin + ?Sized> Future for WriteVectoredFuture<'_, T> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let bufs = self.bufs;
+        Pin::new(&mut *self.writer).poll_write_vectored(cx, bufs)
+    }
+}