4
\$\begingroup\$

This is my first attempt to provide some syntactic sugar for doing mutually independent loop iterations in parallel. Thanks to Java 8 lambdas, I can write the parallel loops in pretty elegant fashion compared to pre-lambda Java's.

Also, I have declared the actual forp as synchronized, which leads to the question will it be sufficient to make sure that two or more threads trying to use forp will get to it in some non-overlapping order?

My code is as follows:

ParallelFor.java:

package net.coderodde.util;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ForkJoinPool;
import static net.coderodde.util.Utilities.checkNotNull;
import static net.coderodde.util.Utilities.checkRange;

/**
 * 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;

    static {
        pool = new ForkJoinPool();
    }

    /**
     * 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 outputList the list of output data, may be <code>null</code> 
     *                   whenever no output is needed.
     * @param body       the actual code. May be specified in the form of a 
     *                   lambda expression.
     */
    public static synchronized final <I, O> 
        void forp(final List<I> inputList,
                  final List<O> outputList,
                  final ParallelLoopBody<I, O> body) {
            // Create the tasks.
            final List<MyTask<I, O>> tasks = new ArrayList<>(inputList.size());

            if (outputList == null) {
                // Tasks with no output.
                for (int i = 0; i < inputList.size(); ++i) {
                    tasks.add(new MyTask<>(body, inputList.get(i)));
                }
            } else {
                outputList.clear();

                // Tasks with output.
                for (int i = 0; i < inputList.size(); ++i) {
                    outputList.add(null);
                    tasks.add(new MyTask<>(body, 
                                           inputList.get(i), 
                                           outputList, 
                                           i));
                }
            }   

            pool.invokeAll(tasks);
    }

    /**
     * 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<Object> {

        /**
         * Contains the actual code for transforming input to output.
         */
        private final ParallelLoopBody<I, O> body;

        /**
         * The input datum.
         */
        private final I input;

        /**
         * The list where the output datum is to be placed.
         */
        private final List<O> outputList;

        /**
         * The index in the output list where the output datum is to be placed.
         */
        private final int outputIndex;

        /**
         * Constructs a new task which requires output.
         * 
         * @param body        the loop body implementation.
         * @param input       the input datum.
         * @param outputList  output list.
         * @param outputIndex index within the output list.
         */
        MyTask(final ParallelLoopBody<I, O> body, 
             final I input, 
             final List<O> outputList,
             final int outputIndex) {
            checkNotNull(body, "The parallel loop body is null.");
            checkNotNull(outputList, "The output list is null.");
            checkRange(outputIndex, 
                       outputList.size(), 
                       "The output index is outside the range " +
                       "[0, " + outputList.size() + "); index: " + 
                       outputIndex + ".");
            this.body = body;
            this.outputList = outputList;
            this.outputIndex = outputIndex;
            this.input = input;
        }

        /**
         * Creates a task that does not produce any output.
         * 
         * @param body  the loop body implementation.
         * @param input the input datum.
         */
        MyTask(final ParallelLoopBody<I, O> body,
               final I input) {
            checkNotNull(body, "The parallel loop body is null.");
            this.body = body;
            this.input = input;
            this.outputList = null;
            this.outputIndex = -1;
        }

        /**
         * 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 Object call() throws Exception {
            final O output = body.execute(input);

            if (outputList != null) {
                outputList.set(outputIndex, output);
            }

            return null;
        }
    }
}

ParallelLoopBody.java:

package net.coderodde.util;

/**
 * This interface defines the API for parallel for loop body implementation.
 * 
 * @author Rodion Efremov
 * @version I
 *
 * @param <I> the type of input data.
 * @param <O> the type of output data.
 */
public interface ParallelLoopBody<I, O> {

    /**
     * Transform the input datum to output datum.
     * 
     * @param input the input datum.
     * @return output datum.
     */
    public O execute(final I input);
}

Utilities.java:

package net.coderodde.util;

import java.util.List;

/**
 * This class contains some utility methods.
 * 
 * @author Rodion Efremov
 * @version I
 */
public class Utilities {

    /**
     * Checks that the input reference is not null.
     * 
     * @param reference the reference to check.
     * @param message the error message printed when the test fails.
     */
    public static final void checkNotNull(final Object reference,
                                          final String message) {
        if (reference == null) {
            throw new NullPointerException(message);
        }
    }

    /**
     * Checks that the input index is within range <tt>[0, size)</tt>.
     * 
     * @param index the index to check.
     * @param size the size of the container indexed.
     * @param message the error message printed when the test fails.
     */
    public static final void checkRange(final int index, 
                                        final int size,
                                        final String message) {
        if (index < 0 || index >= size) {
            throw new IndexOutOfBoundsException(message);
        }
    }

