Massive tasks alternative pattern for Runnable or Callable

Question

For massive parallel computing I tend to use executors and callables. When I have thousand of objects to be computed I feel not so good to instantiate thousand of Runnables for each object.

So I have two approaches to solve this:

I. Split the workload into a small amount of x-workers giving y-objects each. (splitting the object list into x-partitions with y/x-size each)

public static <V> List<List<V>> partitions(List<V> list, int chunks) {
      final ArrayList<List<V>> lists = new ArrayList<List<V>>();
      final int size = Math.max(1, list.size() / chunks + 1);
      final int listSize = list.size();
      for (int i = 0; i <= chunks; i++) {
         final List<V> vs = list.subList(Math.min(listSize, i * size), Math.min(listSize, i * size + size));
         if(vs.size() == 0) break;
         lists.add(vs);
      }
      return lists;
   }

II. Creating x-workers which fetch objects from a queue.

Questions:

• Is creating thousand of Runnables really expensive and to be avoided?
• Is there a generic pattern/recommendation how to do it by solution II?
• Are you aware of a different approach?

Answer 1

Creating thousands of Runnable (objects implementing Runnable) is not more expensive than creating a normal object.

Creating and running thousands of Threads can be very heavy, but you can use Executors with a pool of threads to solve this problem.

Answer 2

As for the different approach, you might be interested in java 8's parallel streams.

Answer 3

Combining various answers here :

Is creating thousand of Runnables really expensive and to be avoided?

No, it's not in and of itself. It's how you will make them execute that may prove costly (spawning a few thousand threads certainly has its cost). So you would not want to do this :

List<Computation> computations = ...
List<Thread> threads = new ArrayList<>();
for (Computation computation : computations) {
    Thread thread = new Thread(new Computation(computation));
    threads.add(thread);
    thread.start();
}
// If you need to wait for completion:
for (Thread t : threads) {
    t.join();
}

