char/avril

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

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

Charlotte Somport workers extension to typescriptb7bd857

main
branchesmainall refs →
9.4 KiB276 linesraw
1use std::cell::RefCell;
2use std::collections::HashMap;
3use std::convert::Infallible;
4use std::io;
5use std::rc::Rc;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::sync::{Arc, Mutex};
8use std::thread::JoinHandle;
9
10use avril_caps::{Capabilities, Category, Pledge};
11use avril_core::{Extension, Op, v8};
12use avril_host::environment::Environment;
13use avril_host::messaging::{Broker, Endpoint};
14use futures::channel::oneshot;
15use futures::task::AtomicWaker;
16use tokio::sync::Notify;
17use tokio_util::sync::CancellationToken;
18
19const MAX_ID: u64 = 9_007_199_254_740_991;
20
21pub struct Request {
22    pub specifier: String,
23    pub referrer: String,
24    pub name: String,
25    pub caps: Arc<Capabilities>,
26    pub environment: Environment,
27}
28
29/// Runs on the child isolate thread. The host owns loading and runtime assembly.
30pub trait Launch: Send + Sync {
31    fn run(
32        &self,
33        request: Request,
34        control: Arc<Control>,
35        inbox: Arc<Endpoint>,
36    ) -> Result<(), Box<dyn std::error::Error>>;
37}
38
39#[derive(Default)]
40pub struct Control {
41    stopped: AtomicBool,
42    terminated: Arc<AtomicBool>,
43    processes: CancellationToken,
44    isolate: Mutex<Option<v8::IsolateHandle>>,
45    notify: Notify,
46    waker: AtomicWaker,
47}
48
49impl Control {
50    pub fn is_stopped(&self) -> bool {
51        self.stopped.load(Ordering::SeqCst)
52    }
53
54    pub fn termination_flag(&self) -> Arc<AtomicBool> {
55        self.terminated.clone()
56    }
57
58    pub fn process_cancellation(&self) -> CancellationToken {
59        self.processes.clone()
60    }
61
62    pub fn set_isolate(&self, isolate: Option<v8::IsolateHandle>) {
63        let mut handle = self.isolate.lock().unwrap();
64        *handle = isolate;
65        if self.is_stopped()
66            && let Some(handle) = &*handle
67        {
68            handle.terminate_execution();
69        }
70    }
71
72    pub fn register_waker(&self, waker: &std::task::Waker) {
73        self.waker.register(waker);
74    }
75
76    fn stop(&self, terminate: bool) {
77        self.stopped.store(true, Ordering::SeqCst);
78        self.processes.cancel();
79        if terminate {
80            self.terminated.store(true, Ordering::SeqCst);
81            if let Some(isolate) = &*self.isolate.lock().unwrap() {
82                isolate.terminate_execution();
83            }
84        }
85        self.notify.notify_waiters();
86        self.waker.wake();
87    }
88
89    pub async fn cancelled(&self) {
90        loop {
91            let notified = self.notify.notified();
92            tokio::pin!(notified);
93            notified.as_mut().enable();
94            if self.stopped.load(Ordering::SeqCst) {
95                return;
96            }
97            notified.await;
98        }
99    }
100}
101
102pub struct WorkerScope {
103    pub control: Arc<Control>,
104    pub inbox: Arc<Endpoint>,
105    pub url: String,
106    pub name: String,
107}
108
109struct Child {
110    control: Arc<Control>,
111    inbox: Arc<Endpoint>,
112    outbox: Arc<Endpoint>,
113    thread: Option<JoinHandle<()>>,
114    completion: Option<oneshot::Receiver<Option<String>>>,
115}
116
117impl Child {
118    fn terminate(&self) {
119        self.control.stop(true);
120        self.inbox.close();
121    }
122}
123
124impl Drop for Child {
125    fn drop(&mut self) {
126        self.terminate();
127        if let Some(thread) = self.thread.take() {
128            let _ = thread.join();
129        }
130    }
131}
132
133/// One serialized capability pledge row inherited from the parent scope.
134type PledgeRow = (String, Option<Vec<String>>, Vec<String>);
135
136/// `op_worker_create` arguments: specifier, referrer, worker name, pledges.
137type CreateArgs = (String, String, String, Vec<PledgeRow>);
138
139pub fn init(
140    entry_url: impl Fn() -> String + 'static,
141    caps: Arc<Capabilities>,
142    broker: Arc<Broker>,
143    launch: Arc<dyn Launch>,
144    environment: Rc<RefCell<Environment>>,
145    worker: Option<WorkerScope>,
146) -> Extension {
147    let children = Rc::new(RefCell::new(HashMap::<u64, Child>::new()));
148    let next = Rc::new(std::cell::Cell::new(0u64));
149    let metadata = worker
150        .as_ref()
151        .map(|scope| (scope.inbox.id(), scope.url.clone(), scope.name.clone()));
152    Extension {
153        name: "ext:workers",
154        dependencies: &["ext:base", "ext:url", "ext:web"],
155        ops: vec![
156            Op::new("op_worker_metadata", move |(): ()| {
157                Ok::<_, Infallible>(metadata.clone())
158            }),
159            Op::new("op_worker_create", {
160                let children = children.clone();
161                move |(specifier, referrer, name, restrictions): CreateArgs| {
162                    let environment = environment.borrow().clone();
163                    let caps = Capabilities::inherit(caps.clone());
164                    let launch = launch.clone();
165                    let restrictions = restrictions
166                        .into_iter()
167                        .map(|(category, allow, deny)| {
168                            category
169                                .parse::<Category>()
170                                .map(|category| (category, allow, deny))
171                        })
172                        .collect::<Result<Pledge, _>>()
173                        .map_err(io::Error::other)?;
174                    caps.pledge(restrictions).map_err(io::Error::other)?;
175                    let (parent, inbox) = broker.pair()?;
176                    let control = Arc::new(Control::default());
177                    let (done, completion) = oneshot::channel();
178                    let child_control = control.clone();
179                    let child_inbox = inbox.clone();
180                    let referrer = if referrer.is_empty() {
181                        caps.cwd()
182                            .join("__worker__.js")
183                            .to_string_lossy()
184                            .into_owned()
185                    } else {
186                        referrer
187                    };
188                    let id = next.get();
189                    if id > MAX_ID {
190                        return Err(io::Error::other("Worker ID space exhausted"));
191                    }
192                    next.set(id + 1);
193                    let thread = std::thread::Builder::new()
194                        .name(format!("avril:worker:{id}"))
195                        .spawn(move || {
196                            let result = launch.run(
197                                Request {
198                                    specifier,
199                                    referrer,
200                                    name,
201                                    caps,
202                                    environment,
203                                },
204                                child_control.clone(),
205                                child_inbox.clone(),
206                            );
207                            child_inbox.close();
208                            let error = if child_control.stopped.load(Ordering::SeqCst) {
209                                None
210                            } else {
211                                result.err().map(|error| error.to_string())
212                            };
213                            let _ = done.send(error);
214                        })?;
215                    children.borrow_mut().insert(
216                        id,
217                        Child {
218                            control,
219                            inbox,
220                            outbox: parent.clone(),
221                            thread: Some(thread),
222                            completion: Some(completion),
223                        },
224                    );
225                    Ok::<_, io::Error>((id, parent.id()))
226                }
227            }),
228            Op::asynchronous("op_worker_wait", {
229                let children = children.clone();
230                move |id: u64| {
231                    let completion = children
232                        .borrow_mut()
233                        .get_mut(&id)
234                        .and_then(|child| child.completion.take());
235                    let outbox = children.borrow().get(&id).map(|child| child.outbox.clone());
236                    let children = children.clone();
237                    async move {
238                        let result = match completion {
239                            Some(completion) => completion
240                                .await
241                                .unwrap_or(Some("Worker thread stopped unexpectedly".into())),
242                            None => None,
243                        };
244                        if let Some(outbox) = outbox {
245                            outbox.drained().await;
246                        }
247                        children.borrow_mut().remove(&id);
248                        Ok::<_, Infallible>(result)
249                    }
250                }
251            }),
252            Op::new("op_worker_terminate", move |id: u64| {
253                if let Some(child) = children.borrow().get(&id) {
254                    child.terminate();
255                }
256                Ok::<_, Infallible>(())
257            }),
258            Op::new("op_worker_close", move |(): ()| {
259                if let Some(scope) = &worker {
260                    scope.control.stop(false);
261                    scope.inbox.close();
262                }
263                Ok::<_, Infallible>(())
264            }),
265            Op::new("op_worker_base", move |(): ()| {
266                // The entry URL is the worker API's base, not the constructor's call site.
267                Ok::<_, Infallible>(entry_url())
268            }),
269        ],
270        modules: &[(
271            "ext:workers/main.js",
272            include_str!(concat!(env!("OUT_DIR"), "/main.js")),
273        )],
274        entry_point: Some("ext:workers/main.js"),
275    }
276}