char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somrearrange libs into lib and extensions into extsf4a0f5b

main
branchesmainall refs →
3.2 KiB108 linesraw
1pub mod environment;
2pub mod io;
3pub mod messaging;
4
5use std::future::Future;
6use std::sync::Arc;
7use std::sync::atomic::{AtomicBool, Ordering};
8
9use tokio::runtime::{Handle, Runtime};
10
11/// The Tokio runtime shared by the filesystem and network extensions.
12pub struct Host(Option<Runtime>);
13
14impl Host {
15    pub fn new() -> std::io::Result<Self> {
16        let runtime = tokio::runtime::Builder::new_multi_thread()
17            // A second worker keeps decompression or a large message from stalling other connections.
18            .worker_threads(2)
19            .thread_name("avril:host")
20            .enable_all()
21            .build()?;
22        Ok(Self(Some(runtime)))
23    }
24
25    pub fn handle(&self) -> Handle {
26        self.0.as_ref().unwrap().handle().clone()
27    }
28}
29
30impl Drop for Host {
31    fn drop(&mut self) {
32        // getaddrinfo may still be running in Tokio's blocking pool after an abort.
33        if let Some(runtime) = self.0.take() {
34            runtime.shutdown_background();
35        }
36    }
37}
38
39pub fn interrupted(scope: &avril_core::v8::PinScope, flag: &AtomicBool) -> bool {
40    if flag.load(Ordering::SeqCst) {
41        // V8 may not observe a termination request until this native call returns.
42        scope.terminate_execution();
43        true
44    } else {
45        scope.is_execution_terminating()
46    }
47}
48
49struct WakeThread(std::thread::Thread);
50
51impl std::task::Wake for WakeThread {
52    fn wake(self: Arc<Self>) {
53        self.0.unpark();
54    }
55
56    fn wake_by_ref(self: &Arc<Self>) {
57        self.0.unpark();
58    }
59}
60
61// Worker and REPL loops may already be inside a futures executor. Poll only
62// this op's native work, without entering an executor or pumping JavaScript.
63pub fn wait<F: Future>(future: F, interrupted: impl Fn() -> bool) -> Option<F::Output> {
64    let waker = std::task::Waker::from(Arc::new(WakeThread(std::thread::current())));
65    let mut context = std::task::Context::from_waker(&waker);
66    // A synchronous wait cannot yield to the outer task to replenish its budget.
67    let mut pending = std::pin::pin!(tokio::task::unconstrained(future));
68    loop {
69        if interrupted() {
70            return None;
71        }
72        if let std::task::Poll::Ready(result) = pending.as_mut().poll(&mut context) {
73            return Some(result);
74        }
75        std::thread::park_timeout(std::time::Duration::from_millis(10));
76    }
77}
78
79#[cfg(test)]
80mod tests {
81    use super::{Host, wait};
82    use std::time::{Duration, Instant};
83
84    #[test]
85    fn synchronous_wait_drains_tokio_io_inside_an_async_task() {
86        let host = Host::new().unwrap();
87        let (sender, mut receiver) = tokio::sync::mpsc::channel(1_000);
88        for value in 0..1_000 {
89            sender.try_send(value).unwrap();
90        }
91        drop(sender);
92
93        host.handle().block_on(async {
94            let deadline = Instant::now() + Duration::from_secs(2);
95            let result = wait(
96                async {
97                    let mut values = Vec::new();
98                    while let Some(value) = receiver.recv().await {
99                        values.push(value);
100                    }
101                    values
102                },
103                || Instant::now() >= deadline,
104            );
105            assert_eq!(result, Some((0..1_000).collect()));
106        });
107    }
108}