How to wait for all threads to finish, using ExecutorService?

Question

I need to execute some amount of tasks 4 at a time, something like this:

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
    taskExecutor.execute(new MyTask());
}
//...wait for completion somehow

How can I get notified once all of them are complete? For now I can't think about anything better than setting some global task counter and decrease it at the end of every task, then monitor in infinite loop this counter to become 0; or get a list of Futures and in infinite loop monitor isDone for all of them. What are better solutions not involving infinite loops?

Thanks.

Solution 1

Basically on an ExecutorService you call shutdown() and then awaitTermination():

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
  taskExecutor.execute(new MyTask());
}
taskExecutor.shutdown();
try {
  taskExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
} catch (InterruptedException e) {
  ...
}

Solution 2

Use a CountDownLatch:

CountDownLatch latch = new CountDownLatch(totalNumberOfTasks);
ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
  taskExecutor.execute(new MyTask());
}

try {
  latch.await();
} catch (InterruptedException E) {
   // handle
}

and within your task (enclose in try / finally)

latch.countDown();

Solution 3

ExecutorService.invokeAll() does it for you.

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
List<Callable<?>> tasks; // your tasks
// invokeAll() returns when all tasks are complete
List<Future<?>> futures = taskExecutor.invokeAll(tasks);

Solution 4

You can use Lists of Futures, as well:

List<Future> futures = new ArrayList<Future>();
// now add to it:
futures.add(executorInstance.submit(new Callable<Void>() {
  public Void call() throws IOException {
     // do something
    return null;
  }
}));

then when you want to join on all of them, its essentially the equivalent of joining on each, (with the added benefit that it re-raises exceptions from child threads to the main):

for(Future f: this.futures) { f.get(); }

