Merge pull request gtk-rs#772 from BiagioFesta/wip/bfesta/cancellable-future

gio: `CancellableFuture` ergonomic for cancelling tasks

Author: sdroege

examples/Cargo.toml

@@ -40,3 +40,7 @@ path = "gio_futures_await/main.rs"
 [[bin]]
 name = "gio_task"
 path = "gio_task/main.rs"
+
+[[bin]]
+name = "gio_cancellable_future"
+path = "gio_cancellable_future/main.rs"

examples/gio_cancellable_future/README.md

@@ -0,0 +1,9 @@
+# GIO Futures
+
+This example reads our `Cargo.toml` by executing GIO futures.
+
+Run it by executing:
+
+```bash
+cargo run --bin gio_cancellable_future
+```

examples/gio_cancellable_future/main.rs

@@ -0,0 +1,53 @@
+use std::future::pending;
+use std::thread;
+use std::time::Duration;
+
+use gio::prelude::*;
+
+use futures::prelude::*;
+
+/// A very long task. This task actually never ends.
+async fn a_very_long_task() {
+    println!("Very long task started");
+    pending().await
+}
+
+fn main() {
+    const TIMEOUT: Duration = Duration::from_secs(3);
+
+    let main_ctx = glib::MainContext::default();
+    let main_loop = glib::MainLoop::new(Some(&main_ctx), false);
+    let cancellable = gio::Cancellable::new();
+
+    {
+        let main_loop = main_loop.clone();
+
+        // We wrap `a_very_long_task` inside a `CancellableFuture` controlled by `cancellable`.
+        // The task is cancelled when `.cancel()` is invoked.
+        let cancellable_task = gio::CancellableFuture::new(a_very_long_task(), cancellable.clone())
+            .map(move |res| {
+                if let Err(error) = res {
+                    println!("{:?}", error);
+                }
+
+                main_loop.quit();
+            });
+
+        main_ctx.spawn_local(cancellable_task);
+    }
+
+    // We simulate a timeout here.
+    // After `TIMEOUT` we cancel the pending task.
+    thread::spawn(move || {
+        thread::sleep(TIMEOUT);
+
+        println!(
+            "Timeout ({:?}) elapsed! Cancelling pending task...",
+            TIMEOUT
+        );
+
+        cancellable.cancel();
+    });
+
+    main_loop.run();
+}

gio/Cargo.toml

@@ -45,8 +45,10 @@ futures-util = { version = "0.3", default-features = false }
 ffi = { package = "gio-sys", path = "sys" }
 glib = { path = "../glib" }
 thiserror = "1"
+pin-project-lite = "0.2"
 
 [dev-dependencies]
+futures = "0.3"
 futures-util = { version = "0.3", features = ["io"] }
 gir-format-check = "^0.1"
 serial_test = "0.9"

gio/src/cancellable_future.rs

