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
use {
crate::future::{CatchUnwind, FutureExt},
futures_channel::oneshot::{self, Sender, Receiver},
futures_core::{
future::Future,
task::{LocalWaker, Poll},
},
pin_utils::{unsafe_pinned, unsafe_unpinned},
std::{
any::Any,
fmt,
marker::Unpin,
panic::{self, AssertUnwindSafe},
pin::Pin,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
thread,
},
};
#[must_use = "futures do nothing unless polled"]
#[derive(Debug)]
pub struct RemoteHandle<T> {
rx: Receiver<thread::Result<T>>,
keep_running: Arc<AtomicBool>,
}
impl<T> RemoteHandle<T> {
pub fn forget(self) {
self.keep_running.store(true, Ordering::SeqCst);
}
}
impl<T: Send + 'static> Future for RemoteHandle<T> {
type Output = T;
fn poll(mut self: Pin<&mut Self>, lw: &LocalWaker) -> Poll<T> {
match self.rx.poll_unpin(lw) {
Poll::Ready(Ok(Ok(output))) => Poll::Ready(output),
Poll::Ready(Ok(Err(e))) => panic::resume_unwind(e),
Poll::Ready(Err(e)) => panic::resume_unwind(Box::new(e)),
Poll::Pending => Poll::Pending,
}
}
}
type SendMsg<Fut> = Result<<Fut as Future>::Output, Box<(dyn Any + Send + 'static)>>;
#[must_use = "futures do nothing unless polled"]
pub struct Remote<Fut: Future> {
tx: Option<Sender<SendMsg<Fut>>>,
keep_running: Arc<AtomicBool>,
future: CatchUnwind<AssertUnwindSafe<Fut>>,
}
impl<Fut: Future + fmt::Debug> fmt::Debug for Remote<Fut> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_tuple("Remote")
.field(&self.future)
.finish()
}
}
impl<Fut: Future + Unpin> Unpin for Remote<Fut> {}
impl<Fut: Future> Remote<Fut> {
unsafe_pinned!(future: CatchUnwind<AssertUnwindSafe<Fut>>);
unsafe_unpinned!(tx: Option<Sender<SendMsg<Fut>>>);
unsafe_unpinned!(keep_running: Arc<AtomicBool>);
}
impl<Fut: Future> Future for Remote<Fut> {
type Output = ();
fn poll(mut self: Pin<&mut Self>, lw: &LocalWaker) -> Poll<()> {
if let Poll::Ready(_) = self.tx().as_mut().unwrap().poll_cancel(lw) {
if !self.keep_running().load(Ordering::SeqCst) {
return Poll::Ready(())
}
}
let output = match self.future().poll(lw) {
Poll::Ready(output) => output,
Poll::Pending => return Poll::Pending,
};
drop(self.tx().take().unwrap().send(output));
Poll::Ready(())
}
}
pub(super) fn remote_handle<Fut: Future>(future: Fut) -> (Remote<Fut>, RemoteHandle<Fut::Output>) {
let (tx, rx) = oneshot::channel();
let keep_running = Arc::new(AtomicBool::new(false));
let wrapped = Remote {
future: AssertUnwindSafe(future).catch_unwind(),
tx: Some(tx),
keep_running: keep_running.clone(),
};
(wrapped, RemoteHandle { rx, keep_running })
}