feat(rt): Clean up Timer trait (#3037)

Closes #3028 

BREAKING CHANGE: The return types of `Timer` have been changed.

benches/support/tokiort.rs

@@ -8,6 +8,7 @@ use std::{
 
 use futures_util::Future;
 use hyper::rt::{Sleep, Timer};
+use pin_project_lite::pin_project;
 
 #[derive(Clone)]
 /// An Executor that uses the tokio runtime.
@@ -29,16 +30,16 @@ where
 pub struct TokioTimer;
 
 impl Timer for TokioTimer {
-    fn sleep(&self, duration: Duration) -> Box<dyn Sleep + Unpin> {
-        let s = tokio::time::sleep(duration);
-        let hs = TokioSleep { inner: Box::pin(s) };
-        return Box::new(hs);
+    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            inner: tokio::time::sleep(duration),
+        })
     }
 
-    fn sleep_until(&self, deadline: Instant) -> Box<dyn Sleep + Unpin> {
-        return Box::new(TokioSleep {
-            inner: Box::pin(tokio::time::sleep_until(deadline.into())),
-        });
+    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            inner: tokio::time::sleep_until(deadline.into()),
+        })
     }
 }
 
@@ -59,15 +60,18 @@ where
 
 // Use TokioSleep to get tokio::time::Sleep to implement Unpin.
 // see https://docs.rs/tokio/latest/tokio/time/struct.Sleep.html
-pub(crate) struct TokioSleep {
-    pub(crate) inner: Pin<Box<tokio::time::Sleep>>,
+pin_project! {
+    pub(crate) struct TokioSleep {
+        #[pin]
+        pub(crate) inner: tokio::time::Sleep,
+    }
 }
 
 impl Future for TokioSleep {
     type Output = ();
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.inner.as_mut().poll(cx)
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().inner.poll(cx)
     }
 }
 

src/common/mod.rs

@@ -28,10 +28,3 @@ cfg_proto! {
     pub(crate) use std::marker::Unpin;
 }
 pub(crate) use std::{future::Future, pin::Pin};
-
-pub(crate) fn into_pin<T: ?Sized>(boxed: Box<T>) -> Pin<Box<T>> {
-    // It's not possible to move or replace the insides of a `Pin<Box<T>>`
-    // when `T: !Unpin`, so it's safe to pin it directly without any
-    // additional requirements.
-    unsafe { Pin::new_unchecked(boxed) }
-}

src/common/time.rs

@@ -55,7 +55,7 @@ impl<F> Future for HyperTimeout<F> where F: Future {
 */
 
 impl Time {
-    pub(crate) fn sleep(&self, duration: Duration) -> Box<dyn Sleep + Unpin> {
+    pub(crate) fn sleep(&self, duration: Duration) -> Pin<Box<dyn Sleep>> {
         match *self {
             Time::Empty => {
                 panic!("You must supply a timer.")
@@ -64,7 +64,7 @@ impl Time {
         }
     }
 
-    pub(crate) fn sleep_until(&self, deadline: Instant) -> Box<dyn Sleep + Unpin> {
+    pub(crate) fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>> {
         match *self {
             Time::Empty => {
                 panic!("You must supply a timer.")

src/proto/h1/role.rs

@@ -87,8 +87,7 @@ where
                 }
                 None => {
                     debug!("setting h1 header read timeout timer");
-                    *ctx.h1_header_read_timeout_fut =
-                        Some(crate::common::into_pin(ctx.timer.sleep_until(deadline)));
+                    *ctx.h1_header_read_timeout_fut = Some(ctx.timer.sleep_until(deadline));
                 }
             }
         }

src/proto/h2/ping.rs

@@ -61,7 +61,7 @@ pub(super) fn channel(ping_pong: PingPong, config: Config, __timer: Time) -> (Re
         interval,
         timeout: config.keep_alive_timeout,
         while_idle: config.keep_alive_while_idle,
-        sleep: crate::common::into_pin(__timer.sleep(interval)),
+        sleep: __timer.sleep(interval),
         state: KeepAliveState::Init,
         timer: __timer,
     });

src/rt.rs

@@ -20,16 +20,16 @@ pub trait Executor<Fut> {
 /// A timer which provides timer-like functions.
 pub trait Timer {
     /// Return a future that resolves in `duration` time.
-    fn sleep(&self, duration: Duration) -> Box<dyn Sleep + Unpin>;
+    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Sleep>>;
 
     /// Return a future that resolves at `deadline`.
-    fn sleep_until(&self, deadline: Instant) -> Box<dyn Sleep + Unpin>;
+    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>>;
 
     /// Reset a future to resolve at `new_deadline` instead.
     fn reset(&self, sleep: &mut Pin<Box<dyn Sleep>>, new_deadline: Instant) {
-        *sleep = crate::common::into_pin(self.sleep_until(new_deadline));
+        *sleep = self.sleep_until(new_deadline);
     }
 }
 
 /// A future returned by a `Timer`.
-pub trait Sleep: Send + Sync + Unpin + Future<Output = ()> {}
+pub trait Sleep: Send + Sync + Future<Output = ()> {}