1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
use std::panic::{catch_unwind, AssertUnwindSafe};
use crate::{Promise, PromiseRejection};
impl<T, E> Promise<T, E>
where
T: Send + 'static,
E: PromiseRejection,
{
/// Runs a blocking closure on a thread pool and wraps the result in a
/// [`Promise`].
///
/// Dispatch is selected at compile time based on which runtime features
/// are enabled, with a runtime check when `tokio` is on:
///
/// - `tokio` enabled and called from within a tokio runtime context
/// (detected via `tokio::runtime::Handle::try_current`): dispatches to
/// `tokio::task::spawn_blocking`; a panic in the closure is mapped to
/// [`TaskFailure::Panic`](crate::TaskFailure::Panic), and a cancelled
/// task (runtime shutdown) to
/// [`TaskFailure::Aborted`](crate::TaskFailure::Aborted).
/// - Otherwise, with `smol` enabled: dispatches to `smol::unblock`.
/// - Otherwise: dispatches to `blocking::unblock`, which uses the
/// `blocking` crate's runtime-independent thread pool.
///
/// On the smol and blocking paths, the pool task is detached, and a
/// panic in the closure is caught on the worker thread and mapped to
/// [`TaskFailure::Panic`](crate::TaskFailure::Panic); a closure panic
/// rejects the promise on every path.
///
/// The closure is scheduled synchronously during this call and runs to
/// completion even if the [`Promise`] is dropped, in which case its
/// outcome is discarded. The outer [`Promise`] must still be polled (or
/// awaited) to receive the outcome.
pub fn unblock<F>(f: F) -> Self
where
F: FnOnce() -> Result<T, E> + Send + 'static,
{
#[cfg(feature = "tokio")]
if tokio::runtime::Handle::try_current().is_ok() {
let handle = tokio::task::spawn_blocking(f);
return Self::lazy(
async move { super::eager_with_tokio::map_join_result(handle.await) },
);
}
let (relay, resolve, reject) = Self::with_resolvers();
let run = move || match catch_unwind(AssertUnwindSafe(f)) {
Ok(Ok(value)) => resolve.resolve(value),
Ok(Err(rejection)) => reject.reject(rejection),
Err(panic) => reject.reject(E::task_failed(crate::TaskFailure::from(panic))),
};
#[cfg(feature = "smol")]
smol::unblock(run).detach();
#[cfg(not(feature = "smol"))]
blocking::unblock(run).detach();
relay
}
}
#[cfg(test)]
#[allow(clippy::expect_used)]
mod tests {
use std::{
task::{Context, Waker},
time::{Duration, Instant},
};
use crate::{Promise, PromiseRejection, TaskFailure};
#[derive(Debug)]
enum E {
AlreadyConsumed,
Fail,
TaskFailed(TaskFailure),
}
impl PromiseRejection for E {
fn already_consumed() -> Self {
Self::AlreadyConsumed
}
fn task_failed(failure: TaskFailure) -> Self {
Self::TaskFailed(failure)
}
}
fn wait_settled<T: Send + 'static>(promise: &mut Promise<T, E>) {
let deadline = Instant::now() + Duration::from_secs(5);
while !promise.poll_settled(&mut Context::from_waker(Waker::noop())) {
assert!(Instant::now() < deadline, "promise did not settle in time");
std::thread::yield_now();
}
}
#[test]
fn resolves_value() {
let mut promise: Promise<i32, E> = Promise::unblock(|| Ok(42));
wait_settled(&mut promise);
assert!(matches!(promise.consume(), Some(Ok(42))));
}
#[test]
fn rejects_app_error() {
let mut promise: Promise<i32, E> = Promise::unblock(|| Err(E::Fail));
wait_settled(&mut promise);
assert!(matches!(promise.consume(), Some(Err(E::Fail))));
}
#[test]
fn closure_panic_rejects() {
let mut promise: Promise<i32, E> = Promise::unblock(|| panic!("boom"));
wait_settled(&mut promise);
match promise.consume() {
Some(Err(E::TaskFailed(failure @ TaskFailure::Panic(_)))) => {
assert_eq!(failure.to_string(), "task panicked: boom");
}
other => panic!("expected Err(TaskFailed(Panic(_))), got {other:?}"),
}
}
/// Dropping the outer [`Promise`] abandons the outcome but must not
/// cancel the closure, even one the pool has not started yet, mirroring
/// ECMAScript promise semantics.
#[test]
fn dropped_promise_leaves_the_closure_running() {
let (start_tx, start_rx) = std::sync::mpsc::channel::<()>();
let (done_tx, done_rx) = std::sync::mpsc::channel::<()>();
let promise: Promise<i32, E> = Promise::unblock(move || {
start_rx.recv().ok();
done_tx.send(()).ok();
Ok(0)
});
drop(promise);
start_tx
.send(())
.expect("the closure must still be listening");
done_rx
.recv_timeout(Duration::from_secs(5))
.expect("the closure must run to completion");
}
}