how not to overwhelm java executorservice task queue?

Question

I have the below code snippet, which runs fine. But the problem is it creates and put over 2000 tasks on the executor queue right off the bat.

I need to have a check if the tasks already in the executor queue are complete, only then give it more tasks. It doesnt have to be exact, ie if the queue has <10 tasks left, add 50 more.

So the executor task queue doesnt have so many pending tasks, which will also allow shutdown() to work in a timely manner, otherwise even if called, the executor will still trying to complete all 2000 tasks in its queue first.

What is the best way to accomplish this? thank you

executor = Executors.newFixedThreadPool(numThreads);

while(some_condition==true)
{
    //if(executor < 10 tasks pending)  <---- how do i do this?
    //{                             
        for(int k=0;k<20;k++)
        {  
            Runnable worker = new MyRunnable();
            executor.execute(worker);
        }
    //}
    //else 
    //{
    //      wait(3000);
    //}
}

Update using semaphore:

private final Semaphore semaphore = new Semaphore(10)
executor = new ThreadPoolExecutorWithSemaphoreFromJohnExample();

while(some_condition==true)
{

        Runnable worker = new MyRunnable();
        //So at this point if semaphore is full, then while loop would PAUSE(??) until
        //semaphore frees up again.
          executor.execute(worker);   
}

Gray · Answer 1 · 2017-12-08T18:30:01.667

I have the below code snippet, which runs fine. But the problem is it creates and put over 2000 tasks on the executor queue right off the bat.

One way to do this is to create your own ThreadPoolExecutor with a limited job queue and set a custom RejectedExecutionHandler on it. This allows you to have fine grained control over how many jobs to queue.

You need the custom handler because by default, if the queue is full the ThreadPoolExecutor.submit(...) will throw a RejectedExecutionException. With the custom handler below, when it gets rejected by the queue, the rejection handler just puts it back in, blocking until the queue has space. So no jobs will be rejected/dropped.

Here's approximately how you start your own thread-pool and set your own reject handler.

// you can tune the blocking queue size which is the number of jobs to queue
// when the NUM_THREADS are all working
final BlockingQueue<MyRunnable> queue =
    new ArrayBlockingQueue<MyRunnable>(NUM_JOBS_TO_QUEUE);
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(NUM_THREADS, NUM_THREADS,
       0L, TimeUnit.MILLISECONDS, queue);
// by default (unfortunately) the ThreadPoolExecutor will throw an exception
// when you submit the job that fills the queue, to have it block you do:
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
   public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
      // this will block if the queue is full as opposed to throwing
      executor.getQueue().put(r);
   }
});
...
// now submit all of your jobs and it will block if the queue is full
for(int k = 0; k < 20000000; k++) {  
   Runnable worker = new MyRunnable();
   threadPool.execute(worker);
}

See my answer here for more details about blocking thread-pools:

How can I make ThreadPoolExecutor command wait if there's too much data it needs to work on?

You can also use the ThreadPoolExecutor.CallerRunsPolicy which would cause the caller that is submitting the job into the thread-pool to execute the job. I don't like this solution however because it blocks the caller until the job finishes which might starve the other worker threads. Also, if there are multiple submitters, it might still cause too many threads to run the jobs.

Lastly, notice that I set the core and max thread count in the ThreadPoolExecutor to the same number. Unfortunately, by default, the executor starts the core threads, then fills the queue, and only then does it allocate additional threads up to the max. This is completely counter-intuitive.

but the problem is i dont want to reject the tasks or throw any question, i just want to wait until the task queue has less pending then add those tasks. Will it be okie to create a counter (int tasksRunning = 0), then everytime execute() is called tasksRunning++ and in the runnable.run() when it finishes do tasksRunning-- thanks — user1539050, Jul 19 '12 at 20:55
My solution will not reject the tasks. That's the whole point. I'll make that more plain @user1539050. — Gray, Jul 19 '12 at 20:55
also see http://stackoverflow.com/questions/2001086/how-to-make-threadpoolexecutors-submit-method-block-if-it-is-saturated — wrschneider, Jul 19 '12 at 21:07

score 7 · Accepted Answer · answered Jul 19 '12 at 21:04

7

You can use a simple Semaphore. Upon submitting acquire a new permit and after completion release the permit to allow anyone else awaiting to submit.

private final Semaphore semaphore = new Semaphore(10);//or however you want max queued at any given moment
ThreadPoolExecutor tp= new ThreadPoolExecutor(...){
      public void execute(Runnable r){
          semaphore.acquire();
          super.execute(r);
      }    
      public void afterExecute(Runnable r, Thread t){
         semaphore.release();  
         super.afterExecute(r,t);
      }
};

So here the submitting threads will be suspended if there are no more permits available.

answered Jul 19 '12 at 21:04

John Vint

39,695
7
78
108

Good one John. It would prolly be recommended to catch and handle `RejectedExecutionException` or be careful that your semaphore value was smaller than the blocking queue limit (if any). – Gray Jul 19 '12 at 21:14
Also the semaphore number (10) is not the number of max-queued, it is the number of running + queued. – Gray Jul 19 '12 at 21:15
@Gray You are right about both, good points. I put this up as a starting point but those recommendations should be considered if the OP were to implement this. – John Vint Jul 19 '12 at 21:22
thank you john and gary, i think this is exactly what I need. But to confirm can you take a look at my reply with some additional questions in the code. thanks – user1539050 Jul 19 '12 at 21:50

score 3 · Answer 3 · answered Jul 20 '12 at 07:55

I usually throttle such systems by using a object 'pool queue' for the task objects - a BlockingQueue that is filled up with X tasks at startup. Anything that wants to submit a task to the threads has to get one from the pool queue, load it up with data and then submit it.

When the task is completed and results in it processed, it is pushed back onto the pool queue for re-use.

If the pool empties, submitting threads block on the pool queue until some tasks are returned.

This is essentially a form of semaphore control as suggested by @John Vint, but has some further advantages - no continual create/GC of the runnables, for example. I like to dump PooolQueue.size to a GUI status bar on a timer, so I can see how 'busy' the system is, (and also to quickly detect any object leaks:)

score 2 · Answer 4 · answered Jul 29 '13 at 15:28

2

You will be better of setting a Rejection policy, since you don't want to overwhelm the ThreadPool, I have found that the best way to accomplish this without complicating yourself much is by doing something like this:

final ThreadPoolExecutor executor=(ThreadPoolExecutor)Executors.newFixedThreadPool(THREADS_COUNT);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

What will happen is that once all Threads are busy, the caller's thread will execute the task. Here is a reference to such policy CallerRunsPolicy JavaDoc

answered Jul 29 '13 at 15:28

Guido Medina

416
4
8

Not going to downvote this, but just want to note that casting it is dangerous because it isn't necessarily a ThreadPoolExecutor – KyleM Dec 18 '13 at 20:17
ExecutorService which is the type of the return instance doesn't have a setter for Rejection Execution policy, if you don't down cast it you won't be able to do that. In this case to troutle the execution by not queueing more jobs.. – Guido Medina Dec 29 '13 at 12:54
Executors in general are tricky and need some work from Oracle/Sun, like the fact that there isn't a Rejection policy to block at the queue while the threads are picking Job, I have to do something NOT recommended to support such functionality at our company, until framework writers make it better. – Guido Medina Dec 29 '13 at 12:56

how not to overwhelm java executorservice task queue?

4 Answers4