How do get a Callable task submit to an ExecutorService timeout

Question

I submit Callable tasks (using submit()) to an implementation of ExecutionService. Very occasionally I seem to be encountering a deadlock but cannot work where or why it is happening so I would like to set a timeout on the task, I'm not clear how do it ?

Should I

1. Use invokeAny() on the ExecutionService instead of submit() when submitting the task and set a timeout. I submit many tasks one at a time over time using submit(), can I use invokeAny() like this as well, I'm cautious because I cannot understand why there isn't a submit() method that takes a timeout.
2. Modify keepAliveTime in my constructor of my ExecutorService (but I think this is doing something else
3. Modify my actual Callable implementation, but if its is deadlocked it cant undeadlock itself.

Option 1 seems the only viable solution but is it ?

More details

I thought it may be worth explaining how the process works in more detail in case it helps with solution.

Callable task P1 is started and works on a folder, and all the files and folders within it and starts grouping the songs into groups, its runs inside ExecutorService ES1, and just the one single instance of P1 is submitted to ES1.

We also have three other Callable classes: P2, P3, and P4 - each of these has their own associated Executor Service, ES2, ES3, Es4). Once P1 has created a group it then submits a task to the associated ES with the group passed as data, i.e. it could submit an instance of P2 to E2, P3 or to P3 or P4 to E4, which one it chooses depends on details of the grouping, P2, P3 and P4 all do different things.

Assume it had submitted an instance of P2, P2 will finish processing by submitting P3 to E3 or P4 to E4. Its a one way pipeline P3 can only submit to P4, and once all tasks have been submitted to P4 and P4 has finished all the tasks the processing has finished.

We complete processing by constructing ES1, ES2, ES3 and ES4, submitting task to P1, then calling shutdown() on each ExecutorService in turn so, shutdown() will not return until P1 has finished submitting all groups, it then calls shutdown() on ES2 which will not return until ES2 has cleared it queue of P2 tasks ecetera.

Very occasionally everything just stops Im assuming some process is stopping other processes from continuing, so at this point I want a way of cancelling processes that take too long so others can continue, this is significantly less bad then it just hanging indefinitently

Update on Answer

I tried using invokeAny() as suggested, it sort of works. If P1 submits an instance of P2 to E2 it then waits before completing, that is sort of okay because when using submit() it just returns any way there it does not further processing but there are two issues:

1. Each ExecutorService uses a bounded queue of 500, the idea being that if P2 is much slower than P1 we dont keep stacking things onto ES2 and eventually run out of memory. So now P1's don't finish until the task they call has finished the queues are effectively smaller because they dont just consist of tasks waiting for a slot on ES2 to finish but they contain tasks that have already submitted to ES2 but are waiting for it to finish.

2. The pipeline is chained so if we use invokeAny on tasks submitted from P1, and tasks submitted from P2 and P3 and P4 then when a task is submitted from P1 to P2 it will not return until subsequent processing completes from E4 !

Answer 1

You could use guava's MoreExecutors ListeningExecutorService. It will not magically solve your problems but can provide some aid:

1) You could set a timeout for each Callable invoked via invokeAll. If a callable isn't finished by given time, it should be killed.

2) You could create a global map of all ListenableFutures where each of them would register a flag on creation and clear that flag on completion. This way you would know which of those futures didn't finish helping to narrow down the problem.

Answer 2

I think the best way is to find and fix deadlock. You can't just kill the thread. You should implement some kind of task cancellation in task and ASK this task to cancel what it is doing. But if it is deadlocked you can't do anything.You can use jsconsole to detect a deadlock

Using invokeAny with timeout blocks thread until one of the submitted tasks completes successfully or timeout expires. If timeout expires you'll get TimeoutException, but your tasks will be running. ExecutorService will ask them to cancel with Future.cancel(true). Internally it interruptes the thread, setting task's thread isInterrupted flag to true. If you are using blocking methods inside your task, which responds to interruptions, they will throw Interrupted exception. Otherwise you should check interrupted status inside your task and, if it returns true, respond to it accordingly. If there are no blocking methods or checking interrupted status, this cancellation will take no effect.

Answer 3

Not sure if this will help you but I managed to quickly and easily find deadlocks before with this recipe:

• Debug in eclipse
• Reproduce the "hang"
• Select the server instance and click the "pause" button in the Eclipse debugger. This will pause all threads.
• Now scroll down the thread list. Deadlocked threads are marked red. Each thread shows the locks it holds and the locks it is waiting for.
• Profit!

Answer 4

Fixing the deadlock situation would be always advisable. But to make an alternative you can use future object of callable to set the timeout duration.

Checkout following solution for single callable instance. Same you can implement for number of callables using List of Future calls. It is using future.get(...) method, where we set the timeout. If the callable doesn't finishes execution by the set timeout, thread will finish its execution.

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ThreadTimeOut{
   public static void main(String[] args) throws Exception {
    ExecutorService executor = Executors.newSingleThreadExecutor();
    Future<String> future = executor.submit(new Task());

    try {
        System.out.println("Started..");
        System.out.println(future.get(3, TimeUnit.SECONDS));
        System.out.println("Finished!");
    } catch (TimeoutException e) {
        System.out.println("Terminated!");
    }

    executor.shutdownNow();
    }
}

class Task implements Callable<String> {
   @Override
   public String call() throws Exception {
       Thread.sleep(4000); // Just to demo a long running task of 4 seconds.
       return "Ready!";
   }
}

Answer 5

The comment by @PhilipWhitehouse works once I made a couple of amendments.

In summary create a custom ThreadPool that enscapulates a ScheduledExecutorPool so that a timeout can be set when the tasks are submitted.

Full solution here:

How can I make shutdown work properly with this custom ExecutorService?