Nested parallel streams in Java

Question

I want to understand the ordering constraints between nested streams in Java.

Example 1:

public static void main(String[] args) {
    IntStream.range(0, 10).forEach(i -> {
        System.out.println(i);
        IntStream.range(0, 10).forEach(j -> {
            System.out.println("    " + i + " " + j);
        });
    });
}

This code executes deterministically, so the inner loop runs forEach on each j before the outer loop runs its own forEach on the next i:

Example 2:

public static void main(String[] args) {
    IntStream.range(0, 10).parallel().forEach(i -> {
        System.out.println(i);
        IntStream.range(0, 10).parallel().forEach(j -> {
            System.out.println("    " + i + " " + j);
        });
    });
}

If the streams are made parallel() as in this second example, I could imagine the inner workers blocking as they wait for threads to become available in the outer work queue, since the outer work queue threads have to block on the completion of the inner stream, and the default thread pool only has a limited number of threads. However, deadlock does not appear to occur:

Both streams share the same default thread pool, yet they generate different work units. Each outer work unit can only complete after all inner units for that outer work unit have completed, since there is a completion barrier at the end of each parallel stream.

How is the coordination between these inner and outer streams managed across the shared pool of worker threads, without any sort of deadlock?

"Both streams": no, there are multiple pipelines here (1+10 at least). Why do you think there would be a need to coordinate between outer an inner stream tasks here? The only thing that can be deterministic, afaik, is that for any given i iteration, the corresponding j iterations will run before that i iteration completes, with no guaranteed order of any form (for i or j executions) — ernest_k
@ernest_k Because there is a limited number of workers in the default thread pool, and all those threads can end up blocking in the outer stream, waiting for each corresponding inner stream to complete. However, the inner streams can't complete if there are no available (idling/quiescent) worker threads in the default thread pool that can process their items -- this results in deadlock. See akuzminykh's response though -- apparently this possibility for deadlock is detected, and extra worker threads are created as needed. — Luke Hutchison

akuzminykh akuzminykh · Accepted Answer · 2020-07-01T06:11:16

The thread pool behind parallel streams is the common pool, which you can get with ForkJoinPool.commonPool(). It usually uses NumberOfProcessors - 1 workers. To resolve dependencies like you've described, it's able to dynamically create additional workers if (some) current workers are blocked and a deadlock becomes possible.

However, this is not the answer for your case.

Tasks in a ForkJoinPool have two important functionalities:

They can create subtasks and split the current task into smaller pieces (fork).
They can wait for the subtasks (join).

When a thread executes such a task A and joins a subtask B, it doesn't just wait blocking for the subtask to finish its execution but executes another task C in the meantime. When C is finished, the thread comes back to A and checks if B is finished. Note that B and C can (and most likely are) the same task. If B is finished, then A has successfully waited for/joined it (non-blocking!). Check out this guide if the previous explanation is not clear.

Now when you use a parallel stream, the range of the stream is split into tasks recursively until the tasks become so small that they can be executed sequentially more efficiently. Those tasks are put into a work queue (there is one for each worker) in the common pool. So, what IntStream.range(0, 100).parallel().forEach does is splitting up the range recursively until it's not worth it anymore. Each final task, or rather bunch of iterations, can be executed sequentially with the provided code in forEach. At this point the workers in the common pool can just execute those tasks until all are done and the stream can return. Note that the calling thread helps out with the execution by joining subtasks!

Now each of those tasks uses a parallel stream itself in your case. The procedure is the same; split it up into smaller tasks and put those tasks into a work queue in the common pool. From the ForkJoinPool's perspective those are just additional tasks on top of the already present ones. The workers just keep executing/joining tasks until all are done and the outer stream can return.

This is what you see in the output: There is no deterministic behaviour, no fixed order. Also there cannot occur a deadlock because in the given use case there won't be blocking threads.

You can check the explanation with the following code:

    public static void main(String[] args) {
        IntStream.range(0, 10).parallel().forEach(i -> {
            IntStream.range(0, 10).parallel().forEach(j -> {
                for (int x = 0; x < 1e6; x++) { Math.sqrt(Math.log(x)); }
                System.out.printf("%d %d %s\n", i, j, Thread.currentThread().getName());
                for (int x = 0; x < 1e6; x++) { Math.sqrt(Math.log(x)); }
            });
        });
    }

You should notice that the main thread is involved in the execution of the inner iterations, so it is not (!) blocked. The common pool workers just pick tasks one after another until all are finished.

Nested parallel streams in Java

1 Answers