Concurrency – Virtual Threads and Structured Concurrency

This chapter includes 16 problems that briefly introduce virtual threads and structured concurrency.

If you don’t have a background in concurrency in Java, then I strongly recommend postponing this chapter until after you have read some good introductory coverage on the topic. For instance, you could try out Chapters 10 and 11 from Java Coding Problems, First Edition.

Virtual threads are one of the most important and astonishing features added by Java in the last few years. They have a significant impact on how we will continue to write and understand concurrent code from this point forward. In this chapter, you’ll learn, step by step, every single detail of this topic and the structured concurrency paradigm.

After this chapter, you’ll be quite knowledgeable in working with virtual threads and structured concurrency.

Problems

Use the following problems to test your programming prowess in virtual threads and structured concurrency in Java. I strongly encourage you to give each problem a try before you turn to the solutions and download the example programs:

209. Explaining concurrency vs. parallelism: Provide a brief but meaningful explanation of concurrency vs. parallelism.
210. Introducing structured concurrency: Write an example highlighting the main issues of “unstructured” concurrency. Moreover, provide an introduction to the structured concurrency paradigm.
211. Introducing virtual threads: Explain and exemplify the main concepts of virtual threads.
212. Using the ExecutorService for virtual threads: Write several examples that highlight the task-per-thread model via ExecutorService and virtual threads.
213. Explaining how virtual threads work: Provide comprehensive coverage of how virtual threads work internally.
214. Hooking virtual threads and sync code: Explain...

209. Explaining concurrency vs. parallelism

Before tackling the main topic of this chapter, structured concurrency, let’s forget about structure, and let’s keep only concurrency. Next, let’s put concurrency against parallelism, since these two notions are often a source of confusion.

Both of them, concurrency and parallelism, use tasks as the main unit of work. However, the way that they handle these tasks makes them very different.

In the case of parallelism, a task is split into subtasks across multiple CPU cores. These subtasks are computed in parallel, and each of them represents a partial solution for the given task. By joining these partial solutions, we obtain the solution. Ideally, solving a task in parallel should result in less wall-clock time than in the case of solving the same task sequentially. In a nutshell, in parallelism, at least two threads run at the same time, which means that parallelism can solve a single task faster.

In the...

210. Introducing structured concurrency

If you are as old as I am, then most probably you started programming with a language such as BASIC or a similar unstructured programming language. At that time, an application was just a sequence of lines that defined a sequential logic/behavior via a bunch of GOTO statements, driving the flow by jumping like a kangaroo back and forward between the code lines. Well, in Java, the building blocks of a typical concurrent code are so primitive that the code looks somewhat like unstructured programming because it is hard to follow and understand. Moreover, a thread dump of a concurrent task doesn’t provide the needed answers.

