ExecutorService have invokeAll method, and documentation say next:
Executes the given tasks, returning a list of Futures holding their status and results when all complete or the timeout expires, whichever happens first. Future.isDone is true for each element of the returned list. Upon return, tasks that have not completed are cancelled. Note that a completed task could have terminated either normally or by throwing an exception. The results of this method are undefined if the given collection is modified while this operation is in progress
Consider the following code:
public class Main {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(3);
List<Callable<Integer>> tasks = Arrays.asList(
Main::veryLongCalculations,
Main::veryLongCalculations,
Main::veryLongCalculations);
try {
List<Future<Integer>> resultTasks = executorService.invokeAll(tasks, 2, TimeUnit.SECONDS);
for (Future<Integer> task: resultTasks) {
System.out.println("is done: " + task.isDone() + ", canceled: " + task.isCancelled());
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static Integer veryLongCalculations() {
while (!Thread.currentThread().isInterrupted()) {
//very long calculations
}
System.out.println("interrupted");
return 0;
}
}
If we run this code, the following output will be displayed on the screen:
interrupted
interrupted
interrupted
is done: true, canceled: true
is done: true, canceled: true
is done: true, canceled: true
The tasks were clearly running longer than the timeout and so they were cancelled. Status of tasks is completed. Everything worked exactly as expected.
But if we use ForkJoinPool as ExecutorService (ExecutorService executorService = Executors.newWorkStealingPool()) then the output will be as like this:
interrupted
interrupted
is done: true, canceled: true
is done: true, canceled: true
is done: false, canceled: false
One of the tasks is never cancelled, which contradicts the documentation But if we change sdk from java17 to java11, then it will work fine. I began to understand and saw that ForkJoinPool inherits from the AbstractExecutorService class, which implements the invokeAll method. But in java 17 this method is overloaded in the ForkJoinPool itself. Here is the overloaded method:
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());
try {
for (Callable<T> t : tasks) {
ForkJoinTask<T> f =
new ForkJoinTask.AdaptedInterruptibleCallable<T>(t);
futures.add(f);
externalSubmit(f);
}
long startTime = System.nanoTime(), ns = nanos;
boolean timedOut = (ns < 0L);
for (int i = futures.size() - 1; i >= 0; --i) {
Future<T> f = futures.get(i);
if (!f.isDone()) {
if (timedOut)
ForkJoinTask.cancelIgnoringExceptions(f);
else {
((ForkJoinTask<T>)f).awaitPoolInvoke(this, ns);
if ((ns = nanos - (System.nanoTime() - startTime)) < 0L)
timedOut = true;
}
}
}
return futures;
} catch (Throwable t) {
for (Future<T> e : futures)
ForkJoinTask.cancelIgnoringExceptions(e);
throw t;
}
}
On the first iteration, the timedOut variable will always be false, so one task will never be cancelled. Is this a bug? Or is there some other reason why it works this way? This behavior is clearly contrary to the documentation, which says that all tasks should have a completed status and should be canceled if the timeout expires.