Java 8 mulithreading: How can I achieve parallelism along with a timeout for individual threads?

Question

Summary of what I want to achieve:

I want to execute N tasks in parallel such that no individual task should run for more than two seconds (we can mark such tasks as failed). As an output I want to return the output of successful tasks and status of failed tasks as failed. Also a timeout of one task should not lead to a circuit break, i.e., other tasks execution should not stop.

---

Note: I am restricted to use Java 8.

I referenced this article for parallel processing. I am doing a similar kind of parallel processing as given in the example in this article:

public void parallelProcessing() {
    try {
        ExecutorService executorService = Executors.newWorkStealingPool(10);

        List<CompletableFuture<Integer>> futuresList = new ArrayList<CompletableFuture<Integer>>();
        futuresList.add(CompletableFuture.supplyAsync(()->(addFun1(10, 5)), executorService));
        futuresList.add(CompletableFuture.supplyAsync(()->(subFun1(10, 5)), executorService));
        futuresList.add(CompletableFuture.supplyAsync(()->(mulFun1(10, 5)), executorService));

        CompletableFuture<Void> allFutures = CompletableFuture.allOf(futuresList.toArray(new CompletableFuture[futuresList.size()]));
        CompletableFuture<List<Integer>> allCompletableFuture = allFutures.thenApply(future -> futuresList.stream().map(completableFuture -> completableFuture.join())
                .collect(Collectors.toList()));
        CompletableFuture<List<Integer>> completableFuture = allCompletableFuture.toCompletableFuture();
        List<Integer> finalList = (List<Integer>) completableFuture.get();
    } catch (Exception ex) {

    }
}


public static Integer addFun1(int a, int b) {
    System.out.println(Thread.currentThread().getName());

    for (int i = 0; i < 10; i++) {

        System.out.print(Thread.currentThread().getName() + i);
    }

    return a + b;
}

public static Integer subFun1(int a, int b) {

    System.out.println(Thread.currentThread().getName());

    for (int i = 0; i < 10; i++) {

        System.out.print(Thread.currentThread().getName() + i);
    }

    return a - b;
}


public static Integer mulFun1(int a, int b) {

    System.out.println(Thread.currentThread().getName());

    for (int i = 0; i < 10; i++) {

        System.out.print(Thread.currentThread().getName() + i);
    }

    return a * b;
}

This works fine. But I want to set a timeout for an individual thread. I know I can use an overloaded get function in the last line. But that would set the timeout for combined futures, right? E.g., if I want no individual thread should be blocked for more than 2 seconds, and if I set a 2 seconds timeout in the last line, it will be combined timeout, right?

get(long timeout, TimeUnit unit)

Here's what I want to achieve as a final outcome:

Suppose there are five threads and four complete on time, one timeout (due to running more than two seconds). In this case, I want to send the output of four threads and send the error for the fifth thread in the result.

My input/output format is in the following way:

Sample input: List<Input> each item is run in a separate thread, where each input has a uniqueIdentifier.

Sample output: List<Output> such that:

Output :{
    uniqueIdentifier: // Same as input to map for which input this output was generated
    result: success/fail // This Field I want to add. Currently it's not there
    data: {
        // From output, e.g., addFun1 and subFun1
    }
}

Answer 1

The semantics of what you want to achieve matter very much. On one hand, you say that you want an alternative for orTimeout for Java 8; on the other hand you kind of imply that you want to drop execution of a certain CompletableFuture if it goes beyond a certain threshold.

These are very different things, because orTimeout says in the documentation:

Exceptionally completes this CompletableFuture with a TimeoutException if not otherwise completed before the given timeout.

So something like:

CompletableFuture<Integer> addAsy =
    supplyAsync(() -> addFun1(10,5), executorService)
    .orTimeout(5, TimeUnit.MILLISECONDS);

will result in a CompletableFuture that is completed exceptionally (assuming that addFun1 takes more than 5 ms). At the same time, this:

CompletableFuture<Void> allFutures = CompletableFuture
        .allOf(futuresList.toArray(new CompletableFuture[0]));

as the documentation states in the allOf:

... If any of the given CompletableFutures complete exceptionally, then the returned CompletableFuture also does so...

means that allFutures is a CompletableFuture that is completed exceptionally too (because addAsy is).

Now, because you have this:

    CompletableFuture<List<Integer>> allCompletableFuture = allFutures.thenApply(future -> {
        return futuresList.stream().map(CompletableFuture::join)
            .collect(Collectors.toList());
    });

And again, the documentation of thenApply says:

