qubit_tokio_executor/
tokio_execution.rs

1/*******************************************************************************
2 *
3 *    Copyright (c) 2025 - 2026 Haixing Hu.
4 *
5 *    SPDX-License-Identifier: Apache-2.0
6 *
7 *    Licensed under the Apache License, Version 2.0.
8 *
9 ******************************************************************************/
10use std::{
11    future::Future,
12    pin::Pin,
13    task::{
14        Context,
15        Poll,
16    },
17};
18
19use tokio::task::{
20    JoinError,
21    JoinHandle,
22};
23
24/// Future-backed execution returned by [`TokioExecutor`](super::TokioExecutor).
25///
26/// This struct **implements [`Future`]**:
27/// [`Output`](std::future::Future::Output) is [`Result<R, E>`](Result) — the
28/// success value `R` or the callable's error `E`. Await this type (on a
29/// Tokio-driven async context) to receive that result; until then the underlying
30/// blocking task may still be running.
31///
32/// Tokio join errors are not part of that `Result`: task panics are resumed and
33/// cancellations panic when the future is awaited.
34///
35/// # Type Parameters
36///
37/// * `R` - The task success value.
38/// * `E` - The task error value.
39///
40pub struct TokioExecution<R, E> {
41    /// Tokio join handle for the blocking task.
42    handle: JoinHandle<Result<R, E>>,
43}
44
45impl<R, E> TokioExecution<R, E> {
46    /// Creates a Tokio execution wrapper.
47    ///
48    /// # Parameters
49    ///
50    /// * `handle` - The Tokio join handle that produces the task result.
51    ///
52    /// # Returns
53    ///
54    /// A future-backed execution wrapper.
55    #[inline]
56    pub(crate) fn new(handle: JoinHandle<Result<R, E>>) -> Self {
57        Self { handle }
58    }
59
60    /// Returns whether the Tokio task has finished.
61    ///
62    /// # Returns
63    ///
64    /// `true` if the underlying Tokio task has completed.
65    #[inline]
66    pub fn is_finished(&self) -> bool {
67        self.handle.is_finished()
68    }
69
70    /// Requests cancellation of the underlying Tokio task.
71    ///
72    /// Tokio can cancel a blocking task only before it starts. If the blocking
73    /// closure is already running, this request is best-effort and awaiting the
74    /// execution will still wait for the closure to finish.
75    ///
76    /// # Returns
77    ///
78    /// `true` after the cancellation request has been sent to Tokio.
79    #[inline]
80    pub fn cancel(&self) -> bool {
81        self.handle.abort();
82        true
83    }
84}
85
86impl<R, E> Future for TokioExecution<R, E> {
87    type Output = Result<R, E>;
88
89    /// Polls the underlying Tokio task.
90    ///
91    /// # Parameters
92    ///
93    /// * `cx` - Async task context used to register the current waker.
94    ///
95    /// # Returns
96    ///
97    /// `Poll::Ready` with the callable result when the Tokio task completes,
98    /// or `Poll::Pending` while the task is still running.
99    ///
100    /// # Panics
101    ///
102    /// Panics if Tokio reports the blocking task was cancelled. If the task
103    /// panicked, this method resumes the original panic payload.
104    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
105        let this = self.get_mut();
106        match Pin::new(&mut this.handle).poll(cx) {
107            Poll::Ready(Ok(result)) => Poll::Ready(result),
108            Poll::Ready(Err(error)) => handle_join_error(error),
109            Poll::Pending => Poll::Pending,
110        }
111    }
112}
113
114/// Converts a Tokio join error into this execution's panic behavior.
115///
116/// # Parameters
117///
118/// * `error` - Tokio join error returned while awaiting the blocking task.
119///
120/// # Returns
121///
122/// This function never returns normally for a join error; its return type
123/// matches the call site.
124///
125/// # Panics
126///
127/// Resumes the task panic when Tokio reports a panic, or panics with a
128/// cancellation message when the task was cancelled.
129fn handle_join_error<R, E>(error: JoinError) -> Poll<Result<R, E>> {
130    if error.is_panic() {
131        std::panic::resume_unwind(error.into_panic());
132    }
133    panic!("tokio execution was cancelled before completion");
134}
qubit_tokio_executor/tokio_execution.rs

qubit_tokio_executor/
tokio_execution.rs
