Set a timeout on individual threads when using CompletableFuture using a ThreadPoolExecutor

Question

I have a program which executes things asynchronously using a ThreadPoolExecutor. I use CompletableFutures in Java 8 to schedule these tasks and then have them executed by the threads available in the thread pool.

My code looks like this:

public class ThreadTest {

    public void print(String m) {
        System.out.println(m);
    }

    public class One implements Callable<Integer> {

        public Integer call() throws Exception {
            print("One...");
            Thread.sleep(6000);
            print("One!!");
            return 100;
        }
    }

    public class Two implements Callable<String> {

        public String call() throws Exception {
            print("Two...");
            Thread.sleep(1000);
            print("Two!!");
            return "Done";
        }
    }

    @Test
    public void poolRun() throws InterruptedException, ExecutionException {
        int n = 3;
        // Build a fixed number of thread pool
        ExecutorService pool = Executors.newFixedThreadPool(n);

        CompletableFuture futureOne = CompletableFuture.runAsync(() -> new One());
        // Wait until One finishes it's task.
        CompletableFuture futureTwo = CompletableFuture.runAsync(() -> new One());
        // Wait until Two finishes it's task.
        CompletableFuture futureTwo = CompletableFuture.runAsync(() -> new Two());

        CompletableFuture.allOf(new CompletableFuture[]{futureOne, futureTwo, futureThree}).get();
        pool.shutdown();
    }
}

I need to set a timeout on each individual thread, for example to timeout at 10 minutes. I looked into the .get(TimeUnit timeUnit) method for CompletableFuture, but I wasn't sure if that sets a timeout on the thread pool or on the individual thread itself.

https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html#get-long-java.util.concurrent.TimeUnit-

Or should I be changing the way I use the executor service to set timeouts on individual threads?

Thanks!

Answer 1

CompletableFuture.get does not stop the thread running your task. The calling thread waits as long as you specify for the result, and if it times out it will throw an exception. But the thread running the task will continue until it is done. Here's the underlying reality: Java will not allow you to arbitrarily a terminate a task at any time. There was a time when this was part of the API, there were Thread.suspend/resume/stop methods on the Thread class. These were deprecated because there is no way to know if a suspended or stopped thread was holding locks that could block execution of other threads. So it's inherently unsafe to stop a thread at arbitrary times and places. You end up with deadlock in your program.

See here: https://docs.oracle.com/javase/8/docs/technotes/guides/concurrency/threadPrimitiveDeprecation.html

The same argument applies to any of the pools and executors and other classes you find in the concurrent package. You cannot arbitrarily stop a thread, or a task.

You must put the logic for stopping and completion into the task itself, or you simply must wait until it is done. In this case you have one that runs for a second and one for 6 seconds. You can use things like mutexes and semaphores, you can use many of the things in the concurrent package and the concurrent.locks package, all that are useful to coordinate threads and to pass information about where they are.

But you will not find a method anywhere that allows you to kill an arbitrary thread at any point in time, except for the ones that were previously deprecated as listed above, and those methods you are encouraged to stay away from.

Future.cancel will stop a task from starting, and it will try to interrupt a task that is running, but all it does it stop a thread (by causing it to throw InterruptedException) that is currently blocked on an interruptible method call like Thread.sleep(), Object.wait(), or Condition.await(). If your task is doing anything else it will not stop until it completes or until it calls an interruptible method call.

This will work on the code above since you are calling Thread.sleep. But once you have your task doing work, it will behave as I described.