Merge pull request #24 from yoshuawuyts/pch/futures-time

import futures-time

Author: pchickey

Cargo.toml

@@ -2,7 +2,6 @@
 name = "wstd"
 version.workspace = true
 license.workspace = true
-repository = "https://github.com/yoshuawuyts/wstd"
 documentation = "https://docs.rs/wstd"
 description = "An async standard library for Wasm Components and WASI 0.2"
 readme = "README.md"
@@ -16,13 +15,16 @@ authors = [
 [features]
 
 [dependencies]
+futures-core.workspace = true
 http.workspace = true
+pin-project-lite.workspace = true
 slab.workspace = true
 wasi.workspace = true
 wstd-macro.workspace = true
 
 [dev-dependencies]
 anyhow.workspace = true
+futures-lite.workspace = true
 serde_json.workspace = true
 test-programs-artifacts.workspace = true
 wasmtime.workspace = true
@@ -41,12 +43,16 @@ resolver = "2"
 version = "0.4.0"
 edition = "2021"
 license = "MIT OR Apache-2.0 OR Apache-2.0 WITH LLVM-exception"
+repository = "https://github.com/yoshuawuyts/wstd"
 
 [workspace.dependencies]
 anyhow = "1"
 cargo_metadata = "0.18.1"
+futures-core = "0.3.19"
+futures-lite = "1.12.0"
 heck = "0.5"
 http = "1.1"
+pin-project-lite = "0.2.8"
 quote = "1.0"
 serde_json = "1"
 slab = "0.4.9"

src/future/delay.rs

@@ -0,0 +1,64 @@
+use futures_core::ready;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use pin_project_lite::pin_project;
+
+pin_project! {
+    /// Suspends a future until the specified deadline.
+    ///
+    /// This `struct` is created by the [`delay`] method on [`FutureExt`]. See its
+    /// documentation for more.
+    ///
+    /// [`delay`]: crate::future::FutureExt::delay
+    /// [`FutureExt`]: crate::future::futureExt
+    #[must_use = "futures do nothing unless polled or .awaited"]
+    pub struct Delay<F, D> {
+        #[pin]
+        future: F,
+        #[pin]
+        deadline: D,
+        state: State,
+    }
+}
+
+/// The internal state
+#[derive(Debug)]
+enum State {
+    Started,
+    PollFuture,
+    Completed,
+}
+
+impl<F, D> Delay<F, D> {
+    pub(super) fn new(future: F, deadline: D) -> Self {
+        Self {
+            future,
+            deadline,
+            state: State::Started,
+        }
+    }
+}
+
+impl<F: Future, D: Future> Future for Delay<F, D> {
+    type Output = F::Output;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.project();
+        loop {
+            match this.state {
+                State::Started => {
+                    ready!(this.deadline.as_mut().poll(cx));
+                    *this.state = State::PollFuture;
+                }
+                State::PollFuture => {
+                    let value = ready!(this.future.as_mut().poll(cx));
+                    *this.state = State::Completed;
+                    return Poll::Ready(value);
+                }
+                State::Completed => panic!("future polled after completing"),
+            }
+        }
+    }
+}

src/future/future_ext.rs

@@ -0,0 +1,76 @@
+use super::{Delay, Timeout};
+use std::future::{Future, IntoFuture};
+
+/// Extend `Future` with time-based operations.
+pub trait FutureExt: Future {
+    /// Return an error if a future does not complete within a given time span.
+    ///
+    /// Typically timeouts are, as the name implies, based on _time_. However
+    /// this method can time out based on any future. This can be useful in
+    /// combination with channels, as it allows (long-lived) futures to be
+    /// cancelled based on some external event.
+    ///
+    /// When a timeout is returned, the future will be dropped and destructors
+    /// will be run.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// use wstd::prelude::*;
+    /// use wstd::time::{Instant, Duration};
+    /// use std::io;
+    ///
+    /// #[wstd::main]
+    /// async fn main() {
+    ///     let res = async { "meow" }
+    ///         .delay(Duration::from_millis(100))  // longer delay
+    ///         .timeout(Duration::from_millis(50)) // shorter timeout
+    ///         .await;
+    ///     assert_eq!(res.unwrap_err().kind(), io::ErrorKind::TimedOut); // error
+    ///
+    ///     let res = async { "meow" }
+    ///         .delay(Duration::from_millis(50))    // shorter delay
+    ///         .timeout(Duration::from_millis(100)) // longer timeout
+    ///         .await;
+    ///     assert_eq!(res.unwrap(), "meow"); // success
+    /// }
+    /// ```
+    fn timeout<D>(self, deadline: D) -> Timeout<Self, D::IntoFuture>
+    where
+        Self: Sized,
+        D: IntoFuture,
+    {
+        Timeout::new(self, deadline.into_future())
+    }
+
+    /// Delay resolving the future until the given deadline.
+    ///
+    /// The underlying future will not be polled until the deadline has expired. In addition
+    /// to using a time source as a deadline, any future can be used as a
+    /// deadline too. When used in combination with a multi-consumer channel,
+    /// this method can be used to synchronize the start of multiple futures and streams.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// use wstd::prelude::*;
+    /// use wstd::time::{Instant, Duration};
+    ///
+    /// #[wstd::main]
+    /// async fn main() {
+    ///     let now = Instant::now();
+    ///     let delay = Duration::from_millis(100);
+    ///     let _ = async { "meow" }.delay(delay).await;
+    ///     assert!(now.elapsed() >= delay);
+    /// }
+    /// ```
+    fn delay<D>(self, deadline: D) -> Delay<Self, D::IntoFuture>
+    where
+        Self: Sized,
+        D: IntoFuture,
+    {
+        Delay::new(self, deadline.into_future())
+    }
+}
+
+impl<T> FutureExt for T where T: Future {}

