time: remove Box from Sleep (#3278)

Removes the box from `Sleep`, taking advantage of intrusive wakers. The
`Sleep` future is now `!Unpin`.

Closes #3267

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

tokio-stream/src/throttle.rs

@@ -14,14 +14,8 @@ pub(super) fn throttle<T>(duration: Duration, stream: T) -> Throttle<T>
 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<T> {
-        // `None` when duration is zero.
-        delay: Option<Sleep>,
+        #[pin]
+        delay: Sleep,
         duration: Duration,
 
         // Set to true when `delay` has returned ready, but `stream` hasn't.
@@ -75,23 +69,29 @@ impl<T: Unpin> Throttle<T> {
 impl<T: Stream> Stream for Throttle<T> {
     type Item = T::Item;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
-        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<Option<Self::Item>> {
+        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)
+}

tokio-stream/src/timeout.rs

@@ -15,6 +15,7 @@ pin_project! {
     pub struct Timeout<S> {
         #[pin]
         stream: Fuse<S>,
+        #[pin]
         deadline: Sleep,
         duration: Duration,
         poll_deadline: bool,
@@ -42,22 +43,24 @@ impl<S: Stream> Timeout<S> {
 impl<S: Stream> Stream for Timeout<S> {
     type Item = Result<S::Item, Elapsed>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        match self.as_mut().project().stream.poll_next(cx) {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        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())));
         }