Structured Concurrency – a new approach to multi-threading in Java

Running multiple things in parallel

When working with microservices, you frequently need to perform some operations in parallel. Either there is a set of HTTP requests to be sent to other services, or you want to offload some costly calculations to a separate thread, so that you can complete other operations in the meantime. This can be done through spawning a system thread, with the Thread class or an ExecutorService. There are several disadvantages to this.

First, there is no easy way to control how other tasks behave when one of them is cancelled:

try (var executorService = Executors.newCachedThreadPool()) {
    Future<Invoice> invoiceFuture = executorService.submit(
        () -> fetchFromInvoiceService(invoiceId)
    );
    Future<PaymentStatus> paymentStatusFuture = executorService.submit(
        () -> fetchFromPaymentsService(invoiceId)
    );

    processInvoicePayment(invoiceFuture.get(), paymentStatusFuture.get());
}

If the invoice service is down, when you invoke get() you will get an exception. However, this exception does not interrupt the call to the payments service, which now exists in a separate thread and you’ll still need to wait for it to finish. If it’s just a simple HTTP request, this might not be a big problem, but imagine this was some resource heavy calculation now running without any need.

We could of course wrap each get() in try ... catch blocks and cancel all other threads in case of failure, but the code becomes increasingly complex as the number of subtasks increases.

Accordingly, if the call to payments fail early, you still need to wait for invoiceFuture to finish before you get informed of the exception.

Another issue with the code above is that for the HTTP calls, database queries and other i/o the thread will become blocked waiting for operation to finish, which is a waste of CPU time. Instead of blocking a platform thread, it would be much more efficient to use the new virtual threads feature from Java 21.

The new Structured Concurrency model, featured as a preview in Java 21, is the way to address all of those problems.

Structured Concurrency: Task Scopes and Subtasks

As of Java SE 21, which is the current LTS, the new concurrency model is a preview feature. To enable it, you have to add --enable-preview to the compiler params. For example in a Spring Boot Maven project you need to add jvmArguments to pom.xml:

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <configuration>
                <jvmArguments>--enable-preview</jvmArguments>
            </configuration>
        </plugin>
    </plugins>
</build>

The central API, which functions as an alternative to the old ExecutorService is the StructuredTaskScope class. Instead of Futures, it gives us Subtasks.

public void processInvoice(long invoiceId) throws ExecutionException, InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        StructuredTaskScope.Subtask<Invoice> invoiceSubtask = scope.fork(
            () -> fetchFromInvoiceService(invoiceId)
        );
        StructuredTaskScope.Subtask<PaymentStatus> paymentStatusSubtask = scope.fork(
            () -> fetchFromPayments(invoiceId)
        );
        scope.join().throwIfFailed();

        processInvoicePayment(invoiceSubtask.get(), paymentStatusSubtask.get());
    }
}

A few things are worth noting here. First, you can explicitly specify the policy, which governs what happens when one of the tasks fails or finishes. ShutdownOnFailure means that the first task to fail automatically cancels all other. The alternative is ShutdownOnSuccess, which cancels all the other tasks if one of them finishes successfully. This behavior is useful for “invoke any” scenarios.

You can also create your own custom policies by extending StructuredTaskScope.

You also need to call the join() method of the scope to wait for the subtasks. There is also a joinUntil(), which allows you to specify a deadline instead of waiting forever.

ShutdownOnSuccess has a result() method, which returns one of the successfully finished tasks.

public WeatherForecast getFirstForecast(List<WeatherService> weatherServices, Instant deadline) {
    try (var scope = new StructuredTaskScope.ShutdownOnSuccess<WeatherForecast>()) {
        for (var weatherService : weatherServices) {
            scope.fork(() -> getWeather(weatherService.url()));
        }

        return scope.joinUntil(deadline).result();
    } catch (ExecutionException | InterruptedException | TimeoutException e) {
        // handle exceptions here
        throw new RuntimeException(e);
    }
}

Behind the scenes, StructuredTaskScope uses virtual threads, which are beneficial for naturally async i/o operations, but not necesarrily for CPU-bound code. You can override this by passsing a ThreadFactory:

try (var scope = new StructuredTaskScope.ShutdownOnFailure(null, Thread.ofPlatform().factory())) {
    // ...
}

Scoped Values

Another preview feature in Java 21 is Scoped Values: a new way of providing shared context that can be inherited by subtasks. It’s essentially like a thread-local variable but with extra safety.

Let’s return to the Invoice and Payment example and say we need to store some context information about the current user (let’s call it UserContext). Instead of passing it to each subtask as a parameter, we can wrap it in a Scoped Value:

private final static ScopedValue<UserContext> USER_CONTEXT = ScopedValue.newInstance();

Then in order to set the context, we wrap our processInvoice function:

var userContext = new UserContext(userId, /* ... */);
ScopedValue.where(USER_CONTEXT, userContext).run(() -> {
    processInvoice(invoiceId);
});

… and finally, down the call stack, we can access the user context:

private PaymentStatus fetchFromPaymentsService(long invoiceId) {
    long userId = USER_CONTEXT.get().userId();
    // authorize and send the HTTP call
    // ...
}

The value won’t leak outside of the function call.

We didn’t need to change anything in processInvoice, because Scoped Values are designed to work with Structured Concurrency. The StructuredTaskScope handles inheritance for us.

Conclusion

Structured Concurrency is the new, improved way of writing parallel code. It allows us to define failure handling policies, conditions for short-circuiting, and also – in combination with Scoped Values – gives us ways to inherit the parent state in the subtasks, and all of that without having to use complicated reactive libraries.

In Java 21 it’s still a preview, but it definitely looks promising, especially if you want to take advantage of virtual threads.

References

• JEP 453: Structured Concurrency (Preview)
• JEP 446: Scoped Values (Preview)
• Loom is Blooming by José Paumard & Remi Forax (Devoxx 2022 talk)