Java Structured Concurrency

Categories:
Tags:

… continued from the previous post.

Structured Concurrency Preview

I’ve written about structured concurrency, and Java has a preview API1 StructuredTaskScope2:

 1import com.fillmore_labs.blog.jvt.Slow2;
 2import java.util.concurrent.StructuredTaskScope;
 3
 4void main() {
 5  var deadline = Instant.now().plusMillis(100L);
 6  try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
 7    for (int i = 0; i < 1_000; i++) {
 8      // var queryStart = Instant.now();
 9      scope.fork(
10          () -> {
11            Slow2.fibonacci(27);
12            // var duration = Duration.between(queryStart, Instant.now());
13            return null;
14          });
15    }
16
17      scope.joinUntil(deadline);
18  }
19}

In Java structured concurrency includes cancelation via thread interruption, aborting the unfinished calculations. We use our old recursive Fibonacci calculation as Slow2, made cancelable with:

    if (Thread.interrupted()) {
      throw new InterruptedException();
    }

When we run this, it exits after around 100 Milliseconds:

> bazel run //:try6
INFO: Running command line: bazel-bin/try6
*** Finished 129 runs (871 canceled) in 113.267ms - avg 50.995ms, stddev 16.769ms

Which shows us that all virtual threads are started, even though we could only finish 129. Extending the deadline to run to completion gives:

*** Finished 1000 runs (0 canceled) in 373.313ms - avg 172.825ms, stddev 92.234ms

flame graph of run 6

So, Thread.interrupted() is not free (it’s the blue areas on top), but performant enough to call it often.

Another Example

Mirroring our Go experiments we define a task and a function calling it:

 1import java.time.Duration;
 2import java.time.Instant;
 3import java.util.concurrent.StructuredTaskScope;
 4
 5final Duration processingTime = Duration.ofSeconds(1);
 6
 7void main() throws Exception {
 8  try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
 9    var start = Instant.now();
10
11    scope.fork(
12        () -> {
13          task("task1", processingTime.dividedBy(3), null);
14          return null;
15        });
16
17    scope.fork(
18        () -> {
19          task("task2", processingTime.dividedBy(2), new TestException("task2 failed"));
20          return null;
21        });
22
23    scope.fork(
24        () -> {
25          task("task3", processingTime, null);
26          return null;
27        });
28
29    scope.join();
30
31    var result = scope.exception();
32    var duration = Duration.between(start, Instant.now());
33    System.out.println(STR."*** Got \"\{result}\" in \{duration}");
34  }
35}
36
37void task(String name, Duration processingTime, Exception result) throws Exception {
38  Thread.sleep(processingTime);
39
40  if (result != null) {
41    throw result;
42  }
43}
44
45static class TestException extends Exception {
46  TestException(String message) {
47    super(message);
48  }
49}

Running this, we see similar results as in our previous experiment:

> bazel run //:try7
INFO: Running command line: bazel-bin/try7
*** Got "com.fillmore_labs.blog.jvt.TestException: task2 failed" in 520,398ms

So ShutdownOnFailure closely mimics Go’s errgroup.

Summary

Java seems to bet on structured concurrency, at least for virtual threads in non-library code. It uses thread interruption as a means of cancelation, which requires having a handle to the running thread. We might eventually see support for context propagation, e.g. from OpenTelemetry, for the new constructs. This is conceptually very different for Go’s context, which is just hierarchically passed down and cancels tasks, including subtasks, regardless of whether the canceler is aware of them.

  1. Ron Pressler, Alan Bateman. 2023. Structured Concurrency (Second Preview). In JDK Enhancement Proposals — September 2023 — JEP 462 — <openjdk.org/jeps/462> ↩︎

  2. The code is available on GitHub at github.com/fillmore-labs/blog-javavirtualthreads. ↩︎