How to check when all CompleteableFuture are done?

user_10255450 0
Viewed 332

I have a Stream<Item> which I'm mapping to a CompleteableFuture<ItemResult>

What I'd like to do is to know when all the futures are completed. One may suggest to:

  1. collect all the futures to an array and use CompleteableFuture.allOf(). This is somewhat problematic since there could be hundreds of thousands of items
  2. just continue with forEach(CompleteableFuture::join). This is problematic too as calling forEach with join will just block the stream and it will be essentially a serial processing and not concurrent
  3. Inject a poisoned item in the end of the stream. This could work but it's not that elegant in my view
  4. check if the executor queue is empty - This is quite limiting because I might use more than one executor in the future. Also, the queue can be momentarily empty
  5. Monitor the database instead and check the number of new items

I feel like all the suggested solutions aren't good enough.

What is the appropriate way to monitor the futures?

Thanks

EDIT:
another (vague) idea I had in mind is to use a counter and wait for it to go down to zero. But again, need to check that it's not a momentarily 0..

completable-future asynchronous multithreading java
1 Answers

Disclaimer: I'm not sure whether Phaser is the right tool here, and if yes, whether it's better to have one root with multiple children or to chain them like I'm proposing below, so feel free to correct me.

Here's one approach that uses Phaser.

A Phaser has a limited number of parties, so we need to create a new child Phaser if that limit is about to get reached:

private Phaser register(Phaser phaser) {
    if (phaser.getRegisteredParties() < 65534) {
        // warning: side-effect,
        // conflicts with AtomicReference#updateAndGet recommendation,
        // might not fit well if the Stream is parallel:
        phaser.register();
        return phaser;
    } else {
        return new Phaser(phaser, 1);
    }
}

Register each CompletableFuture against that Phaser chain, and deregister once done:

private void register(CompletableFuture<?> future, AtomicReference<Phaser> phaser) {
    Phaser registeredPhaser = phaser.updateAndGet(this::register);
    future
            .thenRun(registeredPhaser::arriveAndDeregister)
            .exceptionally(e -> {
                // log e?
                registeredPhaser.arriveAndDeregister();
                return null;
            });
}

Wait for all futures to be finished:

private <T> void await(Stream<CompletableFuture<T>> futures) {
    Phaser rootPhaser = new Phaser(1);
    AtomicReference<Phaser> phaser = new AtomicReference<>(rootPhaser);
    futures.forEach(future -> register(future, phaser));
    rootPhaser.arriveAndAwaitAdvance();
    rootPhaser.arriveAndDeregister();
}

Example:

ExecutorService executor = Executors.newFixedThreadPool(500);

// creating fake stream with 500,000 futures:
Stream<CompletableFuture<Integer>> stream = IntStream
        .rangeClosed(1, 500_000)
        .mapToObj(i -> CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(10);
                if (i % 50_000 == 0) {
                    System.out.println(Thread.currentThread().getName() + ": " + i);
                }
                return i;
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
        }, executor));

// usage:
await(stream);
System.out.println("Done");

Outputs:

pool-1-thread-348: 50000
pool-1-thread-395: 100000
pool-1-thread-333: 150000
pool-1-thread-30: 200000
pool-1-thread-120: 250000
pool-1-thread-10: 300000
pool-1-thread-241: 350000
pool-1-thread-340: 400000
pool-1-thread-283: 450000
pool-1-thread-176: 500000
Done
user_12253280
Related