[JavaSpecialists] Issue 338 - Adjusting Number of Carrier Threads Dynamically
[338] Adjusting Number of Carrier Threads Dynamically
Abstract:
Java 24 added the VirtualThreadSchedulerMXBean, which gives us insight into the pool that is executing our virtual threads. We can even change the number of carrier threads at runtime. In this newsletter we discover how this can get us out of a pickle.
Welcome to the 338th edition of The Java(tm) Specialists' Newsletter, typed by hand on my trusty MacBookPro M2 Max. In years past, I always dreamed of the latest and greatest MacBookPro, expecting that it would be so much faster than before. And every time I eventually took the plunge to upgrade, I was left disappointed. That envy is gone. I am perfectly content with my M2 Max with 96 GB of RAM and an 8TB SSD. Sure, the M5 Max might be faster for some tasks, but a lot of my waiting nowadays is anyway on external services like LLMs and agents. My M2 Max is definitely fast enough for that task!
We had an interesting two months here on Crete. In the past, we could count on the weather being mild until the middle of October. However, this year, the storms rolled in three weeks early. You might have seen it on the news and I received several concerned WhatsApp messages from friends around the world.
javaspecialists.teachable.com: Please visit our new self-study course catalog to see how you can upskill your Java knowledge.
Adjusting Number of Carrier Threads Dynamically
One of the challenges of my recent agentic coding is that I don't spot as many Java peculiarities. Yes, I still look at the code, but not with as much care as before. I use agents to churn out boring SpringBoot code for apps that make my (real) life better. One of my best decisions ever was to totally ignore many of the popular Java frameworks out there, and instead to focus on the code that is truly important to companies that want to squeeze the last ounce of performance out of Java.
However, the other day I noticed the VirtualThreadSchedulerMXBean interface whilst hacking away at a new exercise for my revised Mastering Virtual Threads in Java Course. Even though agents should make course material construction easier, I've found this not to be the case. The secret to a great programming course is not how much material you include, but rather, what you choose to leave out. Less is more. And LLMs struggle with that.
The new MXBean has been in Java since Java 24 and has the following properties:
public interface VirtualThreadSchedulerMXBean {
int getParallelism();
void setParallelism(int size);
int getPoolSize();
int getMountedVirtualThreadCount();
long getQueuedVirtualThreadCount();
}The method that piqued my interest was setParallelism(). Could this be used to change the number of carrier threads dynamically? And would we ever want to do this?
To illustrate, please have a look at VirtualThreadCarrierThreadNumber. It uses countCarriers() to estimate the number of carrier threads based on the toString() output of 10000 virtual threads. We can do this using a concurrent gatherer.
import jdk.management.VirtualThreadSchedulerMXBean;
import java.lang.management.ManagementFactory;
import java.time.Duration;
import java.util.stream.Gatherers;
import java.util.stream.IntStream;
public class VirtualThreadCarrierThreadNumber {
public static void main(String... args)
throws InterruptedException {
// since Java 24
var vt = ManagementFactory.getPlatformMXBean(
VirtualThreadSchedulerMXBean.class);
System.out.printf("%s: %d%n", vt, countCarriers());
vt.setParallelism(24);
System.out.printf("%s: %d%n", vt, countCarriers());
vt.setParallelism(1);
System.out.printf("%s: %d%n", vt, countCarriers());
System.out.println("Sleeping for 62 seconds");
for (int i = 0; i < 62; i++) {
if ((i + 1) % 10 == 0) System.out.print('#');
else System.out.print(".");
Thread.sleep(Duration.ofSeconds(1));
}
System.out.println();
System.out.printf("%s: %d%n", vt, countCarriers());
try {
vt.setParallelism(0); // parallelism must be at least 1
} catch (IllegalArgumentException e) {
System.out.println(e); // expected
}
}
private static int countCarriers() {
return IntStream.range(0, 100_000)
.boxed()
.gather(Gatherers.mapConcurrent(10_000,
i -> Thread.currentThread()
.toString()
.replaceAll(".*@", "")))
.distinct()
.toList()
.size();
}
}In the output we see that the number of carrier threads does not immediately decrease once we set it to 1. Instead, there is a 60 second timeout before they are stopped. Whilst not serious, this could be considered a bug. Shouldn't the carrier thread pool have a timeout of 0 for its threads if they exceed parallelism? Here is the output:
[parallelism=4, size=0, mounted=0, queued=0]: 4 [parallelism=24, size=4, mounted=1, queued=1]: 24 [parallelism=1, size=24, mounted=0, queued=0]: 1 Sleeping for 62 seconds .........#.........#.........#.........#.........#.........#.. [parallelism=1, size=0, mounted=0, queued=0]: 1 java.lang.IllegalArgumentException
So the question is - "is this useful?" I asked ChatGPT what a good use case for setParallelism() and after much prompting it came up with 40 different scenarios. But it didn't think of what I will describe here.
Java has had the ability to increase the number of threads in a ForkJoinPool since Java 7. The first class to do this was the Phaser synchronizer. It used a ManagedBlocker to give hints to the ForkJoinPool when additional threads should be spun up to keep the parallelism high. This mechanism was used for parallel streams, and now, for the carrier thread pool. For example, if a virtual thread pins a carrier thread, some mechanisms would start additional carrier threads. See for example Object.wait() in Java 21.
Another place that used to employ the ManagedBlocker to keep parallelism high is file IO. When we read a large or slow file (e.g. on an NFS) and a carrier thread got pinned, it would start up new threads. However, with disks being so fast, they decided to remove this functionality in Java 23. We can read in JDK-8329593: "1. Drop the begin/end compensate around ops on files opened for buffered I/O. Sergery Kuksenko analysis have demonstrated that there isn't any benefit or there is negative benefit, esp. when reading/writing a small number of bytes."
We would hardly ever notice this change, except on systems that happen to read from a large slow file and where we only have a single carrier thread. One of my readers had their JVM frequently kicked out of their cluster for blocking longer than allowed. We can see this behaviour in the following demo. Please run it with -XX:ActiveProcessorCount=1 to simulate a single core machine. It also needs to run on a Unix type of system that has access to the mkfifo tool.
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import static java.util.concurrent.Executors.*;
/**
* In Java 21, blocking file I/O in java.io (e.g.
* FileInputStream.read()) was wrapped in
* jdk.internal.misc.Blocker.begin()/end(). On a virtual
* thread, this could increase the number of carrier threads
* to keep the parallelism high. Since Java 23, this
* compensation was removed from java.io. With only one
* carrier thread, a virtual thread stuck in a file read now
* freezes all the other virtual threads, potentially
* getting our JVM kicked out of clusters for being
* unresponsive.
*
* To make "slow file" deterministic, we read from a named
* pipe (FIFO), which blocks in the read() system call until
* a writer sends data. A file on a slow NFS share or a
* stalled disk behaves the same way.
*
* Run with -XX:ActiveProcessorCount=1 to simulate a single core
* machine
* Run with Java 21/22: the ticker keeps ticking whilst the
* reader is blocked.
* Run with Java 23+: the ticker stops until the read
* completes.
* Only works on Unix-like systems (needs mkfifo).
*/
public class VirtualThreadFileBlockingDemo {
private static final long BLOCK_MS = 3000;
private static final long TICKER_PERIOD = 250;
public static void main(String... args)
throws Exception {
if (Runtime.getRuntime().availableProcessors() != 1) {
System.err.println("Please use JVM flag " +
"-XX:ActiveProcessorCount=1");
System.exit(1);
}
System.out.printf("Java Version: %s, cores: %d%n",
Runtime.getRuntime().version(),
Runtime.getRuntime().availableProcessors());
var dir = Files.createTempDirectory("vtfifo");
var fifo = dir.resolve("slowfile");
new ProcessBuilder("mkfifo", fifo.toString())
.inheritIO().start().waitFor();
try {
run(fifo);
} finally {
Files.deleteIfExists(fifo);
Files.deleteIfExists(dir);
}
}
private static void run(Path fifo) throws Exception {
var start = System.nanoTime();
var readerBlocked = new AtomicBoolean(false);
var ticksWhileBlocked = new AtomicInteger();
try (var ticker = newSingleThreadScheduledExecutor(
Thread.ofVirtual().name("ticker").factory())
) {
ticker.scheduleAtFixedRate(() -> {
System.out.printf("%5dms tick on %s%n",
ms(start), Thread.currentThread());
if (readerBlocked.get())
ticksWhileBlocked.incrementAndGet();
}, 0, TICKER_PERIOD, TimeUnit.MILLISECONDS);
Thread.sleep(600); // let the ticker get going
var reader = Thread.startVirtualThread(() -> {
System.out.printf("%5dms reader opening " +
"slow file%n", ms(start));
readerBlocked.set(true);
try (var in = new FileInputStream(
fifo.toFile())) {
var buf = new byte[8192];
// carrier thread is stuck here
var n = in.read(buf);
readerBlocked.set(false);
System.out.printf(
"%5dms reader got %d bytes%n",
ms(start), n);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
});
// A platform thread acts as the slow "disk",
// supplying data later
Thread.ofPlatform().name("slow-disk").start(() -> {
try (var out = new FileOutputStream(
fifo.toFile())) {
Thread.sleep(BLOCK_MS);
out.write(new byte[1024]);
} catch (IOException e) {
throw new UncheckedIOException(e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
reader.join();
Thread.sleep(600);
}
var expected = BLOCK_MS / TICKER_PERIOD;
System.out.printf(
"Ticks whilst reader was blocked: %d " +
"(expected ~%d)%n",
ticksWhileBlocked.get(), expected);
System.out.println(
ticksWhileBlocked.get() >= expected / 2
? "OK: carrier was compensated, other " +
"virtual threads kept running"
: "STARVED: the single carrier was stuck " +
"in file I/O");
}
private static long ms(long start) {
return TimeUnit.NANOSECONDS.toMillis(
System.nanoTime() - start);
}
}Here is the output for Java 21 LTS:
Java Version: 21.0.12.1+12-LTS, cores: 1
3ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
256ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
506ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
609ms reader opening slow file
756ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
1003ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
1252ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
1503ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
1752ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
2006ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-3
2254ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-3
2506ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-3
2755ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-3
3007ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-3
3256ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-3
3506ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
3616ms reader got 1024 bytes
3757ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
4006ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
Ticks whilst reader was blocked: 12 (expected ~12)
OK: carrier was compensated, other virtual threads kept runningHere is the output for Java 23 (and it has been the same since then):
Java Version: 23.0.2+7-FR, cores: 1
3ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
257ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
507ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-1
609ms reader opening slow file
3616ms reader got 1024 bytes
3616ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3617ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3617ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3617ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3617ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3618ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3618ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3618ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3618ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3618ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3618ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3619ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
3755ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
4007ms tick on VirtualThread[#14,ticker]/runnable@ForkJoinPool-1-worker-2
Ticks whilst reader was blocked: 0 (expected ~12)
STARVED: the single carrier was stuck in file I/OMy recommendation has been for a while to always have at least two carrier threads, even with a single-core machine. The operating system is good at scheduling the multiple threads and we can thus avoid starvation. Typically we do that by starting the JVM with the system property -Djdk.virtualThreadScheduler.parallelism=2. That works fine, but the MXBean would allow us to do that inside code. Unfortunately the MXBean is only available since Java 24, so we will need to use reflection to make our code backwards compatible. Also, the trick will only work for Java 24 onwards. I think that is OK, because no one would be running Java 23 in production anyway. It would be either Java 21, 25 or the latest release. Here is how we could set it at runtime:
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.lang.management.ManagementFactory;
import java.lang.management.PlatformManagedObject;
public class SaferVirtualThreadFileBlockingDemo {
public static void main(String... args)
throws Exception {
setParallelismToAtLeast2();
VirtualThreadFileBlockingDemo.main(args);
}
private static void setParallelismToAtLeast2()
throws ReflectiveOperationException {
try {
// Using reflection so that it compiles and runs
// in pre Java 24 versions
var mxBeanIntf = Class.forName(
"jdk.management." +
"VirtualThreadSchedulerMXBean")
.asSubclass(PlatformManagedObject.class);
var vt = ManagementFactory.getPlatformMXBean(mxBeanIntf);
var getParallelism = MethodHandles.lookup().findVirtual(
mxBeanIntf, "getParallelism",
MethodType.methodType(int.class)
);
var setParallelism = MethodHandles.lookup().findVirtual(
mxBeanIntf, "setParallelismToAtLeast2",
MethodType.methodType(void.class,
int.class)
);
try {
if ((int) getParallelism.invoke(vt) == 1) {
setParallelism.invoke(vt, 2);
}
} catch (RuntimeException | Error e) {
throw e;
} catch (Throwable e) {
throw new AssertionError(e);
}
} catch (ClassNotFoundException ignore) { }
}
}Now our Java 25 output is the same as Java 21. Of course, it can happen that multiple virtual threads get stuck on long buffered file IO operations, but in that case we might have a different problem to solve and perhaps platform threads would be a better solution.
Kind regards
Heinz
P.S. Of the 40 "use cases" that ChatGPT suggested for the setParallelism() method, there was one good one. Say we have a system that is hung up because of a livelock involving our virtual threads. We can connect via jconsole, head over to the MXBean and increase the parallelism temporarily. This might break out of the livelock and let the system continue, whilst we come up with a good fix.
