reactor: don't use async fn for sleep

Author: Keruspe

src/implementors/async_io.rs

@@ -3,8 +3,11 @@ use async_io::{Async, Timer};
 use async_trait::async_trait;
 use futures_core::Stream;
 use std::{
+    future::Future,
     io,
     net::{SocketAddr, TcpStream},
+    pin::Pin,
+    task::{self, Context, Poll},
     time::{Duration, Instant},
 };
 
@@ -21,8 +24,8 @@ impl Reactor for AsyncIO {
         Async::new(socket)
     }
 
-    async fn sleep(&self, dur: Duration) {
-        Timer::after(dur).await;
+    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> {
+        TimerTask(Timer::after(dur))
     }
 
     fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> {
@@ -34,6 +37,17 @@ impl Reactor for AsyncIO {
     }
 }
 
+pub(crate) struct TimerTask<T: Future + Unpin>(pub(crate) T);
+
+impl<T: Future + Unpin> Future for TimerTask<T> {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        task::ready!(Pin::new(&mut self.0).poll(cx));
+        Poll::Ready(())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/implementors/smol.rs

@@ -1,6 +1,8 @@
 //! smol implementation of async runtime definition traits
 
-use crate::{AsyncIOHandle, Executor, IOHandle, Reactor, Runtime, RuntimeKit, Task, sys::IO};
+use crate::{
+    AsyncIOHandle, Executor, IOHandle, Reactor, Runtime, RuntimeKit, Task, TimerTask, sys::IO,
+};
 use async_trait::async_trait;
 use futures_core::Stream;
 use smol::{Async, Timer};
@@ -83,8 +85,8 @@ impl Reactor for Smol {
         Async::new(socket)
     }
 
-    async fn sleep(&self, dur: Duration) {
-        Timer::after(dur).await;
+    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> {
+        TimerTask(Timer::after(dur))
     }
 
     fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> {

src/implementors/tokio.rs

@@ -126,8 +126,8 @@ impl Reactor for Tokio {
         }
     }
 
-    async fn sleep(&self, dur: Duration) {
-        tokio::time::sleep(dur).await;
+    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> {
+        tokio::time::sleep(dur)
     }
 
     fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> {

src/runtime.rs

@@ -58,8 +58,8 @@ impl<RK: RuntimeKit + Sync + 'static> Reactor for Runtime<RK> {
         self.kit.register(socket)
     }
 
-    async fn sleep(&self, dur: Duration) {
-        self.kit.sleep(dur).await;
+    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> {
+        self.kit.sleep(dur)
     }
 
     fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> {
@@ -119,8 +119,8 @@ impl<E: Executor + Sync, R: Reactor + Sync> Reactor for RuntimeParts<E, R> {
         self.reactor.register(socket)
     }
 
-    async fn sleep(&self, dur: Duration) {
-        self.reactor.sleep(dur).await;
+    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> {
+        self.reactor.sleep(dur)
     }
 
     fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> {

src/traits/reactor.rs

@@ -23,7 +23,9 @@ pub trait Reactor {
         Self: Sized;
 
     /// Sleep for the given duration
-    async fn sleep(&self, dur: Duration);
+    fn sleep(&self, dur: Duration) -> impl Future<Output = ()>
+    where
+        Self: Sized;
 
     /// Stream that yields at every given interval
     fn interval(&self, dur: Duration) -> impl Stream<Item = Instant>