How to await a list of ListenableFuture with a timeout

user_864027 0
Viewed 4190

I'm working on a problem where I have a List<ListenableFuture<T>>. I would like to aggregate the results of all of these futures into a List<T> with a timeout. The naive approach would be something like:

List<T> blockForResponses(List<ListenableFuture<T>> futures, long timeoutMillis) {
    return futures.stream()
        .map(future -> future.get(timeoutMillis,TimeUnit.MILLISECONDS)
        .collect(Collectors.toList());
}

This doesn't work because it waits for the timeout for each future and I want that to be the timeout for the entire list. Manually keeping track of how much time has passed also doesn't work because if the first one times out, there won't be any time left to try the others.

The solution I'm looking for would enforce a timeout on all of the futures and return when the timeout had elapsed or all of the futures in the list were complete. Then I could inspect each future in the list myself to aggregate the results and check which ones timed out.

future guava java
2 Answers

This problem turned out to be simpler than I thought. I was able to use the Futures.allAsList method and then catch the TimeoutException:

List<T> blockForResponses(List<ListenableFuture<T>> futures, long timeoutMillis) {
    ListenableFuture<List<T>> futureOfList = Futures.allAsList(futures);
    List<T> responses;
    try {
        responses = futureOfList.get(timeoutMillis, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
        responses = new ArrayList<>();
        for (ListenableFuture<T> future : futures) {
            if (future.isDone()) {
                responses.add(Uninterruptibles.getUninterruptibly(future));
            }
        }
    }
    return responses;
}
user_8640270

So I fiddled around a bit (Haven't used Guava's Listenable interface until this evening) but I think this may work for you: package basic;

import static com.google.common.util.concurrent.Futures.catching;
import static com.google.common.util.concurrent.Futures.successfulAsList;
import static com.google.common.util.concurrent.Futures.withTimeout;

import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

public class FuturesExample {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ListeningScheduledExecutorService scheduledExecutorService = MoreExecutors
                .listeningDecorator(Executors.newScheduledThreadPool(20));
        List<ListenableFuture<Integer>> list = new LinkedList<>();
        for (int i = 1; i <= 4; i++) {
            list.add(catching(
                    withTimeout(scheduledExecutorService.submit(getCallback(i * 1000, i)), 3, TimeUnit.SECONDS,
                            scheduledExecutorService),
                    TimeoutException.class, exception -> 0, scheduledExecutorService));
        }
        ListenableFuture<List<Integer>> result = successfulAsList(list);
        Optional<Integer> sum = result.get().stream().reduce(Integer::sum);
        System.out.println(sum.orElse(-1));
        scheduledExecutorService.shutdownNow();
    }

    private static Callable<Integer> getCallback(int timeout, int value) {
        return () -> {
            Thread.sleep(timeout);
            return value;
        };
    }
}

edit: code is a bit cleaner when using static imports for Futures

user_51029320
Related