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 →
15.0 KiB453 linesraw
1use std::cell::RefCell;
2use std::collections::{BTreeMap, HashMap, VecDeque};
3use std::convert::Infallible;
4use std::rc::Rc;
5use std::sync::{Arc, Mutex, Weak};
6
7use avril_core::{Op, message, v8};
8use tokio::sync::Notify;
9
10type Id = u64;
11
12const MAX_ID: u64 = 9_007_199_254_740_991;
13
14fn u64_arg<'s, 'i>(
15    scope: &mut v8::PinScope<'s, 'i>,
16    value: v8::Local<'s, v8::Value>,
17) -> Option<u64> {
18    match avril_core::serde_v8::from_v8::<u64>(scope, value) {
19        Ok(id) => Some(id),
20        Err(error) => {
21            let message = v8::String::new(scope, &error.to_string())?;
22            scope.throw_exception(v8::Exception::type_error(scope, message));
23            None
24        }
25    }
26}
27
28fn port_id<'s, 'i>(
29    scope: &mut v8::PinScope<'s, 'i>,
30    value: v8::Local<'s, v8::Value>,
31) -> Option<Id> {
32    u64_arg(scope, value)
33}
34
35struct Packet {
36    message: message::Message,
37    _transfers: Vec<Arc<Endpoint>>,
38}
39
40#[derive(Default)]
41struct State {
42    queue: VecDeque<Packet>,
43    epoch: u64,
44    closed: bool,
45}
46
47pub struct Endpoint {
48    id: Id,
49    broker: Weak<Broker>,
50    peer: Mutex<Weak<Endpoint>>,
51    state: Mutex<State>,
52    notify: Notify,
53}
54
55impl Endpoint {
56    pub fn close(&self) {
57        let mut state = self.state.lock().unwrap();
58        state.closed = true;
59        state.queue.clear();
60        drop(state);
61        self.notify.notify_waiters();
62        if let Some(peer) = self.peer.lock().unwrap().upgrade() {
63            peer.notify.notify_waiters();
64        }
65    }
66
67    pub fn id(&self) -> Id {
68        self.id
69    }
70
71    pub async fn drained(&self) {
72        loop {
73            let notified = self.notify.notified();
74            tokio::pin!(notified);
75            notified.as_mut().enable();
76            if self.state.lock().unwrap().queue.is_empty() {
77                return;
78            }
79            notified.await;
80        }
81    }
82
83    async fn wait(&self, epoch: u64) -> bool {
84        loop {
85            let notified = self.notify.notified();
86            tokio::pin!(notified);
87            notified.as_mut().enable();
88            let peer_closed = self
89                .peer
90                .lock()
91                .unwrap()
92                .upgrade()
93                .is_none_or(|peer| peer.state.lock().unwrap().closed);
94            {
95                let state = self.state.lock().unwrap();
96                if state.closed || state.epoch != epoch {
97                    return false;
98                }
99                if !state.queue.is_empty() {
100                    return true;
101                }
102                if peer_closed {
103                    return false;
104                }
105            }
106            notified.await;
107        }
108    }
109}
110
111impl Drop for Endpoint {
112    fn drop(&mut self) {
113        if let Some(broker) = self.broker.upgrade()
114            && let Ok(mut registry) = broker.0.lock()
115        {
116            registry.ports.remove(&self.id);
117        }
118        if let Some(peer) = self.peer.get_mut().unwrap().upgrade() {
119            peer.notify.notify_waiters();
120        }
121    }
122}
123
124#[derive(Default)]
125struct Registry {
126    next: Id,
127    ports: HashMap<Id, Weak<Endpoint>>,
128    broadcasts: HashMap<Vec<u16>, BTreeMap<Id, Weak<Endpoint>>>,
129}
130
131#[derive(Default)]
132pub struct Broker(Mutex<Registry>);
133
134impl Broker {
135    pub fn pair(self: &Arc<Self>) -> std::io::Result<(Arc<Endpoint>, Arc<Endpoint>)> {
136        let mut registry = self.0.lock().unwrap();
137        let id = registry.next;
138        if id > MAX_ID - 1 {
139            return Err(std::io::Error::other("MessagePort ID space exhausted"));
140        }
141        registry.next = id + 2;
142        let first = Arc::new(Endpoint {
143            id,
144            broker: Arc::downgrade(self),
145            peer: Mutex::default(),
146            state: Mutex::default(),
147            notify: Notify::new(),
148        });
149        let second = Arc::new(Endpoint {
150            id: id + 1,
151            broker: Arc::downgrade(self),
152            peer: Mutex::default(),
153            state: Mutex::default(),
154            notify: Notify::new(),
155        });
156        *first.peer.lock().unwrap() = Arc::downgrade(&second);
157        *second.peer.lock().unwrap() = Arc::downgrade(&first);
158        registry.ports.insert(id, Arc::downgrade(&first));
159        registry.ports.insert(id + 1, Arc::downgrade(&second));
160        Ok((first, second))
161    }
162
163    pub fn endpoint(&self, id: Id) -> Option<Arc<Endpoint>> {
164        self.0
165            .lock()
166            .unwrap()
167            .ports
168            .get(&id)
169            .and_then(Weak::upgrade)
170    }
171}
172
173struct Subscription {
174    id: Id,
175    name: Vec<u16>,
176    sender: Arc<Endpoint>,
177    broker: Arc<Broker>,
178}
179
180impl Drop for Subscription {
181    fn drop(&mut self) {
182        self.sender.close();
183        let Ok(mut registry) = self.broker.0.lock() else {
184            return;
185        };
186        if let Some(group) = registry.broadcasts.get_mut(&self.name) {
187            group.remove(&self.id);
188            if group.is_empty() {
189                registry.broadcasts.remove(&self.name);
190            }
191        }
192    }
193}
194
195struct Ports {
196    broker: Arc<Broker>,
197    owned: RefCell<HashMap<Id, Arc<Endpoint>>>,
198    moving: RefCell<Vec<Arc<Endpoint>>>,
199    broadcasts: RefCell<HashMap<Id, Subscription>>,
200}
201
202impl Ports {
203    fn resolve(&self, id: Id) -> Option<Arc<Endpoint>> {
204        if let Some(endpoint) = self.owned.borrow().get(&id).cloned() {
205            return Some(endpoint);
206        }
207        self.broker.endpoint(id)
208    }
209}
210
211impl Drop for Ports {
212    fn drop(&mut self) {
213        for (_, endpoint) in self.owned.get_mut().drain() {
214            endpoint.close();
215        }
216    }
217}
218
219pub fn ops(broker: Arc<Broker>) -> Vec<Op> {
220    let ports = Rc::new(Ports {
221        broker: broker.clone(),
222        owned: RefCell::default(),
223        moving: RefCell::default(),
224        broadcasts: RefCell::default(),
225    });
226    vec![
227        Op::raw("op_broadcast_open", {
228            let ports = ports.clone();
229            move |scope, args| {
230                let value = v8::Local::<v8::String>::try_from(args.get(0)).ok()?;
231                let mut name = vec![0; value.length()];
232                value.write_v2(scope, 0, &mut name, v8::WriteFlags::empty());
233                let (sender, receiver) = match ports.broker.pair() {
234                    Ok(pair) => pair,
235                    Err(error) => {
236                        let message = v8::String::new(scope, &error.to_string())?;
237                        scope.throw_exception(v8::Exception::error(scope, message));
238                        return None;
239                    }
240                };
241                let id = receiver.id;
242                ports.owned.borrow_mut().insert(id, receiver.clone());
243                ports
244                    .broker
245                    .0
246                    .lock()
247                    .unwrap()
248                    .broadcasts
249                    .entry(name.clone())
250                    .or_default()
251                    .insert(id, Arc::downgrade(&receiver));
252                ports.broadcasts.borrow_mut().insert(
253                    id,
254                    Subscription {
255                        id,
256                        name,
257                        sender,
258                        broker: ports.broker.clone(),
259                    },
260                );
261                Some(v8::Number::new(scope, id as f64).into())
262            }
263        }),
264        Op::new("op_broadcast_close", {
265            let ports = ports.clone();
266            move |id: Id| {
267                ports.broadcasts.borrow_mut().remove(&id);
268                Ok::<_, Infallible>(())
269            }
270        }),
271        Op::raw("op_broadcast_send", {
272            let ports = ports.clone();
273            move |scope, args| {
274                let id = port_id(scope, args.get(0))?;
275                let name = ports.broadcasts.borrow().get(&id)?.name.clone();
276                let transfer = v8::Array::new(scope, 0);
277                let callbacks = [
278                    v8::Local::<v8::Function>::try_from(args.get(2)).ok()?,
279                    v8::Local::<v8::Function>::try_from(args.get(3)).ok()?,
280                    v8::Local::<v8::Function>::try_from(args.get(4)).ok()?,
281                    v8::Local::<v8::Function>::try_from(args.get(5)).ok()?,
282                    v8::Local::<v8::Function>::try_from(args.get(6)).ok()?,
283                ];
284                let message = message::serialize(scope, args.get(1), transfer, callbacks)?;
285                let recipients: Vec<_> = ports
286                    .broker
287                    .0
288                    .lock()
289                    .unwrap()
290                    .broadcasts
291                    .get(&name)
292                    .into_iter()
293                    .flat_map(|group| group.iter())
294                    .filter(|(recipient, _)| **recipient != id)
295                    .filter_map(|(_, endpoint)| endpoint.upgrade())
296                    .collect();
297                for endpoint in recipients {
298                    let mut state = endpoint.state.lock().unwrap();
299                    if !state.closed {
300                        state.queue.push_back(Packet {
301                            message: message
302                                .try_clone()
303                                .expect("Broadcast messages have no transfers"),
304                            _transfers: Vec::new(),
305                        });
306                    }
307                    drop(state);
308                    endpoint.notify.notify_waiters();
309                }
310                Some(v8::undefined(scope).into())
311            }
312        }),
313        Op::new("op_port_pair", {
314            let ports = ports.clone();
315            move |(): ()| {
316                let (a, b) = ports.broker.pair()?;
317                ports
318                    .owned
319                    .borrow_mut()
320                    .extend([(a.id, a.clone()), (b.id, b.clone())]);
321                Ok::<_, std::io::Error>((a.id, b.id))
322            }
323        }),
324        Op::new("op_port_adopt", {
325            let ports = ports.clone();
326            move |id: Id| {
327                let endpoint = ports
328                    .resolve(id)
329                    .ok_or_else(|| std::io::Error::other("MessagePort endpoint is closed"))?;
330                ports.owned.borrow_mut().insert(id, endpoint.clone());
331                ports.moving.borrow_mut().retain(|port| port.id != id);
332                Ok::<_, std::io::Error>(endpoint.state.lock().unwrap().epoch)
333            }
334        }),
335        Op::new("op_port_move", {
336            let ports = ports.clone();
337            move |id: Id| {
338                let endpoint = ports
339                    .owned
340                    .borrow_mut()
341                    .remove(&id)
342                    .ok_or_else(|| std::io::Error::other("Transferred port is not owned"))?;
343                endpoint.state.lock().unwrap().epoch += 1;
344                endpoint.notify.notify_waiters();
345                ports.moving.borrow_mut().push(endpoint);
346                Ok::<_, std::io::Error>(())
347            }
348        }),
349        Op::new("op_port_close", {
350            let ports = ports.clone();
351            move |id: Id| {
352                if let Some(endpoint) = ports.owned.borrow_mut().remove(&id) {
353                    endpoint.close();
354                }
355                Ok::<_, Infallible>(())
356            }
357        }),
358        Op::new("op_port_peer", {
359            let ports = ports.clone();
360            move |id: Id| {
361                Ok::<_, Infallible>(ports.resolve(id).and_then(|endpoint| {
362                    endpoint.peer.lock().unwrap().upgrade().map(|peer| peer.id)
363                }))
364            }
365        }),
366        Op::asynchronous_unreferenced("op_port_wait", {
367            let ports = ports.clone();
368            move |(id, epoch): (Id, u64)| {
369                let endpoint = ports.resolve(id);
370                async move {
371                    Ok::<_, Infallible>(match endpoint {
372                        Some(endpoint) => endpoint.wait(epoch).await,
373                        None => false,
374                    })
375                }
376            }
377        }),
378        Op::asynchronous("op_port_wait_ref", {
379            let ports = ports.clone();
380            move |(id, epoch): (Id, u64)| {
381                let endpoint = ports.resolve(id);
382                async move {
383                    Ok::<_, Infallible>(match endpoint {
384                        Some(endpoint) => endpoint.wait(epoch).await,
385                        None => false,
386                    })
387                }
388            }
389        }),
390        Op::raw("op_port_send", {
391            let ports = ports.clone();
392            move |scope, args| {
393                let id_value = args.get(0);
394                let id = if id_value.is_null() {
395                    None
396                } else {
397                    Some(port_id(scope, id_value)?)
398                };
399                let value = args.get(1);
400                let transfer = v8::Local::<v8::Array>::try_from(args.get(2)).ok()?;
401                let callbacks = [
402                    v8::Local::<v8::Function>::try_from(args.get(3)).ok()?,
403                    v8::Local::<v8::Function>::try_from(args.get(4)).ok()?,
404                    v8::Local::<v8::Function>::try_from(args.get(5)).ok()?,
405                    v8::Local::<v8::Function>::try_from(args.get(6)).ok()?,
406                    v8::Local::<v8::Function>::try_from(args.get(7)).ok()?,
407                ];
408                let serialized = message::serialize(scope, value, transfer, callbacks);
409                let transfers = std::mem::take(&mut *ports.moving.borrow_mut());
410                let packet = Packet {
411                    message: serialized?,
412                    _transfers: transfers,
413                };
414                let endpoint = id.and_then(|id| ports.resolve(id));
415                if let Some(endpoint) = endpoint
416                    && !endpoint.state.lock().unwrap().closed
417                    && let Some(peer) = endpoint.peer.lock().unwrap().upgrade()
418                {
419                    let mut state = peer.state.lock().unwrap();
420                    if !state.closed {
421                        state.queue.push_back(packet);
422                    }
423                    drop(state);
424                    peer.notify.notify_waiters();
425                }
426                Some(v8::undefined(scope).into())
427            }
428        }),
429        Op::raw("op_port_take", {
430            let ports = ports.clone();
431            move |scope, args| {
432                let id = port_id(scope, args.get(0))?;
433                let epoch = u64_arg(scope, args.get(1))?;
434                let host = v8::Local::<v8::Function>::try_from(args.get(2)).ok()?;
435                let error = v8::Local::<v8::Function>::try_from(args.get(3)).ok()?;
436                let Some(endpoint) = ports.resolve(id) else {
437                    return Some(v8::null(scope).into());
438                };
439                let mut state = endpoint.state.lock().unwrap();
440                if state.epoch != epoch || state.closed {
441                    return Some(v8::null(scope).into());
442                }
443                let packet = state.queue.pop_front();
444                drop(state);
445                endpoint.notify.notify_waiters();
446                match packet {
447                    Some(packet) => message::deserialize(scope, packet.message, host, error),
448                    None => Some(v8::null(scope).into()),
449                }
450            }
451        }),
452    ]
453}