Returns a new CompletionStage that, when this stage completes normally, is executed with this stage's result as the argument to the supplied function.

Your allFutures did not complete normally, as such this is not even called.

So you need to understand what exactly you want to achieve. For a backport of orTimeout you could start by looking here.

---

You still need some kind of a backport for orTimeout. I will use the method as if it already exists.

static void parallelProcessing() throws Exception {

    ExecutorService executorService = Executors.newFixedThreadPool(10);

    List<CompletableFuture<Integer>> futuresList = new ArrayList<>();
    futuresList.add(CompletableFuture.supplyAsync(() -> addFun1(10,5), executorService).orTimeout(2, TimeUnit.SECONDS));
    futuresList.add(CompletableFuture.supplyAsync(() -> subFun1(10,5), executorService));
    futuresList.add(CompletableFuture.supplyAsync(() -> mulFun1(10,5), executorService));


    CompletableFuture<Void> all = CompletableFuture.allOf(futuresList.toArray(new CompletableFuture[0]));
    Map<Boolean, List<CompletableFuture<Integer>>> map =
            all.thenApply(x -> both(futuresList)).exceptionally(x -> both(futuresList)).get();

    List<CompletableFuture<Integer>> failed = map.get(Boolean.TRUE);
    List<CompletableFuture<Integer>> ok = map.get(Boolean.FALSE);

    System.out.println("failed = " + failed.size());
    System.out.println("ok = " + ok.size());

}

private static Map<Boolean, List<CompletableFuture<Integer>>> both(
        List<CompletableFuture<Integer>> futuresList) {
    return futuresList.stream().collect(Collectors.partitioningBy(
                    CompletableFuture::isCompletedExceptionally
            ));
}

Answer 2

Suppose you want 10 threads running and want a returned value, you can use the Callable<Boolean> interface, submit it to ExecutorService, and then get the result using Future#get returned as a Boolean.

Here is an example usage.

final int NUM_THREADS = 10;
List<Boolean> results = new ArrayList<Boolean>();
List<Callable<Boolean>> callables = new ArrayList<Callable<Boolean>>();

for(int i=0; i<NUM_THREADS; ++i)
{
    callables.add(new Callable<Boolean>()
    {
        public Boolean call()
        {
            // Add your task here
            return isTaskCompleted;
        }
    });
}

ExecutorService executorService = ExecutorService.newFixedThreadPool(NUM_THREADS); // Run 10 threads

for(Callable<Boolean> callable:callables)
{
    Future<Boolean> future = executor.submit(callable);
    try
    {
        results.add(future.get(2, TimeUnit.SECONDS)); // Timeout 2 seconds and add the result
    }
    catch(Exception ex)
    {
        results.add(false); // Set result to false if task throw TimeOutExeption
    }
}

If you want more information about these classes you can read this book: O'Reilly - Learning Java, Chapter 9: Threads.

Answer 3

The following is a single-file mre (paste the entire code into RunParallelTasks.java and run). It is a prototype of the structure I suggested in my comment aimed to achieve the required functionality by using simple means:

import java.util.Optional;

public class RunParallelTasks {

    public static void main(String[] args) {

        new Thread(()->{
            long duration = 3000;
            Callback<Long> cb = new LongTask(duration);
            Output<Long> output = new TaskExecuter<Long>().work(cb);
            System.out.println( output);
        }).start();

        new Thread(()->{
            long duration = 300;
            Callback<Long> cb = new LongTask(duration);
            Output<Long> output = new TaskExecuter<Long>().work(cb);
            System.out.println( output);
        }).start();

        new Thread(()->{
            long duration = 4000;
            Callback<Long> cb = new LongTask(duration);
            Output<Long> output = new TaskExecuter<Long>().work(cb);
            System.out.println( output);
        }).start();

        new Thread(()->{
            long duration = 1000;
            Callback<Long> cb = new LongTask(duration);
            Output<Long> output = new TaskExecuter<Long>().work(cb);
            System.out.println( output);
        }).start();

    }
}

class TaskExecuter<T>{

    private static final long TIMEOUT = 2000;//millis
    private T value = null;
    public Output<T> work(Callback<T> call){

        Thread t = new Thread(()->{
            value = call.work();
        });
        t.start();

        try {
            t.join(TIMEOUT);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }

        return new Output<>(t.getId(), value == null ?  Optional.empty() : Optional.of(value)) ;
    }
}

interface Callback<T> {
    T work();
}

class LongTask implements Callback<Long>{

    private final long durationInMillis;

    public LongTask(long durationInMillis) {
        this.durationInMillis = durationInMillis;
    }

