io: remove unsafe pin-projections and remove manual Unpin implementations (#1588)

* Removes most pin-projection related unsafe code.

* Removes manual Unpin implementations.
  As references always implement Unpin, there is no need to implement
  Unpin manually.

* Adds tests to check that Unpin requirement does not change accidentally 
  because changing Unpin requirements will be breaking changes.

tokio-io/Cargo.toml

@@ -20,14 +20,13 @@ Core I/O primitives for asynchronous I/O in Rust.
 categories = ["asynchronous"]
 
 [features]
-util = ["memchr", "pin-utils", "pin-project"]
+util = ["memchr", "pin-project"]
 
 [dependencies]
 bytes = "0.4.7"
 log = "0.4"
 futures-core-preview = "=0.3.0-alpha.18"
 memchr = { version = "2.2", optional = true }
-pin-utils = { version = "=0.1.0-alpha.4", optional = true }
 pin-project = { version = "=0.4.0-beta.1", optional = true }
 
 [dev-dependencies]

tokio-io/src/io/buf_reader.rs

@@ -1,7 +1,7 @@
 use super::DEFAULT_BUF_SIZE;
 use crate::{AsyncBufRead, AsyncRead, AsyncWrite};
 use futures_core::ready;
-use pin_utils::{unsafe_pinned, unsafe_unpinned};
+use pin_project::{pin_project, project};
 use std::io::{self, Read};
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -23,7 +23,9 @@ use std::{cmp, fmt};
 /// discarded. Creating multiple instances of a `BufReader` on the same
 /// stream can cause data loss.
 // TODO: Examples
+#[pin_project]
 pub struct BufReader<R> {
+    #[pin]
     inner: R,
     buf: Box<[u8]>,
     pos: usize,
@@ -31,10 +33,6 @@ pub struct BufReader<R> {
 }
 
 impl<R: AsyncRead> BufReader<R> {
-    unsafe_pinned!(inner: R);
-    unsafe_unpinned!(pos: usize);
-    unsafe_unpinned!(cap: usize);
-
     /// 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 {
@@ -74,7 +72,7 @@ impl<R: AsyncRead> BufReader<R> {
     ///
     /// It is inadvisable to directly read from the underlying reader.
     pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R> {
-        self.inner()
+        self.project().inner
     }
 
     /// Consumes this `BufWriter`, returning the underlying reader.
@@ -93,9 +91,10 @@ impl<R: AsyncRead> BufReader<R> {
 
     /// Invalidates all data in the internal buffer.
     #[inline]
-    fn discard_buffer(mut self: Pin<&mut Self>) {
-        *self.as_mut().pos() = 0;
-        *self.cap() = 0;
+    fn discard_buffer(self: Pin<&mut Self>) {
+        let me = self.project();
+        *me.pos = 0;
+        *me.cap = 0;
     }
 }
 
@@ -109,7 +108,7 @@ impl<R: AsyncRead> AsyncRead for BufReader<R> {
         // (larger than our internal buffer), bypass our internal buffer
         // entirely.
         if self.pos == self.cap && buf.len() >= self.buf.len() {
-            let res = ready!(self.as_mut().inner().poll_read(cx, buf));
+            let res = ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
             self.discard_buffer();
             return Poll::Ready(res);
         }
@@ -126,14 +125,15 @@ impl<R: AsyncRead> AsyncRead for BufReader<R> {
 }
 
 impl<R: AsyncRead> AsyncBufRead for BufReader<R> {
+    #[project]
     fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
-        let Self {
+        #[project]
+        let BufReader {
             inner,
             buf,
             cap,
             pos,
-        } = unsafe { self.get_unchecked_mut() };
-        let mut inner = unsafe { Pin::new_unchecked(inner) };
+        } = self.project();
 
         // If we've reached the end of our internal buffer then we need to fetch
         // some more data from the underlying reader.
@@ -141,14 +141,15 @@ impl<R: AsyncRead> AsyncBufRead for BufReader<R> {
         // to tell the compiler that the pos..cap slice is always valid.
         if *pos >= *cap {
             debug_assert!(*pos == *cap);
-            *cap = ready!(inner.as_mut().poll_read(cx, buf))?;
+            *cap = ready!(inner.poll_read(cx, buf))?;
             *pos = 0;
         }
         Poll::Ready(Ok(&buf[*pos..*cap]))
     }
 
-    fn consume(mut self: Pin<&mut Self>, amt: usize) {
-        *self.as_mut().pos() = cmp::min(self.pos + amt, self.cap);
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        let me = self.project();
+        *me.pos = cmp::min(*me.pos + amt, *me.cap);
     }
 }
 
@@ -170,7 +171,7 @@ impl<R: AsyncRead + AsyncWrite> AsyncWrite for BufReader<R> {
     }
 }
 
-impl<R: AsyncRead + fmt::Debug> fmt::Debug for BufReader<R> {
+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)
@@ -181,3 +182,13 @@ impl<R: AsyncRead + fmt::Debug> fmt::Debug for BufReader<R> {
             .finish()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<BufReader<()>>();
+    }
+}

tokio-io/src/io/buf_stream.rs

@@ -16,7 +16,7 @@ use std::{
 /// one in the other so that both directions are buffered. See their documentation for details.
 #[pin_project]
 #[derive(Debug)]
-pub struct BufStream<RW: AsyncRead + AsyncWrite>(#[pin] BufReader<BufWriter<RW>>);
+pub struct BufStream<RW>(#[pin] BufReader<BufWriter<RW>>);
 
 impl<RW: AsyncRead + AsyncWrite> BufStream<RW> {
     /// Wrap a type in both [`BufWriter`] and [`BufReader`].
@@ -69,3 +69,13 @@ impl<RW: AsyncBufRead + AsyncRead + AsyncWrite> AsyncBufRead for BufStream<RW> {
         self.project().0.consume(amt)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<BufStream<()>>();
+    }
+}

tokio-io/src/io/buf_writer.rs

@@ -1,7 +1,7 @@
 use super::DEFAULT_BUF_SIZE;
 use crate::{AsyncBufRead, AsyncRead, AsyncWrite};
 use futures_core::ready;
-use pin_utils::{unsafe_pinned, unsafe_unpinned};
+use pin_project::{pin_project, project};
 use std::fmt;
 use std::io::{self, Write};
 use std::pin::Pin;
@@ -28,16 +28,15 @@ use std::task::{Context, Poll};
 /// [`flush`]: super::AsyncWriteExt::flush
 ///
 // TODO: Examples
+#[pin_project]
 pub struct BufWriter<W> {
+    #[pin]
     inner: W,
     buf: Vec<u8>,
     written: usize,
 }
 
 impl<W: AsyncWrite> BufWriter<W> {
-    unsafe_pinned!(inner: W);
-    unsafe_unpinned!(buf: Vec<u8>);
-
     /// 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 {
@@ -53,13 +52,14 @@ impl<W: AsyncWrite> BufWriter<W> {
         }
     }
 
+    #[project]
     fn flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        let Self {
-            inner,
+        #[project]
+        let BufWriter {
+            mut inner,
             buf,
             written,
-        } = unsafe { self.get_unchecked_mut() };
-        let mut inner = unsafe { Pin::new_unchecked(inner) };
+        } = self.project();
 
         let len = buf.len();
         let mut ret = Ok(());
@@ -102,7 +102,7 @@ impl<W: AsyncWrite> BufWriter<W> {
     ///
     /// It is inadvisable to directly write to the underlying writer.
     pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut W> {
-        self.inner()
+        self.project().inner
     }
 
     /// Consumes this `BufWriter`, returning the underlying writer.
@@ -127,21 +127,23 @@ impl<W: AsyncWrite> AsyncWrite for BufWriter<W> {
         if self.buf.len() + buf.len() > self.buf.capacity() {
             ready!(self.as_mut().flush_buf(cx))?;
         }
-        if buf.len() >= self.buf.capacity() {
-            self.inner().poll_write(cx, buf)
+
+        let me = self.project();
+        if buf.len() >= me.buf.capacity() {
+            me.inner.poll_write(cx, buf)
         } else {
-            Poll::Ready(self.buf().write(buf))
+            Poll::Ready(me.buf.write(buf))
         }
     }
 
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         ready!(self.as_mut().flush_buf(cx))?;
-        self.inner().poll_flush(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.inner().poll_shutdown(cx)
+        self.get_pin_mut().poll_shutdown(cx)
     }
 }
 
@@ -170,7 +172,7 @@ impl<W: AsyncWrite + AsyncBufRead> AsyncBufRead for BufWriter<W> {
     }
 }
 
-impl<W: AsyncWrite + fmt::Debug> fmt::Debug for BufWriter<W> {
+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)
@@ -182,3 +184,13 @@ impl<W: AsyncWrite + fmt::Debug> fmt::Debug for BufWriter<W> {
             .finish()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<BufWriter<()>>();
+    }
+}

tokio-io/src/io/chain.rs

@@ -1,26 +1,22 @@
 use crate::{AsyncBufRead, AsyncRead};
 use futures_core::ready;
-use pin_utils::{unsafe_pinned, unsafe_unpinned};
+use pin_project::{pin_project, project};
 use std::fmt;
 use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
 /// Stream for the [`chain`](super::AsyncReadExt::chain) method.
+#[pin_project]
 #[must_use = "streams do nothing unless polled"]
 pub struct Chain<T, U> {
+    #[pin]
     first: T,
+    #[pin]
     second: U,
     done_first: bool,
 }
 
-impl<T, U> Unpin for Chain<T, U>
-where
-    T: Unpin,
-    U: Unpin,
-{
-}
-
 pub(super) fn chain<T, U>(first: T, second: U) -> Chain<T, U>
 where
     T: AsyncRead,
@@ -38,10 +34,6 @@ where
     T: AsyncRead,
     U: AsyncRead,
 {
-    unsafe_pinned!(first: T);
-    unsafe_pinned!(second: U);
-    unsafe_unpinned!(done_first: bool);
-
     /// Gets references to the underlying readers in this `Chain`.
     pub fn get_ref(&self) -> (&T, &U) {
         (&self.first, &self.second)
@@ -62,10 +54,8 @@ where
     /// 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>) {
-        unsafe {
-            let Self { first, second, .. } = self.get_unchecked_mut();
-            (Pin::new_unchecked(first), Pin::new_unchecked(second))
-        }
+        let me = self.project();
+        (me.first, me.second)
     }
 
     /// Consumes the `Chain`, returning the wrapped readers.
@@ -93,17 +83,19 @@ where
     U: AsyncRead,
 {
     fn poll_read(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &mut [u8],
     ) -> Poll<io::Result<usize>> {
-        if !self.done_first {
-            match ready!(self.as_mut().first().poll_read(cx, buf)?) {
-                0 if !buf.is_empty() => *self.as_mut().done_first() = true,
+        let me = self.project();
+
+        if !*me.done_first {
+            match ready!(me.first.poll_read(cx, buf)?) {
+                0 if !buf.is_empty() => *me.done_first = true,
                 n => return Poll::Ready(Ok(n)),
             }
         }
-        self.second().poll_read(cx, buf)
+        me.second.poll_read(cx, buf)
     }
 }
 
@@ -112,14 +104,14 @@ where
     T: AsyncBufRead,
     U: AsyncBufRead,
 {
+    #[project]
     fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
-        let Self {
+        #[project]
+        let Chain {
             first,
             second,
             done_first,
-        } = unsafe { self.get_unchecked_mut() };
-        let first = unsafe { Pin::new_unchecked(first) };
-        let second = unsafe { Pin::new_unchecked(second) };
+        } = self.project();
 
         if !*done_first {
             match ready!(first.poll_fill_buf(cx)?) {
@@ -133,10 +125,21 @@ where
     }
 
     fn consume(self: Pin<&mut Self>, amt: usize) {
-        if !self.done_first {
-            self.first().consume(amt)
+        let me = self.project();
+        if !*me.done_first {
+            me.first.consume(amt)
         } else {
-            self.second().consume(amt)
+            me.second.consume(amt)
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Chain<(), ()>>();
+    }
+}

tokio-io/src/io/copy.rs

@@ -81,3 +81,14 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Copy<'_, PhantomPinned, PhantomPinned>>();
+    }
+}

tokio-io/src/io/flush.rs

@@ -20,8 +20,6 @@ where
     Flush { a }
 }
 
-impl<A> Unpin for Flush<'_, A> where A: Unpin + ?Sized {}
-
 impl<A> Future for Flush<'_, A>
 where
     A: AsyncWrite + Unpin + ?Sized,
@@ -33,3 +31,14 @@ where
         Pin::new(&mut *me.a).poll_flush(cx)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Flush<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/lines.rs

@@ -2,23 +2,24 @@ use super::read_line::read_line_internal;
 use crate::AsyncBufRead;
 
 use futures_core::{ready, Stream};
+use pin_project::{pin_project, project};
 use std::io;
 use std::mem;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
 /// Stream for the [`lines`](crate::io::AsyncBufReadExt::lines) method.
+#[pin_project]
 #[derive(Debug)]
 #[must_use = "streams do nothing unless polled"]
 pub struct Lines<R> {
+    #[pin]
     reader: R,
     buf: String,
     bytes: Vec<u8>,
     read: usize,
 }
 
-impl<R: Unpin> Unpin for Lines<R> {}
-
 pub(crate) fn lines<R>(reader: R) -> Lines<R>
 where
     R: AsyncBufRead,
@@ -34,14 +35,16 @@ where
 impl<R: AsyncBufRead> Stream for Lines<R> {
     type Item = io::Result<String>;
 
+    #[project]
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let Self {
+        #[project]
+        let Lines {
             reader,
             buf,
             bytes,
             read,
-        } = unsafe { self.get_unchecked_mut() };
-        let reader = unsafe { Pin::new_unchecked(reader) };
+        } = self.project();
+
         let n = ready!(read_line_internal(reader, cx, buf, bytes, read))?;
         if n == 0 && buf.is_empty() {
             return Poll::Ready(None);
@@ -55,3 +58,13 @@ impl<R: AsyncBufRead> Stream for Lines<R> {
         Poll::Ready(Some(Ok(mem::replace(buf, String::new()))))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Lines<()>>();
+    }
+}

tokio-io/src/io/read.rs

@@ -28,9 +28,6 @@ pub struct Read<'a, R: ?Sized> {
     buf: &'a mut [u8],
 }
 
-// forward Unpin
-impl<R: Unpin + ?Sized> Unpin for Read<'_, R> {}
-
 impl<R> Future for Read<'_, R>
 where
     R: AsyncRead + Unpin + ?Sized,
@@ -42,3 +39,14 @@ where
         Pin::new(&mut *me.reader).poll_read(cx, me.buf)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Read<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/read_exact.rs

@@ -37,9 +37,6 @@ fn eof() -> io::Error {
     io::Error::new(io::ErrorKind::UnexpectedEof, "early eof")
 }
 
-// forward Unpin
-impl<A: Unpin + ?Sized> Unpin for ReadExact<'_, A> {}
-
 impl<A> Future for ReadExact<'_, A>
 where
     A: AsyncRead + Unpin + ?Sized,
@@ -64,3 +61,14 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadExact<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/read_line.rs

@@ -11,15 +11,13 @@ 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> {
+pub struct ReadLine<'a, R: ?Sized> {
     reader: &'a mut R,
     buf: &'a mut String,
     bytes: Vec<u8>,
     read: usize,
 }
 
-impl<R: ?Sized + Unpin> Unpin for ReadLine<'_, R> {}
-
 pub(crate) fn read_line<'a, R>(reader: &'a mut R, buf: &'a mut String) -> ReadLine<'a, R>
 where
     R: AsyncBufRead + ?Sized + Unpin,
@@ -69,3 +67,14 @@ impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadLine<'_, R> {
         read_line_internal(Pin::new(reader), cx, buf, bytes, read)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadLine<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/read_to_end.rs

@@ -13,8 +13,6 @@ pub struct ReadToEnd<'a, R: ?Sized> {
     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,
@@ -97,3 +95,14 @@ where
         read_to_end_internal(Pin::new(&mut this.reader), cx, this.buf, this.start_len)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadToEnd<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/read_to_string.rs

@@ -9,15 +9,13 @@ 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> {
+pub struct ReadToString<'a, R: ?Sized> {
     reader: &'a mut R,
     buf: &'a mut String,
     bytes: Vec<u8>,
     start_len: usize,
 }
 
-impl<R: ?Sized + Unpin> Unpin for ReadToString<'_, R> {}
-
 pub(crate) fn read_to_string<'a, R>(reader: &'a mut R, buf: &'a mut String) -> ReadToString<'a, R>
 where
     R: AsyncRead + ?Sized + Unpin,
@@ -70,3 +68,14 @@ where
         read_to_string_internal(Pin::new(reader), cx, buf, bytes, *start_len)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadToString<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/read_until.rs

@@ -9,15 +9,13 @@ 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> {
+pub struct ReadUntil<'a, R: ?Sized> {
     reader: &'a mut R,
     byte: u8,
     buf: &'a mut Vec<u8>,
     read: usize,
 }
 
-impl<R: ?Sized + Unpin> Unpin for ReadUntil<'_, R> {}
-
 pub(crate) fn read_until<'a, R>(
     reader: &'a mut R,
     byte: u8,
@@ -73,3 +71,14 @@ impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadUntil<'_, R> {
         read_until_internal(Pin::new(reader), cx, *byte, buf, read)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<ReadUntil<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/shutdown.rs

@@ -20,8 +20,6 @@ where
     Shutdown { a }
 }
 
-impl<A> Unpin for Shutdown<'_, A> where A: Unpin + ?Sized {}
-
 impl<A> Future for Shutdown<'_, A>
 where
     A: AsyncWrite + Unpin + ?Sized,
@@ -33,3 +31,14 @@ where
         Pin::new(&mut *me.a).poll_shutdown(cx)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Shutdown<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/take.rs

@@ -1,21 +1,21 @@
 use crate::{AsyncBufRead, AsyncRead};
 use futures_core::ready;
-use pin_utils::{unsafe_pinned, unsafe_unpinned};
+use pin_project::{pin_project, project};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::{cmp, io};
 
 /// Stream for the [`take`](super::AsyncReadExt::take) method.
+#[pin_project]
 #[derive(Debug)]
 #[must_use = "streams do nothing unless you `.await` or poll them"]
 pub struct Take<R> {
+    #[pin]
     inner: R,
     // Add '_' to avoid conflicts with `limit` method.
     limit_: u64,
 }
 
-impl<R: Unpin> Unpin for Take<R> {}
-
 pub(super) fn take<R: AsyncRead>(inner: R, limit: u64) -> Take<R> {
     Take {
         inner,
@@ -24,9 +24,6 @@ pub(super) fn take<R: AsyncRead>(inner: R, limit: u64) -> Take<R> {
 }
 
 impl<R: AsyncRead> Take<R> {
-    unsafe_pinned!(inner: R);
-    unsafe_unpinned!(limit_: u64);
-
     /// Returns the remaining number of bytes that can be
     /// read before this instance will return EOF.
     ///
@@ -66,7 +63,7 @@ impl<R: AsyncRead> Take<R> {
     /// 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.inner()
+        self.project().inner
     }
 
     /// Consumes the `Take`, returning the wrapped reader.
@@ -81,7 +78,7 @@ impl<R: AsyncRead> AsyncRead for Take<R> {
     }
 
     fn poll_read(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &mut [u8],
     ) -> Poll<Result<usize, io::Error>> {
@@ -89,17 +86,19 @@ impl<R: AsyncRead> AsyncRead for Take<R> {
             return Poll::Ready(Ok(0));
         }
 
-        let max = std::cmp::min(buf.len() as u64, self.limit_) as usize;
-        let n = ready!(self.as_mut().inner().poll_read(cx, &mut buf[..max]))?;
-        *self.as_mut().limit_() -= n as u64;
+        let me = self.project();
+        let max = std::cmp::min(buf.len() as u64, *me.limit_) as usize;
+        let n = ready!(me.inner.poll_read(cx, &mut buf[..max]))?;
+        *me.limit_ -= n as u64;
         Poll::Ready(Ok(n))
     }
 }
 
 impl<R: AsyncBufRead> AsyncBufRead for Take<R> {
+    #[project]
     fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
-        let Self { inner, limit_ } = unsafe { self.get_unchecked_mut() };
-        let inner = unsafe { Pin::new_unchecked(inner) };
+        #[project]
+        let Take { inner, limit_ } = self.project();
 
         // Don't call into inner reader at all at EOF because it may still block
         if *limit_ == 0 {
@@ -111,10 +110,21 @@ impl<R: AsyncBufRead> AsyncBufRead for Take<R> {
         Poll::Ready(Ok(&buf[..cap]))
     }
 
-    fn consume(mut self: Pin<&mut Self>, amt: usize) {
+    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, self.limit_) as usize;
-        *self.as_mut().limit_() -= amt as u64;
-        self.inner().consume(amt);
+        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<()>>();
     }
 }

tokio-io/src/io/write.rs

@@ -21,9 +21,6 @@ where
     Write { writer, buf }
 }
 
-// forward Unpin
-impl<W: Unpin + ?Sized> Unpin for Write<'_, W> {}
-
 impl<W> Future for Write<'_, W>
 where
     W: AsyncWrite + Unpin + ?Sized,
@@ -35,3 +32,14 @@ where
         Pin::new(&mut *me.writer).poll_write(cx, me.buf)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<Write<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/io/write_all.rs

@@ -20,8 +20,6 @@ where
     WriteAll { writer, buf }
 }
 
-impl<W: ?Sized + Unpin> Unpin for WriteAll<'_, W> {}
-
 impl<W> Future for WriteAll<'_, W>
 where
     W: AsyncWrite + Unpin + ?Sized,
@@ -44,3 +42,14 @@ where
         Poll::Ready(Ok(()))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        use std::marker::PhantomPinned;
+        crate::is_unpin::<WriteAll<'_, PhantomPinned>>();
+    }
+}

tokio-io/src/lib.rs

@@ -38,3 +38,7 @@ pub use self::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufS
 
 // Re-export `Buf` and `BufMut` since they are part of the API
 pub use bytes::{Buf, BufMut};
+
+#[cfg(feature = "util")]
+#[cfg(test)]
+fn is_unpin<T: Unpin>() {}