ExecutorCompletionService? Why do need one if we have invokeAll?

Question

If we use an ExecutorCompletionService we can submit a series of tasks as Callables and get the result interacting with the CompletionService as a queue.

But there is also the invokeAll of ExecutorService that accepts a Collection of tasks and we get a list of Future to retrieve the results.

As far as I can tell, there is no benefit in using one or over the other (except that we avoid a for loop using an invokeAll that we would have to submit the tasks to the CompletionService) and essentially they are the same idea with a slight difference.

So why are there 2 different ways to submit a series of tasks? Am I correct that performance wise they are equivalent? Is there a case that one is more suitable than the other? I can't think of one.

Solution 1

Using a ExecutorCompletionService.poll/take, you are receiving the Futures as they finish, in completion order (more or less). Using ExecutorService.invokeAll, you do not have this power; you either block until are all completed, or you specify a timeout after which the incomplete are cancelled.

---

static class SleepingCallable implements Callable<String> {

  final String name;
  final long period;

  SleepingCallable(final String name, final long period) {
    this.name = name;
    this.period = period;
  }

  public String call() {
    try {
      Thread.sleep(period);
    } catch (InterruptedException ex) { }
    return name;
  }
}

Now, below I will demonstrate how invokeAll works:

final ExecutorService pool = Executors.newFixedThreadPool(2);
final List<? extends Callable<String>> callables = Arrays.asList(
    new SleepingCallable("quick", 500),
    new SleepingCallable("slow", 5000));
try {
  for (final Future<String> future : pool.invokeAll(callables)) {
    System.out.println(future.get());
  }
} catch (ExecutionException | InterruptedException ex) { }
pool.shutdown();

This produces the following output:

C:\dev\scrap>java CompletionExample
... after 5 s ...
quick
slow

---

Using CompletionService, we see a different output:

final ExecutorService pool = Executors.newFixedThreadPool(2);
final CompletionService<String> service = new ExecutorCompletionService<String>(pool);
final List<? extends Callable<String>> callables = Arrays.asList(
    new SleepingCallable("slow", 5000),
    new SleepingCallable("quick", 500));
for (final Callable<String> callable : callables) {
  service.submit(callable);
}
pool.shutdown();
try {
  while (!pool.isTerminated()) {
    final Future<String> future = service.take();
    System.out.println(future.get());
  }
} catch (ExecutionException | InterruptedException ex) { }

This produces the following output:

C:\dev\scrap>java CompletionExample
... after 500 ms ...
quick
... after 5 s ...
slow

Note the times are relative to program start, not the previous message.

---

You can find full code on both here.

Solution 2

So why are there 2 different ways to submit a series of tasks? Am I correct that performance wise they are equivalent? Is there a case that one is more suitable than the other? I can't think of one.

By using an ExecutorCompletionService, you can get immediately notified when each of your jobs completes. In comparison, ExecutorService.invokeAll(...) waits for all of your jobs to complete before returning the collection of Futures. This means that (for example), if all but one job completes in 10 minutes but 1 job takes 30 minutes, you will get no results for 30 minutes.

// this waits until _all_ of the jobs complete
List<Future<Object>> futures = threadPool.invokeAll(...);

Instead, when you use a ExecutorCompletionService, you will be able to get the jobs as soon as each of them completes which allows you to (for example) send them on for processing into another thread pool, log results immediately, etc..

ExecutorService threadPool = Executors.newFixedThreadPool(2);
ExecutorCompletionService<Result> compService
      = new ExecutorCompletionService<Result>(threadPool);
for (MyJob job : jobs) {
    compService.submit(job);
}
// shutdown the pool but the jobs submitted continue to run
threadPool.shutdown();
while (true) {
    Future<Result> future;
    // if pool has terminated (all jobs finished after shutdown) then poll() else take()
    if (threadPool.isTerminated()) {
        future = compService.poll();
        if (future == null) {
            break;
        }
    } else {
        // the take() blocks until any of the jobs complete
        // this joins with the jobs in the order they _finish_
        future = compService.take();
    }
    // this get() won't block
    Result result = future.get();
    // you can then put the result in some other thread pool or something
    // to immediately start processing it
    someOtherThreadPool.submit(new SomeNewJob(result));
}

