CompletableFuture return all finished result after timeout

user_1887253 0
Viewed 1535

I want to have a list of comletablefutures I want to wait. With following code.

  public static <T> CompletableFuture<List<T>> finishAllQuery(
  List<CompletableFuture<T>> futures) {
CompletableFuture<Void> allDoneFuture =
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
return allDoneFuture.thenApply(
    v -> futures.stream().filter(Objects::nonNull)
        .map(future -> future.join())
        .collect(Collectors.toList())
);
  }
CompletableFuture<List<Result>> allResponse = finishAllQuery(futures);
allResponse.get(5, milliseconds)

The problem is that among all the futures, some of them can be slow, I want that after the expiration time, the get method return with all completed results. Is there a way to do that?

Thanks a lot.

completable-future java
2 Answers

This should be handled by finishAllQuery itself. E.g., starting with Java 9, you can use

public static <T> CompletableFuture<List<T>> finishAllQuery(
    List<CompletableFuture<T>> futures, long timeOut, TimeUnit unit) {

    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
        .completeOnTimeout(null, timeOut, unit)
        .thenApply(v -> futures.stream()
            .filter(CompletableFuture::isDone)
            .map(CompletableFuture::join)
            .collect(Collectors.toList()));
}

With completeOnTimeout, we can force the completion of the future with a predefined value after the timeout elapsed. Here, we just use null, as the result value of allOf doesn’t matter anyway.

We just have to add a filter condition, to skip all futures which are not completed yet, as otherwise, join would block the thread.

This can be used like

CompletableFuture<List<Result>> allResponse
    = finishAllQuery(futures, 5, TimeUnit.MILLISECONDS);
List<Result> list = allResponse.join(); // will wait at most 5 milliseconds

For Java 8, we can use

static <T> CompletableFuture<T> completeOnTimeout(
    CompletableFuture<T> cf, T value, long timeOut, TimeUnit unit) {

    ScheduledExecutorService e = Executors.newSingleThreadScheduledExecutor();
    ScheduledFuture<Boolean> job = e.schedule(() -> cf.complete(value), timeOut, unit);
    return cf.whenComplete((x,y) -> { job.cancel(false); e.shutdown(); });
}

for the missing feature, which requires a small rewrite:

public static <T> CompletableFuture<List<T>> finishAllQuery(
    List<CompletableFuture<T>> futures, long timeOut, TimeUnit unit) {
    return completeOnTimeout(
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])),
        null, timeOut, unit)
        .thenApply(v -> futures.stream()
            .filter(CompletableFuture::isDone)
            .map(CompletableFuture::join)
            .collect(Collectors.toList()));
}

The way, the caller uses the method, doesn’t change.

For production use, it’s worth rewriting the completeOnTimeout method to reuse the ScheduledExecutorService, but this also requires adding shutdown code or a thread factory creating daemon threads. With Java 9 or newer, you get that for free.

user_27114880

If you don't necessarily want all futures to complete, then you should not use allOf(). You can instead use CompletionService and then use the poll() method with timeout. So the flow will be:

  1. Wrap your ExcecutorService in a CompletionService
  2. Submit N tasks to CompletionService. Then poll for individual tasks with timeout. Something like:

elapsedTime = 0; availableTime = 5ms; completedFutures = List();

while (elapsedTime < availableTime):
    remainingTime = availableTime - elapsedTime; 
    startTime = currentTime();
    completedFutures.add(CompletionService.poll(timeout=remainingTime));
    elapsedTime += (currentTime() - startTime)
user_3731510
Related