Basically the trick is to call .get() on each Future one at a time, instead of infinite looping calling isDone() on (all or each). So you're guaranteed to "move on" through and past this block as soon as the last thread finishes. The caveat is that since the .get() call re-raises exceptions, if one of the threads dies, you would raise from this possibly before the other threads have finished to completion [to avoid this, you could add a catch ExecutionException around the get call]. The other caveat is it keeps a reference to all threads so if they have thread local variables they won't get collected till after you get past this block (though you might be able to get around this, if it became a problem, by removing Future's off the ArrayList). If you wanted to know which Future "finishes first" you could use some something like https://stackoverflow.com/a/31885029/32453

Solution 5

In Java8 you can do it with CompletableFuture:

ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
                               .map(task -> CompletableFuture.runAsync(task, es))
                               .toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();    
es.shutdown();

View more solutions

Comments

[Lordn__n]

There aren't 4 tasks. There are "some number of tasks" done 4 at a time.

[user2012801]

Only instead of CountDownLatch(4) I need to use number of total threads created, right?

[ChssPly76]

Sorry, I misunderstood the question. Yes, number of tasks should be the argument to CountDownLatch constructor

[user2012801]

Looks nice and simple, only I have feeling that shutdown here is used not for what it was designed (or it is fine?).

[Lordn__n]

It's a common pattern. Transient threadpools like the above are good for several reasons and I pretty much always prefer them to persistent threadpools where a lot more can go wrong or at least it can be harder to figure out. The above is really straightforward which is why I like it.

[wvdschel]

I find this solution more elegant than the others, it looks like it was made for this purpose, and it's simple and straightforward.

[Lordn__n]

What if you don't know the number of tasks before you start?

[matt b]

this is exactly what shutdown / awaitTermination are meant for

[ChssPly76]

@cletus - then you don't use a CountDownLatch :-) Mind you, I'm not arguing that this approach is better than yours. However, I found that in real life scenarios I do know the number of tasks, thread pool settings do need to be configurable per deployment, and pools can be reused. So I usually have thread pools injected by Spring and setting them as prototypes and manually shutting them down only in order to wait for threads to finish seems less than ideal.

[sjlee]

It is a good pattern if this task handling is a one-time event. If this is done repeatedly during the same runtime, however, it is not optimal, as you would create and tear down threads repeatedly every time it is executed.

[象嘉道]

This pattern is incomplete: Some unsubmitted tasks may be never executed.

[inanutshellus]

@Kejia It is not incomplete... Think about it this way: Your main thread is spawning off subthreads in a loop. When it's done adding, you run this code. You might see output out of order, but all of your tasks will have been spun off. Thanks @Cletus!

[נשמה קשוחה]

Cool, can never remember the name of this data structure. However, only suitable if you know beforehand the amount of tasks that will get queued.

[Sam Harwell]

I'm looking for any official documentation that Long.MAX_VALUE, TimeUnit.NANOSECONDS is equivalent to not having a timeout.

[Brad]

@serg is this still considered the best answer over using invokeAll() as proposed by sjlee?

[rogerdpack]

I can't believe that you have to use shutdown in order to join on all the current threads (after using shutdown, you cannot use the executor again ever). Suggest using list of Future's instead...

[rogerdpack]

The difficulty comes if/when you have the start the "4" threads one at a time, piece-wise, then join/let finish all 4...

[rogerdpack]

yeah you'd think you'd be able to hit the barrier with the current thread, and all the child threads, then when you passed it you'd know the child threads were done...

[Andrew Mao]

This does seem to be better than shutdown/awaitTermination. CyclicBarrier also seems to be useful when you need to run threads in groups at a time.

[jontejj]

As I'm testing this out I found that the executor finishes before all tasks have completed. If I use Thread.sleep(100) before calling shutdown() it works. So I suspect that there may be some timing issues with your code if the time between adding tasks to the shutdown call is too small.

[Nazgul]

this wont work if you dont want to shutdown the executor. For a perpetually running batch kind of thing u need to submit jobs and wait for them to finish before jumping ahead. In Such a case a latch or a barrier makes more sense than a shutdown.

[gerrytan]

IMO it's probably not a wise thing to NOT have a timeout. There's always a likelihood your task is stuck / deadlocking and the user would wait forever.

[Mukul Goel]

@rogerdpack: I am still learning this executors and stuff. But in response to what you ask. Should the 4 threads at a time not be part of a batch task which is executed using the answer above?

[Konstantin]

This method will only work if you know the number of tasks before hand.

[Olivier Faucheux]

Not a good idea, see stackoverflow.com/a/7271685/1166992 and the javadoc: "Returns the approximate number of threads that are actively executing tasks."

[Zero3]

@ashutosh I can't find any documentation on this. It appears to be a method of a Spring library, and not plain Java functionality.

[beluchin]

@SamHarwell see the java.util.concurrent package documentation under the Timing section: To wait "forever", you can use a value of Long.MAX_VALUE

[ToolmakerSteve]

To know which "finishes first", use ExecutorCompletionService.take: stackoverflow.com/a/11872604/199364

[ToolmakerSteve]

Took me awhile to see how this would solve OP's question. First, note that this wrapping is of each task, not of the code that starts all the tasks. Presumably, each start would increment a counter, and each finish would decrement that counter, or would increment a completed counter. So after starting them all, at each notification, could determine whether all tasks have completed. Note that it is vital to use try/finally so that a finished notification (or an alternative notification in catch block) is given even if a task fails. Otherwise, would wait forever.

[ToolmakerSteve]

It is good that you show use of future.get -- good alternative to know about. But why do you consider it better to wait forever, than to set some maximum acceptable timeout? More importantly, there is no reason to do all this logic, when one can simply give a really, really long time to awaitTermination, if you want to wait (essentially forever) until all tasks complete.

[ToolmakerSteve]

On seeing this, I first wondered if it would be a problem if some release calls happen before the acquire call, but after reading Semaphore documentation, I see that is okay.

[ToolmakerSteve]

@ashutosh 1) That requires springframework. 2) Doesn't look any simpler to me than shutdown/awaitTermination.

[ToolmakerSteve]

