• 1

• 31.7k
• 15
• 77
• 202

The previous and initial iteration at Parallel for loop in Java 8.

Changes are as follows:

1. ParallelLoopBody removed; java.util.Function used instead.
2. The actual parallel loop synchronizes on a class monitor.
3. 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;
 }
}