Java - CompletableFutures - How can i cancel all futures if there are exceptions

user_2716177 0
Viewed 859

I have a method (included below) to return the values of a list of CompletableFutures.

The method is supposed to:

  1. be able to timeout after a given time.
  2. be able to cancel all futures if there are more than n amount of exceptions.

The first point works well and indeed bombs out after it passed the timeout limit. (I still need to call exectuorService.shutdownNow() afterwards to return to the caller). The problem I'm having is with the second thing I'm trying to accomplish.

Lets say i have a list of 20,000 futures and all of them will have an exception, then why let all of them execute, if I see that there are too many exceptions then i assume that something is wrong with all of the futures andI want to cancel them.

In addition i would love to have a timeout on each future individually how long it may take, but this also would'nt work, unassuming for the same reason outlined below.

It seems that the reason is, because when I call allDoneFuture.thenApply(), at this point it waits and lets all the futures complete, either successfully or exceptionally. Only after all of them completed does it go through each future and fetches its result. At that point what good does it do to cancel, when they have completed already.

I would much appreciate if someone can show me how to accomplish this specific need: "Monitor the exceptions, and the individual timeouts, and based on that cancel all others".

Thanks.

Below is the method I wrote:

/**
     * @param futures a list of completable futures
     * @param timeout how long to allow the futures to run before throwing exception
     * @param timeUnit unit of timeout
     * @param allowedExceptions how many of the futures do we tolerate exceptions,
     * NOTE: if an exception is thrown from the futures it will return null, until it reaches the allowedExceptions threshold
     * */
    public static <T> List<T> extractFromFutures(List<CompletableFuture<T>> futures, int timeout, TimeUnit timeUnit, int allowedExceptions) {
        CompletableFuture<Void> allDoneFuture = CompletableFuture
                .allOf(futures.toArray(new CompletableFuture[futures.size()]));
        try {
            AtomicInteger exceptionCount = new AtomicInteger(0);
            return allDoneFuture.thenApply(v ->//when all are done
                    futures.stream().
                            map(future -> {
                                try {
                                    //if only I could set an individual timeout
                                    return future.get(timeout, timeUnit);
                                } catch (Exception e) {
                                    future.cancel(true);
                                    int curExceptionCnt = exceptionCount.incrementAndGet();
                                    if(curExceptionCnt >= allowedExceptions){
                                        //I would've hoped that it will throw it to the calling try-catch 
                                        //and then cancel all futures, but it doesn't
                                        throw new RuntimeException(e);
                                    }
                                    else{
                                        return null;
                                    }
                                }
                            }).
                            collect(Collectors.<T>toList())
            ).get(timeout, timeUnit);
        } catch (Exception e) {
            allDoneFuture.cancel(true);
            throw new RuntimeException(e);
        }
    }
completable-future multithreading java
2 Answers

To cancel all of the remaining futures after a certain number of exceptions you can call exceptionally on each of them and increment the exception count and possibly cancel them inside of that.

For individual timeouts you could create a class that holds the future with its timeout then sort them based on the timeout and call get with the timeout minus the elapsed time.

static class FutureWithTimeout<T> {
    CompletableFuture<T> f;
    long timeout;
    TimeUnit timeUnit;

    FutureWithTimeout(CompletableFuture<T> f, long timeout, TimeUnit timeUnit) {
        this.f = f;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
    }
}
public static <T> List<T> extractFromFutures(List<FutureWithTimeout<T>> futures, int allowedExceptions) {
    AtomicInteger exceptionCount = new AtomicInteger(0);
    futures.forEach(f -> f.f.exceptionally(t -> {
        if(exceptionCount.getAndIncrement() == allowedExceptions){
            futures.forEach(c -> c.f.cancel(false));
        }
        return null;
    }));
    long t = System.nanoTime();
    return futures.stream()
        .sorted(Comparator.comparingLong(f -> f.timeUnit.toNanos(f.timeout)))
        .map(f -> {
            try {
                return f.f.get(Math.max(0, f.timeUnit.toNanos(f.timeout) - (System.nanoTime() - t)), 
                    TimeUnit.NANOSECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException ex) {
                f.f.cancel(false);
                return null;
            }
        })
        .collect(Collectors.toList());
}

Note that this may return the list in a different order than it was passed in. If you need it in the same order then you could change the map().collect() to a forEachOrdered and then re map them into their results after without sorting.

Also the mayInterruptIfRunning parameter to cancel has no effect on CompletableFuture so I changed it to false.

user_31797590

CompletableFuture completely ignores any call to cancel(true). I don't know why (presumably to simplify the API), but it sucks. If you want to make futures actually cancelable (where you can either manually check for interruption, or accept cancellation by blocking on a lock), then you have to use Future, not CompletableFuture.

user_39509820
Related