char/avril

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

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

Charlotte Somport stdio extension to typescriptf20b64f

main
branchesmainall refs →
14.0 KiB406 linesraw
1use std::cell::RefCell;
2use std::convert::Infallible;
3use std::future::Future;
4use std::io::{self, Read, Write};
5use std::rc::Rc;
6use std::sync::Arc;
7use std::sync::atomic::AtomicBool;
8use std::sync::mpsc::{self, Sender, SyncSender};
9
10use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
11use avril_core::{Extension, Op};
12use futures::channel::oneshot;
13use futures::future::{Either, select};
14
15enum WriteReply {
16    Async(oneshot::Sender<io::Result<()>>),
17    Blocking(SyncSender<io::Result<()>>),
18}
19
20struct WriteRequest {
21    bytes: Option<Vec<u8>>,
22    reply: WriteReply,
23}
24
25struct Output(Sender<WriteRequest>);
26
27async fn response<T>(reply: oneshot::Receiver<io::Result<T>>) -> io::Result<T> {
28    reply
29        .await
30        .map_err(|_| io::Error::other("Stdio worker stopped"))?
31}
32
33impl Output {
34    fn new(name: &str, mut writer: impl Write + Send + 'static) -> io::Result<Self> {
35        let (sender, requests) = mpsc::channel::<WriteRequest>();
36        std::thread::Builder::new()
37            .name(name.into())
38            .spawn(move || {
39                let mut failure: Option<(io::ErrorKind, String)> = None;
40                for request in requests.iter() {
41                    let result = if let Some((kind, message)) = &failure {
42                        Err(io::Error::new(*kind, message.clone()))
43                    } else {
44                        (|| {
45                            if let Some(bytes) = request.bytes {
46                                writer.write_all(&bytes)?;
47                            }
48                            writer.flush()
49                        })()
50                    };
51                    if let Err(error) = &result {
52                        failure = Some((error.kind(), error.to_string()));
53                    }
54                    match request.reply {
55                        WriteReply::Async(reply) => {
56                            let _ = reply.send(result);
57                        }
58                        WriteReply::Blocking(reply) => {
59                            let _ = reply.send(result);
60                        }
61                    }
62                }
63            })?;
64        Ok(Self(sender))
65    }
66
67    fn submit(&self, bytes: Option<Vec<u8>>) -> io::Result<oneshot::Receiver<io::Result<()>>> {
68        let (reply, receiver) = oneshot::channel();
69        // Stream writes are serialized by the sink; raw ops must not block the isolate on capacity.
70        self.0
71            .send(WriteRequest {
72                bytes,
73                reply: WriteReply::Async(reply),
74            })
75            .map_err(|_| io::Error::other("Stdio worker stopped"))?;
76        Ok(receiver)
77    }
78
79    fn blocking_write(&self, bytes: Option<Vec<u8>>) -> io::Result<()> {
80        let (reply, receiver) = mpsc::sync_channel(1);
81        self.0
82            .send(WriteRequest {
83                bytes,
84                reply: WriteReply::Blocking(reply),
85            })
86            .map_err(|_| io::Error::other("Stdio worker stopped"))?;
87        receiver
88            .recv()
89            .map_err(|_| io::Error::other("Stdio worker stopped"))?
90    }
91
92    fn flush(&self) -> io::Result<()> {
93        self.blocking_write(None)
94    }
95}
96
97struct ReadRequest {
98    limit: usize,
99    reply: oneshot::Sender<io::Result<Vec<u8>>>,
100}
101
102enum InputState {
103    Idle,
104    Reading(oneshot::Sender<()>),
105    Canceled,
106}
107
108struct PendingRead(Rc<Stdio>);
109
110impl Drop for PendingRead {
111    fn drop(&mut self) {
112        if matches!(*self.0.input.borrow(), InputState::Reading(_)) {
113            self.0.cancel_read();
114        }
115    }
116}
117
118pub struct Stdio {
119    stdin: SyncSender<ReadRequest>,
120    stdin_available: bool,
121    stdout: Output,
122    stderr: Output,
123    input: RefCell<InputState>,
124    interrupted: Arc<AtomicBool>,
125}
126
127impl Stdio {
128    pub fn new() -> io::Result<Rc<Self>> {
129        Self::new_with_stdin(true, Arc::default())
130    }
131
132    pub fn new_for_repl() -> io::Result<Rc<Self>> {
133        Self::new_with_stdin(false, Arc::default())
134    }
135
136    pub fn new_for_worker(interrupted: Arc<AtomicBool>) -> io::Result<Rc<Self>> {
137        Self::new_with_stdin(false, interrupted)
138    }
139
140    pub fn interrupt_flag(&self) -> Arc<AtomicBool> {
141        self.interrupted.clone()
142    }
143
144    fn new_with_stdin(stdin_available: bool, interrupted: Arc<AtomicBool>) -> io::Result<Rc<Self>> {
145        let (stdin, requests) = mpsc::sync_channel::<ReadRequest>(1);
146        if stdin_available {
147            // An outstanding OS read cannot be interrupted portably; never join this worker.
148            std::thread::Builder::new()
149                .name("avril:stdin".into())
150                .spawn(move || {
151                    for request in requests {
152                        let result = (|| {
153                            let mut bytes = vec![0; request.limit];
154                            let mut stdin = io::stdin().lock();
155                            let count = loop {
156                                match stdin.read(&mut bytes) {
157                                    Err(error) if error.kind() == io::ErrorKind::Interrupted => {
158                                        continue;
159                                    }
160                                    result => break result?,
161                                }
162                            };
163                            bytes.truncate(count);
164                            Ok(bytes)
165                        })();
166                        let _ = request.reply.send(result);
167                    }
168                })?;
169        } else {
170            drop(requests);
171        }
172        Ok(Rc::new(Self {
173            stdin,
174            stdin_available,
175            stdout: Output::new("avril:stdout", io::stdout())?,
176            stderr: Output::new("avril:stderr", io::stderr())?,
177            input: RefCell::new(InputState::Idle),
178            interrupted,
179        }))
180    }
181
182    fn read(
183        self: &Rc<Self>,
184        limit: usize,
185    ) -> impl Future<Output = io::Result<Option<ToJsBuffer>>> + use<> {
186        let started = (|| {
187            if !self.stdin_available {
188                return Err(io::Error::new(
189                    io::ErrorKind::Unsupported,
190                    "Guest stdin in the REPL is not yet implemented; the REPL owns standard input",
191                ));
192            }
193            if limit == 0 {
194                return Ok(None);
195            }
196            let (reply, receiver) = oneshot::channel();
197            let (cancel, canceled) = oneshot::channel();
198            {
199                let mut input = self.input.borrow_mut();
200                match &*input {
201                    InputState::Reading(_) => {
202                        return Err(io::Error::new(
203                            io::ErrorKind::WouldBlock,
204                            "A stdin read is already pending",
205                        ));
206                    }
207                    InputState::Canceled => return Ok(None),
208                    InputState::Idle => *input = InputState::Reading(cancel),
209                }
210            }
211            let pending = PendingRead(self.clone());
212            // Only one read can be submitted; cancellation permanently closes the input.
213            self.stdin
214                .send(ReadRequest {
215                    limit: limit.min(64 * 1024),
216                    reply,
217                })
218                .map_err(|_| io::Error::other("Stdin worker stopped"))?;
219            Ok(Some((pending, receiver, canceled)))
220        })();
221        async move {
222            let Some((pending, receiver, canceled)) = started? else {
223                return Ok(if limit == 0 {
224                    Some(ToJsBuffer::from(Vec::new()))
225                } else {
226                    None
227                });
228            };
229            let result = match select(receiver, canceled).await {
230                Either::Left((result, _)) => result
231                    .map_err(|_| io::Error::other("Stdin worker stopped"))
232                    .and_then(|result| result)
233                    .map(|bytes| {
234                        if bytes.is_empty() {
235                            None
236                        } else {
237                            Some(ToJsBuffer::from(bytes))
238                        }
239                    }),
240                Either::Right(_) => Ok(None),
241            };
242            {
243                let mut input = pending.0.input.borrow_mut();
244                if matches!(*input, InputState::Reading(_)) {
245                    *input = InputState::Idle;
246                }
247            }
248            result
249        }
250    }
251
252    fn cancel_read(&self) {
253        if let InputState::Reading(sender) = self.input.replace(InputState::Canceled) {
254            let _ = sender.send(());
255        }
256    }
257
258    fn output(&self, stderr: bool) -> &Output {
259        if stderr { &self.stderr } else { &self.stdout }
260    }
261
262    pub fn flush(&self) -> io::Result<()> {
263        let stdout = self.stdout.flush();
264        let stderr = self.stderr.flush();
265        stdout.and(stderr)
266    }
267
268    pub fn flush_interruptible(&self, interrupted: impl Fn() -> bool) -> io::Result<()> {
269        let stdout = self.stdout.submit(None);
270        let stderr = self.stderr.submit(None);
271        avril_host::wait(
272            async {
273                let stdout = response(stdout?).await;
274                let stderr = response(stderr?).await;
275                stdout.and(stderr)
276            },
277            interrupted,
278        )
279        .unwrap_or(Ok(()))
280    }
281
282    pub fn print(&self, stderr: bool, text: String) -> io::Result<()> {
283        self.output(stderr).blocking_write(Some(text.into_bytes()))
284    }
285
286    pub fn extension(self: &Rc<Self>) -> Extension {
287        let read = self.clone();
288        let read_sync = self.clone();
289        let write_sync = self.clone();
290        let cancel = self.clone();
291        let write = self.clone();
292        let flush = self.clone();
293        Extension {
294            name: "ext:stdio",
295            dependencies: &["ext:webstreams", "ext:file"],
296            ops: vec![
297                Op::asynchronous("op_stdio_read", move |limit: usize| read.read(limit)),
298                Op::with_scope("op_stdio_read_sync", move |scope, limit: usize| {
299                    let pending = read_sync.read(limit);
300                    Ok::<_, Infallible>(
301                        avril_host::wait(pending, || {
302                            avril_host::interrupted(scope, &read_sync.interrupted)
303                        })
304                        .map(avril_host::io::response),
305                    )
306                }),
307                Op::with_scope(
308                    "op_stdio_write_sync",
309                    move |scope, (stderr, bytes): (bool, JsBuffer)| {
310                        let count = bytes.len();
311                        let pending = write_sync.output(stderr).submit(Some(bytes.to_vec()));
312                        let result = avril_host::wait(
313                            async { response(pending?).await.map(|()| count) },
314                            || avril_host::interrupted(scope, &write_sync.interrupted),
315                        );
316                        Ok::<_, Infallible>(result.map(avril_host::io::response))
317                    },
318                ),
319                Op::new("op_stdio_cancel_read", move |_: ()| {
320                    cancel.cancel_read();
321                    Ok::<_, io::Error>(())
322                }),
323                Op::asynchronous(
324                    "op_stdio_write",
325                    move |(stderr, bytes): (bool, JsBuffer)| {
326                        let pending = write.output(stderr).submit(Some(bytes.to_vec()));
327                        async move { response(pending?).await }
328                    },
329                ),
330                Op::asynchronous("op_stdio_flush", move |stderr: bool| {
331                    let pending = flush.output(stderr).submit(None);
332                    async move { response(pending?).await }
333                }),
334            ],
335            modules: &[
336                (
337                    "ext:stdio/main.js",
338                    include_str!(concat!(env!("OUT_DIR"), "/main.js")),
339                ),
340                (
341                    "ext:stdio/sync.js",
342                    include_str!(concat!(env!("OUT_DIR"), "/sync.js")),
343                ),
344            ],
345            entry_point: Some("ext:stdio/main.js"),
346        }
347    }
348}
349
350#[cfg(test)]
351mod tests {
352    use super::*;
353
354    fn input() -> (Rc<Stdio>, mpsc::Receiver<ReadRequest>) {
355        let (stdin, requests) = mpsc::sync_channel(1);
356        let stdio = Rc::new(Stdio {
357            stdin,
358            stdin_available: true,
359            stdout: Output::new("test:stdout", io::sink()).unwrap(),
360            stderr: Output::new("test:stderr", io::sink()).unwrap(),
361            input: RefCell::new(InputState::Idle),
362            interrupted: Arc::default(),
363        });
364        (stdio, requests)
365    }
366
367    #[test]
368    fn reads_reserve_input_before_polling_and_empty_reads_do_not_consume_it() {
369        let (stdio, requests) = input();
370        let first = stdio.read(2);
371        let request = requests.try_recv().unwrap();
372        assert_eq!(request.limit, 2);
373        let second = avril_host::wait(stdio.read(1), || false).unwrap();
374        assert_eq!(second.err().unwrap().kind(), io::ErrorKind::WouldBlock);
375        let empty = avril_host::wait(stdio.read(0), || false).unwrap().unwrap();
376        assert!(empty.is_some());
377        assert!(requests.try_recv().is_err());
378        request.reply.send(Ok(vec![65, 66])).unwrap();
379        assert!(
380            avril_host::wait(first, || false)
381                .unwrap()
382                .unwrap()
383                .is_some()
384        );
385        let next = stdio.read(1);
386        let request = requests.try_recv().unwrap();
387        request.reply.send(Ok(Vec::new())).unwrap();
388        assert!(avril_host::wait(next, || false).unwrap().unwrap().is_none());
389    }
390
391    #[test]
392    fn an_interrupted_read_does_not_leave_a_second_reader_on_the_native_queue() {
393        let (stdio, requests) = input();
394        let first = stdio.read(1);
395        let request = requests.try_recv().unwrap();
396        assert!(avril_host::wait(first, || true).is_none());
397        assert!(
398            avril_host::wait(stdio.read(1), || false)
399                .unwrap()
400                .unwrap()
401                .is_none()
402        );
403        assert!(requests.try_recv().is_err());
404        assert!(request.reply.send(Ok(vec![65])).is_err());
405    }
406}