    @Override
    public Long work() {
        try {
            Thread.sleep(durationInMillis);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
        return durationInMillis;
    }
}

class Output<T> {

    private final long id;
    private boolean success = false;
    private T data;

    public Output(long id, Optional<T> op) {
        this.id = id;
        if(!op.isEmpty()) {
            data = op.get();
            success = true;
        }
    }
    
    //todo add getters 
    
    @Override
    public String toString() {
        return "task "+ id+ (success ? " Completed, returned "+data : " Failed" );
    }
}

Answer 4

I believe that in order to achieve canceling a task if execution takes too long, you need two tasks:

1. The task itself doing the computation
2. Another task, which takes care to cancel the first, if it takes too long

This is inspired by my answer here, at least for now i still have not come up with a better way to do it.

Let's say this is Output:

public class Output {

    private final String uniqueIdentifier;
    private final boolean result;
    private final Object data;

    //all arguments constructor and getters

    @Override
    public String toString() {
        return "Output{" +
                "uniqueIdentifier='" + uniqueIdentifier + '\'' +
                ", result=" + result +
                ", data=" + data +
                '}';
    }
}

For simplicity i'll use only add integers task from your example, wrapping it in a Supplier.

public class AddIntegerTask implements Supplier<Integer> {

    private static final long NANOSECONDS_IN_SECOND = 1_000_000_000;

    private final String uniqueIdentifier;
    private final boolean tooLong;
    private final int a;
    private final int b;

    public AddIntegerTask(boolean tooLong, int a, int b) {
        this.uniqueIdentifier = UUID.randomUUID().toString();
        this.tooLong = tooLong;
        this.a = a;
        this.b = b;
    }

    @Override
    public Integer get() {
        long nanoseconds = this.tooLong ? 3 * NANOSECONDS_IN_SECOND : NANOSECONDS_IN_SECOND;
        long start = System.nanoTime();
        long toEnd = start + nanoseconds;
        //simulate long execution
        while (System.nanoTime() <= toEnd) {
            //check for interruption at crucial points
            boolean interrupted = Thread.currentThread().isInterrupted();
            if (interrupted) {
                //custom exception extending RuntimeException
                throw new TooLongExecutionException();
            }
        }
        return this.a + this.b;
    }

    public String getUniqueIdentifier() {
        return this.uniqueIdentifier;
    }
}

Most important here is, that you need to check the current thread for interruption at key moments in your own implementation.

The cancel task is quite straightforward:

public class CancelTask implements Runnable {

    private final Future<?> future;

    public CancelTask(Future<?> future) {
        this.future = future;
    }

    @Override
    public void run() {
        this.future.cancel(true);
    }
}

Wrap the canceling of a Future in a Runnable, so it can be scheduled for execution with approproate delay.

And the Runnable, which will wrap the result in an Output, and will be submitted for execution:

public class MyRunnable implements Runnable {

    private final Map<String, Output> outputMap;
    private final AddIntegerTask calcFunction;
    private final CountDownLatch latch;

    public MyRunnable(Map<String, Output> outputMap, AddIntegerTask calcFunction, CountDownLatch latch) {
        this.outputMap = outputMap;
        this.calcFunction = calcFunction;
        this.latch = latch;
    }

    @Override
    public void run() {
        String uniqueIdentifier = this.calcFunction.getUniqueIdentifier();
        Output output;
        try {
            Integer result = this.calcFunction.get();
            output = new Output(uniqueIdentifier, true, result);
        } catch (TooLongExecutionException exc) {
            output = new Output(uniqueIdentifier, false, null);
        }
        this.outputMap.replace(uniqueIdentifier, output);
        this.latch.countDown();
    }
}

Things to note here: CountDownLatch, it looks to me that you know the number of tasks beforehand, so it's a good choice to force main thread to wait until all tasks have finished. TooLongExecutionException is a custom exception extending RuntimeException. If the job completes, set as successful with result, if it was interrupted set to not successful without result.

And a main to combine and test all that:

public class CancelingMain {

    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);

        int taskCount = 6;
        CountDownLatch latch = new CountDownLatch(taskCount);
        long start = System.nanoTime();
        Map<String, Output> outputMap = new LinkedHashMap<>();
        for (int i = 1; i <= taskCount; i++) {
            boolean tooLong = i % 2 == 0;
            AddIntegerTask task = new AddIntegerTask(tooLong, 10, 7);
            outputMap.put(task.getUniqueIdentifier(), null);

            MyRunnable runnable = new MyRunnable(outputMap, task, latch);
            Future<?> future = executorService.submit(runnable);
            //schedule cancel task to run once, 2 seconds after scheduling
            executorService.schedule(new CancelTask(future), 2, TimeUnit.SECONDS);
        }
        latch.await();
        System.out.println("execution took - " + (System.nanoTime() - start) / 1_000_000_000D);
        executorService.shutdown();
        outputMap.values().forEach(System.out::println);
    }
}