Let’s follow a snippet of Java concurrent code and stop every time we have a question (always check the code below the question). The task is to concurrently load three testers by ID and team them up in a testing team. First, let’s list the server code (we will use this simple code to serve us in this...

211. Introducing virtual threads

Java allows us to write multithreaded applications via the java.lang.Thread class. These are classical Java threads that are basically just thin wrappers of OS (kernel) threads. As you’ll see, these classical Java threads are referred to as platform threads, and they have been available for quite a long time (since JDK 1.1, as the following diagram reveals):

Figure 10.4: JDK multithreading evolution

Next, let’s move on to JDK 19 virtual threads.

What’s the problem with platform (OS) threads?

OS threads are expensive in every single way, or more specifically, they are costly in terms of time and space. Creating OS threads is, therefore, a costly operation that requires a lot of stack space (around 20 megabytes) to store their context, Java call stacks, and additional resources. Moreover, the OS thread scheduler is responsible for scheduling Java threads, which is another costly operation that requires moving...

212. Using the ExecutorService for virtual threads

Virtual threads allow us to write more expressive and straightforward concurrent code. Thanks to the massive throughput obtained via virtual threads, we can easily adopt the task-per-thread model (for an HTTP server, this means a request per thread, for a database, this means a transaction per thread, and so on). In other words, we can assign a new virtual thread for each concurrent task.

Trying to use the task-per-thread model with platform threads will result in a throughput limited by the number of hardware cores – this is explained by Little’s law (https://en.wikipedia.org/wiki/Little%27s_law), L = λW, or throughput equals average concurrency multiplied by latency.

Whenever possible, it is recommended to avoid interacting with threads directly. JDK sustains this via the ExecutorService/Executor API. More precisely, we are used to submitting a task (Runnable/Callable) to an ExecutorService/Executor...

213. Explaining how virtual threads work

Now that we know how to create and start a virtual thread, let’s see how they actually work.

Let’s start with a meaningful diagram:

Figure 10.7: How virtual threads work

As you can see, Figure 10.7 is similar to Figure 10.6, except that we have added a few more elements.

First of all, notice that the platform threads run under a ForkJoinPool umbrella. This is a First-In-First-Out (FIFO) dedicated fork/join pool, dedicated to scheduling and orchestrating the relationships between virtual threads and platform threads (detailed coverage of Java’s fork/join framework is available in Java Coding Problems, First Edition, Chapter 11).

Important note

This dedicated ForkJoinPool is controlled by the JVM, and it acts as the virtual thread scheduler based on a FIFO queue. Its initial capacity (i.e., the number of threads) is equal to the number of available cores, and it can be increased to...

Join our book community on Discord

https://packt.link/vODyi

This chapter includes 24 problems covering a wide range of functional programming topics. We will start by introducing the JDK 16 mapMulti(), and continue with a handful of problems for working with predicates (Predicate), functions, and collectors.If you don't have a background in functional programming in Java then I strongly recommend postponing this chapter until you read Chapter 8 and Chapter 9 from Java Coding Problems, First Edition.At the end of this chapter, you'll be deeply skilled in functional style programming in Java.

Problems

Use the following problems to test your programming prowess in functional-style programming in Java. I strongly encourage you to give each problem a try before you turn to the solutions and download the example programs:

1. 178. Working with mapMulti(): Explain and exemplify the JDK 16, mapMulti(). Provide a brief introduction, explain how it works in comparison with flatMap(), and point out when mapMulti() is a good fit.
2. 179. Streaming custom code to map: Imagine a class that shapes some blog posts. Each post is identified by a unique integer id and it has several properties including its tags. The tags of each post are actually represented as a string of tags separated by hashtag (#). Whenever we need the list of tags for a given post, we can call the allTags() helper method. Our goal is to write a stream pipeline that extracts from this list of tags a Map<String, List<Integer>> containing for each tag (key) the list of posts (value).
3. 180. Exemplifying method reference...

196. Throwing checked exceptions from lambdas

Let's suppose that we have the following lambda:

static void readFiles(List<Path> paths) {
  paths.forEach(p -> {
    try {
      readFile(p);
    } catch (IOException e) {
      ... // what can we throw here?
    }
  });
}

What can we throw in the catch block? Everybody knows the answer … we can throw an unchecked exception such as RuntimeException:

static void readFiles(List<Path> paths) {
  paths.forEach(p -> {
    try {
      readFile(p);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  });
}

Also, everybody knows that we cannot throw a checked exception such as IOException. The following snippet of code will not compile:

static void readFiles(List<Path> paths) {
  paths.forEach(p -> {
    try {
      readFile(p);
    } catch (IOException e) {
      throw new IOException(e);
    }
  });
}

Can we change this rule? Can we come up with a hack that allows us to throw checked...

Summary

This chapter covered 24 problems. Most of them were focused on working with predicates, functions, and collectors, but we also covered JDK 16 mapMulti(), refactoring imperative code to functional, and much more.

218. Hooking task state

Starting with JDK 19, we can rely on Future.state(). This method computes the state of a Future based on the well-known get(), isDone(), and isCancelled(), returning a Future.State enum entry, as follows:

• CANCELLED – the task was canceled.
• FAILED – the task was completed exceptionally (with an exception).
• RUNNING – the task is still running (has not been completed).
• SUCCESS – the task was completed normally with a result (no exception).

In the following snippet of code, we analyze the state of loading the testing team members and act accordingly:

public static TestingTeam buildTestingTeam() 
       throws InterruptedException {
  List<String> testers = new ArrayList<>();
  try (ExecutorService executor 
      = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<String>> futures = executor.invokeAll(
      List.of(() -> fetchTester(Integer.MAX_VALUE)...

219. Combining newVirtualThreadPerTaskExecutor() and streams

Streams and newVirtualThreadPerTaskExecutor() is a handy combination. Here is an example that relies on IntStream to submit 10 simple tasks and collect the returned List of Future instances:

try (ExecutorService executor 
      = Executors.newVirtualThreadPerTaskExecutor()) {
  List<Future<String>> futures = IntStream.range(0, 10)
    .mapToObj(i -> executor.submit(() -> {
       return Thread.currentThread().toString() 
         + "(" + i + ")";
  })).collect(toList());
  // here we have the following snippet of code
}

Next, we wait for each Future to complete by calling the get() method:

  futures.forEach(f -> {
    try {
      logger.info(f.get());
    } catch (InterruptedException | ExecutionException ex) {
      // handle exception
    }
  });

Moreover, using stream pipelines is quite useful in combination with invokeAll(). For instance, the following stream...

220. Introducing a scope object (StructuredTaskScope)

So far, we have covered a bunch of problems that use virtual threads directly or indirectly via an ExecutorService. We already know that virtual threads are cheap to create and block and that an application can run millions of them. We don’t need to reuse them, pool them, or do any fancy stuff. Use and throw is the proper and recommended way to deal with virtual threads. This means that virtual threads are very useful for expressing and writing asynchronous code, which is commonly based on a lot of threads that are capable of blocking/unblocking several times in a short period. On the other hand, we know that OS threads are expensive to create, very expensive to block, and are not easy to put into an asynchronous context.

Before virtual threads (so for many, many years), we had to manage the life cycle of OS threads via an ExecutorService/Executor, and we could write asynchronous (or reactive) code via callbacks (you...

221. Introducing ShutdownOnSuccess

In the previous problem, we introduced StructuredTaskScope and used it to solve a task via a single virtual thread (a single Subtask). Basically, we fetched the tester with ID 1 from our server (we had to wait until this one was available). Next, let’s assume that we still need a single tester, but not mandatorily the one with ID 1. This time, it could be any of IDs 1, 2, or 3. We simply take the first one that is available from these three, and we cancel the other two requests.

Especially for such scenarios, we have an extension of StructuredTaskScope called StructuredTaskScope.ShutdownOnSuccess. This scope is capable of returning the result of the first task that completes successfully and interrupts the rest of the threads. It follows the “invoke any” model and can be used as follows:

public static TestingTeam buildTestingTeam() 
       throws InterruptedException, ExecutionException {
  try (ShutdownOnSuccess scope...

222. Introducing ShutdownOnFailure

As its name suggests, StructuredTaskScope.ShutdownOnFailure is capable of returning the exception of the first subtask that completes exceptionally and interrupts the rest of the subtasks (threads). For instance, we may want to fetch the testers with IDs 1, 2, and 3. Since we need exactly these three testers, we want to be informed if any of them are not available and, if so, cancel everything (i.e., the remaining threads). The code looks as follows:

public static TestingTeam buildTestingTeam() 
       throws InterruptedException, ExecutionException {
  try (ShutdownOnFailure scope 
      = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<String> subtask1 
      = scope.fork(() -> fetchTester(1));
    Subtask<String> subtask2 
      = scope.fork(() -> fetchTester(2));
    Subtask<String> subtask3 
      = scope.fork(() -> fetchTester(Integer.MAX_VALUE));
    scope.join();
    logger.info(() -> "Subtask...

223. Combining StructuredTaskScope and streams

If you prefer functional programming, then you’ll be happy to see that streams can be used with StructuredTaskScope as well. For instance, here we rewrite the application from Problem 221, using a stream pipeline to fork our tasks:

public static TestingTeam buildTestingTeam() 
       throws InterruptedException, ExecutionException {
  try (ShutdownOnSuccess scope 
      = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
    Stream.of(1, 2, 3)
      .<Callable<String>>map(id -> () -> fetchTester(id))
      .forEach(scope::fork);
    scope.join();
    String result = (String) scope.result();
    logger.info(result);
    return new TestingTeam(result);
  }
}

Moreover, we can use stream pipelines to collect results and exceptions, as follows:

public static TestingTeam buildTestingTeam() 
    throws InterruptedException, ExecutionException {
 try (ShutdownOnSuccess scope 
  = new StructuredTaskScope...

224. Observing and monitoring virtual threads

Observing and monitoring virtual threads can be done in several ways. First, we can use Java Flight Recorder (JFR) – we introduced this tool in Chapter 6, Problem 143.

Using JFR

Among its reach list of events, JFR can monitor and record the following events related to virtual threads:

• jdk.VirtualThreadStart – this event is recorded when a virtual thread starts (by default, it is disabled)
• jdk.VirtualThreadEnd – this event is recorded when a virtual thread ends (by default, it is disabled)
• jdk.VirtualThreadPinned – this event is recorded when a virtual thread is parked while pinned (by default, it is enabled with a threshold of 20 ms)
• jdk.VirtualThreadSubmitFailed – this event is recorded if a virtual thread cannot be started or unparked (by default, it is enabled)

You can find all the JFR events at https://sap.github.io/SapMachine/jfrevents/.

We start...

Summary

This chapter covered 16 introductory problems about virtual threads and structured concurrency. You can see this chapter as preparation for the next one, which will cover more detailed aspects of these two topics.

Join our community on Discord

Join our community’s Discord space for discussions with the author and other readers:

https://discord.gg/8mgytp5DGQ