diff --git a/tokio-stream/src/lib.rs b/tokio-stream/src/lib.rs index ae8b2e2e4..15794a5ae 100644 --- a/tokio-stream/src/lib.rs +++ b/tokio-stream/src/lib.rs @@ -937,7 +937,8 @@ pub trait StreamExt: Stream { /// use std::time::Duration; /// # let int_stream = stream::iter(1..=3); /// - /// let mut int_stream = int_stream.timeout(Duration::from_secs(1)); + /// let int_stream = int_stream.timeout(Duration::from_secs(1)); + /// tokio::pin!(int_stream); /// /// // When no items time out, we get the 3 elements in succession: /// assert_eq!(int_stream.try_next().await, Ok(Some(1))); @@ -981,7 +982,8 @@ pub trait StreamExt: Stream { /// use tokio_stream::StreamExt; /// /// # async fn dox() { - /// let mut item_stream = futures::stream::repeat("one").throttle(Duration::from_secs(2)); + /// let item_stream = futures::stream::repeat("one").throttle(Duration::from_secs(2)); + /// tokio::pin!(item_stream); /// /// loop { /// // The string will be produced at most every 2 seconds diff --git a/tokio-stream/src/throttle.rs b/tokio-stream/src/throttle.rs index 525763dca..99f3e0e00 100644 --- a/tokio-stream/src/throttle.rs +++ b/tokio-stream/src/throttle.rs @@ -14,14 +14,8 @@ pub(super) fn throttle(duration: Duration, stream: T) -> Throttle where T: Stream, { - let delay = if duration == Duration::from_millis(0) { - None - } else { - Some(tokio::time::sleep_until(Instant::now() + duration)) - }; - Throttle { - delay, + delay: tokio::time::sleep_until(Instant::now() + duration), duration, has_delayed: true, stream, @@ -33,8 +27,8 @@ pin_project! { #[derive(Debug)] #[must_use = "streams do nothing unless polled"] pub struct Throttle { - // `None` when duration is zero. - delay: Option, + #[pin] + delay: Sleep, duration: Duration, // Set to true when `delay` has returned ready, but `stream` hasn't. @@ -75,23 +69,29 @@ impl Throttle { impl Stream for Throttle { type Item = T::Item; - fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - if !self.has_delayed && self.delay.is_some() { - ready!(Pin::new(self.as_mut().project().delay.as_mut().unwrap()).poll(cx)); - *self.as_mut().project().has_delayed = true; + fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { + let mut me = self.project(); + let dur = *me.duration; + + if !*me.has_delayed && !is_zero(dur) { + ready!(me.delay.as_mut().poll(cx)); + *me.has_delayed = true; } - let value = ready!(self.as_mut().project().stream.poll_next(cx)); + let value = ready!(me.stream.poll_next(cx)); if value.is_some() { - let dur = self.duration; - if let Some(ref mut delay) = self.as_mut().project().delay { - delay.reset(Instant::now() + dur); + if !is_zero(dur) { + me.delay.reset(Instant::now() + dur); } - *self.as_mut().project().has_delayed = false; + *me.has_delayed = false; } Poll::Ready(value) } } + +fn is_zero(dur: Duration) -> bool { + dur == Duration::from_millis(0) +} diff --git a/tokio-stream/src/timeout.rs b/tokio-stream/src/timeout.rs index 303142f01..9ebbaa233 100644 --- a/tokio-stream/src/timeout.rs +++ b/tokio-stream/src/timeout.rs @@ -15,6 +15,7 @@ pin_project! { pub struct Timeout { #[pin] stream: Fuse, + #[pin] deadline: Sleep, duration: Duration, poll_deadline: bool, @@ -42,22 +43,24 @@ impl Timeout { impl Stream for Timeout { type Item = Result; - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - match self.as_mut().project().stream.poll_next(cx) { + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let me = self.project(); + + match me.stream.poll_next(cx) { Poll::Ready(v) => { if v.is_some() { - let next = Instant::now() + self.duration; - self.as_mut().project().deadline.reset(next); - *self.as_mut().project().poll_deadline = true; + let next = Instant::now() + *me.duration; + me.deadline.reset(next); + *me.poll_deadline = true; } return Poll::Ready(v.map(Ok)); } Poll::Pending => {} }; - if self.poll_deadline { - ready!(Pin::new(self.as_mut().project().deadline).poll(cx)); - *self.as_mut().project().poll_deadline = false; + if *me.poll_deadline { + ready!(me.deadline.poll(cx)); + *me.poll_deadline = false; return Poll::Ready(Some(Err(Elapsed::new()))); } diff --git a/tokio-test/src/io.rs b/tokio-test/src/io.rs index 6705f80c3..77adfc3ee 100644 --- a/tokio-test/src/io.rs +++ b/tokio-test/src/io.rs @@ -67,7 +67,7 @@ enum Action { struct Inner { actions: VecDeque, waiting: Option, - sleep: Option, + sleep: Option>>, read_wait: Option, // rx: mpsc::UnboundedReceiver, rx: Pin + Send>>, @@ -370,7 +370,7 @@ impl AsyncRead for Mock { Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { if let Some(rem) = self.inner.remaining_wait() { let until = Instant::now() + rem; - self.inner.sleep = Some(time::sleep_until(until)); + self.inner.sleep = Some(Box::pin(time::sleep_until(until))); } else { self.inner.read_wait = Some(cx.waker().clone()); return Poll::Pending; @@ -415,7 +415,7 @@ impl AsyncWrite for Mock { Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { if let Some(rem) = self.inner.remaining_wait() { let until = Instant::now() + rem; - self.inner.sleep = Some(time::sleep_until(until)); + self.inner.sleep = Some(Box::pin(time::sleep_until(until))); } else { panic!("unexpected WouldBlock"); } diff --git a/tokio-util/src/time/delay_queue.rs b/tokio-util/src/time/delay_queue.rs index 4edd5cd64..e2c8f3137 100644 --- a/tokio-util/src/time/delay_queue.rs +++ b/tokio-util/src/time/delay_queue.rs @@ -138,7 +138,7 @@ pub struct DelayQueue { expired: Stack, /// Delay expiring when the *first* item in the queue expires - delay: Option, + delay: Option>>, /// Wheel polling state wheel_now: u64, @@ -342,9 +342,9 @@ impl DelayQueue { let delay_time = self.start + Duration::from_millis(when); if let Some(ref mut delay) = &mut self.delay { - delay.reset(delay_time); + delay.as_mut().reset(delay_time); } else { - self.delay = Some(sleep_until(delay_time)); + self.delay = Some(Box::pin(sleep_until(delay_time))); } } @@ -553,7 +553,7 @@ impl DelayQueue { let next_deadline = self.next_deadline(); if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) { // This should awaken us if necessary (ie, if already expired) - delay.reset(deadline); + delay.as_mut().reset(deadline); } } @@ -759,7 +759,7 @@ impl DelayQueue { // We poll the wheel to get the next value out before finding the next deadline. let wheel_idx = self.wheel.poll(self.wheel_now, &mut self.slab); - self.delay = self.next_deadline().map(sleep_until); + self.delay = self.next_deadline().map(|when| Box::pin(sleep_until(when))); if let Some(idx) = wheel_idx { return Poll::Ready(Some(Ok(idx))); diff --git a/tokio/src/macros/select.rs b/tokio/src/macros/select.rs index 2131c8907..ca4f9630c 100644 --- a/tokio/src/macros/select.rs +++ b/tokio/src/macros/select.rs @@ -76,7 +76,8 @@ /// /// #[tokio::main] /// async fn main() { -/// let mut sleep = time::sleep(Duration::from_millis(50)); +/// let sleep = time::sleep(Duration::from_millis(50)); +/// tokio::pin!(sleep); /// /// while !sleep.is_elapsed() { /// tokio::select! { @@ -109,7 +110,8 @@ /// /// #[tokio::main] /// async fn main() { -/// let mut sleep = time::sleep(Duration::from_millis(50)); +/// let sleep = time::sleep(Duration::from_millis(50)); +/// tokio::pin!(sleep); /// /// loop { /// tokio::select! { @@ -226,7 +228,8 @@ /// #[tokio::main] /// async fn main() { /// let mut stream = stream::iter(vec![1, 2, 3]); -/// let mut sleep = time::sleep(Duration::from_secs(1)); +/// let sleep = time::sleep(Duration::from_secs(1)); +/// tokio::pin!(sleep); /// /// loop { /// tokio::select! { diff --git a/tokio/src/sync/mod.rs b/tokio/src/sync/mod.rs index a183fe6ed..a953c66b3 100644 --- a/tokio/src/sync/mod.rs +++ b/tokio/src/sync/mod.rs @@ -359,7 +359,8 @@ //! let mut conf = rx.borrow().clone(); //! //! let mut op_start = Instant::now(); -//! let mut sleep = time::sleep_until(op_start + conf.timeout); +//! let sleep = time::sleep_until(op_start + conf.timeout); +//! tokio::pin!(sleep); //! //! loop { //! tokio::select! { @@ -371,14 +372,14 @@ //! op_start = Instant::now(); //! //! // Restart the timeout -//! sleep = time::sleep_until(op_start + conf.timeout); +//! sleep.set(time::sleep_until(op_start + conf.timeout)); //! } //! _ = rx.changed() => { //! conf = rx.borrow().clone(); //! //! // The configuration has been updated. Update the //! // `sleep` using the new `timeout` value. -//! sleep.reset(op_start + conf.timeout); +//! sleep.as_mut().reset(op_start + conf.timeout); //! } //! _ = &mut op => { //! // The operation completed! diff --git a/tokio/src/time/driver/entry.rs b/tokio/src/time/driver/entry.rs index 87ba0c176..bcad988ef 100644 --- a/tokio/src/time/driver/entry.rs +++ b/tokio/src/time/driver/entry.rs @@ -367,6 +367,8 @@ pub(super) struct TimerEntry { /// Initial deadline for the timer. This is used to register on the first /// poll, as we can't register prior to being pinned. initial_deadline: Option, + /// Ensure the type is !Unpin + _m: std::marker::PhantomPinned, } unsafe impl Send for TimerEntry {} @@ -556,6 +558,7 @@ impl TimerEntry { driver, inner: StdUnsafeCell::new(TimerShared::new()), initial_deadline: Some(deadline), + _m: std::marker::PhantomPinned, } } diff --git a/tokio/src/time/driver/sleep.rs b/tokio/src/time/driver/sleep.rs index 9f358c34e..69a6e6d41 100644 --- a/tokio/src/time/driver/sleep.rs +++ b/tokio/src/time/driver/sleep.rs @@ -1,9 +1,9 @@ use crate::time::driver::{Handle, TimerEntry}; use crate::time::{error::Error, Duration, Instant}; +use pin_project_lite::pin_project; use std::future::Future; use std::pin::Pin; - use std::task::{self, Poll}; /// Waits until `deadline` is reached. @@ -57,22 +57,24 @@ pub fn sleep(duration: Duration) -> Sleep { sleep_until(Instant::now() + duration) } -/// Future returned by [`sleep`](sleep) and -/// [`sleep_until`](sleep_until). -#[derive(Debug)] -#[must_use = "futures do nothing unless you `.await` or poll them"] -pub struct Sleep { - deadline: Instant, +pin_project! { + /// Future returned by [`sleep`](sleep) and + /// [`sleep_until`](sleep_until). + #[derive(Debug)] + #[must_use = "futures do nothing unless you `.await` or poll them"] + pub struct Sleep { + deadline: Instant, - // The link between the `Sleep` instance and the timer that drives it. - // This will be unboxed in tokio 1.0 - entry: Pin>, + // The link between the `Sleep` instance and the timer that drives it. + #[pin] + entry: TimerEntry, + } } impl Sleep { pub(crate) fn new_timeout(deadline: Instant) -> Sleep { let handle = Handle::current(); - let entry = Box::pin(TimerEntry::new(&handle, deadline)); + let entry = TimerEntry::new(&handle, deadline); Sleep { deadline, entry } } @@ -96,16 +98,19 @@ impl Sleep { /// /// This function can be called both before and after the future has /// completed. - pub fn reset(&mut self, deadline: Instant) { - self.entry.as_mut().reset(deadline); - self.deadline = deadline; + pub fn reset(self: Pin<&mut Self>, deadline: Instant) { + let me = self.project(); + me.entry.reset(deadline); + *me.deadline = deadline; } - fn poll_elapsed(&mut self, cx: &mut task::Context<'_>) -> Poll> { + fn poll_elapsed(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { + let me = self.project(); + // Keep track of task budget let coop = ready!(crate::coop::poll_proceed(cx)); - self.entry.as_mut().poll_elapsed(cx).map(move |r| { + me.entry.poll_elapsed(cx).map(move |r| { coop.made_progress(); r }) diff --git a/tokio/src/time/interval.rs b/tokio/src/time/interval.rs index e12a3f339..10de4e9d7 100644 --- a/tokio/src/time/interval.rs +++ b/tokio/src/time/interval.rs @@ -101,7 +101,7 @@ pub fn interval_at(start: Instant, period: Duration) -> Interval { assert!(period > Duration::new(0, 0), "`period` must be non-zero."); Interval { - delay: sleep_until(start), + delay: Box::pin(sleep_until(start)), period, } } @@ -110,7 +110,7 @@ pub fn interval_at(start: Instant, period: Duration) -> Interval { #[derive(Debug)] pub struct Interval { /// Future that completes the next time the `Interval` yields a value. - delay: Sleep, + delay: Pin>, /// The duration between values yielded by `Interval`. period: Duration, @@ -127,7 +127,7 @@ impl Interval { // The next interval value is `duration` after the one that just // yielded. let next = now + self.period; - self.delay.reset(next); + self.delay.as_mut().reset(next); // Return the current instant Poll::Ready(now) diff --git a/tokio/tests/async_send_sync.rs b/tokio/tests/async_send_sync.rs index c345d2acb..671fa4a70 100644 --- a/tokio/tests/async_send_sync.rs +++ b/tokio/tests/async_send_sync.rs @@ -93,6 +93,14 @@ macro_rules! assert_value { AmbiguousIfSync::some_item(&f); }; }; + ($type:ty: Unpin) => { + #[allow(unreachable_code)] + #[allow(unused_variables)] + const _: fn() = || { + let f: $type = todo!(); + require_unpin(&f); + }; + }; } macro_rules! async_assert_fn { ($($f:ident $(< $($generic:ty),* > )? )::+($($arg:ty),*): Send & Sync) => { @@ -280,6 +288,12 @@ async_assert_fn!(tokio::time::timeout_at(Instant, BoxFutureSend<()>): Send & !Sy async_assert_fn!(tokio::time::timeout_at(Instant, BoxFuture<()>): !Send & !Sync); async_assert_fn!(tokio::time::Interval::tick(_): Send & Sync); +assert_value!(tokio::time::Interval: Unpin); +async_assert_fn!(tokio::time::sleep(Duration): !Unpin); +async_assert_fn!(tokio::time::sleep_until(Instant): !Unpin); +async_assert_fn!(tokio::time::timeout(Duration, BoxFuture<()>): !Unpin); +async_assert_fn!(tokio::time::timeout_at(Instant, BoxFuture<()>): !Unpin); +async_assert_fn!(tokio::time::Interval::tick(_): !Unpin); async_assert_fn!(tokio::io::AsyncBufReadExt::read_until(&mut BoxAsyncRead, u8, &mut Vec): !Unpin); async_assert_fn!(tokio::io::AsyncBufReadExt::read_line(&mut BoxAsyncRead, &mut String): !Unpin); async_assert_fn!(tokio::io::AsyncReadExt::read(&mut BoxAsyncRead, &mut [u8]): !Unpin); diff --git a/tokio/tests/time_sleep.rs b/tokio/tests/time_sleep.rs index 20e2b1c6b..27362581d 100644 --- a/tokio/tests/time_sleep.rs +++ b/tokio/tests/time_sleep.rs @@ -317,15 +317,15 @@ async fn drop_after_reschedule_at_new_scheduled_time() { let start = tokio::time::Instant::now(); - let mut a = tokio::time::sleep(Duration::from_millis(5)); - let mut b = tokio::time::sleep(Duration::from_millis(5)); - let mut c = tokio::time::sleep(Duration::from_millis(10)); + let mut a = Box::pin(tokio::time::sleep(Duration::from_millis(5))); + let mut b = Box::pin(tokio::time::sleep(Duration::from_millis(5))); + let mut c = Box::pin(tokio::time::sleep(Duration::from_millis(10))); let _ = poll!(&mut a); let _ = poll!(&mut b); let _ = poll!(&mut c); - b.reset(start + Duration::from_millis(10)); + b.as_mut().reset(start + Duration::from_millis(10)); a.await; drop(b); @@ -334,12 +334,13 @@ async fn drop_after_reschedule_at_new_scheduled_time() { #[tokio::test] async fn drop_from_wake() { use std::future::Future; + use std::pin::Pin; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::task::Context; let panicked = Arc::new(AtomicBool::new(false)); - let list: Arc>> = Arc::new(Mutex::new(Vec::new())); + let list: Arc>>>> = Arc::new(Mutex::new(Vec::new())); let arc_wake = Arc::new(DropWaker(panicked.clone(), list.clone())); let arc_wake = futures::task::waker(arc_wake); @@ -349,9 +350,9 @@ async fn drop_from_wake() { let mut lock = list.lock().unwrap(); for _ in 0..100 { - let mut timer = tokio::time::sleep(Duration::from_millis(10)); + let mut timer = Box::pin(tokio::time::sleep(Duration::from_millis(10))); - let _ = std::pin::Pin::new(&mut timer).poll(&mut Context::from_waker(&arc_wake)); + let _ = timer.as_mut().poll(&mut Context::from_waker(&arc_wake)); lock.push(timer); } @@ -366,7 +367,10 @@ async fn drop_from_wake() { ); #[derive(Clone)] - struct DropWaker(Arc, Arc>>); + struct DropWaker( + Arc, + Arc>>>>, + ); impl futures::task::ArcWake for DropWaker { fn wake_by_ref(arc_self: &Arc) {