char/avril

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

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

Charlotte Somport net extension to typescript1f64888

main
branchesmainall refs →
17.5 KiB473 linesraw
1use std::collections::HashMap;
2use std::convert::Infallible;
3use std::io;
4use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
5use std::sync::{Arc, Mutex};
6
7use avril_caps::{Capabilities, PathTarget};
8use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
9use avril_core::{Extension, Op};
10use avril_net::{Address, Endpoint, Socket};
11use serde::Serialize;
12use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf};
13use tokio::sync::{Mutex as AsyncMutex, oneshot};
14use tokio_util::sync::CancellationToken;
15
16use avril_host::io::response;
17
18mod udp;
19
20const CHUNK: usize = 64 * 1024;
21const MAX_ID: u64 = 9_007_199_254_740_991;
22type ConnectionInfo = (u64, Endpoint, Endpoint);
23type ListenerInfo = (u64, Endpoint);
24
25fn closed() -> io::Error {
26    io::Error::new(io::ErrorKind::NotConnected, "Socket is closed")
27}
28
29#[derive(Clone)]
30enum Authority {
31    Net {
32        caps: Arc<Capabilities>,
33        hostname: String,
34        port: u16,
35    },
36    Unix {
37        caps: Arc<Capabilities>,
38        target: PathTarget,
39    },
40}
41impl Authority {
42    fn check(&self) -> io::Result<()> {
43        let result = match self {
44            Self::Net { caps, hostname, port } => caps.check_net(hostname, *port),
45            Self::Unix { caps, target } => caps.check_write(target),
46        };
47        result.map_err(|error| io::Error::new(io::ErrorKind::PermissionDenied, error))
48    }
49}
50fn authority(
51    mut address: Address,
52    caps: &Arc<Capabilities>,
53    connecting: bool,
54) -> io::Result<(Address, Authority)> {
55    let authority = match &mut address {
56        Address::Tcp { hostname, port } => Authority::Net {
57            caps: caps.clone(),
58            hostname: hostname.clone(),
59            port: *port,
60        },
61        Address::Unix { path } => {
62            if !cfg!(unix) {
63                return Err(io::Error::new(
64                    io::ErrorKind::Unsupported,
65                    "Unix sockets are not supported on this platform",
66                ));
67            }
68            let label = path
69                .as_ref()
70                .filter(|path| !path.is_empty())
71                .ok_or_else(|| {
72                    io::Error::new(io::ErrorKind::InvalidInput, "Missing Unix socket path")
73                })?;
74            let target = if connecting {
75                PathTarget::resolve(caps.cwd().join(label))?
76            } else {
77                PathTarget::for_entry(caps.cwd().join(label))?
78            };
79            *path = Some(
80                target
81                    .as_path()
82                    .to_str()
83                    .ok_or_else(|| {
84                        io::Error::new(
85                            io::ErrorKind::InvalidInput,
86                            "Unix socket paths must be UTF-8",
87                        )
88                    })?
89                    .to_owned(),
90            );
91            Authority::Unix { caps: caps.clone(), target }
92        }
93    };
94    authority.check()?;
95    Ok((address, authority))
96}
97fn hostname(value: &str) -> &str {
98    value
99        .strip_prefix('[')
100        .and_then(|value| value.strip_suffix(']'))
101        .unwrap_or(value)
102}
103
104struct Connection {
105    reader: AsyncMutex<ReadHalf<Socket>>,
106    writer: AsyncMutex<WriteHalf<Socket>>,
107    write_closed: AtomicBool,
108    canceled: CancellationToken,
109    authority: Authority,
110    local: Endpoint,
111    remote: Endpoint,
112}
113impl Connection {
114    fn new(
115        socket: Socket,
116        local: Endpoint,
117        remote: Endpoint,
118        authority: Authority,
119        canceled: CancellationToken,
120    ) -> Self {
121        let (reader, writer) = tokio::io::split(socket);
122        Self {
123            reader: AsyncMutex::new(reader),
124            writer: AsyncMutex::new(writer),
125            write_closed: AtomicBool::new(false),
126            canceled,
127            authority,
128            local,
129            remote,
130        }
131    }
132}
133struct Listener {
134    socket: avril_net::Listener,
135    authority: Authority,
136    canceled: CancellationToken,
137}
138struct Pending {
139    receiver: AsyncMutex<Option<oneshot::Receiver<io::Result<Connection>>>>,
140    canceled: CancellationToken,
141}
142enum Resource {
143    Connection(Arc<Connection>),
144    Listener(Arc<Listener>),
145    Pending(Arc<Pending>),
146    Datagram(Arc<udp::Datagram>),
147}
148impl Resource {
149    fn cancel(&self) {
150        match self {
151            Self::Connection(value) => value.canceled.cancel(),
152            Self::Listener(value) => value.canceled.cancel(),
153            Self::Pending(value) => value.canceled.cancel(),
154            Self::Datagram(value) => value.canceled.cancel(),
155        }
156    }
157}
158// Spawned I/O must not keep the registry alive: its drop cancels worker resources.
159struct Sockets {
160    runtime: tokio::runtime::Handle,
161    caps: Arc<Capabilities>,
162    resources: Mutex<HashMap<u64, Resource>>,
163    next: AtomicU64,
164    canceled: CancellationToken,
165}
166impl Drop for Sockets {
167    fn drop(&mut self) {
168        self.canceled.cancel();
169    }
170}
171impl Sockets {
172    fn id(&self) -> io::Result<u64> {
173        self.next
174            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |id| {
175                (id <= MAX_ID).then_some(id + 1)
176            })
177            .map_err(|_| io::Error::other("Socket IDs exhausted"))
178    }
179    fn connection(&self, id: u64) -> io::Result<Arc<Connection>> {
180        match self.resources.lock().unwrap().get(&id) {
181            Some(Resource::Connection(value)) => Ok(value.clone()),
182            _ => Err(closed()),
183        }
184    }
185    fn insert(&self, connection: Connection) -> io::Result<ConnectionInfo> {
186        let id = self.id()?;
187        let row = (id, connection.local.clone(), connection.remote.clone());
188        self.resources
189            .lock()
190            .unwrap()
191            .insert(id, Resource::Connection(Arc::new(connection)));
192        Ok(row)
193    }
194    fn listen(&self, target: Address) -> io::Result<ListenerInfo> {
195        let (target, mut authority) = authority(target, &self.caps, false)?;
196        let (socket, local) =
197            avril_net::Listener::bind(&self.runtime, &target).map_err(io::Error::other)?;
198        if let (Authority::Net { port, .. }, Endpoint::Tcp(selected)) = (&mut authority, &local) {
199            *port = selected.port();
200        }
201        authority.check()?;
202        let id = self.id()?;
203        self.resources.lock().unwrap().insert(
204            id,
205            Resource::Listener(Arc::new(Listener {
206                socket,
207                authority,
208                canceled: self.canceled.child_token(),
209            })),
210        );
211        Ok((id, local))
212    }
213    fn start(&self, target: Address) -> io::Result<u64> {
214        let (target, authority) = authority(target, &self.caps, true)?;
215        let id = self.id()?;
216        let canceled = self.canceled.child_token();
217        let (sender, receiver) = oneshot::channel();
218        self.resources.lock().unwrap().insert(
219            id,
220            Resource::Pending(Arc::new(Pending {
221                receiver: AsyncMutex::new(Some(receiver)),
222                canceled: canceled.clone(),
223            })),
224        );
225        self.runtime.spawn(async move {
226            let result = async {
227                let (socket,local,remote): (Socket,Endpoint,Endpoint) = match &target {
228                    Address::Tcp { hostname:host,port } => {
229                        let socket=tokio::net::TcpStream::connect((hostname(host),*port)).await?;
230                        let local=Endpoint::Tcp(socket.local_addr()?);
231                        let remote=Endpoint::Tcp(socket.peer_addr()?);
232                        (socket.into(),local,remote)
233                    }
234                    #[cfg(unix)]
235                    Address::Unix { path:Some(path) } => {
236                        let socket=tokio::net::UnixStream::connect(path).await?;
237                        let local=Endpoint::Unix {path:socket.local_addr()?.as_pathname().map(|path|path.to_string_lossy().into_owned())};
238                        (socket.into(),local,Endpoint::Unix {path:Some(path.clone())})
239                    }
240                    _ => return Err(io::Error::new(io::ErrorKind::Unsupported,"Unsupported socket transport")),
241                };
242                authority.check()?;
243                Ok(Connection::new(socket,local,remote,authority,canceled.clone()))
244            };
245            let value = tokio::select! { biased; _=canceled.cancelled()=>Err(closed()), value=result=>value };
246            let _ = sender.send(value);
247        });
248        Ok(id)
249    }
250    fn close(&self, id: u64) -> io::Result<()> {
251        let value = self
252            .resources
253            .lock()
254            .unwrap()
255            .remove(&id)
256            .ok_or_else(closed)?;
257        value.cancel();
258        Ok(())
259    }
260}
261
262async fn reply<T: Serialize>(
263    task: tokio::task::JoinHandle<io::Result<T>>,
264) -> Result<impl Serialize, Infallible> {
265    Ok(response(task.await.unwrap_or_else(|error| {
266        Err(io::Error::other(error.to_string()))
267    })))
268}
269
270pub fn init(runtime: tokio::runtime::Handle, caps: Arc<Capabilities>) -> Extension {
271    let sockets = Arc::new(Sockets {
272        runtime: runtime.clone(),
273        caps,
274        resources: Mutex::new(HashMap::new()),
275        next: AtomicU64::new(0),
276        canceled: CancellationToken::new(),
277    });
278    let mut ops = vec![
279        Op::new("op_net_listen", {
280            let sockets = sockets.clone();
281            move |target: Address| Ok::<_, Infallible>(response(sockets.listen(target)))
282        }),
283        Op::new("op_net_connect_start", {
284            let sockets = sockets.clone();
285            move |target: Address| Ok::<_, Infallible>(response(sockets.start(target)))
286        }),
287        Op::asynchronous("op_net_connected", {
288            let sockets = sockets.clone();
289            move |id: u64| {
290                let pending = match sockets.resources.lock().unwrap().get(&id) {
291                    Some(Resource::Pending(value)) => Some(value.clone()),
292                    _ => None,
293                };
294                let weak = Arc::downgrade(&sockets);
295                reply(runtime.spawn(async move {
296                    let pending = pending.ok_or_else(closed)?;
297                    let receiver = pending.receiver.lock().await.take().ok_or_else(closed)?;
298                    let connection = tokio::select! {
299                        biased;
300                        _ = pending.canceled.cancelled() => return Err(closed()),
301                        value = receiver => value.map_err(|_|closed())??,
302                    };
303                    connection.authority.check()?;
304                    let sockets = weak.upgrade().ok_or_else(closed)?;
305                    let row = (id, connection.local.clone(), connection.remote.clone());
306                    let mut resources = sockets.resources.lock().unwrap();
307                    if pending.canceled.is_cancelled() || !resources.contains_key(&id) {
308                        return Err(closed());
309                    }
310                    resources.insert(id, Resource::Connection(Arc::new(connection)));
311                    Ok(row)
312                }))
313            }
314        }),
315        Op::asynchronous("op_net_accept", {
316            let sockets = sockets.clone();
317            let runtime = sockets.runtime.clone();
318            move |id: u64| {
319                let listener = match sockets.resources.lock().unwrap().get(&id) {
320                    Some(Resource::Listener(value)) => Some(value.clone()),
321                    _ => None,
322                };
323                let weak = Arc::downgrade(&sockets);
324                let canceled = sockets.canceled.child_token();
325                reply(runtime.spawn(async move {
326                    let listener = listener.ok_or_else(closed)?;
327                    listener.authority.check()?;
328                    let accepted = tokio::select! {
329                        biased;
330                        _ = listener.canceled.cancelled() => return Err(closed()),
331                        value = listener.socket.accept() => value?,
332                    };
333                    listener.authority.check()?;
334                    if listener.canceled.is_cancelled() {
335                        return Err(closed());
336                    }
337                    let connection = Connection::new(
338                        accepted.socket,
339                        accepted.local,
340                        accepted.remote,
341                        listener.authority.clone(),
342                        canceled,
343                    );
344                    weak.upgrade().ok_or_else(closed)?.insert(connection)
345                }))
346            }
347        }),
348        Op::asynchronous("op_net_read", {
349            let sockets = sockets.clone();
350            let runtime = sockets.runtime.clone();
351            move |(id, count): (u64, usize)| {
352                let connection = sockets.connection(id);
353                reply(runtime.spawn(async move {
354                    let connection = connection?;
355                    connection.authority.check()?;
356                    let work = async {
357                        let mut reader = connection.reader.lock().await;
358                        connection.authority.check()?;
359                        let mut bytes = vec![0; count.min(CHUNK)];
360                        if bytes.is_empty() {
361                            return Ok(Some(ToJsBuffer::from(bytes)));
362                        }
363                        let count = reader.read(&mut bytes).await?;
364                        connection.authority.check()?;
365                        if count == 0 {
366                            return Ok(None);
367                        }
368                        bytes.truncate(count);
369                        Ok(Some(ToJsBuffer::from(bytes)))
370                    };
371                    tokio::select! {
372                        biased;
373                        _ = connection.canceled.cancelled() => Err(closed()),
374                        value = work => value,
375                    }
376                }))
377            }
378        }),
379        Op::asynchronous("op_net_write", {
380            let sockets = sockets.clone();
381            let runtime = sockets.runtime.clone();
382            move |(id, bytes): (u64, JsBuffer)| {
383                let connection = sockets.connection(id);
384                let bytes = bytes[..bytes.len().min(CHUNK)].to_vec();
385                reply(runtime.spawn(async move {
386                    let connection = connection?;
387                    let work = async {
388                        let mut writer = connection.writer.lock().await;
389                        connection.authority.check()?;
390                        if connection.write_closed.load(Ordering::Acquire) {
391                            return Err(closed());
392                        }
393                        writer.write(&bytes).await
394                    };
395                    tokio::select! {
396                        biased;
397                        _ = connection.canceled.cancelled() => Err(closed()),
398                        value = work => value,
399                    }
400                }))
401            }
402        }),
403        Op::asynchronous("op_net_writev", {
404            let sockets = sockets.clone();
405            let runtime = sockets.runtime.clone();
406            move |(id, buffers): (u64, Vec<JsBuffer>)| {
407                let connection = sockets.connection(id);
408                let segments = avril_host::io::bound_vectored(buffers);
409                reply(runtime.spawn(async move {
410                    let connection = connection?;
411                    let work = async {
412                        let mut writer = connection.writer.lock().await;
413                        connection.authority.check()?;
414                        if connection.write_closed.load(Ordering::Acquire) {
415                            return Err(closed());
416                        }
417                        if segments.is_empty() {
418                            return Ok(0);
419                        }
420                        let slices = segments
421                            .iter()
422                            .map(|segment| io::IoSlice::new(segment))
423                            .collect::<Vec<_>>();
424                        writer.write_vectored(&slices).await
425                    };
426                    tokio::select! {
427                        biased;
428                        _ = connection.canceled.cancelled() => Err(closed()),
429                        value = work => value,
430                    }
431                }))
432            }
433        }),
434        Op::asynchronous("op_net_shutdown", {
435            let sockets = sockets.clone();
436            let runtime = sockets.runtime.clone();
437            move |id: u64| {
438                let connection = sockets.connection(id);
439                reply(runtime.spawn(async move {
440                    let connection = connection?;
441                    let work = async {
442                        let mut writer = connection.writer.lock().await;
443                        connection.authority.check()?;
444                        if !connection.write_closed.swap(true, Ordering::AcqRel) {
445                            writer.shutdown().await?;
446                        }
447                        Ok(())
448                    };
449                    tokio::select! {
450                        biased;
451                        _ = connection.canceled.cancelled() => Err(closed()),
452                        value = work => value,
453                    }
454                }))
455            }
456        }),
457        Op::new("op_net_close", {
458            let sockets = sockets.clone();
459            move |id: u64| Ok::<_, Infallible>(response(sockets.close(id)))
460        }),
461    ];
462    ops.extend(udp::ops(sockets));
463    Extension {
464        name: "ext:net",
465        dependencies: &["ext:file", "ext:web", "ext:webstreams"],
466        ops,
467        modules: &[(
468            "ext:net/main.js",
469            include_str!(concat!(env!("OUT_DIR"), "/main.js")),
470        )],
471        entry_point: None,
472    }
473}