Java ProcessBuilder, Virtual Threads and Structured Concurrency

A few utility methods to run a process from Java with structured concurrency, using some newer Java features.
Very suitable for long-running batch processing.

Requires Java 25 with --enable-preview.

Context

We have thousands of edge devices in the field and use a reverse tunnel to access and manage them. The tool for this is a shell executable.
Connectivity is ofcourse not perfect. So errors and unresponsive processes will happen. Good observability, setting timeouts and always cleaning them up is essential.
The executable is 100 MB. But even if it were less; you can't leak resources if you want your service to have a high uptime.

Ofcourse a forked posix process is connected to its service parent through stdin, stdout, stderr and exit code. And this can can continue for a while and create a lot of output. So we have to account for ongoing incoming data. Not necessarily just wait until it's all there in memory.
Handling ongoing incoming data is basically stream processing. So we probably need some form of that.

Our services are JVM based. So we'll have to work with the building blocks that gives us.
The JVM has had Process since java 1.0, ProcessBuilder since 1.5, Process.waitFor since 9. So you easily find yourself having to do old school things like implement Runnable, managing Threads, pumping bytes from one java.io stream to another, ...
The primitives are all there. But it's verbose and easy to make mistakes.

Thankfully Java improved a lot. Especially structured concurrency and virtual threads and managed resources (AutoClosable and try with resource) wil help. The idea for this tiny library was to leverage those. And design an api that is solid, that provides sane defaults but also allows customising fully and is well testable.

So that's what drove this implementation.

The initial implementation and test were written by hand.
AI helped in:

• analysing corner cases, digging into JVM implementation specifics
• extend the tests,
• write a stream gatherer example.
• writing the generic types of callbacks that throw exceptions. I've been writing other JVM languages like Kotlin and Scala for a long time so I didn't have those Java language idioms at hand.
• turning a void method into a Callable by making it return @Nullable Void. I didn't have that idiom at hand either.
• clever Perl tricks to simulate process behaviours. That $| = 1; turns on autoflush to stdout.

This repo is mostly a blogpost. But hopefully you'll find the implementation useful as well. The implementation is so small you can just copy it.

Design

Our requirements:

• The java.lang.ProcessBuilder api requires us to read the standard output and standard error of the process in separate threads.
• Those threads should always be cleaned up in time.
• We need to always destroy the process after we are done with it. For error cases as well.
• Running a process is unpredictable; we need to specify a timeout. Because a hanging process is a resource leak.
• child processes should be destroyed as well
• we want the option of not reading stdin, stdout, stderr all at once in memory
• We want the process id before we start waiting for the process to finish. For observability, and so we can test error scenarios where the process is killed by the OS or by Kubernetes.

Structured concurrency can help us with this. The blocking calls can be handled with virtual threads, which are lightweight so we can afford one per stream. And the structured concurrency helps us with error handling and making sure to clean up all threads.

How to use

In memory example:

    var toZip = "zip me";

    String[] cmd = {"gzip", "-c"};
    try (RunningProcess runningProcess = startProcess(cmd, toZip.getBytes(StandardCharsets.UTF_8), Duration.ofSeconds(5), Duration.ofSeconds(5))) {

        IO.println("started process with pid: " + runningProcess.getProcess().pid());
        var result = runningProcess.waitFor();
        IO.println("exit value: " + result.exitValue());
        var zipped = result.stdout();

        var unzippedAgain = unzip(zipped);

        IO.println("as expected: "+ unzippedAgain.equals(toZip));
    }