stryba's Semaphore solution is more flexible, as it accomplishes the same result (no need to shutdown) as CountDownLatch, without needing to know how many tasks there are.

[Praveen Kumar]

Is it better than using infinite loop while checking isTerminated() ?

[Saurabh]

+1,thanks it helped, also referred this howtodoinjava.com/core-java/interviews-questions/…

[Coder]

If I don't want to shutdown because I want to keep my executor alive. How can I know the maximum number of pending tasks that could I have there?

[Tim Malone]

Hi Vlad, welcome to StackOverflow. Can you please edit your answer to explain how this answers the question, and what the code does? Code only answers are discouraged here. Thank you!

[AlikElzin-kilaka]

I think that when the futures are returned, the tasks haven't been completed. They might complete in the future and you'll have a link to the result. That's why it's called a Future. You have the method Future.get(), which will wait for the task to finish to get a result.

[hinneLinks]

@SamHarwell TimeUnit.NANOSECONDS and Long.MAX_VALUE equals 106,751 Days or 292 years (TimeUnit.NANOSECONDS.toDays(Long.MAX_VALUE);), that should be enough, or use some of the bigger TimeUnits.

[gstackoverflow]

Actually it is wrong answer. CyclicBarrier designed for portions. CountDownLatch designed for waiting event

[DSchmidt]

Very clean. Does work flawlessly even on Android. Just had to use runOnUiThread() in onSuccess().

[Manish Kumar Sharma]

This is not different from already presented solutions here. Your just solution is the same as presented by @sjlee

[eli-bd]

When sizing thread pools, it is often useful to base the size on the number of logical cores in the machine running the application. In Java, you can get that value by calling Runtime.getRuntime().availableProcessors(). In this case, it would be Math.min(4, Runtime.getRuntime().availableProcessors());

[Hulk]

@AlikElzin-kilaka Quote from the JavaDocs (linked in the answer): "Executes the given tasks, returning a list of Futures holding their status and results when all complete. Future.isDone() is true for each element of the returned list."

[Mashrur]

Not sure why you need to check for done when according to oracle doc, invokeAll will return only "when all complete or the timeout expires, whichever happens first"

[Gili]

Another major advantage of this approach is that it works with shared Executors that cannot simply be shut down because they are used by others.

[mattvonb]

This is a very elegant solution.

[Jobs]

executor.shutdown(); is stopping my program !!

[user2862544]

ExecutorService es = Executors.newFixedThreadPool(4); List< Future<?>> futures = new ArrayList<>(); for(Runnable task : taskList) { futures.add(es.submit(task)); } for(Future<?> future : futures) { try { future.get(); }catch(Exception e){ // do logging and nothing else } }

[The Gilbert Arenas Dagger]

Note that executorService.invokeAll will wait for all threads to complete, but you will still need to call executorService.shutdown to clean up your thread pool.

[Mobigital]

Java apis are pretty terrible. Thank you for this insight.

[GabrielBB]

This post talks about concurrency. Parallelism != Concurrency

[Gaurav]

@cletus is awaitTermination() necessary after shutdown()?

[Gaurav]

@AdamSkywalker is awaitTermination() necessary after es.shutdown()?

[Gaurav]

is awaitTermination() necessary after shutdown()/

[AdamSkywalker]

@gaurav when you call shutdown, some tasks may not be finished yet. So awaitTermination will block the calling thread until everything is done. It depends on whether you need to wait for results in this thread or not.

[Gaurav]

@AdamSkywalker great answer. makes sense to not call awaitTermination() if I dont need to wait for the results.

[nagendra547]

@AdamSkywalker Great answer!! Elegant!

[Alfredo Zuloaga]

What if you simply add finally and there put your latch.CountDown()

[Siddhartha]

Even better to make the future.get call timed future.get(10, TimeUnit.SECONDS); and catch the TimeoutException

[ryanlee]

this method has one serious problem is that once the thread pool commit task error and refuse it, it can never go out because the countDown in task can not be executed