Parallel for loop in Java 8 - follow-up

Question

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;
    }
}

Answer 1

MyTask seems unneccesary. You can make tasks a list of Callable<O> and then add tasks with a lambda instead:

tasks.add(() -> body.apply(inputList.get(i)));

You're using forEach on the stream, but the operation you're doing more closely matches map. I'd try to use that instead.

The stream api can already do this with parallelStream and map:

public static final <I, O> List<O> forp(final List<I> inputList, final Function<I, O> body) {
    return inputList.parallelStream()
        .map(body)
        .collect(Collectors.toList());
}

You're catching exceptions and silently replacing with null. If this is what you want, document it in the javadoc. Alternatively, you could throw an exception from forp.

The thread-safety problems are going to come from other threads manipulating inputList or side-effects executed within body. The synchronization you have isn't preventing this, but I would say it's the caller's responsibility to not do anything like that. You could actually just remove the synchronization.