Java – Complete Actively Executing Tasks in ExecutorService While Halting Waiting Tasks

javamultithreading

I am using an ExecutorService (a ThreadPoolExecutor) to run (and queue) a lot of tasks. I am attempting to write some shut down code that is as graceful as possible.

ExecutorService has two ways of shutting down:

I can call ExecutorService.shutdown() and then ExecutorService.awaitTermination(...).
I can call ExecutorService.shutdownNow().

According to the JavaDoc, the shutdown command:

Initiates an orderly shutdown in which previously submitted
tasks are executed, but no new tasks will be accepted.

And the shutdownNow command:

Attempts to stop all actively executing tasks, halts the
processing of waiting tasks, and returns a list of the tasks that were
awaiting execution.

I want something in between these two options.

I want to call a command that:
a. Completes the currently active task or tasks (like shutdown).
b. Halts the processing of waiting tasks (like shutdownNow).

For example: suppose I have a ThreadPoolExecutor with 3 threads. It currently has 50 tasks in the queue with the first 3 actively running. I want to allow those 3 active tasks to complete but I do not want the remaining 47 tasks to start.

I believe I can shutdown the ExecutorService this way by keeping a list of Future objects around and then calling cancel on all of them. But since tasks are being submitted to this ExecutorService from multiple threads, there would not be a clean way to do this.

I'm really hoping I'm missing something obvious or that there's a way to do it cleanly.

Thanks for any help.

Best Answer

I ran into this issue recently. There may be a more elegant approach, but my solution is to first call shutdown(), then pull out the BlockingQueue being used by the ThreadPoolExecutor and call clear() on it (or else drain it to another Collection for storage). Finally, calling awaitTermination() allows the thread pool to finish what's currently on its plate.

For example:

public static void shutdownPool(boolean awaitTermination) throws InterruptedException {

    //call shutdown to prevent new tasks from being submitted
    executor.shutdown();

    //get a reference to the Queue
    final BlockingQueue<Runnable> blockingQueue = executor.getQueue();

    //clear the Queue
    blockingQueue.clear();
    //or else copy its contents here with a while loop and remove()

    //wait for active tasks to be completed
    if (awaitTermination) {
        executor.awaitTermination(SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);
    }
}

This method would be implemented in the directing class wrapping the ThreadPoolExecutor with the reference executor.

It's important to note the following from the ThreadPoolExecutor.getQueue() javadoc:

Access to the task queue is intended primarily for debugging and monitoring. This queue may be in active use. Retrieving the task queue does not prevent queued tasks from executing.

This highlights the fact that additional tasks may be polled from the BlockingQueue while you drain it. However, all BlockingQueue implementations are thread-safe according to that interface's documentation, so this shouldn't cause problems.

Related Solutions

Java Multithreading ExecutorService – How to Stop Submitted Tasks in Java Executors

Just because you call cancel() on Future doesn't mean that the task will stop automatically. You have to do some work within the task to make sure that it will stop:

Use cancel(true) so that an interrupt is sent to the task.
Handle InterruptedException. If a function in your task throws an InterruptedException, make sure you exit gracefully as soon as possible upon catching the exception.
Periodically check Thread.currentThread().isInterrupted() if the task does continuous computation.

For example:

class LongTask implements Callable<Double> {
    public Double call() {
        
         // Sleep for a while; handle InterruptedException appropriately
         try {
             Thread.sleep(10000);
         } catch (InterruptedException ex) {
             System.out.println("Exiting gracefully!");
             return null;
         }


         // Compute for a while; check Thread.isInterrupted() periodically
         double sum = 0.0;
         for (long i = 0; i < 10000000; i++) {
             sum += 10.0
             if (Thread.currentThread().isInterrupted()) {
                 System.out.println("Exiting gracefully");
                 return null;
             }
         }

         return sum;
    } 
}

Also, as other posts have mentioned: ConcurrentModificationException can be thrown even if using the thread-safe Vector class, because iterators you obtain from Vector are not thread-safe, and thus need to be synchronized. The enhanced for-loop uses iterators, so watch out:

final Vector<Double> vector = new Vector<Double>();
vector.add(1.0);
vector.add(2.0);

// Not thread safe!  If another thread modifies "vector" during the loop, then
// a ConcurrentModificationException will be thrown.
for (Double num : vector) {
    System.out.println(num);
}

// You can try this as a quick fix, but it might not be what you want:
synchronized (vector) {    // "vector" must be final
    for (Double num : vector) {
        System.out.println(num);
    }
}

Java Multithreading – Force Stopping ExecutorService Threads

Instant thread termination is never guaranteed, unless the thread checks periodically for isInterrupted() flag (or is waiting in interruptable method, i.e. which throws InterruptedException).

Consider implementing your worker threads in manner, when they check periodically for isInterrupted(). This may be something like that:

public void run() { 
  byte[] data;
  do {
     data = receiveDataChunk(timeout);
     processData(data);
  } while(!isInterrupted() && data != null);
}

Best Answer

Related Solutions

Java Multithreading ExecutorService – How to Stop Submitted Tasks in Java Executors

Java Multithreading – Force Stopping ExecutorService Threads

Related Question