The performance difference you’re observing comes from how the ForkJoinPool.commonPool() handles nested parallel streams. When both the outer and inner streams use parallelStream() without a custom thread pool, they compete for the same limited set of threads in the common pool. This leads to thread contention and under-utilization of CPU resources because the pool cannot effectively manage the nested parallelism.

By creating a new ForkJoinPool for each inner stream, you provide dedicated threads for inner tasks, avoiding contention with the outer stream’s threads. This allows both levels of parallelism to utilize the CPU cores fully. Obviously as you’ve noticed it results in a substantial performance boost despite the overhead of creating new pools.

Possible Improvements:

You can avoid the overhead of creating multiple pools by using a shared custom ForkJoinPool for all inner streams. This approach eliminates the pool creation overhead while still providing separate threads for inner parallelism, leading to even better performance.

public class NestedPerf {
  private static final ForkJoinPool innerPool = new ForkJoinPool();

  @Benchmark
  public void testingSharedInnerPool(Blackhole bh){
    StateData.outerLoop.parallelStream().unordered().forEach(i -> {
      innerPool.submit(() -> StateData.innerLoop.parallelStream().unordered().forEach(j -> {
        try {
          Thread.sleep(5);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
      })).join();
      bh.consume(i);
    });
  }
}

Alternatively, you could flatten the parallelism to a single level by combining the outer and inner loops into one parallel stream. This method can effectively utilize the parallel stream without nested parallelism, often resulting in the fastest execution time because it maximizes CPU utilization and minimizes overhead.

@Benchmark
public void testingFlattenedStream(Blackhole bh){
  StateData.outerLoop.parallelStream()
    .flatMap(i -> StateData.innerLoop.stream())
    .unordered()
    .forEach(j -> {
      try {
        Thread.sleep(5);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
      bh.consume(j);
    });
}

If you’re going for simplicity and speed then flattening is likely going to win out.

Why Throughput Increases

Your tasks are simply a call to Thread::sleep. That blocks the calling thread, which means the OS will not schedule the thread for execution until the specified duration elapses. This leaves the CPU free to execute any other threads. In other words, your tasks are not CPU-bound and thus do not burden the CPU. Which means throwing more threads at your set of tasks is going to increase throughput without overwhelming the CPU.

By using multiple fork-join pools, you are effectively increasing the number of threads available to execute your tasks. It’s not much different from simply increasing the number of threads in a single pool. Whether you have 1 pool with 15 threads or 3 pools with 5 threads each, you still end up with a total of 15 threads.

Let’s say you have 10 tasks that each sleep for 5 milliseconds. If you have 5 threads to execute those tasks, then you’ll roughly see:

Start 5 tasks => Wait 5 ms => Start 5 tasks => Wait 5 ms => Done!

But if you have 10 threads you’ll roughly see:

Start 10 tasks => Wait 5 ms => Done!

The first takes a total of 10 milliseconds to execute every task, the second only takes 5 milliseconds. And that’s basically where the increased throughput is coming from in your tests.

Maintaining Parallelism

All that said, a ForkJoinPool has a set level of parallelism. One way it tries to maintain this parallelism is by spawning a new thread (if the maximum number of threads hasn’t already been reached) when one of its threads is blocked. From the documentation:

[A ForkJoinPool] attempts to maintain enough active (or available) threads by dynamically adding, suspending, or resuming internal worker threads, even if some tasks are stalled waiting to join others. However, no such adjustments are guaranteed in the face of blocked I/O or other unmanaged synchronization. The nested ForkJoinPool.ManagedBlocker interface enables extension of the kinds of synchronization accommodated.

You’re calling Thread::sleep in an unmanaged way. In other words, you’re blocking the threads of the pool in such a way that the pool cannot compensate. To prevent that, consider using a ManagedBlocker. Here’s an example implementation:

import java.time.Duration;
import java.util.concurrent.ForkJoinPool;

public class SleepManagedBlocker implements ForkJoinPool.ManagedBlocker {

  private final Duration sleepDuration;
  private boolean slept; // Does this need to be volatile?

  public SleepManagedBlocker(Duration slepDuration) {
    this.sleepDuration = slepDuration;
  }

  @Override
  public boolean block() throws InterruptedException {
    if (!slept) {
      slept = true;
      Thread.sleep(sleepDuration);
    }
    return slept;
  }

  @Override
  public boolean isReleasable() {
    return slept;
  }
}

Then you would replace the Thread.sleep(5) calls with:

ForkJoinPool.managedBlock(new SleepManagedBlocker(Duration.ofMillis(5)))

You should see similar throughput increases in your tests without needing to using multiple fork-join pools.

JMH Benchmarks

Here is a benchmark showing the effect of using ManagedBlocker in this case. It was compiled and executed on Java 23.

import java.time.Duration;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Warmup;
import org.openjdk.jmh.infra.Blackhole;

@Fork(value = 1, jvmArgsAppend = {"-Djava.util.concurrent.ForkJoinPool.common.maximumSpares=1024"})
@Warmup(iterations = 5)
@Measurement(iterations = 5)
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
public class FJPBenchmarks {

  @Benchmark
  public void runTest(TestState state, Blackhole bh) {
    state.executeOuterLoop(bh);
  }

  @State(Scope.Benchmark)
  public static class TestState {

    private static final Duration SLEEP_DURATION = Duration.ofMillis(5);
    private static final int OUTER_LOOP_COUNT = 32;
    private static final int INNER_LOOP_COUNT = 32;

    @Param({"sequential", "parallel"})
    private String sequentialMode;

    @Param({"common", "separate"})
    private String poolMode;

    @Param({"raw", "managed"})
    private String sleepMode;

    void executeOuterLoop(Blackhole bh) {
      IntStream.range(0, OUTER_LOOP_COUNT)
          .unordered()
          .parallel()
          .forEach(i -> {
            executeInnerLoop(createInnerLoop());
            bh.consume(i);
          });
    }

    IntStream createInnerLoop() {
      var stream = IntStream.range(0, INNER_LOOP_COUNT).unordered();
      return switch (sequentialMode) {
        case "sequential" -> stream.sequential();
        case "parallel" -> stream.parallel();
        default -> throw new IllegalStateException("bad sequentialMode: " + sequentialMode);
      };
    }

    void executeInnerLoop(IntStream loop) {
      var sleeper = getSleeper();
      switch (poolMode) {
        case "common" -> loop.forEach(_ -> sleeper.sleepUnchecked());
        case "separate" -> {
          try (var pool = new ForkJoinPool()) {
            loop.forEach(_ -> pool.submit(sleeper::sleepUnchecked).join());
          }
        }
        default -> throw new IllegalStateException("bad poolMode: " + poolMode);
      }
    }

    Sleeper getSleeper() {
      return switch (sleepMode) {
        case "raw" -> () -> Thread.sleep(SLEEP_DURATION);
        case "managed" -> () -> ForkJoinPool.managedBlock(new SleepManagedBlocker());
        default -> throw new IllegalStateException("bad sleepMode: " + sleepMode);
      };
    }

    @FunctionalInterface
    interface Sleeper {
  
      void sleep() throws InterruptedException;

      default Void sleepUnchecked() {
        try {
          sleep();
        } catch (InterruptedException ex) {
          throw new RuntimeException(ex);
        }
        return null;
      }
    }

    static class SleepManagedBlocker implements ForkJoinPool.ManagedBlocker {

      private boolean slept;

      @Override
      public boolean block() throws InterruptedException {
        if (!slept) {
          slept = true;
          Thread.sleep(SLEEP_DURATION);
        }
        return true;
      }

      @Override
      public boolean isReleasable() {
        return slept;
      }
    }
  }
}

Results (from executing the benchmark on a computer with 8 processors):

Benchmark              (poolMode)  (sequentialMode)  (sleepMode)   Mode  Cnt   Score   Error  Units
FJPBenchmarks.runTest      common        sequential          raw  thrpt    5   1.463 � 0.022  ops/s
FJPBenchmarks.runTest      common        sequential      managed  thrpt    5   5.858 � 0.026  ops/s
FJPBenchmarks.runTest      common          parallel          raw  thrpt    5   1.454 � 0.044  ops/s
FJPBenchmarks.runTest      common          parallel      managed  thrpt    5  35.997 � 0.234  ops/s
FJPBenchmarks.runTest    separate        sequential          raw  thrpt    5   1.426 � 0.325  ops/s
FJPBenchmarks.runTest    separate        sequential      managed  thrpt    5   1.348 � 0.157  ops/s
FJPBenchmarks.runTest    separate          parallel          raw  thrpt    5  13.505 � 1.175  ops/s
FJPBenchmarks.runTest    separate          parallel      managed  thrpt    5  16.864 � 0.186  ops/s