I am using LinkedHashMap in order to keep the tasks in their order of submission.

Answer 5

Here's a fleshed out version of what I suggested in a comment to the question. The idea is to wrap a call to get(long timeout, TimeUnit unit) into another future. I encapsulate the required logic into a BetterFuture class, which delegates to a CompletableFuture under the hood:

import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.runAsync;
import static java.util.stream.Stream.concat;

import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.stream.Stream;

public class BetterFuture<T> {
    private final CompletableFuture<T> delegate;

    private BetterFuture(CompletableFuture<T> delegate) {
        this.delegate = delegate;
    }

    public static <T> BetterFuture<T> completed(T value) {
        return new BetterFuture<>(completedFuture(value));
    }

    public static <T> BetterFuture<T> future(Executor executor, Callable<T> callable) {
        CompletableFuture<T> delegate = new CompletableFuture<T>();

        runAsync(() -> {
            try {
                delegate.complete(callable.call());
            } catch (Throwable e) {
                delegate.completeExceptionally(e);
            }
        }, executor);

        return new BetterFuture<>(delegate);
    }

    public static <T> BetterFuture<Optional<T>> future(Executor executor, Callable<T> callable, Duration timeout) {
        return future(executor, () -> future(executor, callable).get(timeout));
    }

    public <R> BetterFuture<R> map(Function<T, R> fn) {
        return new BetterFuture<>(delegate.thenApply(fn));
    }

    public <R> BetterFuture<R> andThen(Function<T, BetterFuture<R>> fn) {
        return new BetterFuture<>(
            delegate.thenCompose(value -> fn.apply(value).delegate));
    }

    public static <T> BetterFuture<Stream<T>> collect(Stream<BetterFuture<T>> futures) {
        return futures
            .map(future -> future.map(Stream::of))
            .reduce(
                BetterFuture.completed(Stream.empty()),
                (future1, future2) ->
                future1
                    .andThen(stream1 ->
                future2
                    .map(stream2 ->
                concat(stream1, stream2)))
            );
    }

    public T get() throws ExecutionException, InterruptedException {
        return delegate.get();
    }

    public Optional<T> get(Duration timeout) throws ExecutionException, InterruptedException {
        try {
            return Optional.of(delegate.get(timeout.toMillis(), TimeUnit.MILLISECONDS));
        } catch (TimeoutException e) {
            return Optional.empty();
        }
    }

}

Most of the methods just delegate to the underlying CompletableFuture without adding much additional functionality.

To start an async task with a timeout, use the method

<T> BetterFuture<Optional<T>> future(Executor executor, Callable<T> callable, Duration timeout) 

If a timeout occurs, it completes with empty and with an optional of T otherwise.

In addition the method

public static <T> BetterFuture<Stream<T>> collect(Stream<BetterFuture<T>> futures) 

provides a convenient way for collecting a stream of futures into a future of a stream of the same type:

Stream<BetterFuture<Optional<String>>> futures = ...
BetterFuture<Stream<Optional<String>>> futureStream = BetterFuture.collect(futures);

Here's a full fledged examples where the first future times out and the second one completes successfully:

@Test
public void timeoutTest() throws ExecutionException, InterruptedException {
    ExecutorService executor = Executors.newCachedThreadPool();

    BetterFuture<Optional<String>> fa = BetterFuture.future(executor, () -> {
        Thread.sleep(3000);
        return "a";
    }, Duration.ofSeconds(2));

    BetterFuture<Optional<String>> fb = BetterFuture.future(executor, () -> {
        Thread.sleep(1000);
        return "b";
    }, Duration.ofSeconds(2));

    Stream<BetterFuture<Optional<String>>> futures = Stream.of(fa, fb);
    BetterFuture<Stream<Optional<String>>> c = BetterFuture.collect(futures);

    System.out.println(c.get().toList());
}

When run, it prints

[Optional.empty, Optional[b]]

As a final note, the implementation does nothing about the running threads when a timeout occurs. That is, it only times out the future but it does not interrupt the running thread. The thread will keep running in the background until in completes naturally.

Answer 6

