Execute action when ThreadPoolExecutor has no active workers

Question

I have a cached thread pool where new tasks are spawned in rather unpredictable manner. These tasks don't generate any results (they are Runnables rather than Callables).

I would like to have an action to be executed whenever the pool has no active workers.

However I don't want to shutdown the pool (and obviously use awaitTermination) because I would have to reinitialize it again when a new task arrives (as it could arrive unpredictably, even during the shutdown).

I came up with the following possible approaches:

1. Have an extra thread (outside the pool) which is spawned whenever a new task is spawned AND the ThreadPoolExecutor had no active workers. It should then continually check the getActiveWorkers() until it returns 0 and if yes, execute the desired action.

2. Have some thread-safe queue (which one?), where the Future of every newly spawned task is added. Whenever there's at least one entry in the queue, spawn an extra thread (outside the pool) which waits until the queue is empty and executes the desired action.

3. Implement a PriorityBlockingQueue to use with the pool and assign the worker threads higher priority than to the thread (now from inside the pool) which executes the desired action.

My question: I was wondering if there is some cleaner solution, which uses some nice synchronization object (like CountDownLatch, which however cannot be used here, because I don't know the number of tasks in advance) ?

Answer 1

If I were you, I would implement a decorator for your thread pool that keeps track of the scheduled tasks and slighlig modifies the tasks that are run. This way, whenever a Runnable is scheduled, you can instead schedule another, decoarated Runnable which is capable of tracing its own process.

This decorator would look something like:

class RunnableDecorator implements Runnable {

  private final Runnable delegate;

  // this task counter must be increased on any 
  // scheduling of a task by the thread pool
  private final AtomicInteger taskCounter;

  // Constructor omitted

  @Override
  public void run() {
    try {
      delegate.run();
    } finally {
      if (taskCounter.decrementAndGet() == 0) {
        // spawn idle action
      }
    }
  }
}

Of course, the thread pool has to increment the counter every time a task is scheduled. Thus, the logic for this must not be added to the Runnable but to the ThreadPool. Finally, it is up to you to decide if you want to run the idle action in the same thread or if you want to provide a reference to the executing thread pool to run a new thread. If you decide the latter, note however that the completion of the idle action would then trigger another idle action. You might however also provide a method for a sort of raw scheduling. You could also add the decoration to the thread queue what however makes it harder to provide this sort of raw scheduling.

This approach is non-blocking and does not mess with your code base too much. Note that the tread pool does not start an action when it is created and therefore empty by definition.

Answer 2

If you look at the source behind Executors.newCachedThreadPool(), you can see how it's created with a ThreadPoolExecutor. Using that, override the execute and afterExecute methods to add a counter. This way the increment and decrement logic is isolated in one location. Ex:

ExecutorService executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
        new SynchronousQueue<Runnable>()) {
    private AtomicInteger counter = new AtomicInteger(0);

    @Override
    public void execute(Runnable r) {
        counter.incrementAndGet();
        super.execute(r);
    }

    @Override
    public void afterExecute(Runnable r, Throwable t) {
        if (counter.decrementAndGet() == 0) {
            // thread pool is idle - do something
        }
        super.afterExecute(r, t);
    }
};