@@ -0,0 +1,195 @@
+// Take a look at the license at the top of the repository in the LICENSE file.
+
+use crate::cancellable::CancellableExtManual;
+use crate::cancellable::CancelledHandlerId;
+use crate::prelude::CancellableExt;
+use crate::Cancellable;
+use crate::IOErrorEnum;
+use pin_project_lite::pin_project;
+use std::fmt::Debug;
+use std::fmt::Display;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+// rustdoc-stripper-ignore-next
+/// Indicator that the [`CancellableFuture`] was cancelled.
+pub struct Cancelled;
+
+pin_project! {
+    // rustdoc-stripper-ignore-next
+    /// A future which can be cancelled via [`Cancellable`].
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use futures::FutureExt;
+    /// # use gio::traits::CancellableExt;
+    /// # use gio::CancellableFuture;
+    /// let l = glib::MainLoop::new(None, false);
+    /// let c = gio::Cancellable::new();
+    ///
+    /// l.context().spawn_local(CancellableFuture::new(async { 42 }, c.clone()).map(|_| ()));
+    /// c.cancel();
+    ///
+    /// ```
+    pub struct CancellableFuture<F> {
+        #[pin]
+        future: F,
+
+        #[pin]
+        waker_handler_cb: Option<CancelledHandlerId>,
+
+        cancellable: Cancellable,
+    }
+}
+
+impl<F> CancellableFuture<F> {
+    // rustdoc-stripper-ignore-next
+    /// Creates a new `CancellableFuture` using a [`Cancellable`].
+    ///
+    /// When [`cancel`](CancellableExt::cancel) is called, the future will complete
+    /// immediately without making any further progress. In such a case, an error
+    /// will be returned by this future (i.e., [`Cancelled`]).
+    pub fn new(future: F, cancellable: Cancellable) -> Self {
+        Self {
+            future,
+            waker_handler_cb: None,
+            cancellable,
+        }
+    }
+
+    // rustdoc-stripper-ignore-next
+    /// Checks whether the future has been cancelled.
+    ///
+    /// This is a shortcut for `self.cancellable().is_cancelled()`
+    ///
+    /// Note that all this method indicates is whether [`cancel`](CancellableExt::cancel)
+    /// was called. This means that it will return true even if:
+    ///   * `cancel` was called after the future had completed.
+    ///   * `cancel` was called while the future was being polled.
+    #[inline]
+    pub fn is_cancelled(&self) -> bool {
+        self.cancellable.is_cancelled()
+    }
+
+    // rustdoc-stripper-ignore-next
+    /// Returns the inner [`Cancellable`] associated during creation.
+    #[inline]
+    pub fn cancellable(&self) -> &Cancellable {
+        &self.cancellable
+    }
+}
+
+impl<F> Future for CancellableFuture<F>
+where
+    F: Future,
+{
+    type Output = Result<<F as Future>::Output, Cancelled>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if self.is_cancelled() {
+            return Poll::Ready(Err(Cancelled));
+        }
+
+        let mut this = self.as_mut().project();
+
+        match this.future.poll(cx) {
+            Poll::Ready(out) => Poll::Ready(Ok(out)),
+
+            Poll::Pending => {
+                if let Some(prev_handler) = this.waker_handler_cb.take() {
+                    this.cancellable.disconnect_cancelled(prev_handler);
+                }
+
+                let canceller_handler_id = this.cancellable.connect_cancelled({
+                    let w = cx.waker().clone();
+                    move |_| w.wake()
+                });
+
+                match canceller_handler_id {
+                    Some(canceller_handler_id) => {
+                        *this.waker_handler_cb = Some(canceller_handler_id);
+                        Poll::Pending
+                    }
+
+                    None => Poll::Ready(Err(Cancelled)),
+                }
+            }
+        }
+    }
+}
+
+impl From<Cancelled> for glib::Error {
+    fn from(_: Cancelled) -> Self {
+        glib::Error::new(IOErrorEnum::Cancelled, "Task cancelled")
+    }
+}
+
+impl std::error::Error for Cancelled {}
+
+impl Debug for Cancelled {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Task cancelled")
+    }
+}
+
+impl Display for Cancelled {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        Debug::fmt(self, f)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Cancellable;
+    use super::CancellableExt;
+    use super::CancellableFuture;
+    use super::Cancelled;
+    use futures_channel::oneshot;
+
+    #[test]
+    fn cancellable_future_ok() {
+        let ctx = glib::MainContext::new();
+        let c = Cancellable::new();
+        let (tx, rx) = oneshot::channel();
+
+        {
+            ctx.spawn_local(async {
+                let cancellable_future = CancellableFuture::new(async { 42 }, c);
+                assert!(!cancellable_future.is_cancelled());
+
+                let result = cancellable_future.await;
+                assert!(matches!(result, Ok(42)));
+
+                tx.send(()).unwrap();
+            });
+        }
+
+        ctx.block_on(rx).unwrap()
+    }
+
+    #[test]
+    fn cancellable_future_cancel() {
+        let ctx = glib::MainContext::new();
+        let c = Cancellable::new();
+        let (tx, rx) = oneshot::channel();
+
+        {
+            let c = c.clone();
+            ctx.spawn_local(async move {
+                let cancellable_future = CancellableFuture::new(std::future::pending::<()>(), c);
+
+                let result = cancellable_future.await;
+                assert!(matches!(result, Err(Cancelled)));
+
+                tx.send(()).unwrap();
+            });
+        }
+
+        std::thread::spawn(move || c.cancel()).join().unwrap();
+
+        ctx.block_on(rx).unwrap();
+    }
+}

gio/src/lib.rs

@@ -20,6 +20,9 @@ pub use action_entry::{ActionEntry, ActionEntryBuilder};
 pub use application::{ApplicationBusyGuard, ApplicationHoldGuard};
 mod async_initable;
 mod cancellable;
+mod cancellable_future;
+pub use crate::cancellable_future::CancellableFuture;
+pub use crate::cancellable_future::Cancelled;
 mod converter;
 mod data_input_stream;
 mod dbus;