It totally depends on the tasks / calculations you are running in parallel if you can really cancel them. Keep in mind that the Java Runtime is not an operating system, and you cannot forcible kill a thread like you could do with a process.

So if you want to interrupt long-running calculations, you will have to write time in a way so they regularly check if they should stop execution. For waiting on some other stuff (sleep, sync on other threads etc.) it is a totally different strategy: you can interrupt such threads and the code receives a InterruptedException that can be used to really stop the code with much less cooperation from the code.

I prepared a small example here to show you the difference:

package examples.stackoverflow.q71322315;

import java.util.concurrent.*;

public class Q71322315 {
    public static final long COUNTER = 10000000000L;
    public static final boolean SLEEP = false;

    private static final ExecutorService taskExec = Executors.newCachedThreadPool();

    public static void timedRun(Runnable r, long timeout, TimeUnit unit) throws InterruptedException {
        Future<?> task = taskExec.submit(r);
        try {
            task.get(timeout, unit);
            System.out.println("completed");
        } catch (TimeoutException e) {
            // task will be cancelled below
            System.out.println("timeout");
        } catch (ExecutionException e) {
            System.out.println("exex");
            // exception thrown in task; rethrow
            throw new RuntimeException(e.getCause());
        } finally {
            // Harmless if task already completed
            task.cancel(true); // interrupt if running
        }
    }

    public static void main(String[] args) throws InterruptedException {
        timedRun(new Task(SLEEP), 2000, TimeUnit.MILLISECONDS);
        taskExec.shutdown();
        System.out.println("finish");
    }
    private static class Task implements Runnable {

        private final boolean sleep;
        private Task(boolean sleep) {
            this.sleep = sleep;
        }
        @Override
        public void run() {
            try {
                if (sleep) {
                    Thread.sleep(5000L);
                } else {
                    longRunningMethod(COUNTER);
                }
                System.out.println("Success");
            } catch (Exception e) {
                e.printStackTrace();
                if (e instanceof InterruptedException) {
                    Thread.currentThread().interrupt();
                }
            }
        }

        private void longRunningMethod(long counter) {
            for (long y = 0; y < counter; y++) {
                Math.sqrt(y);
            }
        }
    }
}

The example is based on some example code of the already mentioned Java Concurrency in Practice - "7.10 Cancelling a task using Future."

The code as above performs a long running calculation that doesn't care about any interruptions. (You might have to increase the value of COUNTER, just add zeros at the end until the whole method takes longer than 2 seconds.)

You will see that you first get the "timeout" message that indicates that the task wasn't finished in the wanted timeout. But the code continues running and also prints out "finish" and "Success".

When you flip the SLEEP constant to true it uses an interruptible call to Thread.sleep() instead and the output will not contain the "Success" message.

---

After you managed to build a cancellable / interruptible computation you can then set up multiple threads that each execute the timedRun execution in parallel, so the tasks are started in parallel and also interrupted after the timeout.

This does not yet include the collection of the results, but instead of the sysouts for completed and timeout you can collect results or count the timed out tasks.

(And if you want to use that code in production, please clean it up very thoroughly, it has really some huge smells that should never land in any production ready code ;-)

Answer 7

We had similar requirement where we need capture timeout of each thread and ignore the results. Java 8 doesn't have this in built. One of the ways we achieved it,

List<CompletableFuture<?>> futures = new ArrayList<>();
List<?> results = new ArrayList<>(); // It can be anything you collect
futures.add(asyncService.fetchMethod()
.acceptEither(
    timeoutAfter(timeout, TimeUnit.SECONDS),
    results:add)
.handle(
   (result, ex) -> {
     //Handle the timeout exception
        results.add(...);
       return result
    });
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

---

private <T> CompletableFuture<T> timeoutAfter(long timeout, TimeUnit unit) {
  CompletableFuture<T> result = new CompletableFuture<>();
  // We need a separate executor here
  scheduledExecutor.schedule(
    () -> result.completeExceptionally(new TimeoutException()), timeout, unit);
  );
  return result;
}

Answer 8

Here is what helped me.

Problem in hand: Given X seconds as timeout return the value from the Task which completes first, in case none of the Task is able to finish then return default value.

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class Test {

    public static void main(String[] args) {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> func("Task 1", 1000));
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> func("Task 2", 2000));
        String str = null;
        try {
            str = (String) CompletableFuture.anyOf(f1, f2).get(3, TimeUnit.SECONDS);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            str = "Default";
            e.printStackTrace();
        }

        System.out.println(str);
    }

    public static String func(String task, int sleepTime) {
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return task;
    }
}