Number of active tasks using ThreadPoolExecutor

Question

I am using a ThreadPoolExecutor to execute tasks in my Java application. I have a requirement where I want to get the number of active tasks in the queue at any point in time in the executor queue . I looked up at the javadoc for ThreadPoolExecutor and found two relevant methods: getTaskCount() and getCompletedTaskCount().

Per the documentation, I could get the number of scheduled tasks and completed tasks from the above two methods respectively. But I am not able to find a solution for getting the number of active tasks in the queue at any point in time. I can do something like:

getTaskCount() = getCompletedTaskCount() + failed tasks + active tasks

But the number of failed tasks is not directly available to arrive at the intended calculation.

Am I missing something here ?

Solution 1

I don't think you need to know the failed count with the calculation you're trying to use.

long submitted = executor.getTaskCount();
long completed = executor.getCompletedTaskCount();
long notCompleted = submitted - completed; // approximate

Would be (approximately) sufficient.

---

Alternatively, you can use getQueue() with size():

int queued = executor.getQueue().size();
int active = executor.getActiveCount();
int notCompleted = queued + active; // approximate

This answer presumes you're looking for a "not yet completed" count. Your question contradicts itself so I'm not completely certain what you're asking. Reply to my comment on your question if this is incorrect, and I'll update this answer accordingly.

Solution 2

Have you tried using the beforeExecute and afterExecute methods? These are called before and after a task is executed. The after execute method even supplies a throwable as a second argument, so you know when a task has failed.

You could add a hook so that beforeExecute increments the value of the active tasks, and afterExecute decrements it. Ofcourse, these methods are called on their respective fields, so that you would have to synchronize the result on a mutual lock Object.

To use these methods, just override the ThreadPoolExecutor object of your choice and add the hook there.

For instance, the following code should hopefully work:

public class MyExecutor extends ThreadPoolExecutor {
       //Lock object used for synchronization
       private final Object lockObject = new Object();
       //Contains the active task count
       private int activeTaskCount = 0;
       //Failed task count
       private int failedTaskCount = 0;
       private int succeededTaskCount = 0;

       public MyExecutor () {
            //call super here with your parameters;
       }

       public int getActiveTaskCount(){
           synchronized(lockObject){
              return activeTaskCount;
           }
       } 

       public int getFailedTaskCount(){
           synchronized(lockObject){
              return failedTaskCount ;
           }
       } 

       public int getSucceededTaskCount(){
           synchronized(lockObject){
              return succeededTaskCount ;
           }
       } 

       protected void beforeExecute(Thread t,
                             Runnable r){
            super.beforeExecute(t,r);
            synchronized(lockObject){
                activeTaskCount++;
            }
       }

       protected void afterExecute(Runnable r,Throwable t){
            super.afterExecute(r,t);
            synchronized(lockObject){
                activeTaskCount--;
                if(t!=null){
                    failedTaskCount++;
                }else{
                    succeededTaskCount++;
                }
            }
       }

}

Comments

[Rob Hruska]

Your statement ("active tasks in the queue") contradicts itself: do you want "active tasks" or "tasks in the queue"? Because tasks in the queue are not active, they are queued.

[Rob Hruska]

Additionally, can you clarify what you mean by a "failed task"? What constitutes a failure is completely up to you; ThreadPoolExecutor really only has the notion that a thread completed or that some (non-business-logic) system error occurred during execution. If you need to check a return status or catch a checked exception to determine failure, you need to use Futures.

[invinc4u]

My apologies for not being precise with the requirements. I am basically trying to find the number of tasks that have been scheduled but have not yet been completed, at run-time. And i would consider "failed task" as any task that was scheduled but did not successfully complete due to some error.

[Rob Hruska]

This seems unnecessary; ThreadPoolExecutor already provides getActiveCount().

[Tovi7]

Yes but not a failed count, which this method CAN provide. Check the comments in the afterExecute method. You can even check which runnables are running if you keep them all in a set. This method allows for more possibilities than I posted in my simple sample.

[Rob Hruska]

I did miss your comment there; it might serve better for example's sake to actually demonstrate the failed count with your example rather than the active count. Additionally, the t == null check would only be checking for serious errors (unchecked, i.e. RuntimeExceptions, Errors) that get thrown from the execution (and not business logic "failures"). The notion of what "failed" and "succeeded" is solely up to the developer using the API (and should be done using Futures instead of throwing RuntimeExceptions from an executing thread).

[Tovi7]

I have edited my code to include a failed and succeeded count. I agree though with your comment on how failures should be handled is application-specific.

[invinc4u]

The second solution was the one that i was looking for and it works for me . Thanks for your help.

[user482594]

Remember that some Queue implementations do not support O(1) operation for size() method, and this can take longer than expected.