Which executor does CompletableFuture.allOf use?

Question

Suppose we have two executors, 1 and 2.

We can configure which executor to use when doing

CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(()-> {return 1;}, executor1) //executor1
CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(()-> {return 2;}, executor1) //executor1
CompletableFuture<Integer> cf3 = CompletableFuture.supplyAsync(()-> {return 3;}, executor2) //executor2

but which thread executor uses CompletableFuture static method allOf?

CompletableFuture.allOf(cf1, cf2, cf3)

Thanks!

Answer 1

The answer of Ivan Gammel is not exact.

There is indeed no executor associated with the CompletableFuture returned by allOf(), as in fact, there isn't ever an executor associated with any CompletableFuture.

A task is associated with an executor, as it is running inside of it, but the association is inverse: the executor has a list of tasks to execute.

A task can also be associated with a CompletableFuture, which it will complete when the task finishes. The CompletableFuture itself does not keep a reference to the task or executor that were used for its creation. It may however keep references to tasks and optionally executors used in dependent stages.

The CompletableFuture returned by allOf() will be completed by a task, which is a dependant stage of the original CompletableFutures. In your example, this task can be executed by:

• executor1, if the third task finished first;
• executor2, if the 2 first tasks finished before the third one; or
• the original thread, if all tasks finished before you called allOf().

This can be seen by adding a dependent thenRun() stage to the allOf() call:

public class CompletableFutureAllOfCompletion {
    private ExecutorService executor1 = Executors.newFixedThreadPool(2);
    private ExecutorService executor2 = Executors.newFixedThreadPool(2);
    private Random random = new Random();

    public static void main(String[] args) {
        new CompletableFutureAllOfCompletion().run();
    }

    public void run() {
        CompletableFuture<Integer> cf1 = supplyAsync(this::randomSleepAndReturn, executor1);
        CompletableFuture<Integer> cf2 = supplyAsync(this::randomSleepAndReturn, executor1);
        CompletableFuture<Integer> cf3 = supplyAsync(this::randomSleepAndReturn, executor2);
        randomSleepAndReturn();
        CompletableFuture.allOf(cf1, cf2, cf3)
                .thenRun(() -> System.out.println("allOf() commpleted on "
                        + Thread.currentThread().getName()));

        executor1.shutdown();
        executor2.shutdown();
    }

    public int randomSleepAndReturn() {
        try {
            final long millis = random.nextInt(1000);
            System.out.println(
                    Thread.currentThread().getName() + " waiting for " + millis);
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 0;
    }
}

Some possible outputs:

Completing on first executor (third task finished first):

pool-1-thread-1 waiting for 937
pool-1-thread-2 waiting for 631
main waiting for 776
pool-2-thread-1 waiting for 615
allOf() commpleted on pool-1-thread-1

Completing on second executor (first and second task finished before the third one):

pool-1-thread-1 waiting for 308
pool-1-thread-2 waiting for 788
main waiting for 389
pool-2-thread-1 waiting for 863
allOf() commpleted on pool-2-thread-1

Completing on main thread (all tasks finished before allOf().thenRun()):

pool-1-thread-1 waiting for 168
pool-1-thread-2 waiting for 292
main waiting for 941
pool-2-thread-1 waiting for 188
allOf() commpleted on main

How to control the executor that will be used after allOf() (or anyOf())

Since there is no guarantee on the executor that will be used, a call to one of those methods should be followed by a *Async(, executor) call to control which executor will be used.

If you need to return the resulting CompletableFuture of one of those calls, just add a thenApplyAsync(i -> i, executor) before returning it.

Answer 2

There's no executor associated with CompletableFuture#allOf, it just produces the CompletableFuture that will wait for the completion of the dependencies in the same thread where you will call the CompletableFuture#get().

In your example, the tasks behind cf1 and cf2 will be still executed by executor1, the task in cf2 will be executed by executor2, the result of allOf(..).get() will be returned in current thread and no additional threads will be started behind the scene.

Here's the example, how you can observe the actual behavior in your IDE, by setting a breakpoint on System.out.println line and checking the list of active threads.

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.function.Supplier;

import static java.util.concurrent.CompletableFuture.supplyAsync;

public class ExecutorTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Executor executor1 = Executors.newSingleThreadExecutor();
        Executor executor2 = Executors.newSingleThreadExecutor();
        CompletableFuture<Integer> cf1 = supplyAsync(run(1), executor1); //executor1
        CompletableFuture<Integer> cf2 = supplyAsync(run(2), executor1); //executor1
        CompletableFuture<Integer> cf3 = supplyAsync(run(3), executor2); //executor2
        CompletableFuture<Void> result = CompletableFuture.allOf(cf1, cf2, cf3);
        new Thread(() -> {
            try {
                result.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }).start();
        System.out.println("Waiting now...");
    }

    private static Supplier<Integer> run(int result) {
        return () -> runDelayed(result);
    }

    private static int runDelayed(int result) {
        try {
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return result;
    }

}