The previous and initial iteration at Parallel for loop in Java 8.
Changes are as follows:
ParallelLoopBody
removed;java.util.Function
used instead.- The actual parallel loop synchronizes on a class monitor.
forp
does not ask for the output list in arguments; instead, it creates a new list of output data on each call.
The modified files:
ParallelFor.java
:
package net.coderodde.util;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.function.Function;
import static net.coderodde.util.Utilities.checkNotNull;
/**
* This class implements the parallel for loop. It is assumed that two distinct
* tasks in the loop are independent, i.e., one task needs no output data from
* another task.
*
* @author Rodion Efremov
* @version I
*/
public class ParallelFor {
/**
* The thread pool doing the actual work.
*/
private static final ForkJoinPool POOL;
private static final Object PARALLEL_LOCK;
static {
POOL = new ForkJoinPool();
PARALLEL_LOCK = new Object();
}
/**
* Implements the actual parallel for loop.
*
* @param <I> the type of input data.
* @param <O> the type of output data.
* @param inputList the list of individual arguments to the routine to
* parallelize.
* @param body the actual function transforming an input datum into a
* output datum. May be specified in the form of a
* lambda expression.
* @return the list of output data in the same order as input data.
*/
public static final <I, O>
List<O> forp(final List<I> inputList,
final Function<I, O> body) {
synchronized (PARALLEL_LOCK) {
// Create the tasks.
final List<MyTask<I, O>> tasks = new ArrayList<>(inputList.size());
for (int i = 0; i < inputList.size(); ++i) {
tasks.add(new MyTask<>(body, inputList.get(i)));
}
final List<Future<O>> futures = POOL.invokeAll(tasks);
final List<O> output = new ArrayList<>(futures.size());
futures.stream().forEach((future) -> {
try {
final O out = future.get();
output.add(out);
} catch (final InterruptedException | ExecutionException e) {
output.add(null);
}
});
return output;
}
}
/**
* Implements the actual individual task.
*
* @param <I> the type of input data.
* @param <O> the type of output data.
*/
private static class MyTask<I, O> implements Callable<O> {
/**
* Contains the actual code for transforming input to output.
*/
private final Function<I, O> body;
/**
* The input datum.
*/
private final I input;
/**
* Constructs a new task which requires output.
*
* @param body the loop body implementation.
* @param input the input datum.
*/
MyTask(final Function<I, O> body, final I input) {
checkNotNull(body, "The parallel loop body is null.");
this.body = body;
this.input = input;
}
/**
* Runs this task by transforming input using the loop body to output
* and stores it in the output list (if needed).
*
* @return <code>null</code> as a dummy return value.
*/
@Override
public O call() throws Exception {
return body.apply(input);
}
}
}
Demo.java
:
package net.coderodde.util;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import static net.coderodde.util.ParallelFor.forp;
import static net.coderodde.util.Utilities.listsEqual;
/**
* This class contains the entry point to the demo demonstrating the parallel
* for loop and provides some demo-related utility methods.
*
* @author Rodion Efremov
* @versio I
*/
public class Demo {
private static final int INPUT_LENGTH = 50;
private static final int MIN = 10;
private static final int MAX = 40;
public static void main(final String... args) {
final List<Integer> input = new ArrayList<>();
final List<Long> output = new ArrayList<>();
final long seed = System.currentTimeMillis();
final Random rnd = new Random(seed);
final List<Integer> inputList = getRandomInputList(INPUT_LENGTH,
MIN,
MAX,
rnd);
final List<Long> serialResult = new ArrayList<>();
System.out.println("Seed: " + seed);
long ta = System.currentTimeMillis();
for (final Integer i : inputList) {
serialResult.add(hardFibonacciWork(i));
}
long tb = System.currentTimeMillis();
System.out.println("Serial time: " + (tb - ta) + " ms.");
ta = System.currentTimeMillis();
//// CHECK THIS OUT: MY FUNKY PARALLEL FOR
final List<Long> parallelResult =
forp(inputList, (i) -> hardFibonacciWork(i));
//////////////////////////////////
tb = System.currentTimeMillis();
System.out.println("Parallel time: " + (tb - ta) + " ms.");
boolean identical = listsEqual(serialResult, parallelResult);
System.out.println("Output lists identical: " + identical);
if (identical) {
for (int i = 0; i < inputList.size(); ++i) {
System.out.printf("%2d: %-10d %-10d\n", inputList.get(i),
serialResult.get(i),
parallelResult.get(i));
}
}
}
private static long hardFibonacciWork(final int i) {
if (i <= 0) {
return 0L;
}
if (i == 1) {
return 1L;
}
return hardFibonacciWork(i - 1) + hardFibonacciWork(i - 2);
}
private static List<Integer> getRandomInputList(final int size,
final int min,
final int max,
final Random rnd) {
final List<Integer> list = new ArrayList<>(size);
for (int i = 0; i < size; ++i) {
list.add(rnd.nextInt(max - min + 1) + min);
}
return list;
}
}