Implement a Non-Blocking Child Process Interface

.lune/hello_lune.luau

@@ -129,7 +129,7 @@ end
 ]]
 
 print("Sending 4 pings to google 🌏")
-local result = process.spawn("ping", {
+local result = process.exec("ping", {
 	"google.com",
 	"-c 4",
 })

crates/lune-std-process/Cargo.toml

@@ -20,6 +20,8 @@ directories = "5.0"
 pin-project = "1.0"
 os_str_bytes = { version = "7.0", features = ["conversions"] }
 
+bstr = "1.9"
+
 tokio = { version = "1", default-features = false, features = [
     "io-std",
     "io-util",
@@ -29,3 +31,4 @@ tokio = { version = "1", default-features = false, features = [
 ] }
 
 lune-utils = { version = "0.1.0", path = "../lune-utils" }
+bytes = "1.6.0"

crates/lune-std-process/src/lib.rs

@@ -1,27 +1,32 @@
 #![allow(clippy::cargo_common_metadata)]
 
 use std::{
+    cell::RefCell,
     env::{
         self,
         consts::{ARCH, OS},
     },
     path::MAIN_SEPARATOR,
     process::Stdio,
+    rc::Rc,
 };
 
 use mlua::prelude::*;
 
 use lune_utils::TableBuilder;
 use mlua_luau_scheduler::{Functions, LuaSpawnExt};
+use options::ProcessSpawnOptionsStdio;
 use os_str_bytes::RawOsString;
-use tokio::io::AsyncWriteExt;
+use stream::{ChildProcessReader, ChildProcessWriter};
+use tokio::{io::AsyncWriteExt, process::Child};
 
 mod options;
+mod stream;
 mod tee_writer;
 mod wait_for_child;
 
 use self::options::ProcessSpawnOptions;
-use self::wait_for_child::{wait_for_child, WaitForChildResult};
+use self::wait_for_child::wait_for_child;
 
 use lune_utils::path::get_current_dir;
 
@@ -73,6 +78,7 @@ pub fn module(lua: &Lua) -> LuaResult<LuaTable> {
         .with_value("cwd", cwd_str)?
         .with_value("env", env_tab)?
         .with_value("exit", process_exit)?
+        .with_async_function("exec", process_exec)?
         .with_async_function("spawn", process_spawn)?
         .build_readonly()
 }
@@ -141,11 +147,16 @@ fn process_env_iter<'lua>(
     })
 }
 
-async fn process_spawn(
+async fn process_exec(
     lua: &Lua,
     (program, args, options): (String, Option<Vec<String>>, ProcessSpawnOptions),
 ) -> LuaResult<LuaTable> {
-    let res = lua.spawn(spawn_command(program, args, options)).await?;
+    let res = lua
+        .spawn(async move {
+            let cmd = spawn_command(program, args, options.clone()).await?;
+            wait_for_child(cmd, options.stdio.stdout, options.stdio.stderr).await
+        })
+        .await?;
 
     /*
         NOTE: If an exit code was not given by the child process,
@@ -168,22 +179,109 @@ async fn process_spawn(
         .build_readonly()
 }
 
+#[allow(clippy::await_holding_refcell_ref)]
+async fn process_spawn(
+    lua: &Lua,
+    (program, args, options): (String, Option<Vec<String>>, ProcessSpawnOptions),
+) -> LuaResult<LuaTable> {
+    // Spawn does not accept stdio options, so we remove them from the options
+    // and use the defaults instead
+    let mut spawn_options = options.clone();
+    spawn_options.stdio = ProcessSpawnOptionsStdio::default();
+
+    let (stdin_tx, stdin_rx) = tokio::sync::oneshot::channel();
+    let (stdout_tx, stdout_rx) = tokio::sync::oneshot::channel();
+    let (stderr_tx, stderr_rx) = tokio::sync::oneshot::channel();
+    let (code_tx, code_rx) = tokio::sync::broadcast::channel(4);
+    let code_rx_rc = Rc::new(RefCell::new(code_rx));
+
+    tokio::spawn(async move {
+        let mut child = spawn_command(program, args, spawn_options)
+            .await
+            .expect("Could not spawn child process");
+        stdin_tx
+            .send(child.stdin.take())
+            .expect("Stdin receiver was unexpectedly dropped");
+        stdout_tx
+            .send(child.stdout.take())
+            .expect("Stdout receiver was unexpectedly dropped");
+        stderr_tx
+            .send(child.stderr.take())
+            .expect("Stderr receiver was unexpectedly dropped");
+
+        let res = child
+            .wait_with_output()
+            .await
+            .expect("Failed to get status and output of spawned child process");
+
+        let code = res
+            .status
+            .code()
+            .unwrap_or(i32::from(!res.stderr.is_empty()));
+
+        code_tx
+            .send(code)
+            .expect("ExitCode receiver was unexpectedly dropped");
+    });
+
+    TableBuilder::new(lua)?
+        .with_value(
+            "stdout",
+            ChildProcessReader(
+                stdout_rx
+                    .await
+                    .expect("Stdout sender unexpectedly dropped")
+                    .unwrap(),
+            ),
+        )?
+        .with_value(
+            "stderr",
+            ChildProcessReader(
+                stderr_rx
+                    .await
+                    .expect("Stderr sender unexpectedly dropped")
+                    .unwrap(),
+            ),
+        )?
+        .with_value(
+            "stdin",
+            ChildProcessWriter(
+                stdin_rx
+                    .await
+                    .expect("Stdin sender unexpectedly dropped")
+                    .unwrap(),
+            ),
+        )?
+        .with_async_function("status", move |lua, ()| {
+            let code_rx_rc_clone = Rc::clone(&code_rx_rc);
+            async move {
+                let code = code_rx_rc_clone
+                    .borrow_mut()
+                    .recv()
+                    .await
+                    .expect("Code sender unexpectedly dropped");
+
+                TableBuilder::new(lua)?
+                    .with_value("code", code)?
+                    .with_value("ok", code == 0)?
+                    .build_readonly()
+            }
+        })?
+        .build_readonly()
+}
+
 async fn spawn_command(
     program: String,
     args: Option<Vec<String>>,
     mut options: ProcessSpawnOptions,
-) -> LuaResult<WaitForChildResult> {
+) -> LuaResult<Child> {
     let stdout = options.stdio.stdout;
     let stderr = options.stdio.stderr;
     let stdin = options.stdio.stdin.take();
 
     let mut child = options
         .into_command(program, args)
-        .stdin(if stdin.is_some() {
-            Stdio::piped()
-        } else {
-            Stdio::null()
-        })
+        .stdin(Stdio::piped())
         .stdout(stdout.as_stdio())
         .stderr(stderr.as_stdio())
         .spawn()?;
@@ -193,5 +291,5 @@ async fn spawn_command(
         child_stdin.write_all(&stdin).await.into_lua_err()?;
     }
 
-    wait_for_child(child, stdout, stderr).await
+    Ok(child)
 }

crates/lune-std-process/src/stream.rs

@@ -0,0 +1,52 @@
+use bstr::BString;
+use bytes::BytesMut;
+use mlua::prelude::*;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+
+const CHUNK_SIZE: usize = 8;
+
+#[derive(Debug, Clone)]
+pub struct ChildProcessReader<R: AsyncRead>(pub R);
+#[derive(Debug, Clone)]
+pub struct ChildProcessWriter<W: AsyncWrite>(pub W);
+
+impl<R: AsyncRead + Unpin> ChildProcessReader<R> {
+    pub async fn read(&mut self, chunk_size: Option<usize>) -> LuaResult<Vec<u8>> {
+        let mut buf = BytesMut::with_capacity(chunk_size.unwrap_or(CHUNK_SIZE));
+        self.0.read_buf(&mut buf).await?;
+
+        Ok(buf.to_vec())
+    }
+
+    pub async fn read_to_end(&mut self) -> LuaResult<Vec<u8>> {
+        let mut buf = vec![];
+        self.0.read_to_end(&mut buf).await?;
+
+        Ok(buf)
+    }
+}
+
+impl<R: AsyncRead + Unpin + 'static> LuaUserData for ChildProcessReader<R> {
+    fn add_methods<'lua, M: LuaUserDataMethods<'lua, Self>>(methods: &mut M) {
+        methods.add_async_method_mut("read", |lua, this, chunk_size: Option<usize>| async move {
+            Ok(lua.create_buffer(this.read(chunk_size).await?))
+        });
+
+        methods.add_async_method_mut("readToEnd", |lua, this, ()| async {
+            Ok(lua.create_buffer(this.read_to_end().await?))
+        });
+    }
+}
+
+impl<W: AsyncWrite + Unpin> ChildProcessWriter<W> {
+    pub async fn write(&mut self, data: BString) -> LuaResult<()> {
+        self.0.write_all(data.as_ref()).await?;
+        Ok(())
+    }
+}
+
+impl<W: AsyncWrite + Unpin + 'static> LuaUserData for ChildProcessWriter<W> {
+    fn add_methods<'lua, M: LuaUserDataMethods<'lua, Self>>(methods: &mut M) {
+        methods.add_async_method_mut("write", |_, this, data| async { this.write(data).await });
+    }
+}

crates/lune/src/tests.rs

@@ -138,13 +138,16 @@ create_tests! {
     process_cwd: "process/cwd",
     process_env: "process/env",
     process_exit: "process/exit",
-    process_spawn_async: "process/spawn/async",
-    process_spawn_basic: "process/spawn/basic",
-    process_spawn_cwd: "process/spawn/cwd",
-    process_spawn_no_panic: "process/spawn/no_panic",
-    process_spawn_shell: "process/spawn/shell",
-    process_spawn_stdin: "process/spawn/stdin",
-    process_spawn_stdio: "process/spawn/stdio",
+    process_exec_async: "process/exec/async",
+    process_exec_basic: "process/exec/basic",
+    process_exec_cwd: "process/exec/cwd",
+    process_exec_no_panic: "process/exec/no_panic",
+    process_exec_shell: "process/exec/shell",
+    process_exec_stdin: "process/exec/stdin",
+    process_exec_stdio: "process/exec/stdio",
+    process_spawn_non_blocking: "process/spawn/non_blocking",
+    process_spawn_status: "process/spawn/status",
+    process_spawn_stream: "process/spawn/stream",
 }
 
 #[cfg(feature = "std-regex")]

scripts/generate_compression_test_files.luau

@@ -108,7 +108,7 @@ local BIN_ZLIB = if process.os == "macos" then "/opt/homebrew/bin/pigz" else "pi
 
 local function checkInstalled(program: string, args: { string }?)
 	print("Checking if", program, "is installed")
-	local result = process.spawn(program, args)
+	local result = process.exec(program, args)
 	if not result.ok then
 		stdio.ewrite(string.format("Program '%s' is not installed\n", program))
 		process.exit(1)
@@ -123,7 +123,7 @@ checkInstalled(BIN_ZLIB, { "--version" })
 -- Run them to generate files
 
 local function run(program: string, args: { string }): string
-	local result = process.spawn(program, args)
+	local result = process.exec(program, args)
 	if not result.ok then
 		stdio.ewrite(string.format("Command '%s' failed\n", program))
 		if #result.stdout > 0 then

tests/datetime/formatLocalTime.luau

@@ -31,7 +31,7 @@ if not runLocaleTests then
 	return
 end
 
-local dateCmd = process.spawn("bash", { "-c", "date +\"%A, %d %B %Y\" --date='@1693068988'" }, {
+local dateCmd = process.exec("bash", { "-c", "date +\"%A, %d %B %Y\" --date='@1693068988'" }, {
 	env = {
 		LC_ALL = "fr_FR.UTF-8 ",
 	},

tests/process/spawn/async.luau → tests/process/exec/async.luau

@@ -4,7 +4,7 @@ local task = require("@lune/task")
 
 local IS_WINDOWS = process.os == "windows"
 
--- Spawning a process should not block any lua thread(s)
+-- Executing a command should not block any lua thread(s)
 
 local SLEEP_DURATION = 1 / 4
 local SLEEP_SAMPLES = 2
@@ -31,7 +31,7 @@ for i = 1, SLEEP_SAMPLES, 1 do
 			table.insert(args, 1, "-Milliseconds")
 		end
 		-- Windows does not have `sleep` as a process, so we use powershell instead.
-		process.spawn("sleep", args, if IS_WINDOWS then { shell = true } else nil)
+		process.exec("sleep", args, if IS_WINDOWS then { shell = true } else nil)
 		sleepCounter += 1
 	end)
 end

tests/process/spawn/basic.luau → tests/process/exec/basic.luau

@@ -2,7 +2,7 @@ local process = require("@lune/process")
 local stdio = require("@lune/stdio")
 local task = require("@lune/task")
 
--- Spawning a child process should work, with options
+-- Executing a comamnd should work, with options
 
 local thread = task.delay(1, function()
 	stdio.ewrite("Spawning a process should take a reasonable amount of time\n")
@@ -12,7 +12,7 @@ end)
 
 local IS_WINDOWS = process.os == "windows"
 
-local result = process.spawn(
+local result = process.exec(
 	if IS_WINDOWS then "cmd" else "ls",
 	if IS_WINDOWS then { "/c", "dir" } else { "-a" }
 )

tests/process/spawn/cwd.luau → tests/process/exec/cwd.luau

@@ -6,7 +6,7 @@ local pwdCommand = if IS_WINDOWS then "cmd" else "pwd"
 local pwdArgs = if IS_WINDOWS then { "/c", "cd" } else {}
 
 -- Make sure the cwd option actually uses the directory we want
-local rootPwd = process.spawn(pwdCommand, pwdArgs, {
+local rootPwd = process.exec(pwdCommand, pwdArgs, {
 	cwd = "/",
 }).stdout
 rootPwd = string.gsub(rootPwd, "^%s+", "")
@@ -27,24 +27,24 @@ end
 
 -- Setting cwd should not change the cwd of this process
 
-local pwdBefore = process.spawn(pwdCommand, pwdArgs).stdout
-process.spawn("ls", {}, {
+local pwdBefore = process.exec(pwdCommand, pwdArgs).stdout
+process.exec("ls", {}, {
 	cwd = "/",
 	shell = true,
 })
-local pwdAfter = process.spawn(pwdCommand, pwdArgs).stdout
+local pwdAfter = process.exec(pwdCommand, pwdArgs).stdout
 assert(pwdBefore == pwdAfter, "Current working directory changed after running child process")
 
 -- Setting the cwd on a child process should properly
 -- replace any leading ~ with the users real home dir
 
-local homeDir1 = process.spawn("echo $HOME", nil, {
+local homeDir1 = process.exec("echo $HOME", nil, {
 	shell = true,
 }).stdout
 
 -- NOTE: Powershell for windows uses `$pwd.Path` instead of `pwd` as pwd would return
 -- a PathInfo object, using $pwd.Path gets the Path property of the PathInfo object
-local homeDir2 = process.spawn(if IS_WINDOWS then "$pwd.Path" else "pwd", nil, {
+local homeDir2 = process.exec(if IS_WINDOWS then "$pwd.Path" else "pwd", nil, {
 	shell = true,
 	cwd = "~",
 }).stdout