🥮 mooncakes.io
    Sign in

    #Process Management (@moonbitlang/async/process)

    Asynchronous process spawning and management for MoonBit with support for pipes, environment variables, and I/O redirection.

    #Quick Start

    #Running Simple Commands

    Execute commands and collect their output:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "simple command execution" {
      let (exit_code, output) = @process.collect_stdout("echo", ["Hello, World!"])
      inspect(exit_code, content="0")
      let text = output.text()
      inspect(text.has_prefix("Hello"), content="true")
    }

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "command with exit code" {
      let exit_code = @process.run("sh", ["-c", "exit 42"])
      inspect(exit_code, content="42")
    }

    #Spawn process in the background

    Spawning process in the background can be achieved by combining @process.run and @async.TaskGroup::spawn_bg. @proces also provides a convenient helper @process.spawn, which accept a task group as its first parameter, and spawn the process in the task group:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "spawn command in background" {
      @async.with_task_group(group => {
        let process = @process.spawn(group, "sh", ["-c", "sleep 0.5; exit 42"])
        // do other stuff while the process is running in the background
        @async.sleep(250)
        // use `.wait()` to wait for the child process, or `.try_wait()` to peek its status
        inspect(process.wait(), content="42")
      })
    }

    Child process follows rules of structured concurrency as well:

    • by default, the task group will wait for the child process
    • the child process will be terminated automatically when the task group terminates (for example when no_wait=true is passed to @process.spawn)

    #Collecting Process Output

    #Collect Standard Output

    Capture stdout from a process:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "collect stdout" {
      let (code, output) = @process.collect_stdout("printf", ["test output"])
      inspect(code, content="0")
      inspect(output.text(), content="test output")
    }

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "collect stdout with args" {
      let (code, output) = @process.collect_stdout("sh", [
        "-c", "echo 'line 1'; echo 'line 2'",
      ])
      inspect(code, content="0")
      let text = output.text()
      inspect(text.contains("line 1"), content="true")
      inspect(text.contains("line 2"), content="true")
    }

    #Collect Standard Error

    Capture stderr from a process:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "collect stderr" {
      let (code, output) = @process.collect_stderr("sh", [
        "-c", "printf 'error message' >&2",
      ])
      inspect(code, content="0")
      inspect(output.text(), content="error message")
    }

    #Collect Both Stdout and Stderr

    Capture both output streams separately:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "collect both streams" {
      let (code, stdout, stderr) = @process.collect_output("sh", [
        "-c", "printf 'out msg'; printf 'err msg' >&2",
      ])
      inspect(code, content="0")
      inspect(stdout.text(), content="out msg")
      inspect(stderr.text(), content="err msg")
    }

    #Collect Merged Output

    Merge stdout and stderr into a single stream:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "collect merged output" {
      let (code, output) = @process.collect_output_merged("sh", [
        "-c", "printf 'ab'; printf 'cd' >&2; printf 'ef'",
      ])
      inspect(code, content="0")
      inspect(output.text(), content="abcdef")
    }

    #Process I/O with Pipes

    #Reading from Process

    Create a pipe to read from a process:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "read from process with pipe" {
      @async.with_task_group(group => {
        let (reader, writer) = @process.read_from_process()
        defer reader.close()
        let _ = @process.spawn(group, "echo", ["Hello from process"], stdout=writer)
        let output = reader.read_all().text()
        inspect(output.has_prefix("Hello"), content="true")
      })
    }

    #Writing to Process

    Create a pipe to write to a process:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "write to process with pipe" {
      @async.with_task_group(group => {
        let (cat_read, we_write) = @process.write_to_process()
        let (we_read, cat_write) = @process.read_from_process()
        let _ = @process.spawn(
          group,
          "cat",
          ["-"],
          stdin=cat_read,
          stdout=cat_write,
        )
        group.spawn_bg(() => {
          defer we_write.close()
          we_write.write(b"test input\n")
        })
        group.spawn_bg(() => {
          defer we_read.close()
          let output = we_read.read_all().text()
          inspect(output.contains("test input"), content="true")
        })
      })
    }

    #File Redirection

    #Redirect from File

    Use a file as process input:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "redirect input from file" {
      @async.with_task_group(root => {
        let input_file = "_build/process_test_input.txt"
        @fs.write_file(input_file, "file content")
        root.add_defer(() => @fs.remove(input_file))
        let (code, output) = @process.collect_stdout(
          "cat",
          [],
          stdin=@process.redirect_from_file(input_file),
        )
        inspect(code, content="0")
        inspect(output.text(), content="file content")
      })
    }

    #Redirect to File

    Write process output to a file:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "redirect output to file" {
      @async.with_task_group(root => {
        let output_file = "_build/process_test_output.txt"
        root.add_defer(() => @fs.remove(output_file))
        let code = @process.run(
          "echo",
          ["test output"],
          stdout=@process.redirect_to_file(output_file),
        )
        inspect(code, content="0")
        let content = @fs.read_file(output_file).text()
        inspect(content.has_prefix("test output"), content="true")
      })
    }

    #File to File Redirection

    Copy file content using process redirection:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "file to file redirection" {
      @async.with_task_group(root => {
        let input_file = "_build/process_redirect_in.txt"
        let output_file = "_build/process_redirect_out.txt"
        @fs.write_file(input_file, "redirect test")
        root.add_defer(() => @fs.remove(input_file))
        root.add_defer(() => @fs.remove(output_file))
        let _ = @process.run(
          "cat",
          [],
          stdin=@process.redirect_from_file(input_file),
          stdout=@process.redirect_to_file(output_file),
        )
        inspect(@fs.read_file(output_file).text(), content="redirect test")
      })
    }

    #Environment Variables

    #Setting Environment Variables

    Pass custom environment variables to processes:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "set environment variable" {
      let (code, output) = @process.collect_stdout("sh", ["-c", "echo $MY_VAR"], extra_env={
        "MY_VAR": "my_value",
      })
      inspect(code, content="0")
      inspect(output.text().trim(), content="my_value")
    }

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "multiple environment variables" {
      let (code, output) = @process.collect_stdout(
        "sh",
        ["-c", "echo $VAR1-$VAR2"],
        extra_env={ "VAR1": "first", "VAR2": "second" },
      )
      inspect(code, content="0")
      inspect(output.text().trim(), content="first-second")
    }

    #Isolated Environment

    Run process without inheriting parent environment:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "isolated environment" {
      let (code, output) = @process.collect_stdout(
        "env",
        [],
        extra_env={ "ONLY_VAR": "only_value" },
        inherit_env=false,
      )
      inspect(code, content="0")
      let text = output.text()
      inspect(text.contains("ONLY_VAR=only_value"), content="true")
      // Parent environment variables won't be present
      inspect(text.contains("PATH="), content="false")
    }

    #Working Directory

    #Change Working Directory

    Execute processes in a specific directory:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "set working directory" {
      let (code, output) = @process.collect_stdout("pwd", [], cwd="src")
      inspect(code, content="0")
      let text = output.text().trim()
      // Check that the path ends with "src" (works across OSes)
      inspect(text.has_suffix("src"), content="true")
    }

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "relative path in cwd" {
      let (code, output) = @process.collect_stdout("ls", [], cwd="src")
      inspect(code, content="0")
      let text = output.text()
      // Should list contents of src directory
      let has_content = text.length() > 0
      inspect(has_content, content="true")
    }

    #Asynchronous Process Management

    #Spawn and Wait

    Spawn processes asynchronously and wait for completion:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "spawn and wait" {
      let exit_code = @process.run("sleep", ["0.1"])
      inspect(exit_code, content="0")
    }

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "wait for specific exit code" {
      let exit_code = @process.run("sh", ["-c", "exit 5"])
      inspect(exit_code, content="5")
    }

    #Spawn Orphan Process

    Start a orphan process with unbounded lifetime. The orphan process will not be automatically cancelled, and may live longer than the main process. It is recommended to use @process.spawn or @process.run whenever possible, use @process.spawn_orphan only when it is absolutely necessary:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "spawn orphan and wait later" {
      let pid = @process.spawn_orphan("sh", ["-c", "sleep 0.1; exit 7"])

      // Do other work...
      @async.sleep(50)

      // Wait for the process to complete
      let exit_code = @process.wait_pid(pid)
      inspect(exit_code, content="7")
    }

    #Advanced Usage

    #Merge Output Streams

    Combine stdout and stderr into one stream:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "merge stdout and stderr" {
      @async.with_task_group(group => {
        let (reader, writer) = @process.read_from_process()
        defer reader.close()
        let _ = @process.spawn(
          group,
          "sh",
          ["-c", "echo 'to stdout'; echo 'to stderr' >&2"],
          stdout=writer,
          stderr=writer,
        )
        let output = reader.read_all().text()
        inspect(output.contains("to stdout"), content="true")
        inspect(output.contains("to stderr"), content="true")
      })
    }

    #Multiple Processes Sharing Output

    Run multiple processes writing to the same pipe:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "multiple processes to one pipe" {
      @async.with_task_group(root => {
        let (reader, writer) = @process.read_from_process(shared=true)
        root.spawn_bg(no_wait=true, () => {
          defer reader.close()
          let output = reader.read_all().text()
          inspect(output.contains("first"), content="true")
          inspect(output.contains("second"), content="true")
        })
        @async.with_task_group(group => {
          // The write must be manually closed after all children process are spawned
          // in the shared case
          defer writer.close()
          @process.spawn(group, "echo", ["first"], stdout=writer) |> ignore
          @process.spawn(group, "echo", ["second"], stdout=writer) |> ignore
        })
      })
    }

    #Types Reference

    #ProcessInput

    Trait for types that can be used as process input:

    • Created with write_to_process()
    • Created with redirect_from_file(path)
    • Implemented by @pipe.PipeRead

    #ProcessOutput

    Trait for types that can be used as process output:

    • Created with read_from_process()
    • Created with redirect_to_file(path)
    • Implemented by @pipe.PipeWrite

    #Best Practices

    1. Always close pipes with defer reader.close() or defer writer.close()
    2. Use collect functions for simple output capture
    3. Use task groups when managing multiple processes
    4. Handle exit codes appropriately for error detection
    5. Set working directory explicitly when path-dependent
    6. Use environment variables for configuration
    7. Prefer @process.spawn or @process.run. Only use spawn_orphan when necessary.

    #Error Handling

    Process operations handle errors through exit codes:

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "handle process errors" {
      // Non-existent command fails
      @test_util.assert_raise_async(() => @process.run("nonexistent_command", []))
    }

    ///|
    #cfg(all(target="native", not(platform="windows")))
    async test "exit code indicates failure" {
      let exit_code = @process.run("sh", ["-c", "exit 1"])
      let is_failure = exit_code != 0
      inspect(is_failure, content="true")
    }

    For complete examples, see the test files in src/process/.

    ProcessInput

    source
    trait ProcessInput

    An entity that can be used to redirect stdin of a process

    ProcessOutput

    source
    trait ProcessOutput

    An entity that can be used to redirect stdout/stderr of a process

    CancellationHandler

    source
    pub(all) struct CancellationHandler(async (Int) -> Unit)

    A handler function used to stop spawned process on cancellation. The function receive the PID of the process as input.

    Process

    source
    pub struct Process {
      pid : Int
      // private fields
    }

    A handle to a spawned process

    Process::cancel

    source
    fn Process::cancel(self : Process) -> Unit

    Cancel a child process. The method of cancellation is determined by the cancel_handler parameter of @process.spawn. Notice that after .cancel() is called, if may take a while before the child process actually terminates. Use .wait() to wait for actual termination of the child process.

    Process::try_wait

    source
    fn Process::try_wait(self : Process) -> Int? raise

    If the process already terminated, return its exit code. If the process is killed by a signal, the result would be -signal_number. If the process is still running, return None.

    Process::wait

    source
    async fn Process::wait(self : Process) -> Int

    Wait for a process to terminate, return the exit code of the process If the process is killed by a signal, the result would be -signal_number.

    ReadFromProcess

    source
    type ReadFromProcess

    A temporary pipe used to read output from a spawned process

    ReadFromProcess::close

    source
    fn ReadFromProcess::close(self : ReadFromProcess) -> Unit

    ReadFromProcess::drop

    source
    async fn ReadFromProcess::drop(self : ReadFromProcess, len : Int) -> Int

    ReadFromProcess::read

    source
    async fn ReadFromProcess::read(self : ReadFromProcess, dst : FixedArray[Byte], offset? : Int, max_len? : Int) -> Int

    ReadFromProcess::read_all

    source

    ReadFromProcess::read_exactly

    source
    async fn ReadFromProcess::read_exactly(self : ReadFromProcess, len : Int) -> Bytes

    ReadFromProcess::read_some

    source
    async fn ReadFromProcess::read_some(self : ReadFromProcess, max_len? : Int) -> Bytes?

    ReadFromProcess::read_until

    source
    async fn ReadFromProcess::read_until(self : ReadFromProcess, sep : StringView) -> String?

    WriteToProcess

    source
    type WriteToProcess

    A temporary pipe used to write data to a spawned process

    WriteToProcess::close

    source
    fn WriteToProcess::close(self : WriteToProcess) -> Unit

    WriteToProcess::write

    source
    async fn WriteToProcess::write(self : WriteToProcess, data : &
    Data    ) -> Unit

    WriteToProcess::write_once

    source
    async fn WriteToProcess::write_once(self : WriteToProcess, buf : Bytes, offset~ : Int, len~ : Int) -> Int

    WriteToProcess::write_reader

    source
    async fn WriteToProcess::write_reader(self : WriteToProcess, reader : &
    Reader    ) -> Unit

    collect_output

    source
    async fn collect_output(cmd : StringView, args : ArrayView[String], extra_env? : Map[String, String], inherit_env? : Bool, stdin? : &ProcessInput, cwd? : StringView, no_console_window? : Bool) -> (Int, &
    Data    , &
    Data    )

    Run a process and collect its standard output & standard error. Return the exit code of the process, the content of its standard output, and the content of its standard error.

    The meaning of parameters is the same as @process.run

    collect_output_merged

    source
    async fn collect_output_merged(cmd : StringView, args : Array[String], extra_env? : Map[String, String], inherit_env? : Bool, stdin? : &ProcessInput, cwd? : StringView, no_console_window? : Bool) -> (Int, &
    Data    )

    Run a process, merge and collect its standard output & standard error. Return the exit code of the process, and the content of its standard output and standard error.

    The meaning of parameters is the same as @process.run

    collect_stderr

    source
    async fn collect_stderr(cmd : StringView, args : ArrayView[String], extra_env? : Map[String, String], inherit_env? : Bool, stdin? : &ProcessInput, stdout? : &ProcessOutput, cwd? : StringView, no_console_window? : Bool) -> (Int, &
    Data    )

    Run a process and collect its standard error. Return the exit code of the process and the content of its standard error.

    The meaning of parameters is the same as @process.run

    collect_stdout

    source
    async fn collect_stdout(cmd : StringView, args : ArrayView[String], extra_env? : Map[String, String], inherit_env? : Bool, stdin? : &ProcessInput, stderr? : &ProcessOutput, cwd? : StringView, no_console_window? : Bool) -> (Int, &
    Data    )

    Run a process and collect its standard output. Return the exit code of the process and the content of its standard output.

    The meaning of parameters is the same as @process.run

    graceful_cancel

    source
    fn graceful_cancel(timeout~ : Int, signal? : 
    Signal    ) -> CancellationHandler

    A process cancellation handler that first try to gracefully stop the process, and if the process is still running after timeout milliseconds, forcefully stop the process.

    Graceful process termination is implemented by sendig signal to the process. The default signal is SIGTERM on POSIX-like systems and SIGBREAK (aka CTRL_BREAK_EVENT) on Windows.

    Note that on Windows, SIGBREAK is the only signal allowed to be be sent.

    hard_cancel

    source
    fn hard_cancel() -> CancellationHandler

    A process cancellation handler that forcefully stop the process. Implemented via SIGKILL on POSIX-like systems and TerminateProcess on Windows.

    pipe

    source
    fn pipe() -> (&ProcessInput, &ProcessOutput) raise

    Create a temporary pipe for connecting the output of one child process to the input of another child process.

    read_from_process

    source
    fn read_from_process(shared? : Bool) -> (ReadFromProcess, &ProcessOutput) raise

    Create a temporary pipe for reading from stdout/stderr of a process. The return value is a pair (r, w), where r is a temporary pipe that can be used to read process output, and w should be passed to @process.run, @process.spawn etc.

    w is temporary: it can only be passed to one @process.run call. However, it is safe to pass w to both stdout and stderr of the same process.

    If shared is false (the default), the created pipe is unique. In this case, w can only be passed to one child process (but it is ok to pass w to both stdout and stderr of the same process), and w will be closed automatically after the child process started.

    If shared is true, the created pipe is shared: w can be passed to multiple children process to merge and collect their output. In this case, w should be closed manually via w.close() after being passed to the last child process. Note that w can be closed immediately after the last child process started, there is no need to wait for the termination of the child process. If you forgot to close w or close it too late, the read end r will fail to observe EOF from children process.

    redirect_from_file

    source
    async fn redirect_from_file(path : String) -> &ProcessInput

    Redirect the content of a file at path to the stdin of a process.

    redirect_to_file

    source
    async fn redirect_to_file(path : String, append? : Bool, create_mode? : 
    CreateMode    , permission? : Int, shared? : Bool) -> &ProcessOutput

    Redirect the output of a process to the file at path. An output channel that should be passed to child process will be returned.

    The meaning of append, create_mode and permission is the same as @fs.open, see the document of @fs.open for more details.

    If shared is false (the default), the returned channel w is unique. In this case, w can only be passed to one child process (but it is ok to pass w to both stdout and stderr of the same process), and w will be closed automatically after the child process started.

    If shared is true, the returned channel w is shared: it can be passed to multiple children process to merge and collect their output. In this case, w should be closed manually via w.close() after being passed to the last child process. Note that w can be closed immediately after the last child process started, there is no need to wait for the termination of the child process.

    run

    source
    async fn run(cmd : StringView, args : ArrayView[String], extra_env? : Map[String, String], inherit_env? : Bool, stdin? : &ProcessInput, stdout? : &ProcessOutput, stderr? : &ProcessOutput, cwd? : StringView, no_console_window? : Bool, cancel_handler? : CancellationHandler) -> Int

    Execute a system process with command cmd, and provide args as extra arguments to cmd.Both cmd and elements of args` are encoded using UTF8.

    run will block until the new process terminates, and returns the exit status of the process. To run the process in background, use @async.spawn or @async.spawn_bg. If the process is killed by a signal, the result would be -signal_number.

    If inherit_env is true (true by default), the new process will inherit environment variables of current process.

    extra_env, if present, will set extra environment variables for the new process (in addition to those inherited ones, if inherit_env is true). Keys and values of extra_env are encoded using UTF8.

    The standard IO of the new process will be redirected to stdin, stdout and stderr, if set. Standard IO channel can be redirected to one of the following:

    • a temporary pipe created via read_from_process or write_to_process, which can be used to read from/write to the process directly
    • a file on the filesystem, via redirect_to_file or redirect_from_file
    • an existing @pipe.PipeRead or @pipe.PipeWrite, for example redirecting standard error to standard out.

    Note than when passing an existing pipe to the process, the ownership of the pipe is NOT transferred. So the caller should still close the channel manually when apporiate.

    If cwd is present, the spawned command will be executed in the directory specified by cwd.

    When no_console_window is true (false by default), creation of a new console window will be disabled on Windows. By default a new console window may be created when the calling process is a GUI application and the child process is a conlose application. no_console_window has no effect on non-Windows platforms.

    If current task is cancelled while blocking, cancel_handler will be used to automatically stop the process. @process.run will not return until the process terminates, even if cancelled. The default value of cancel_handler is graceful_cancel(timeout=5000) (First try to gracefully terminate the process, and if the process is still running after five seconds, terminate it forcefully).

    If cancel_handler is still running after the process terminates, it will be cancelled.

    spawn

    source
    async fn[X] spawn(group : 
    TaskGroup    [X], cmd : StringView, args : ArrayView[String], extra_env? : Map[String, String], inherit_env? : Bool, stdin? : &ProcessInput, stdout? : &ProcessOutput, stderr? : &ProcessOutput, cwd? : StringView, no_console_window? : Bool, cancel_handler? : CancellationHandler, no_wait? : Bool) -> Process

    Spawn a child process inside a task group. If no_wait is false (fales by default), the task group will only exit when the child process terminates. If no_wait is true, the child process will be terminated automatically when the task group terminates.

    A Process object will be returned, users can retrieve the PID of the spawned process via .pid, or wait for the process to terminate via .wait()/.try_wait().

    All arguments except group and no_wait have the same meaning as @process.run, see @process.run for more details.

    spawn_orphan

    source
    async fn spawn_orphan(cmd : StringView, args : ArrayView[String], extra_env? : Map[String, String], inherit_env? : Bool, stdin? : &ProcessInput, stdout? : &ProcessOutput, stderr? : &ProcessOutput, cwd? : StringView, no_console_window? : Bool) -> Int

    Execute a system process with command cmd, and provide args as extra arguments to cmd.Both cmd and elements of args` are encoded using UTF8. The process ID of the spawned process will be returned.

    The spawned process would be orphan, meaning that it will keep running until completion or explicitly terminated, even if the task that calls spawn_orphan is cancelled. Users are recommended to use @process.run whenever possible, because @process.run has better structured concurrency integration. spawn_orphan should only be used when:

    • the process is intended to be orphan, i.e. keep running after parent process terminates
    • operations on the process ID is needed, such as sending signals to the child process

    The meaning of the arguments is the same as @process.run, see @process.run for more details.

    wait_pid

    source
    async fn wait_pid(pid : Int) -> Int

    Wait for the process with specfic ID to terminate, and return the exit code of the process. If the process is killed by a signal, the result would be -signal_number.

    write_to_process

    source
    fn write_to_process() -> (&ProcessInput, WriteToProcess) raise

    Create a temporary pipe for writing to stdin of a process. The return value is a pair (r, w), where w is a temporary pipe that can be used to write to process output, and r should be passed to @process.run.

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io