How to add futures by for loop for CompletableFuture.allOf?

Question

How can I add futures by CompletableFuture.allOf() when the futures are created within a for-loop? I want to create a bunch of futures which should be executed in parallel. And only when all futures are completed the method should return the result:

// Version 1: execute each task in async and return alls tasks when finished
public Set<Task> getTasks(){
    var executor = Executors.newCachedThreadPool();
    var tasks = new LinkedHashSet<Task>();
    var futures = new ArrayList<CompletableFuture<Set<Task>>>();
    for (var task : user.getTasks()) {
        // all futures are executed in parallel
        futures.add(CompletableFuture.supplyAsync(() -> execute(task), executor));
    }
    for (var f : futures) {
        // this will block as long as each future is finished
        tasks.addAll(f.join());
    }

    return tasks;
}

Or is there another alternative? I have also tried the following, but it also executes the futures one after another (instead of parallel):

// Version 2:
var executor = Executors.newCachedThreadPool();
var tasks = new LinkedHashSet<Task>();
for (var task : user.getTasks()) {
    CompletableFuture.supplyAsync(() -> execute(task), executor)
                     .thenAccept(tasks::addAll).join();
}

---

EDIT: at the end I have two versions which come close the problem I would like to solve. However, I guess version A is not right because parallel threads will add elements to the LinkedHashSet in async mode (which could cause trouble, because LinkedHashSet is not thread safe):

VERSION A (it seems not thread safe):

var executor = Executors.newCachedThreadPool();
var tasks = new LinkedHashSet<Task>();
var futures = new ArrayList<CompletableFuture<Void>>();
for (var t : user.getTasks()) {
    futures.add(CompletableFuture.supplyAsync(() -> execute(t), executor).thenAcceptAsync(tasks::addAll));
}
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

and VERSION B (which could be better, but is a little complex):

var executor = Executors.newCachedThreadPool();
var futures = new ArrayList<CompletableFuture<Set<Task>>>();
for (var t : user.getTasks()) {
    futures.add(CompletableFuture.supplyAsync(() -> execute(t), executor));
}
Set<Task> o = CompletableFuture
                 .allOf(futures.toArray(new CompletableFuture[0]))
                 .thenApplyAsync(v -> futures.stream().flatMap(future -> future.join().stream()))
                 .join().collect(Collectors.toSet());

I cannot find an easier approach..but for completness, I add the following code which is the shortest - however, it uses ForkJoinPool which should be avoided (?) for long running tasks:

        // VERSION C: execute in parallel without suffering from CompletableApi:
        return user.getTasks()
                .parallelStream()
                .flatMap(t -> execute(t).stream())
                .collect(Collectors.toSet());

Answer 1

Your code should work as it is. That is, the for loop in your first example waits for the first future to complete before proceeding to the second future, but in the meantime all the other futures are concurrently running. They typically start to execute as soon as you've called supplyAsync. To prove this, here's a self-contained executable:

import java.time.LocalTime;
import java.util.ArrayList;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Demo {
    public static void main(String[] args) throws InterruptedException {
        var executor = Executors.newCachedThreadPool();
        var results = new ArrayList<String>();
        var futures = new ArrayList<CompletableFuture<String>>();
        futures.add(CompletableFuture.supplyAsync(() -> sleep(2), executor));
        TimeUnit.MILLISECONDS.sleep(100);
        futures.add(CompletableFuture.supplyAsync(() -> sleep(1), executor));

        // All futures are executed in parallel
        for (var f : futures) {
            results.add(f.join());
        }

        results.forEach(System.out::println);
    }

    private static String sleep(int seconds) {
        var start = LocalTime.now();
        try {
            TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException ignored) {
            Thread.currentThread().interrupt();
        }
        var end = LocalTime.now();
        return String.format("Thread %s started at %s and finished at %s",
                Thread.currentThread().getId(), start, end);
    }
}

The output proves that the second future finished before the first, as expected:

Thread 14 started at 17:49:35.202673531 and finished at 17:49:37.206196631
Thread 15 started at 17:49:35.262183490 and finished at 17:49:36.262342704

Answer 2

CompletableFuture.allOf() is pretty simple here when using Stream API:

CompletableFuture.allOf(user.getTasks().stream()
            .map(task -> CompletableFuture.supplyAsync(() -> execute(task), executor))
            .toArray(CompletableFuture[]::new))
            .join();

Answer 3

Of course your second variant will execute one after another:

CompletableFuture.supplyAsync(() -> execute(task), executor)
                 .thenAccept(tasks::addAll)
                 .join();

You join that, blocking the thread.

The second problem is the use of newCachedThreadPool. I'll explain that based on the jdk's HttpClient. In the early version it had that in the documentation that it will use a cached pool, later they removed it from the documentation. Currently it is left in the implementation, but that will be removed also, in time. The problem is that such a pool, when you use it incorrectly, will eat all your resources and kill your application. No more free threads? Sure, I will create a new one, and so on... Eventually this will hurt you. Use a pool with a limited numbers of threads.

---

To answer your question, you are looking for some kind of flatMap, that could do CompletableFuture<Set<X>> to Set<CompletableFuture<X>>. Such a non-blocking method does not exist. You need to call join, but you can delay the call to that join via a trick:

user.getTasks().stream()
               .map(each -> CompletableFuture.supplyAsync(() -> execute(each), executor))
               .flatMap(x -> Stream.of(x).map(CompletableFuture::join))
               .flatMap(Set::stream)
               .collect(Collectors.toSet());

Answer 4

After trying all those versions above I come to the conclustion that the following solution is the best:

// VERSION X is the best
public Set<Task> getTasks(){
    var executor = Executors.newCachedThreadPool();
    var futures = new ArrayList<Future<Set<Task>>>();
    var tasks = new LinkedHashSet<Task>();
    for (var t : user.getTasks()) {
        futures.add(executor.submit(() -> executor(t)));
    }
    for (var f : futures) {
        try {
            tasks.addAll(f.get());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
return tasks;
}

}

It's the best because:

• easy and fast code (no unneeded overhead, lambdas, completableFuture,..)
• no exception is surpressed
• does not stop the execution of further tasks if one task raises an exception

If anyone can convince me to use other versions, then please add arguments.