Wait for all tasks (unknown number) to complete - ThreadPoolExecutor
I ran into this concurrency issue that I scratched my head on several times.
Basically, I want my ThreadPoolExecutor to wait for all tasks (number of unknown tasks) to complete before closing.
public class AutoShutdownThreadPoolExecutor extends ThreadPoolExecutor{
private static final Logger logger = Logger.getLogger(AutoShutdownThreadPoolExecutor.class);
private int executing = 0;
private ReentrantLock lock = new ReentrantLock();
private final Condition newTaskCondition = lock.newCondition();
private final int WAIT_FOR_NEW_TASK = 120000;
public AutoShutdownThreadPoolExecutor(int coorPoolSize, int maxPoolSize, long keepAliveTime,
TimeUnit seconds, BlockingQueue<Runnable> queue) {
super(coorPoolSize, maxPoolSize, keepAliveTime, seconds, queue);
}
@Override
public void execute(Runnable command) {
lock.lock();
executing++;
lock.unlock();
super.execute(command);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
try{
lock.lock();
int count = executing--;
if(count == 0) {
newTaskCondition.await(WAIT_FOR_NEW_TASK, TimeUnit.MILLISECONDS);
if(count == 0){
this.shutdown();
logger.info("Shutting down Executor...");
}
}
}
catch (InterruptedException e) {
logger.error("Sleeping task interrupted", e);
}
finally{
lock.unlock();
}
}
}
The task is supposed to check the task counter (execution), and if it is 0, it locks for a while and later releases its lock so that other tasks can be able to complete and not shut down the executor too soon.
However, this does not happen. All 4 threads in the executor go to the idle state:
"pool-1-thread-4" prio=6 tid=0x034a1000 nid=0x2d0 waiting on condition [0x039cf000]
"pool-1-thread-3" prio=6 tid=0x034d0400 nid=0x1328 waiting on condition [0x0397f000]
"pool-1-thread-2" prio=6 tid=0x03493400 nid=0x15ec waiting on condition [0x0392f000]
"pool-1-thread-1" prio=6 tid=0x034c3800 nid=0x1fe4 waiting on condition [0x038df000]
If I put a log statement (presumably slowing the flow down) in the Runnable class, the problem seems to go away.
public void run() {
// logger.info("Starting task" + taskName);
try {
//doTask();
}
catch (Exception e){
logger.error("task " + taskName + " failed", e);
}
}
The question is similar to this post Java ExecutorService: waitTermination of all recursively created tasks
I took the original poster solution and tried to solve the race issue in afterExecute () but it doesn't work.
Please help shed some light on this. Thank.
source to share
You have your tasks waiting for this newTaskCondition
, but nothing ever signals this condition. So your streams are all piling up, all waiting on newTaskCondition
until the time is up. In addition, blocking afterExecute
delays the execution of the task, so this is probably not what you want to do.
If you want it to wait a while to see if there is still work to be done, that functionality already exists. Just call setKeepAliveTime
to set how long to wait and set allowCoreThreadTimeOut
to ensure all threads (including main threads) can complete.
source to share
I found a very simple solution to my problem.
Even though the number of jobs is unknown in advance, and the goal is to force the main thread to wait for all tasks in the threadPool to complete, the time at which all jobs were submitted (after find-and-submit - the task method has checked all files).
I can just call ThreadPoolExecutor.shutdown () and waitTermination (Long.MAX_VALUE) so that the main thread will wait indefinitely for the ThreadPool to finish executing its tasks.
source to share