src/future/mod.rs

@@ -0,0 +1,35 @@
+//! Asynchronous values.
+//!
+//! # Cancellation
+//!
+//! Futures can be cancelled by dropping them before they finish executing. This
+//! is useful when we're no longer interested in the result of an operation, as
+//! it allows us to stop doing needless work. This also means that a future may cancel at any `.await` point, and so just
+//! like with `?` we have to be careful to roll back local state if our future
+//! halts there.
+//!
+//!
+//! ```no_run
+//! use futures_lite::prelude::*;
+//! use wstd::prelude::*;
+//! use wstd::time::Duration;
+//!
+//! #[wstd::main]
+//! async fn main() {
+//!     let mut counter = 0;
+//!     let value = async { "meow" }
+//!         .delay(Duration::from_millis(100))
+//!         .timeout(Duration::from_millis(200))
+//!         .await;
+//!
+//!     assert_eq!(value.unwrap(), "meow");
+//! }
+//! ```
+
+mod delay;
+mod future_ext;
+mod timeout;
+
+pub use delay::Delay;
+pub use future_ext::FutureExt;
+pub use timeout::Timeout;

src/future/timeout.rs

@@ -0,0 +1,60 @@
+use crate::time::utils::timeout_err;
+
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use pin_project_lite::pin_project;
+
+pin_project! {
+    /// A future that times out after a duration of time.
+    ///
+    /// This `struct` is created by the [`timeout`] method on [`FutureExt`]. See its
+    /// documentation for more.
+    ///
+    /// [`timeout`]: crate::future::FutureExt::timeout
+    /// [`FutureExt`]: crate::future::futureExt
+    #[must_use = "futures do nothing unless polled or .awaited"]
+    pub struct Timeout<F, D> {
+        #[pin]
+        future: F,
+        #[pin]
+        deadline: D,
+        completed: bool,
+    }
+}
+
+impl<F, D> Timeout<F, D> {
+    pub(super) fn new(future: F, deadline: D) -> Self {
+        Self {
+            future,
+            deadline,
+            completed: false,
+        }
+    }
+}
+
+impl<F: Future, D: Future> Future for Timeout<F, D> {
+    type Output = io::Result<F::Output>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+
+        assert!(!*this.completed, "future polled after completing");
+
+        match this.future.poll(cx) {
+            Poll::Ready(v) => {
+                *this.completed = true;
+                Poll::Ready(Ok(v))
+            }
+            Poll::Pending => match this.deadline.poll(cx) {
+                Poll::Ready(_) => {
+                    *this.completed = true;
+                    Poll::Ready(Err(timeout_err("future timed out")))
+                }
+                Poll::Pending => Poll::Pending,
+            },
+        }
+    }
+}

src/lib.rs

@@ -1,4 +1,8 @@
 #![allow(async_fn_in_trait)]
+#![warn(future_incompatible, unreachable_pub)]
+//#![deny(missing_debug_implementations)]
+//#![warn(missing_docs)]
+//#![forbid(rustdoc::missing_doc_code_examples)]
 
 //! An async standard library for Wasm Components and WASI 0.2
 //!
@@ -44,12 +48,21 @@
 //! These are unique capabilities provided by WASI 0.2, and because this library
 //! is specific to that are exposed from here.
 
+pub mod future;
 pub mod http;
 pub mod io;
 pub mod iter;
 pub mod net;
 pub mod rand;
 pub mod runtime;
+pub mod task;
 pub mod time;
 
 pub use wstd_macro::attr_macro_main as main;
+
+pub mod prelude {
+    pub use crate::future::FutureExt as _;
+    pub use crate::http::Body as _;
+    pub use crate::io::AsyncRead as _;
+    pub use crate::io::AsyncWrite as _;
+}

src/runtime/reactor.rs

@@ -75,7 +75,6 @@ impl Reactor {
     pub async fn wait_for(&self, pollable: Pollable) {
         let mut pollable = Some(pollable);
         let mut key = None;
-
         // This function is the core loop of our function; it will be called
         // multiple times as the future is resolving.
         future::poll_fn(|cx| {

src/task/mod.rs

@@ -0,0 +1,7 @@
+//! Types and Traits for working with asynchronous tasks.
+
+mod sleep;
+mod sleep_until;
+
+pub use sleep::{sleep, Sleep};
+pub use sleep_until::{sleep_until, SleepUntil};

src/task/sleep.rs

@@ -0,0 +1,46 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use crate::time::Timer as AsyncTimer;
+use pin_project_lite::pin_project;
+
+use crate::time::{Duration, Instant};
+
+/// Sleeps for the specified amount of time.
+///
+/// This future can be `push_deadline` to be moved
+pub fn sleep(dur: Duration) -> Sleep {
+    Sleep {
+        dur,
+        timer: AsyncTimer::after(dur.into()),
+        completed: false,
+    }
+}
+
+pin_project! {
+    /// Sleeps for the specified amount of time.
+    #[must_use = "futures do nothing unless polled or .awaited"]
+    pub struct Sleep {
+        #[pin]
+        timer: AsyncTimer,
+        completed: bool,
+        dur: Duration,
+    }
+}
+
+impl Future for Sleep {
+    type Output = Instant;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        assert!(!self.completed, "future polled after completing");
+        let this = self.project();
+        match this.timer.poll(cx) {
+            Poll::Ready(instant) => {
+                *this.completed = true;
+                Poll::Ready(instant.into())
+            }
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}