How to wait for list of `Future`s created using different `ExecutorServices`

Question

Ok, so I know the first answer / comment here will be "use one ExecutorService and use invokeAll". However, there is a good reason (which I will not bore people with) for us keeping the thread pools separate.

So I have a list of thread pools (ExecutorServices) and what I need to do is invoke a different Callable on each thread pool using submit (no problem there). Now I have this collection of Future instances, each created on a seperate ExecutorService, and I want to wait for all of them to complete (and be able to provide a timeout at which any not done are cancelled).

Is there an existing class that will do this (wrap a list of Future instances and allow for a wait till all are done)? If not, suggestions on an efficient mechanism would be appreciated.

Was thinking of calling get with a timeout for each but have to do a calculation of the total time passed for each call.

I saw this post Wait Until Any of Future is Done but this extends Future instead of wrapping a list of them.

Solution 1

Per Louis' comment, what I was looking for was Futures.successfulAsList

This allows me to wait for all to complete and then check for any futures that failed.

Guava RULES!

Solution 2

I don't think JDK provides a direct API that lets you do that. However, I think it is equally straightforward to create a simple method that does this. You might want to take a look at the implementation of AbstractExecutorService.invokeAll() to get an idea that this can be done.

Essentially, you would call future.get() on each future, decreasing the wait time by the time it took to wait for the result each time, and before returning from the method cancel all outstanding futures.

Solution 3

Maybe I didn't really get it. However, to me it still sounds as simple as

public <V> List<V> get(List<Future<V>> futures, long timeout, TimeUnit unit)
          throws InterruptedException, ExecutionException, TimeoutException {
    List<V> result = new ArrayList<V>();
    long end = System.nanoTime() + unit.toNanos(timeout);
    for (Future<V> f: futures) {
        result.add(f.get(end - System.nanoTime(), TimeUnit.NANOSECONDS));
    }
    return result;
}

Am I wrong with that?

The question you link is much more complex I think, as they only want to wait for the fastest, and of course have no idea which will be the fastest.

Solution 4

This could use some cleanup, but it should solve your problem. (Some encapsulation omitted for time and space):

public static <T> LatchWithWrappedCallables<T> wrapCallables(Collection<Callable<T>> callablesToWrap)
{
    CountDownLatch latch = new CountDownLatch(callablesToWrap.size());
    List<Callable<T>> wrapped = new ArrayList<Callable<T>>(callablesToWrap.size());
    for (Callable<T> currCallable : callablesToWrap)
    {
        wrapped.add(new CallableCountdownWrapper<T>(currCallable, latch));
    }

    LatchWithWrappedCallables<T> returnVal = new LatchWithWrappedCallables<T>();
    returnVal.latch = latch;
    returnVal.wrappedCallables = wrapped;
    return returnVal;
}

public static class LatchWithWrappedCallables<T>
{
    public CountDownLatch latch;
    public Collection<Callable<T>> wrappedCallables;
}

public static class CallableCountdownWrapper<T> implements Callable<T>
{
    private final Callable<T> wrapped;

    private final CountDownLatch latch;

    public CallableCountdownWrapper(Callable<T> wrapped, CountDownLatch latch)
    {
        this.wrapped = wrapped;
        this.latch = latch;
    }

    @Override
    public T call() throws Exception
    {
        try
        {
            return wrapped.call();
        }
        finally
        {
            latch.countDown();
        }
    }
}

Then your code would call it like this:

Collection<Callable<String>> callablesToWrap = [Your callables that you need to wait for here];
LatchWithWrappedCallables<String> latchAndCallables = wrapCallables(callablesToWrap);

[Submit the wrapped callables to the executors here]

if(latchAndCallables.latch.await(timeToWaitInSec, TimeUnit.SECONDS))
{
    [Handling for timeout here]
}

View more solutions