    /**
     * Checks whether the input lists are of same length and have the same 
     * content according to <code>equals</code>.
     * 
     * @param <E> the list component type.
     * @param lists the actual lists.
     * @return <code>true</code> if the lists are equal, <code>false</code>
     * otherwise.
     */
    public static final <E> boolean listsEqual(final List<E>... lists) {
        if (lists.length == 0) {
            return true;
        }

        for (int i = 0; i < lists.length - 1; ++i) {
            if (lists[i].size() != lists[i + 1].size()) {
                return false;
            }
        }

        for (int i = 0; i < lists[0].size(); ++i) {
            for (int j = 0; j < lists.length - 1; ++j) {
                if (!lists[j].get(i).equals(lists[j + 1].get(i))) {
                    return false;
                }
            }
        }

        return true;
    }
}

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<>();
        final List<Long> parallelResult = 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
        forp(inputList, 
             parallelResult, 
             (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;
    }
}

So, what do you think?

\$\endgroup\$
2
\$\begingroup\$

Your code assumes that List<T>.set(int, T) of the result list is thread safe for different indexes which may not be the case. You resizing will fail when someone uses the fixed-length Arrays.asList(...) to store the result.

Instead create a second list to temporarily store the result (in a thread safe manner) and then fill the passed in outputList using a list iterator:

ListIterator<O> listIt = outputList.listIterator();

for(O result: results){ //results is the temporary result array
    if(listIt.hasNext()){
        listIt.next();
        listIt.set(result);
    }else{
        listIt.add(result);
    }
}

There is already a 1 argument+ return type defined: java.util.Function

\$\endgroup\$
  • \$\begingroup\$ I am struggling to understand how making the store in to a ListIterator instead of directly on to the outputList makes this in any way 'safer'. That makes no sense. Your first paragraph is the reason I +1'd your answer, but the rest of your answer is plain illogical. Note to the OP, do not use the list iterator concept recommended here (as a tool to avoid concurrency problems). \$\endgroup\$ – rolfl Dec 18 '14 at 19:26
  • \$\begingroup\$ @rolfl I meant it after the invokeAll call (when all the tasks are done) in the main method and the listIterator will not resize unless necessary. I also suggested the listIterator to allow a non-randomaccess List to get filled it in O(n). \$\endgroup\$ – ratchet freak Dec 18 '14 at 19:35
  • \$\begingroup\$ OK, about the fill, an addAll() will solve any O(n) concerns you may have. Also, note that the tasks ar eot done after the invokeAll... you will have to wait for each Future to complete, and you are doing that inside a synchronized method, so really, there's no point in that. \$\endgroup\$ – rolfl Dec 18 '14 at 19:41
4
\$\begingroup\$

There are a number of concurrency issues in this code,. I can point out what I see, but correcting them will require some significant rethinking... and determining what the rewrite will need will take more than there's time for in this answer.

Concurrency issues:

  1. outputList is not controlled with any synchronization. Even though you pre-size the list to have a slot 'available' for each task, there's no reason to believe that on the completion of the task, when you run:

        @Override
        public Object call() throws Exception {
            final O output = body.execute(input);
    
            if (outputList != null) {
                outputList.set(outputIndex, output);
            }
    
            return null;
        }
    

    that the copy of the outputlist in your thread is going to be the same copy as the outputList on some other thread. The value you set in that outputList may not be visible to the controlling code.

    Admittedly, pre-sizing the List makes a difference on the entry side, but the exit conditions are not controlled.

    Additionally, note that a user calling your code with a Non-ArrayList (perhaps a LinkedList?) may get different results to one calling with an ArrayList.

  2. Code other than your code can do synchronized (ParallelFor.class) {...} and suddenly your code becomes unrunnable because you cannot get a synchronization lock on it.... Never use a publicly accessible member to control the concurrency of a class unless you need to make it a 'free for all'. Do something like:

    private static final Object PARALLEL_LOCK = new Object();
    

    and then, in your static methods, instead of making them static synchronized, do:

    public static void forp(...) {
        synchronized(PARALLEL_LOCK) {
            ....
        }
    }
    
  3. The static synchronized method means that only one thread in the whole JVM can be scheduling jobs at any one time. Is there even a need to synchronize it at all? The only external fields are the pool and the input parameters. The pool is already safe, and the input parameters are presumed to be unique for each caller... right?

  4. If the input parameters (outputList especially) are used for two separate calls to forp you have a problem, because one set of tasks will be overwriting the data in another set, and you may end up with concurrent modifications, and even ArrayIndexOutOfBounds exceptions

\$\endgroup\$
  • \$\begingroup\$ Is the synchronized (ParallelFor.class) {...} a valid concern? If such code exists, you can assume that it is not using the given classes/methods as intended... With the new way an attacker could still break code by obtaining the PARALLEL_LOCK via reflection, is that then not a valid concern? \$\endgroup\$ – skiwi Dec 19 '14 at 11:01
  • 1
    \$\begingroup\$ @skiwi - The concern is not about an attacker defeating your code, it's about exposing your locks to the world, and, at some point, someone will synchronize on it for a reason that's innocent enough, and causes bugs (or just performance problems) that are very hard to debug. Synchronizing on non-private fields, whether a static class, an instance, or a non-private field is a leading cause of concurrency bugs. \$\endgroup\$ – rolfl Dec 19 '14 at 11:11

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.