diff options
Diffstat (limited to 'bundle/src-tauri/src/runner.rs')
| -rw-r--r-- | bundle/src-tauri/src/runner.rs | 267 |
1 files changed, 267 insertions, 0 deletions
diff --git a/bundle/src-tauri/src/runner.rs b/bundle/src-tauri/src/runner.rs new file mode 100644 index 0000000..e2b492b --- /dev/null +++ b/bundle/src-tauri/src/runner.rs @@ -0,0 +1,267 @@ +use serde::Serialize; +use std::{ + path::PathBuf, + process::Stdio, + sync::{Arc, Mutex}, +}; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWriteExt}, + process::Command, + sync::watch, +}; + +#[derive(Clone, Default)] +pub struct Runner(Arc<Mutex<Option<Control>>>); + +struct Control { + stop: watch::Sender<bool>, + done: watch::Receiver<bool>, +} + +#[derive(Serialize)] +#[serde(tag = "event", rename_all = "camelCase")] +pub enum OutputEvent { + Data { stream: String, bytes: Vec<u8> }, + Finished { code: Option<i32>, stopped: bool }, +} + +pub struct RunResult { + pub code: Option<i32>, + pub stopped: bool, +} + +impl Runner { + pub async fn run( + &self, + executable: PathBuf, + source: PathBuf, + input: String, + output: impl Fn(&str, Vec<u8>) + Send + Sync, + ) -> Result<RunResult, String> { + let (stop, stop_rx) = watch::channel(false); + let (done_tx, done) = watch::channel(false); + { + let mut active = self.0.lock().unwrap(); + if active.is_some() { + return Err("A program is already running.".into()); + } + *active = Some(Control { stop, done }); + } + let result = run_process(executable, source, input, stop_rx, output).await; + self.0.lock().unwrap().take(); + let _ = done_tx.send(true); + result + } + + pub async fn stop(&self) { + let control = self + .0 + .lock() + .unwrap() + .as_ref() + .map(|run| (run.stop.clone(), run.done.clone())); + if let Some((stop, mut done)) = control { + let _ = stop.send(true); + while !*done.borrow_and_update() { + if done.changed().await.is_err() { + break; + } + } + } + } +} + +async fn read_output( + mut pipe: impl AsyncRead + Unpin, + stream: &str, + output: &(impl Fn(&str, Vec<u8>) + Send + Sync), +) -> Result<(), String> { + let mut buffer = [0; 8192]; + loop { + let count = pipe + .read(&mut buffer) + .await + .map_err(|error| error.to_string())?; + if count == 0 { + return Ok(()); + } + output(stream, buffer[..count].to_vec()); + } +} + +async fn run_process( + executable: PathBuf, + source: PathBuf, + input: String, + mut stop: watch::Receiver<bool>, + output: impl Fn(&str, Vec<u8>) + Send + Sync, +) -> Result<RunResult, String> { + let mut command = Command::new(executable); + command + .arg(&source) + .arg("-r") + .current_dir(source.parent().ok_or("Invalid source path.")?) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + #[cfg(windows)] + command.creation_flags(0x08000000); // CREATE_NO_WINDOW + let mut child = command + .spawn() + .map_err(|error| format!("Cannot start compiler: {error}"))?; + let mut stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + let stderr = child.stderr.take().unwrap(); + let write_input = async move { + let result = stdin.write_all(input.as_bytes()).await; + drop(stdin); // EOF must reach programs even when Input is empty. + match result { + Err(error) if error.kind() != std::io::ErrorKind::BrokenPipe => Err(error.to_string()), + _ => Ok(()), // Programs may finish without consuming all input. + } + }; + let wait = async { + tokio::select! { + status = child.wait() => status.map(|status| RunResult { code: status.code(), stopped: false }), + _ = stop.changed() => { + child.kill().await.map_err(|error| error.to_string())?; + child.wait().await.map(|status| RunResult { code: status.code(), stopped: true }) + } + }.map_err(|error| error.to_string()) + }; + let (result, input_result, stdout_result, stderr_result) = tokio::join!( + wait, + write_input, + read_output(stdout, "stdout", &output), + read_output(stderr, "stderr", &output) + ); + let result = result?; + input_result?; + stdout_result?; + stderr_result?; + Ok(result) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::time::{timeout, Duration}; + + struct Source(PathBuf); + impl Source { + fn new(text: &str) -> Self { + static NEXT: AtomicUsize = AtomicUsize::new(0); + let directory = std::env::temp_dir().join(format!( + "cpsc bundle test {} {}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir_all(&directory).unwrap(); + let source = directory.join("program.cps"); + std::fs::write(&source, text).unwrap(); + Self(source) + } + } + impl Drop for Source { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(self.0.parent().unwrap()); + } + } + fn compiler() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(if cfg!(windows) { + "../../build/cpsc.exe" + } else { + "../../build/cpsc" + }) + } + + #[tokio::test] + async fn prefilled_input_empty_eof_and_diagnostics() { + let runner = Runner::default(); + for (program, input, expected, code) in [ + ( + "DECLARE N : INTEGER\nINPUT N\nOUTPUT N\nOUTPUT \"\\U0001F600\"\n", + "42\n", + "42\nš\n", + 0, + ), + ( + "DECLARE S : STRING\nINPUT S\nOUTPUT \"done\"\n", + "", + "done\n", + 0, + ), + ("OUTPUT Missing\n", "", "compilation terminated.", 1), + ( + "OUTPUT 1 DIV 0\n", + "", + "Unhandled exception. Program terminated.", + 1, + ), + ] { + let source = Source::new(program); + let bytes = Mutex::new(Vec::new()); + let result = timeout( + Duration::from_secs(30), + runner.run(compiler(), source.0.clone(), input.into(), |_, chunk| { + bytes.lock().unwrap().extend(chunk) + }), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(result.code, Some(code)); + let text = String::from_utf8(bytes.into_inner().unwrap()) + .unwrap() + .replace("\r\n", "\n"); + if code == 0 { + assert_eq!(text, expected); + } else { + assert!(text.contains(expected), "{text}"); + } + } + } + + #[tokio::test] + async fn output_arrives_before_exit_and_stop_allows_restart() { + let runner = Runner::default(); + let source = Source::new("OUTPUT \"ready\"\nWHILE TRUE\nENDWHILE\n"); + let (send, mut receive) = tokio::sync::mpsc::unbounded_channel(); + let running = runner.clone(); + let path = source.0.clone(); + let task = tokio::spawn(async move { + running + .run(compiler(), path, String::new(), |_, bytes| { + let _ = send.send(bytes); + }) + .await + }); + let mut output = Vec::new(); + timeout(Duration::from_secs(30), async { + while !output.ends_with(b"ready\n") && !output.ends_with(b"ready\r\n") { + output.extend(receive.recv().await.expect("output closed before ready")); + } + }) + .await + .expect("stdout was buffered until exit"); + assert!(runner + .run(compiler(), source.0.clone(), String::new(), |_, _| {}) + .await + .is_err()); + timeout(Duration::from_secs(5), runner.stop()) + .await + .expect("stop hung"); + assert!(task.await.unwrap().unwrap().stopped); + let source = Source::new("OUTPUT \"again\"\n"); + assert_eq!( + runner + .run(compiler(), source.0.clone(), String::new(), |_, _| {}) + .await + .unwrap() + .code, + Some(0) + ); + } +} |