Solution 3

I haven't ever actually used ExecutorCompletionService, but I think the case where this could be more useful than "normal" ExecutorService would be when you want to receive the Futures of completed tasks in completion order. With invokeAll, you just get a list that can contain a mix of incomplete and completed tasks at any given time.

Solution 4

Comparing by Considering only the Order of Results:

When we use CompletionService whenever a job submitted is finished the result will be pushed to the queue (Completion Order). Then the order of the submitted jobs and the returned results is no more same. So if you are concerned about the order which tasks got executed use CompletionService

Where As invokeAll returns a list of Futures representing the tasks, in the same sequential order as produced by the iterator for the given task list, each of which has completed.

View more solutions

Comments

[Cratylus]

So you are saying that in the List<Future> returned from invokeAll if start to iterate over the results, I could block on the first until it finishes, while in the ExecutioncCompletion I would block until any one result is available?Have I got your point?

[Gray]

+1 Yeah that's right @user384706. Underneath the ExecutorCompletionService is a BlockingQueue<Future<V>> so you can wait for the first job to complete instead of all of them.

[obataku]

@user384706 well, using the non-timeout form returns the Futures after all have completed, blocking indefinitely.

[Cratylus]

@Gray:But in invokeAll I don't wait for all to complete either

[Gray]

@user384706 when you call invokeAll it waits for all of the them to complete before finishing. When ExecutorCompletionService.poll/take returns, you know that the resulting get() is not going to block.

[obataku]

@user384706 then perhaps you don't want invokeAll, unless you're OK with cancelling unfinished tasks using a timeout.

[obataku]

@Gray actually I was incorrect... as it turns out, ExecutorService.invokeAll waits for completion.

[Cratylus]

@Gray:Ok, but it will block on take instead

[Gray]

I've added an answer with more details @user384706. Yes it blocks on take().

[Gray]

Btw @veer. take() never returns null.

[obataku]

@Gray yes I know, but for some odd reason I felt compelled to put the assignment into the loop condition :p

[Gray]

Heh. I never put the assignment in the loop condition. A pet peeve I guess. Good answer. :-)

[Cratylus]

@veer:I don't understand why in your first example (invokeAll) it prints slow first.I believe that I have read that the Future are returned in the same order are the Iterator of the passed Collection to invokeAll would return them.So I would expect to see quick first since quick is first in the passed List of the invokeAll

[obataku]

@user384706 good catch, I changed the order I submit them in without updating the output. It was to demonstrate that it waits for both to complete regardless of the order you put them in. I have since updated the output. Feel free to compile and run the full code from the paste.

[obataku]

@user384706 since you appear newer here, please don't forget to mark the answer accepted which helped most in solving the problem.

[Cratylus]

@veer:Good point. Thanks for ALL the great answers here by Gray and esaj as well

[Mr_and_Mrs_D]

What do you mean by "in completion order (more or less)" ?

[obataku]

@Mr_and_Mrs_D the order in which tasks complete i.e. if B finishes before A then the order would be B, A regardless of the order in which they're submit

[Mr_and_Mrs_D]

@oldrinb: thanks - the more or less part confused me - so they are guaranteed to return in the order they finish. You may want to edit this to make it more clear and let me know to delete the comments :)

[Coder]

while(!threadPool.isTerminated()) is not it a busy formal waiting?

[Gray]

It is but take() blocks so it isn't spinning. Did I answer your question @Sergio?

[Coder]

Yes thanks! I was digging on how to limit the blocking queue that there is inside Executors.newFixedThreadPool. In particular I am using the ListenableFuture

[tinkuge]

@Gray I did not understand your explanation of while(!threadPool.isTerminated()). Why is it needed? What purpose does it serve?

[Gray]

isTerminate() is true if the pool has shutdown and all of the jobs have completed. Is that what you are asking @tinkuge?

[tinkuge]

@Gray The edit makes it a lot more clearer, thank you!