diff --git a/tokio-io/Cargo.toml b/tokio-io/Cargo.toml index 65f119e4a..9030e23ae 100644 --- a/tokio-io/Cargo.toml +++ b/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] diff --git a/tokio-io/src/io/buf_reader.rs b/tokio-io/src/io/buf_reader.rs index 738f5a5a2..a6afd7b37 100644 --- a/tokio-io/src/io/buf_reader.rs +++ b/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 { + #[pin] inner: R, buf: Box<[u8]>, pos: usize, @@ -31,10 +33,6 @@ pub struct BufReader { } impl BufReader { - 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 BufReader { /// /// 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 BufReader { /// 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 AsyncRead for BufReader { // (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 AsyncRead for BufReader { } impl AsyncBufRead for BufReader { + #[project] fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - 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 AsyncBufRead for BufReader { // 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 AsyncWrite for BufReader { } } -impl fmt::Debug for BufReader { +impl fmt::Debug for BufReader { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("BufReader") .field("reader", &self.inner) @@ -181,3 +182,13 @@ impl fmt::Debug for BufReader { .finish() } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn assert_unpin() { + crate::is_unpin::>(); + } +} diff --git a/tokio-io/src/io/buf_stream.rs b/tokio-io/src/io/buf_stream.rs index 88624f192..5ffe15e9d 100644 --- a/tokio-io/src/io/buf_stream.rs +++ b/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(#[pin] BufReader>); +pub struct BufStream(#[pin] BufReader>); impl BufStream { /// Wrap a type in both [`BufWriter`] and [`BufReader`]. @@ -69,3 +69,13 @@ impl AsyncBufRead for BufStream { self.project().0.consume(amt) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn assert_unpin() { + crate::is_unpin::>(); + } +} diff --git a/tokio-io/src/io/buf_writer.rs b/tokio-io/src/io/buf_writer.rs index 2e8613b6e..5059d3d74 100644 --- a/tokio-io/src/io/buf_writer.rs +++ b/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 { + #[pin] inner: W, buf: Vec, written: usize, } impl BufWriter { - unsafe_pinned!(inner: W); - unsafe_unpinned!(buf: Vec); - /// 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 BufWriter { } } + #[project] fn flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - 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 BufWriter { /// /// 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 AsyncWrite for BufWriter { 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> { 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> { ready!(self.as_mut().flush_buf(cx))?; - self.inner().poll_shutdown(cx) + self.get_pin_mut().poll_shutdown(cx) } } @@ -170,7 +172,7 @@ impl AsyncBufRead for BufWriter { } } -impl fmt::Debug for BufWriter { +impl fmt::Debug for BufWriter { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("BufWriter") .field("writer", &self.inner) @@ -182,3 +184,13 @@ impl fmt::Debug for BufWriter { .finish() } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn assert_unpin() { + crate::is_unpin::>(); + } +} diff --git a/tokio-io/src/io/chain.rs b/tokio-io/src/io/chain.rs index d7de5f219..99cce4c69 100644 --- a/tokio-io/src/io/chain.rs +++ b/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 { + #[pin] first: T, + #[pin] second: U, done_first: bool, } -impl Unpin for Chain -where - T: Unpin, - U: Unpin, -{ -} - pub(super) fn chain(first: T, second: U) -> Chain 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> { - 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> { - 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::>(); + } +} diff --git a/tokio-io/src/io/copy.rs b/tokio-io/src/io/copy.rs index 688eb5029..0d03c340e 100644 --- a/tokio-io/src/io/copy.rs +++ b/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::>(); + } +} diff --git a/tokio-io/src/io/flush.rs b/tokio-io/src/io/flush.rs index 6a78fb3c3..b7b8f2a03 100644 --- a/tokio-io/src/io/flush.rs +++ b/tokio-io/src/io/flush.rs @@ -20,8 +20,6 @@ where Flush { a } } -impl Unpin for Flush<'_, A> where A: Unpin + ?Sized {} - impl 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::>(); + } +} diff --git a/tokio-io/src/io/lines.rs b/tokio-io/src/io/lines.rs index aa05736ed..364119367 100644 --- a/tokio-io/src/io/lines.rs +++ b/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 { + #[pin] reader: R, buf: String, bytes: Vec, read: usize, } -impl Unpin for Lines {} - pub(crate) fn lines(reader: R) -> Lines where R: AsyncBufRead, @@ -34,14 +35,16 @@ where impl Stream for Lines { type Item = io::Result; + #[project] fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - 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 Stream for Lines { Poll::Ready(Some(Ok(mem::replace(buf, String::new())))) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn assert_unpin() { + crate::is_unpin::>(); + } +} diff --git a/tokio-io/src/io/read.rs b/tokio-io/src/io/read.rs index 62f2fea08..fa3b80416 100644 --- a/tokio-io/src/io/read.rs +++ b/tokio-io/src/io/read.rs @@ -28,9 +28,6 @@ pub struct Read<'a, R: ?Sized> { buf: &'a mut [u8], } -// forward Unpin -impl Unpin for Read<'_, R> {} - impl 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::>(); + } +} diff --git a/tokio-io/src/io/read_exact.rs b/tokio-io/src/io/read_exact.rs index 28b65118a..43b0ef6df 100644 --- a/tokio-io/src/io/read_exact.rs +++ b/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 Unpin for ReadExact<'_, A> {} - impl 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::>(); + } +} diff --git a/tokio-io/src/io/read_line.rs b/tokio-io/src/io/read_line.rs index 99a50be06..b1d7a511c 100644 --- a/tokio-io/src/io/read_line.rs +++ b/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, read: usize, } -impl 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 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::>(); + } +} diff --git a/tokio-io/src/io/read_to_end.rs b/tokio-io/src/io/read_to_end.rs index 2854ef8d4..8c68b8b48 100644 --- a/tokio-io/src/io/read_to_end.rs +++ b/tokio-io/src/io/read_to_end.rs @@ -13,8 +13,6 @@ pub struct ReadToEnd<'a, R: ?Sized> { start_len: usize, } -impl Unpin for ReadToEnd<'_, R> {} - pub(crate) fn read_to_end<'a, R>(reader: &'a mut R, buf: &'a mut Vec) -> 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::>(); + } +} diff --git a/tokio-io/src/io/read_to_string.rs b/tokio-io/src/io/read_to_string.rs index 61c895108..d53b402b1 100644 --- a/tokio-io/src/io/read_to_string.rs +++ b/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, start_len: usize, } -impl 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::>(); + } +} diff --git a/tokio-io/src/io/read_until.rs b/tokio-io/src/io/read_until.rs index d3d501b1e..cc3fe0259 100644 --- a/tokio-io/src/io/read_until.rs +++ b/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, read: usize, } -impl Unpin for ReadUntil<'_, R> {} - pub(crate) fn read_until<'a, R>( reader: &'a mut R, byte: u8, @@ -73,3 +71,14 @@ impl 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::>(); + } +} diff --git a/tokio-io/src/io/shutdown.rs b/tokio-io/src/io/shutdown.rs index 5bc216e4e..8f4e372d5 100644 --- a/tokio-io/src/io/shutdown.rs +++ b/tokio-io/src/io/shutdown.rs @@ -20,8 +20,6 @@ where Shutdown { a } } -impl Unpin for Shutdown<'_, A> where A: Unpin + ?Sized {} - impl 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::>(); + } +} diff --git a/tokio-io/src/io/take.rs b/tokio-io/src/io/take.rs index b0b7975a6..033323578 100644 --- a/tokio-io/src/io/take.rs +++ b/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 { + #[pin] inner: R, // Add '_' to avoid conflicts with `limit` method. limit_: u64, } -impl Unpin for Take {} - pub(super) fn take(inner: R, limit: u64) -> Take { Take { inner, @@ -24,9 +24,6 @@ pub(super) fn take(inner: R, limit: u64) -> Take { } impl Take { - 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 Take { /// 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 AsyncRead for Take { } fn poll_read( - mut self: Pin<&mut Self>, + self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8], ) -> Poll> { @@ -89,17 +86,19 @@ impl AsyncRead for Take { 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 AsyncBufRead for Take { + #[project] fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - 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 AsyncBufRead for Take { 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::>(); } } diff --git a/tokio-io/src/io/write.rs b/tokio-io/src/io/write.rs index 1f27f6056..a208b033f 100644 --- a/tokio-io/src/io/write.rs +++ b/tokio-io/src/io/write.rs @@ -21,9 +21,6 @@ where Write { writer, buf } } -// forward Unpin -impl Unpin for Write<'_, W> {} - impl 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::>(); + } +} diff --git a/tokio-io/src/io/write_all.rs b/tokio-io/src/io/write_all.rs index ceb82e173..3434e4504 100644 --- a/tokio-io/src/io/write_all.rs +++ b/tokio-io/src/io/write_all.rs @@ -20,8 +20,6 @@ where WriteAll { writer, buf } } -impl Unpin for WriteAll<'_, W> {} - impl 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::>(); + } +} diff --git a/tokio-io/src/lib.rs b/tokio-io/src/lib.rs index c7f511048..0f4a82e73 100644 --- a/tokio-io/src/lib.rs +++ b/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() {}