-
Notifications
You must be signed in to change notification settings - Fork 0
Virtual Threads
Virtual Threads
are lightweight threads supporting high-throughput concurrency.
Most changes related to virtual threads in JDK 21 are invisible at an API level
because virtual threads are instances of the familiar Thread class.
To illustrate the changes, we will look at several examples of concurrent programs.
Virtual threads are especially beneficial in the presence of blocking operations like sleep.
To compare different ways to execute tasks concurrently,
we use them through a common interface
ExecutorService
which has implementations for different execution models such as pooling threads or creating new threads for each submitted task.
Sleep Sort is a tongue-in-cheek sorting algorithm that uses concurrent tasks to print each sorted item after waiting some time proportional to the item's size. The idea is to start all concurrent tasks simultaneously so they print their corresponding items in the correct order.
Here is a Java implementation of the Sleep Sort algorithm.
public record SleepSort(ExecutorService executor) {
public static void main(String[] args) {
new SleepSort().sortInputs();
}
private SleepSort() {
this(Executors.newThreadPerTaskExecutor(Thread::new));
}
private void sortInputs() {
try (BufferedReader stdin = new BufferedReader(new InputStreamReader(System.in))) {
stdin.lines().map(Integer::parseInt).forEach(num -> {
executor.submit(() -> {
TimeUnit.SECONDS.sleep(num);
System.out.println(num);
return null;
});
});
executor.shutdown();
executor.awaitTermination(1, TimeUnit.MINUTES);
} catch (IOException | InterruptedException e) {
executor.shutdownNow();
}
}
}We use Executors.newThreadPerTaskExecutor to create a new platform thread for every input number.
In Java, each platform thread is mapped to an operating system thread.
So, we create as many operating system threads as there are input numbers.
For a small number of inputs, this approach works fine.
The following shell command generates ten random numbers between 1 and 10
and then sorts them using our program.
me@java-dev:~/java21-demo$ shuf -r -i 1-10 -n 10 | java -cp target/classes sebfisch.SleepSort
1
2
3
3
5
5
5
8
9
10When sorting a larger number of inputs, however, the time for creating as many operating system threads (for example 10000) seems to interfere with their sleeping time. As a result, numbers are not printed in sorted order.
me@java-dev:~/java21-demo$ time shuf -r -i 1-10 -n 10000 | java -cp target/classes sebfisch.SleepSort
[skipped some output]
9
6
9
8
9
8
7
8
8
8
8
7
6
9
[skipped some output]
real 0m13.356s
user 0m4.547s
sys 0m3.750sWe also print timing information for this run so we can compare it with that of a later implementation.
It is common practice to limit the number of created operating system threads
by creating a pool of threads where each thread is used to execute multiple tasks.
We can change a single line in the constructor of SleepSort to use this execution model with a fixed number of threads.
private SleepSort() {
- this(Executors.newThreadPerTaskExecutor(Thread::new));
+ this(Executors.newFixedThreadPool(10));
}This approach works fine for Sleep Sort as long as the number of sorted inputs does not exceed the number of threads. We can still successfully sort ten random numbers using ten threads as follows.
me@java-dev:~/java21-demo$ shuf -r -i 1-10 -n 10 | java -cp target/classes sebfisch.SleepSort
1
3
3
4
4
4
5
6
6
8However, a thread pool is typically created to use a small number of threads to execute a much larger number of tasks. This approach is problematic when the executed tasks contain blocking operations, like in Sleep Sort. Even twenty random numbers are sorted incorrectly when using a thread pool of size ten.
me@java-dev:~/java21-demo$ shuf -r -i 1-10 -n 20 | java -cp target/classes sebfisch.SleepSort
1
2
2
3
5
6
7
9
7
9
9
5
4
2
2
8
10
10
4
9The first few numbers in the output are sorted correctly. But towards the end of the output, smaller numbers appear between larger ones. While the corresponding tasks have waited the appropriate time, they have been started much too late because the number of threads is insufficient to start all tasks at the same time. Half of the tasks are started later because all threads are blocked when executing the first ten tasks.
In general, using thread pools with tasks that use blocking operations can significantly reduce the throughput of an application. Blocked threads are not available to execute other tasks that would not be blocked.
We can change our program again to create one virtual thread per sorted input.
private SleepSort() {
- this(Executors.newFixedThreadPool(10));
+ this(Executors.newVirtualThreadPerTaskExecutor());
}This implementation is very similar to the first one. The only difference is that each task is executed using a new virtual thread instead of a new platform thread. Let's try using the program to sort 10K random numbers again.
me@java-dev:~/java21-demo$ time shuf -r -i 1-10 -n 10000 | java -cp target/classes sebfisch.SleepSort
[skipped output]
real 0m10.139s
user 0m3.097s
sys 0m1.458sAll numbers are printed in the correct order now, so we do not show the output here. The footprint of 10K virtual threads is small enough to start all of them roughly at the same time. The "real" time is now around the expected ten seconds required for waiting for the largest number. Even 100K numbers can be sorted with this program correctly and the "real" time is still not much more than ten seconds.
me@java-dev:~/java21-demo$ time shuf -r -i 1-10 -n 100000 | java -cp target/classes sebfisch.SleepSort
[skipped output]
real 0m10.670s
user 0m15.926s
sys 0m12.898sWe can use Java's built-in Flight Recorder to collect more detailed information about the execution of our Sleep-Sort program.
me@java-dev:~/java21-demo$ shuf -r -i 1-10 -n 10000 | java -cp target/classes -XX:StartFlightRecording:filename=SleepSort.jfr sebfisch.SleepSort
[skipped output]We can then use
JDK Mission Control
(JMC) to open the generated file SleepSort.jfr and get a detailed overview of the program's execution.
JMC has a Threads section where virtual threads are grouped into a specific thread group.
An important difference between platform threads and virtual threads is how they are mapped to operating system threads. While each new platform thread is executed on a new operating system thread, a large number of virtual threads are executed on a usually much smaller pool of operating system threads, so-called carrier threads.
The other important difference is that blocked virtual threads do not block the operating system thread they are executed on. Instead, virtual threads are mounted on their carrier thread to execute them, and unmounted when they block, to free the carrier thread for the execution of another virtual thread. When a virtual thread is unblocked, it is eventually remounted on a (potentially different) carrier thread to resume execution.
Essentially, blocking operations on virtual threads are implemented with corresponding non-blocking operations, and the Java VM implements asynchronous execution of virtual threads in a way that looks like synchronous execution with blocking operations to application programmers. As a consequence, programmers can use this familiar programming model with a much larger number of threads than was previously practical in Java.
Server programs are typical examples where using a new thread for each request or connection is often conceptually the simplest choice. They also make frequent use of blocking operations for network communication which have the potential to negatively affect application throughput if blocked threads delay the execution of other threads. So, as our next example, we implement and experiment with a TCP server that echoes every received input back to the client that sent it.
public record Server(ServerSocket socket, ExecutorService executor) implements Closeable {
public static void main(String[] args) throws IOException {
try (Server server = new Server()) {
server.start();
}
}
private Server() throws IOException {
this(new ServerSocket(0), Executors.newVirtualThreadPerTaskExecutor());
}
public Server {
System.out.println("listening on port %s".formatted(socket.getLocalPort()));
}
@Override
public void close() throws IOException {
socket.close();
executor.shutdown();
}
private void start() throws IOException {
while (!socket.isClosed()) serve(socket.accept());
}
private void serve(Socket client) {
executor.submit(() -> {
try (BufferedReader reader = new BufferedReader(new InputStreamReader(client.getInputStream()));
PrintWriter writer = new PrintWriter(client.getOutputStream(), true)) {
reader.lines().forEach(writer::println);
} finally {
client.close();
}
return null;
});
}
}We use Executors.newVirtualThreadPerTaskExecutor() to create a new virtual thread for every accepted client connection.
For each of those connections, the server sends back everything it receives using reader.lines().forEach(writer::println).
When we start our server it connects to a random free local port and prints it.
me@java-dev:~/java21-demo$ java -cp target/classes sebfisch.echo.Server
listening on port 45173We can then use Netcat to connect to the server in a different terminal as follows.
me@java-dev:~/java21-demo$ nc localhost 45173
hello
hello
world
world
^CThe server accepts an arbitrary number of concurrent connections.
Instead of creating all those connections by hand, we can use the shell script
echo_clients.sh
to create multiple connections each sending a given number of random messages.
me@java-dev:~/java21-demo$ ./echo_clients.sh
Usage: ./echo_clients.sh <server_port> <number_of_clients> <number_of_messages>Each client waits up to two seconds between sending messages, so we can expect the script to run for up to twenty seconds when starting a single client sending ten messages.
me@java-dev:~/java21-demo$ time ./echo_clients.sh 45173 1 10
real 0m16.038s
user 0m0.075s
sys 0m0.007sEven with ten or a hundred clients, the total run time does not increase significantly.
me@java-dev:~/java21-demo$ time ./echo_clients.sh 45173 10 10
real 0m17.232s
user 0m0.752s
sys 0m0.092s
me@java-dev:~/java21-demo$ time ./echo_clients.sh 45173 100 10
real 0m17.812s
user 0m8.099s
sys 0m1.598sWith 1K and 10K clients, "sys" time gets larger than "real" time, suggesting significant overhead due to a large number of network connections and/or operating system processes.
me@java-dev:~/java21-demo$ time ./echo_clients.sh 45173 1000 10
real 0m22.405s
user 1m48.631s
sys 0m42.325s
me@java-dev:~/java21-demo$ time ./echo_clients.sh 45173 10000 10
real 1m42.008s
user 17m34.711s
sys 4m38.797sThe shell script creates a new operating system process for each client. For a stress test of our server, we might need to re-implement the script in Java.
Write a command line Java program that can be used for testing the echo server. The program should accept four command-line arguments:
- the host to connect to
- the port to connect to
- the number of clients connecting to the given host and port
- the number of messages sent by each client before disconnecting Each client should wait up to one second before sending the first message and up to two seconds before sending subsequent messages. Sent messages should contain a timestamp as well as numbers counting the client and its message. After sending a message, the client should verify that the server sent the same message back.