blob: 05ec85a950ecb1149bd96bfb33bf06a741acd474 [file] [log] [blame]
Stjepan Glavina1479e862019-08-12 20:18:51 +02001//! A single-threaded executor where join handles propagate panics from tasks.
2
Stjepan Glavina1479e862019-08-12 20:18:51 +02003use std::future::Future;
4use std::panic::{resume_unwind, AssertUnwindSafe};
5use std::pin::Pin;
6use std::task::{Context, Poll};
7use std::thread;
8
9use crossbeam::channel::{unbounded, Sender};
10use futures::executor;
11use futures::future::FutureExt;
12use lazy_static::lazy_static;
13
Stjepan Glavinafcfa4ab2019-11-25 18:39:17 +010014type Task = async_task::Task<()>;
15
Stjepan Glavina1479e862019-08-12 20:18:51 +020016/// Spawns a future on the executor.
17fn spawn<F, R>(future: F) -> JoinHandle<R>
18where
19 F: Future<Output = R> + Send + 'static,
20 R: Send + 'static,
21{
22 lazy_static! {
23 // A channel that holds scheduled tasks.
Stjepan Glavinafcfa4ab2019-11-25 18:39:17 +010024 static ref QUEUE: Sender<Task> = {
25 let (sender, receiver) = unbounded::<Task>();
Stjepan Glavina1479e862019-08-12 20:18:51 +020026
27 // Start the executor thread.
28 thread::spawn(|| {
29 for task in receiver {
30 // No need for `catch_unwind()` here because panics are already caught.
31 task.run();
32 }
33 });
34
35 sender
36 };
37 }
38
39 // Create a future that catches panics within itself.
40 let future = AssertUnwindSafe(future).catch_unwind();
41
42 // Create a task that is scheduled by sending itself into the channel.
43 let schedule = |t| QUEUE.send(t).unwrap();
44 let (task, handle) = async_task::spawn(future, schedule, ());
45
46 // Schedule the task by sending it into the channel.
47 task.schedule();
48
49 // Wrap the handle into one that propagates panics.
50 JoinHandle(handle)
51}
52
53/// A join handle that propagates panics inside the task.
54struct JoinHandle<R>(async_task::JoinHandle<thread::Result<R>, ()>);
55
56impl<R> Future for JoinHandle<R> {
57 type Output = Option<R>;
58
59 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
60 match Pin::new(&mut self.0).poll(cx) {
61 Poll::Pending => Poll::Pending,
62 Poll::Ready(None) => Poll::Ready(None),
63 Poll::Ready(Some(Ok(val))) => Poll::Ready(Some(val)),
64 Poll::Ready(Some(Err(err))) => resume_unwind(err),
65 }
66 }
67}
68
69fn main() {
70 // Spawn a future that panics and block on it.
71 let handle = spawn(async {
72 panic!("Ooops!");
73 });
74 executor::block_on(handle);
75}