Structured concurrency in Java with Loom

Contents

Update: JEP 428 brings more support for structured concurrency and is now available in Java 19 EA. Please see this post on how to use structured concurrency.

Introduction - Why Loom

When you create a new Thread in Java a system call is done to the OS telling it to create a new system thread. Creating a system thread is expensive because the call takes up much time, and each thread takes up some memory. Your threads also share the same CPU, so you don't want them to block it, causing other threads to wait unnecessarily.

You can use asynchronous programming to prevent this from happening. You start a thread and tell it what to do when the data arrives. Until the data is available, other threads can use the CPU recourses for their task. In the example below, we use the CompletableFuture to get some data and tell it to print it to the console when the data is available.

1
2
CompletableFuture.supplyAsync(() -> "some data")
                .thenAccept(System.out::println);

Asynchronous programming works fine, but there is another way to work and think about concurrency implemented in Loom called “Structured concurrency”. Loom is a Java enhancement proposal (JEP) for developing concurrent applications. It aims to make it easier to write and debug multithreaded code in Java.

What are virtual Threads

Project Loom introduces virtual threads to Java. A virtual thread looks the same as the threads we are already familiar with in Java, but they work differently. The Thread class we already know is just a tiny wrapper around an expensive to create system thread. A virtual thread is created and managed by the Java virtual machine (JVM). Java doesn't make a system thread for each virtual thread you need. Instead, many virtual threads run on a single system thread called a carrier thread. Using carrier threads makes blocking very cheap! When your virtual thread is waiting on data to be available, another virtual thread can run on the carrier thread.

What is structured concurrency

With Project Loom, we also get a new model named “Structured concurrency” to work with and think about threads. The idea behind Structured concurrency is to make the lifetime of a thread work the same as code blocks in Structured programming. For example, in a Structured programming language like Java, If you call method B inside method A then method B must be finished before you can exit method A. The lifetime of method B can't exceed that of method A.

With Structured concurrency, we want the same kind of rules as with structured programming. When you create virtual thread X inside virtual thread Y, the lifetime of thread X can't exceed that of thread Y. Structured concurrency makes working and thinking about threads a lot easier. When you stop the parent thread Y all its child threads will also be canceled, so you don't have to be afraid of runaway threads still running. The crux of the pattern is to avoid fire and forget concurrency.

Thread Y starts a new thread X; both work separately from each other, but before thread Y can finish, it has to wait for thread X to have completed its work. Let's see what that looks like in Java!

Structured concurrency in Java with Loom

Structured concurrency binds the lifetime of threads to the code block which created them. This binding is implemented by making the ExecutorService Autocloseable, making it possible to use ExecutorServices in a try-with-resources. When all tasks are submitted, the current thread will wait till the tasks are finished and the close method of the ExecutorService is done.

In the following example, we have a try-with-resources that acts as the scope for the threads. We create two threads using the newVirtualThreadPerTaskExecutor(). The current thread will wait until the two submitted threads have finished and we left the try statement.

1
2
3
4
5
try (ExecutorService e = Executors.newVirtualThreadPerTaskExecutor()) {
    e.submit(() -> System.out.println("first virtual thread"));
    e.submit(() -> System.out.println("second virtual thread"));
} // all tasks were submitted and are finished at this point and the 
  // parent thread can continue

Ordered cancellation

Structured concurrency ties your threads to a scope, but all your threads will be canceled in parallel when you exit that scope. This is great but not always optimal behavior. For example, we create in the same scope a thread that writes values to a database (DB), and after that, we start some threads that generate values for the DB thread. All these threads will be closed in parallel when we exit the scope. If the DB thread is closed first, the other threads have nowhere to write to before they are also closed.

We first want to close the threads that generate a value before we close the DB thread. This problem is solved by providing an extra ExecutorService in the try-with-resources. In the example below, we start one thread for each ExecutorService. But in the example, we created a dependency between the executorServices; ExecutorService X can't finish before Y. This example works because the resources in the try are closed in reversed order. First, we wait for ExecutorService Y to close, and then the close method on X will is called.

1
2
3
4
5
try (ExecutorService x = Executors.newVirtualThreadPerTaskExecutor();
     ExecutorService y = Executors.newVirtualThreadPerTaskExecutor()) {
    x.submit(() -> y.awaitTermination(1L, TimeUnit.DAYS));
    y.submit(() -> System.out.println("is Y shutdown? " + y.isShutdown()));
} // both threads are done at this point

Exceptions and structured concurrency

Exceptions and interruptions are or at least feel like something that is still very much in development. Especially when you look at earlier examples like this one, where you could use CompletableFuture.stream. Those methods are no longer available, so we have to do something else.

Let's start with a small example. In the code below, we have a scope that starts three virtual threads, of which the second one throws an exception when it starts. The exception does not propagate to its parent thread, and the other two threads will continue to run. When we leave this scope, all three threads are considered to be finished running.

1
2
3
4
5
try (ExecutorService e = Executors.newVirtualThreadPerTaskExecutor()) {
        e.submit(() -> System.out.println("first thread"));
        e.submit(() -> {throw new RuntimeException();});
        e.submit(() -> System.out.println("third thread"));
}

In the example below, I tried to recreate an example from this post (If you know of a better way to implement this, please let me know). Like the previous example, we start three threads, of which one throws an error. But this time, we will use a stream to get the results from the futures and see if one of the threads has failed.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
try (ExecutorService e = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<String>> futures = e.invokeAll(List.of(
            () -> "first task",
            () -> "second task",
            () -> {
                throw new RuntimeException();
            }
    ));

    String s = futures.stream()
            .map(future -> {
                try {
                    return future.get();
                } catch (InterruptedException | ExecutionException ex) {
                    System.out.println("exception = " + ex);
                }
                return null;
            })
            .filter(Objects::nonNull)
            .collect(Collectors.joining(","));

    System.out.println("s = " + s);
}

In this case, the exception is also not propagated to the parent thread. all threads will be invoked and be finished when we leave the scope of the try-with-resources.

Conclusion

This post looked at the benefits of structured concurrency to the Java language and how it is implemented in Loom. We went over how to create a scope for your threads and have them closed in a specific order. We also saw what happens when one of the virtual threads in a scope throws an error.

References and further reading