Using ExecutorService to repeatedly perform a number of similar tasks in parallel

Question

There is Java code which when simplified looks like this:

while(someCondition)
{
    SomeType a = CalcResult(param1);
    SomeType b = CalcResult(param2);
    SomeType c = CalcResult(param3);

    // Do something with a, b and c
}

CalcResult() is time consuming. The application runs on an SMP system. There is a push to try running all three calculations simultaneously each on its own CPU instead of running them sequentially. It is always those 3 tasks that need to be parallel, not an arbitrary number of tasks (such is the algorithm). Each task may take slightly more or less time than others, but generally the variances are not so large (20-30%).

As they need to return a result, I looked at the Executor service solutions such as this from https://stackoverflow.com/a/9148992/2721750:

ExecutorService executor = Executors.newSingleThreadExecutor();
Callable<Integer> callable = new Callable<Integer>() {
    @Override
    public Integer call() {
        return 2;
    }
};
Future<Integer> future = executor.submit(callable);
// future.get() returns 2
executor.shutdown();

As my experience with Java is mainly in servlet/JSP development I have no experience with threads and not sure if that snippet would work with 3 tasks instead of one.

How can I submit 3 tasks, each with its own parameter value, and wait untill all of them return the result of calculation, simultaneously making sure that creating threads for them does not negate the advantage of running on their own CPUs, i.e. is there a way to create the threads once before the while() loop strarts, then simply shove a new paramN into each of the threads inside the loop, waking them up, and wait until they performed all of the calculations?

Leave the executor around (or use e.g. ForkJoinPool.commonPool()) so you aren't paying to start it up and stop it each time. — Jeffrey Bosboom, Feb 20 '15 at 15:17
Note: a **SingleThreadExecutor** has exactly **one** Thread. So you'll want to use another one. I second Jeffrey's recommendation for ForkJoinPool. — Fildor, Feb 20 '15 at 15:18
Like I said, I have no expeirnce with Java threads whatsoever, so it is still not clear how to use your recommendation. — ajeh, Feb 20 '15 at 15:24
Actually, you're almost there. Calling `get` on the future will block until the result is ready. So if you submit the three tasks and safe the futures, then call the get methods of those three sequentially, you can be sure that all three have finished when the last one returns. If the last one finishes before let's say the second then it will return immediately with the result. For parallelism just don't use a SingleThreadExecutor but for example a Fixed one as recommended in the answers. — Fildor, Feb 20 '15 at 15:30
Got it, thanks! Just struggling with specific implementation as mentioned down at Adrian's answer. — ajeh, Feb 20 '15 at 17:01
*"I have no experience with Java threads"* Then you are playing with fire here. Concurrency is tricky because there are failure modes that are not possible with sequential programming, and yet the runtime does not save you from your own mistakes. So before you go pasting code from SO, you should probably go read at least chapters 1-6 of [*Java Concurrency in Practice*](http://amzn.to/1jyE5Kx) (I am the author) to understand what's going on. — Brian Goetz, Feb 21 '15 at 15:58

Adrian Leonhard · Accepted Answer · 2015-02-20T16:17:36.027

5

Executors.newSingleThreadExecutor() will create only a single thread. What you want is Executors.newFixedThreadPool(3). Call this before the while loop, so the threads are only created once.

Create a Callable wrapper:

class MyCallable implements Callable<V> {
    P p;
    MyCallable(P parameter) {
        p = parameter;
    }
    V call() {
        return CalcResult(p);
    }
}

while loop:

ExecutorService executor = Executors.newFixedThreadPool(3);
while (cond) {
    Future<V> aFuture = executor.submit(new MyCallable(param1));
    Future<V> bFuture = executor.submit(new MyCallable(param2));
    Future<V> cFuture = executor.submit(new MyCallable(param3));

    // this will block until all calculations are finished:
    a = aFuture.get();
    b = bFuture.get();
    c = cFuture.get();

   // do something with a/b/c, e.g. calc new params.
}

edited Feb 20 '15 at 16:17

answered Feb 20 '15 at 15:21

Adrian Leonhard

7,040
2
24
38

Sorry, but I am still not clear how to invoke and wait for completion, would you be willing to add those parts to the answer? Thx! – ajeh Feb 20 '15 at 15:23
3

@ajeh See the second snippet ... each call of `get` will block until the corresponding future has finished its task. So when `cFuture.get()` returns, all three tasks have completed. – Fildor Feb 20 '15 at 15:35
Yes, saw the eidt - thank you! My problem is that `CalcResult` is a library function, which returns a value. Does it need to be `return CalcResult(p)`? – ajeh Feb 20 '15 at 15:40
Was `submit` replaced with `execute` in JDK 8 according to https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executor.html Must be the case, as `submit` symbol is not found. – ajeh Feb 20 '15 at 16:06
1

@ajeh, sorry executor must be declared as an ExecutorService – Adrian Leonhard Feb 20 '15 at 16:11
Still struggling with this. When I declare `class MyCallable implements Callable` compiler errors: `error: method CalcResult in class AsyncThreadBenchmark cannot be applied to given types; return AsyncThreadBenchmark.CalcResult(p); required: java.lang.Integer found: Integer reason: argument mismatch; Integer cannot be converted to java.lang.Integer where Integer is a type-variable: Integer extends Object declared in class MyCallable` Full code at http://pastebin.com/viHhqhaF – ajeh Feb 20 '15 at 16:57
2

@ajeh change `class MyCallable` to `class MyCallable` and move the class definition to a separate file. – Adrian Leonhard Feb 20 '15 at 17:15

score 1 · Answer 2 · answered Feb 20 '15 at 15:26

You could create an executor service for your application, specifically for processing these calculations. The pool size may vary depending on if you ever were going to run multiple calculations at the same time:

ExecutorService service = Executors.newFixedThreadPool(3);

Future<Integer> submitA = service.submit(new Callable<Integer>() {
  @Override
  public Integer call() throws Exception {
    return processA();
  }
});
Future<Integer> submitB = service.submit(new Callable<Integer>() {
  @Override
  public Integer call() throws Exception {
    return processB();
  }
});
Future<Integer> submitC = service.submit(new Callable<Integer>() {
  @Override
  public Integer call() throws Exception {
    return processC();
  }
});

int result = submitA.get() + submitB.get() + submitC.get();

In this case you would want to make sure that you are not creating the Thread pool each time you run the calculation, in that case the impact of creating the pool will be minor compared to the saving - assuming that the running time will be reduced by splitting the task.

Using ExecutorService to repeatedly perform a number of similar tasks in parallel

2 Answers2

Linked