Consider the following parallel stream snippet using both a common and a custom FJP:
import java.time.Instant;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Stream;
public final class Main {
public static void main(String[] args) throws Exception {
log("using common FJP");
work();
log("using custom FJP");
ForkJoinPool forkJoinPool = new ForkJoinPool(4);
try {
forkJoinPool.submit(() -> work()).get();
} finally {
forkJoinPool.shutdown();
}
}
private static void work() {
Stream
.of(0, 1, 2, 3)
.parallel()
.map(index -> {
log("# %d started", index);
LockSupport.parkNanos(500_000_000L);
log("# %d stopped", index);
return index;
})
.reduce(0, Integer::sum);
}
private static void log(String format, Object... args) {
System.out.format(Instant.now() + " [" + Thread.currentThread().getName() + "] " + format + "%n", args);
}
}
This produces the following interesting output:
2022-08-23T19:12:51.173532Z [main] using common FJP
2022-08-23T19:12:51.300260Z [main] # 2 started
2022-08-23T19:12:51.301861Z [ForkJoinPool.commonPool-worker-3] # 1 started
2022-08-23T19:12:51.801748Z [main] # 2 stopped
2022-08-23T19:12:51.802492Z [main] # 3 started
2022-08-23T19:12:51.802570Z [ForkJoinPool.commonPool-worker-3] # 1 stopped
2022-08-23T19:12:51.803850Z [ForkJoinPool.commonPool-worker-3] # 0 started
2022-08-23T19:12:52.303566Z [main] # 3 stopped
2022-08-23T19:12:52.304333Z [ForkJoinPool.commonPool-worker-3] # 0 stopped
2022-08-23T19:12:52.304684Z [main] using custom FJP
2022-08-23T19:12:52.307470Z [ForkJoinPool-1-worker-3] # 2 started
2022-08-23T19:12:52.308441Z [ForkJoinPool-1-worker-7] # 3 started
2022-08-23T19:12:52.309039Z [ForkJoinPool-1-worker-5] # 1 started
2022-08-23T19:12:52.309276Z [ForkJoinPool-1-worker-1] # 0 started
2022-08-23T19:12:52.808504Z [ForkJoinPool-1-worker-3] # 2 stopped
2022-08-23T19:12:52.809272Z [ForkJoinPool-1-worker-7] # 3 stopped
2022-08-23T19:12:52.809541Z [ForkJoinPool-1-worker-5] # 1 stopped
2022-08-23T19:12:52.810086Z [ForkJoinPool-1-worker-1] # 0 stopped
While using the common FJP, why do tasks #2 and #3 execute on
mainrather than an FJP thread? Ismainconsidered a worker of the common FJP? If so, where is this documented?Given it is only a
Runnablesubmitted to the custom FJP and that doesn't expose any details to the FJP that it is using aStream, how doesStream#parallel()determine which FJP to use? (Documentation pointers are appreciated.)