Because it would 1) be unnecessarily costly in terms of OS ressource (native threads, each having a stack on the heap), 2) spam the OS scheduler with a vastly concurrent workload, most certainly leading to plenty of context switchs and associated cache invalidations at the CPU level 3) be a nightmare to catch and deal with exceptions (your threads should probably define an Uncaught exception handler, and you'd have to deal with it manually).

You'd probably prefer an approach where a finite Thread pool (of a few threads, "a few" being closely related to your number of CPU cores) handles many many Callables.

List<Computation> computations = ...
ExecutorService pool = Executors.newFixedSizeThreadPool(someNumber)
List<Future<Result>> results = new ArrayList<>();
for (Computation computation : computations) {
    results.add(pool.submit(new ComputationCallable(computation));
}
for (Future<Result> result : results {
    doSomething(result.get);
}

The fact that you reuse a limited number threads should yield a really nice improvement.

Is there a generic pattern/recommendation how to do it by solution II?

There are. First, your partition code (getting from a List to a List<List>) can be found inside collection tools such as Guava, with more generic and fail-proofed implementations.

But more than this, two patterns come to mind for what you are achieving :

1. Use the Fork/Join Pool with Fork/Join tasks (that is, spawn a task with your whole list of items, and each task will fork sub tasks with half of that list, up to the point where each task manages a small enough list of items). It's divide and conquer. See: http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ForkJoinTask.html

If your computation were to be "add integers from a list", it could look like (there might be a boundary bug in there, I did not really check) :

public static class Adder extends RecursiveTask<Integer> {
protected List<Integer> globalList;
protected int start;
protected int stop;

public Adder(List<Integer> globalList, int start, int stop) {
  super();
  this.globalList = globalList;
  this.start = start;
  this.stop = stop;
  System.out.println("Creating for " + start + " => " + stop);
}

@Override
protected Integer compute() {
  if (stop - start > 1000) {
    // Too many arguments, we split the list
    Adder subTask1 = new Adder(globalList, start, start + (stop-start)/2);
    Adder subTask2 = new Adder(globalList, start + (stop-start)/2, stop);
    subTask2.fork();
    return subTask1.compute() + subTask2.join();
  } else {
    // Manageable size of arguments, we deal in place
    int result = 0;
    for(int i = start; i < stop; i++) {
      result +=i;
    }
    return result;
  }
}
}

public void doWork() throws Exception {
List<Integer> computation = new ArrayList<>();
for(int i = 0; i < 10000; i++) {
  computation.add(i);
}
ForkJoinPool pool = new ForkJoinPool();

RecursiveTask<Integer> masterTask = new Adder(computation, 0, computation.size());
Future<Integer> future = pool.submit(masterTask);
System.out.println(future.get());

}

2. Use Java 8 parallel streams in order to launch multiple parallel computations easily (under the hood, Java parallel streams can fall back to the Fork/Join pool actually).

Others have shown how this might look like.

Are you aware of a different approach?

For a different take at concurrent programming (without explicit task / thread handling), have a look at the actor pattern. https://en.wikipedia.org/wiki/Actor_model Akka comes to mind as a popular implementation of this pattern...

Answer 4

@Aaron is right, you should take a look into Java 8's parallel streams:

void processInParallel(List<V> list) {
    list.parallelStream().forEach(item -> {
        // do something
    });
}

If you need to specify chunks, you could use a ForkJoinPool as described here:

void processInParallel(List<V> list, int chunks) {
    ForkJoinPool forkJoinPool = new ForkJoinPool(chunks);
    forkJoinPool.submit(() -> {
        list.parallelStream().forEach(item -> {
            // do something with each item
        });
    });
}

You could also have a functional interface as an argument:

 void processInParallel(List<V> list, int chunks, Consumer<V> processor) {
    ForkJoinPool forkJoinPool = new ForkJoinPool(chunks);
    forkJoinPool.submit(() -> {
        list.parallelStream().forEach(item -> processor.accept(item));
    });
}

Or in shorthand notation:

void processInParallel(List<V> list, int chunks, Consumer<V> processor) {
    new ForkJoinPool(chunks).submit(() -> list.parallelStream().forEach(processor::accept));
}

And then you would use it like:

processInParallel(myList, 2, item -> {
    // do something with each item
});

Depending on your needs, the ForkJoinPool#submit() returns an instance of ForkJoinTask, which is a Future and you may use it to check for the status or wait for the end of your task.

You'd most probably want the ForkJoinPool instantiated only once (not instantiate it on every method call) and then reuse it to prevent CPU choking if the method is called multiple times.

Answer 5

Is creating thousand of Runnables really expensive and to be avoided?

Not at all, the runnable/callable interfaces have only one method to implement each, and the amount of "extra" code in each task depends on the code you are running. But certainly no fault of the Runnable/Callable interfaces.

Is there a generic pattern/recommendation how to do it by solution II?

Pattern 2 is more favorable than pattern 1. This is because pattern 1 assumes that each worker will finish at the exact same time. If some workers finish before other workers, they could just be sitting idle since they only are able to work on the y/x-size queues you assigned to each of them. In pattern 2 however, you will never have idle worker threads (unless the end of the work queue is reached and numWorkItems < numWorkers).

An easy way to use the preferred pattern, pattern 2, is to use the ExecutorService invokeAll(Collection<? extends Callable<T>> list) method.

Here is an example usage:

List<Callable<?>> workList = // a single list of all of your work
ExecutorService es = Executors.newCachedThreadPool();
es.invokeAll(workList);

Fairly readable and straightforward usage, and the ExecutorService implementation will automatically use solution 2 for you, so you know that each worker thread has their use time maximized.

Are you aware of a different approach?

Solution 1 and 2 are two common approaches for generic work. Now, there are many different implementation available for you choose from (such as java.util.Concurrent, Java 8 parallel streams, or Fork/Join pools), but the concept of each implementation is generally the same. The only exception is if you have specific tasks in mind with non-standard running behavior.