ste 0.1.0-alpha.11

A single-threaded executor with some tricks up its sleeve.
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

use crate::misc::RawSend;
use crate::parker::Parker;
use crate::tag::{with_tag, Tag};
use crate::worker::{Entry, Shared};
use std::future::Future;
use std::pin::Pin;
use std::ptr;
use std::task::{Context, Poll, Waker};

pub(super) struct WaitFuture<'a, F>
where
    F: Future,
{
    /// The future being polled.
    pub(super) future: ptr::NonNull<F>,
    /// Where to store output.
    pub(super) output: ptr::NonNull<Option<F::Output>>,
    pub(super) parker: ptr::NonNull<Parker>,
    pub(super) complete: bool,
    pub(super) shared: &'a Shared,
}

impl<'a, F> Future for WaitFuture<'a, F>
where
    F: Future,
{
    type Output = F::Output;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        unsafe {
            let this = Pin::get_unchecked_mut(self.as_mut());

            if this.complete {
                panic!("task already completed");
            }

            let mut task = into_task(
                RawSend((&mut this.complete).into()),
                RawSend(this.future),
                RawSend(this.output),
                RawSend(cx.waker().into()),
            );
            let entry = Entry::new(&mut task, this.parker);

            this.shared.schedule_in_place(this.parker, entry);

            if this.complete {
                panic!("background thread panicked");
            }

            if let Some(output) = this.output.as_mut().take() {
                this.complete = true;
                Poll::Ready(output)
            } else {
                Poll::Pending
            }
        }
    }
}

unsafe impl<F> Send for WaitFuture<'_, F> where F: Future {}

fn into_task<F>(
    mut complete: RawSend<bool>,
    mut future: RawSend<F>,
    mut output: RawSend<Option<F::Output>>,
    waker: RawSend<Waker>,
) -> impl FnMut(Tag) + Send
where
    F: Future,
{
    use std::panic;

    move |tag| {
        unsafe {
            // Safety: At this point, we know the waker has been
            // replaced by the polling task and can safely deref it into
            // the underlying waker.
            let waker = waker.0.as_ref();

            let mut cx = Context::from_waker(waker);
            let future = Pin::new_unchecked(future.0.as_mut());

            let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
                if let Poll::Ready(ready) = with_tag(tag, || future.poll(&mut cx)) {
                    *output.0.as_mut() = Some(ready);
                }
            }));

            if result.is_err() {
                *complete.0.as_mut() = true;
            }
        }
    }
}