If stdin is too large to hold in memory provide a lazy inputstream instead of the bytes:

    try (var runningProcess = startProcess(cmd, () -> Files.newInputStream(cargo), Duration.ofSeconds(5), Duration.ofSeconds(5))) {

This is essential for management scripts that are larger than the maximum argument size of the executable.

For a long running task there can be more output than we want to load in memory at the same time. Also we may want to respond as lines come in. Here we use Gatherers#windowFixed(int) to batch events before we handle them as they arrive.

    String[] cmd = {"perl", "-e", "$| = 1; for (1..7) { print \"event $_\\n\"; select(undef, undef, undef, 0.25); }"};

    try (var runningProcess = startProcess(cmd, Duration.ofSeconds(10), Duration.ofSeconds(5))) {

        IO.println("streaming events from pid: " + runningProcess.getProcess().pid());
        var start = System.nanoTime();

        var result = runningProcess.waitForLines(
                events -> events
                        .gather(Gatherers.windowFixed(3))
                        .map(batch ->
                                writeBatch(batch, Duration.ofNanos(System.nanoTime() - start))
                        )
                        .reduce(0, Integer::sum),
                Stream::count);

        IO.println("exit value: " + result.exitValue());
        IO.println("events written: " + result.stdout());
        IO.println("lines on stderr: " + result.stderr());
    }

Implementation

The startProcess method returns a RunningProcess that implements AutoCloseable. And thus we support try-with-resources.

    public static RunningProcess startProcess(String[] cmd, @Nullable StdinSource stdin, Duration timeoutAfter, Duration gracePeriod) throws IOException {
        ProcessBuilder processBuilder = new ProcessBuilder(requireNonNull(cmd, "cmd"));

        var process = processBuilder.start();

        return new RunningProcess(process, stdin, timeoutAfter, gracePeriod);
    }

It represents a scope for the resources we need to cleanup: the java.lang.Process plus its children plus the java.util.concurrent.StructuredTaskScope that captures the threads that read standard output and standard error. By using try-with-resources we make sure that both the process is always destroyed and the threads are always finished or interrupted.

Btw if we call startProcess without a try-with-resources statement our IDE suggests introducing it, so that makes the method almost self documenting.

Our coreRunningProcess.waitFor generic types are a bit hard to read. I guess that's unavoidable even in modern java. Basically they're just callbacks that receive the stdout or stderr InputStream and process it on structured concurrency virtual thread. Either in memory or as java streams.
We fork virtual threads to write stdin, read stdout, read stderr and wait for the process to finish. And gather them in a StructuredTaskScope that awaits all four and propagates the first failure

    private <StdoutResult, StderrResult> StreamedResult<StdoutResult, StderrResult> waitFor(
            OutputHandler<StdoutResult> stdoutHandler, OutputHandler<StderrResult> stderrHandler) throws InterruptedException {

        scope.fork(this::readStdin);

        var stdout = scope.fork(() -> stdoutHandler.handle(process.getInputStream()));
        var stderr = scope.fork(() -> stderrHandler.handle(process.getErrorStream()));
        var exitValue = scope.fork(() -> process.waitFor());

        scope.join(); // await all four, or throw TimeoutException / FailedException

        return new StreamedResult<>(exitValue.get(), stdout.get(), stderr.get(), process.pid());
    }

Unfortunately java.lang.Process.destroy() doesn't signal child processes. So we have to do that manually. And then if processes do not terminate gracefully we do that by force after the grace period.

    private void destroyProcessTree() {
        var descendants = process.descendants().toList();
    
        process.destroy();
        descendants.forEach(ProcessHandle::destroy);
    
        if (!awaitExit(descendants)) {
            process.destroyForcibly();
            descendants.forEach(ProcessHandle::destroyForcibly);
        }
    }

Some alternatives that were considered before using virtual threads and structured concurrency:

• We could follow the lead of java.lang.Process.onExit and use CompletableFutures to read the input streams. For instance using CompletableFuture.supplyAsync. That would run the task on the ForkJoinPool.commonPool(). But then we need to take special care to mark the task as blocking. Otherwise our pool will quickly run out. For instance using ForkJoinPool.managedBlock. It's doable but involves more code and joining the 4 CompletableFutures is not that straightforward in Java. Java not having something like a do notation (for comprehension ...) to easily combine futures.
• We could create a Runnable class to read from input stream. But then the threads will have to report back errors to the main thread.

Modern java

The --enable-preview flag is needed because java.util.concurrent.StructuredTaskScope is still a preview api (JEP 505). Virtual threads themselves are final since Java 21.

Java has improved a lot in recent years!

• Virtual threads (JEP 444) and structured concurrency.
• Records and sealed interfaces and pattern matching for data oriented programming.
• try-with-resources for AutoCloseable resources.
• stream gatherers for stream processing (JEP 485)
• var for local variables.
• JSpecify @NullMarked / @Nullable to distinguish mandatory from optional arguments. No JEP 8303099 (Null-Restricted and Nullable Types) yet, unfortunately, that is still a draft.

Some smaller niceties used:

• Compact source files and instance main methods (JEP 512)
• java.lang.IO.println instead of System.out.println
• Module import declarations (JEP 511): import module java.base; replaces a dozen single-type imports in RunningProcess.
• _ for unnamed variables, so a caught exception we deliberately ignore is explicit.
• Markdown documentation comments (JEP 467): /// instead of /** */, with [Process#destroy()] instead of {@link Process#destroy()}.