Parallelism and the Fork-Join Framework

Up til now, we have been working with entirely sequential computation. Our code tells our (single) processor what to do, and then it does it, line by line, stackframe by stackframe.

But suppose our computer has multiple processors, or even was have access to thousands of computers. How can we leverage this extra power?

By analogy, and ignoring the issue of crowding, having more cooks in your kitchen should allow you to get more done more quickly. One cook can peel the potatoes, while the other makes a salad, and so on and so forth. This is the basic idea of parallelism. This also potentially creates issues – one cook cannot simply change the oven temperature for their casserole if another cook is baking a pie at a different temperature. This issue of sharing resources and communicating is one of concurrency. Issues of concurrency we will mostly discuss later in the course, although parallelism and concurrency are always somewhat intertwined.

The example we will repeatedly come back to in these notes is the following: suppose we wanted to sum an array containing n integers. The naive sequential algorithm is of course

sum = 0
for i = 0 to n-1:
    sum = sum + arr[i]
return sum

which runs in O(n) time.

Let’s now explore some possibilities for how we could speed this up with parallel computations!

Parallel Sum Attempt 1

Let’s suppose our machine has exactly 10 processors. One idea would be to split our array into 10 chunks, and pass each chunk to a processor. Once each processor has finished its summation task, we could sum the results. A pseudocode for this idea is below:

total = 0
initialize sub_sums as an array of size 10;
IN PARALLEL:
    processor i computes sub_sums[i] = sum(arr[i*n//10 : (i+1)*n//10]);
WAIT for each processor to finish;
for i = 0 to 9:
    total += sub_sums[i];
return total

We have highlighted the “In parallel” and “waiting” steps in red to emphasize that different programming languages will implement these ideas in different ways. We will see how Java does it later. We also highlight the importance of the WAIT step – without that, the values in sub_sums could be all wrong, since the parallel processes may not be finished yet!

Let’s analyze the runtime of this algorithm. Each processor gets a sublist of length about n/10, so we expect the parallel part of the computation to take about n/10 time, and then after that it just takes constant time to sum the results of the individual processors. This is not too bad! We got roughly a factor 10 speedup over our original runtime, although both are still Θ(n) runtimes.

What if our computer did not have 10 processors, but rather had k processors? How much time could we save with a parallel algorithm of this form? In this case, we would split the list into chunks of size n/k roughly, and pass each chunk to a processor to be run in parallel. Once they have all finished, we would sum the k results. In this case, the parallel part of the computation would take $ O(n/k)$ time, while the for loop to sum up all the results would take O(k) time, for an overall runtime of O(n/k+k).

This is not quite as satisfying as one might hope. Indeed, if we had k = n processors, our runtime is still Θ(n)! It seems like we should certainly be able to do something better than this with so many processors at our disposal.

Exercise: What number k of processors gives the best runtime for this algorithm, and what is that runtime?

Parallel Sum Attempt 2 (Divide and Conquer)

In this section, we will describe a solution that becomes significantly more efficient. Indeed, with O(n) processors, it will turn out that we can sum our list in O(logn) time in parallel!

The idea is one that you have seen before: divide and conquer. More specifically, our algorithm will 1) check if the list is short (has less than some constant C elements), in which case just sum the items sequentially (this is the base case – we will see why later it is important to choose C appropriately) and 2) if the list is not short, will split the list into two equal subparts, and in parallel recurse on those sublists.

Here is a pseudocode implementation of this idea:

Define RecursiveSum(arr, lo, hi):
    total = 0;
    if hi - lo < C:  
        for i = lo to hi - 1:
            total += arr[i];
        return total;
    else:
        mid = (lo + hi)//2;
        IN PARALLEL:
            FirstHalfSum = RecursiveSum(arr, lo, mid);
            SecondHalfSum = RecursiveSum(arr, mid + 1, hi);
        WAIT for parallel calls to finish;
        total = FirstHalfSum + SecondHalfSum;
        return total;

It’s easiest to visualize this as a tree of recursive calls. The below picture illustrates a possible snapshot of our algorithm, in which some of the parallel computations on the right side of the tree have started finishing before those on the left, and our base case is C = 2.

What is the runtime of our algorithm? We will learn more about the precise runtime of parallel algorithms later, but for now we can make some relatively simple observations. 1) if we have only one processor, then the above algorithm is just an overcomplicated way to sum our list sequentially, so takes at least linear time. 2) If we have infinite processors, then every node in the diagram above can be assigned a unique processor which can run independently. Moreover, each of nodes above represents an O(1) amount of work, since we are either splitting the array in half, summing a constant sized array sequentially, or combining the results of two subproblems. So, in this case, our runtime is simply a constant times the height of the tree diagram above, which is O(logn) (exercise: why?)!

So, with lots of processors (exercise: what is the maximum number of processors used in the above algorithm?), we can achieve quite the parallel speed up!

Lower Level Details

We have admittedly in the first section of these notes ignored how parallelism is actually achieved on a computer and some of the finer implementation details and optimizations of these parallel algorithms. Let’s dive into that now. In these notes, we will discuss primarily how Java handles parallelism. In general, a programming language needs a way to tell the operating system that certain subtasks can and should be done in parallel, on different processors, if they are available, and there needs to be a way to communicate vital information between these subtasks happening in parallel, such as if they have concluded yet. We will discuss now how Java handles this with threads.

Threads

The old (sequential) story of how code is run on your computer is that there is 1) a program counter, which is essentially the current line of code being executed, 2) A single call stack, which is a stack of stack frames. Essentially, each stack frame corresponds to a method or function call, and has its own primitive local variables, parameters, and return address. 3) Objects in a memory heap (note: this has nothing to do with the heap data structure), i.e. anything you have to write new ____ for.

In the new story of code execution, we now have a collection of threads, which each have their own individual program counters and call stacks. Importantly, however, the memory heap is shared for all threads, meaning that different threads can all point to the same object.

The fact that threads share one memory heap is important because it allows some sort of communication/ability to work together among threads. For example, in our parallel algorithms for list summation, we can have many different threads reading the same object arr at the same time. But this shared memory heap can also be problematic – for example, we will of course have issues if two different threads are trying to modify the same object at the same time! In general, the issues that can arise with these shared resources (cook sharing one oven, in our analogy) are issues of concurrency, which we will look at in more detail later.

What our program (in Java) can do is create new threads, and then then tell the operatoring system to run these threads. It is then entirely up to the operating system how to assign these individual threads to processors in an efficient way. So although the computer may have only, for example, 32 processors, we may create millions of threads in Java, if we wish. These threads then go into a “thread pool”, and as the programmer we have no control as to the order in which they are executed, or which ones are being executed at the same time. This will also be important later when we discuss concurrency in more detail.

Let’s see how to implement our first algorithm, where we simply split the list into 10 parts and attempt to sum those in parallel, in java.

First, to create a thread object in java, it must extend java.lang.Thread and it must override the run() method, which must take no arguments and return nothing.

class SumThread extends java.lang.Thread {
    int lo; int hi; int[] arr; int ans = 0;
    SumThread(int[] a, int l, int h) {
        lo=l; hi=h; arr=a;
    }
    public void run() { //override, must have this signature
                        // no arguments and no return allowed
                        // must use the object’s fields for input/output
    for(int i=lo; i < hi; i++)
        ans += arr[i];
    }

Now, we can create threads by instantiating SumThread objects, but how do we use them? There are two important methods for Java threads: 1) .start(): this tells Java to call the .run() method in a new thread. This is important – if we just call .run() then we are doing the work in this current thread instead of a new thread. 2) .join(): this tells the current thread to wait for the other thread (the one we called .join on) to finish the computations in its run method. This is important for the “waiting” part of our pseudocode above (in particular, it prevents a “race condition”, which we will discuss more later).

Let’s see now how to implement our first (naive) parallel list summing algorithm above, but with Java threads.

static int parallelSum(int[] arr){ // this method could be anywhere
    int len = arr.length;
    int ans = 0;
    SumThread[] threads = new SumThread[4];
    for(int i=0; i < 10; i++){ // do parallel computations
        threads[i] = new SumThread(arr,i*len/10,(i+1)*len/10);
        threads[i].start(); // start not run
    }
    for(int i=0; i < 10; i++){ // combine results
        threads[i].join(); // wait for thread to finish!
        ans += threads[i].ans;
    }
    return ans;
}

Great! As we saw though, this is not necessarily the most efficient algorithm, especially if our computer has more processors it can access.

Let’s see then how we can accomplish the much better “divide and conquer” algorithm we saw before then!

class SumThread extends java.lang.Thread {
    public void run(){ // override
        if(hi – lo < SEQUENTIAL_CUTOFF) // “base case” 
            for(int i=lo; i < hi; i++) ans += arr[i];
        else {
            SumThread left = new SumThread(arr,lo,(hi+lo)/2); // divide
            SumThread right= new SumThread(arr,(hi+lo)/2,hi); // divide
            left.start(); right.start(); // conquer
            left.join(); right.join(); // wait
            ans = left.ans + right.ans; // combine 
        }
    }
}
int sum(int[] arr){ // just make one thread!
    SumThread t = new SumThread(arr,0,arr.length);
    t.run();
    return t.ans; }

We can see above how this is simply exactly implementing the pseudocode we described earlier. It is important to note that we start both the left and the right threads before calling join on either of them. If we instead had a code that looked like

left.start();
left.join(); 
right.start(); 
right.join();

Then we would’ve just created a very fancy sequential algorithm with a large overhead. It is sequential because the .join() call requires us to wait for left to finish before starting the right subtask. The reason it has high overhead is because there is some inherent behind-the-scences overhead in creating threads in Java (think of this as a non-negligible constant amount of runtime required each time we create a thread). This is the reason we have our SEQUENTIAL_CUTOFF, which is the same as our C before.

Choosing the Sequential Cutoff

As mentioned, creating threads has an overhead. Suppose our list has 2³⁰ items in it, which is about a billion. Then, if we choose our sequential cutoff to be 1, we are creating around a billion threads, which is quite costly. However, a vast majority of those threads will be created at the lowest levels of the recursion, at the leaves of our tree (or rather, in the middle of the two trees stacked on top of each other).

To see this, suppose we set our sequential cutoff to be 2¹⁰ = 1024. In this case, the height of the tree will be 20 rather than 30 (why?), so the number of times we make a new thread will on the order of 2²⁰ or about a million (why?). This will likely be much more efficient, even though we are doing more “sequential work” by setting the sequential cutoff larger.

Of course, we cannot make the sequential cutoff too high either, as then we will approach the runtime of the naive sequential algorithm.

In essence, the issue of what the sequential cutoff is simply the issue of how much overhead thread creation takes. In practice, you can just try a few values and see what works best, although typically somewhere on the order of a thousand seems to work well in Java.

A Minor Optimization

We pause here to mention a small optimization in the above code that should speed up runtime by a constant factor. Our current algorithm creates an initial thread, which creates two new threads (which then create more threads). The problem with this is that our initial thread (and indeed many of the subsequently created threads) is doing no work at all! It is simply sitting idly (twiddling its thumbs, if you will), waiting for the two threads it created to finish. In our “cooks in the kitchen analogy”, this would be like if a chef split the potatoes into two piles, and then assigned each of the piles to a different sous chefs, and then sat there waiting for them to finish. More efficient would be to have the original chef work on one of the piles themselves!

Coming back to our actual problem, a better way of doing things would be to have every thread do some actual work! The way to accomplish this in Java would be to call .run() on one of the two new threads, rather than .start(). This .run() call starts calling that method in the current thread, rather than a new one. This ultimately eliminates about half of the threads we would’ve ended up creating.

When we make this tweak, the body of our code will become:

SumThread left = new SumThread(arr,lo,(hi+lo)/2); // divide
SumThread right= new SumThread(arr,(hi+lo)/2,hi); // divide
left.start(); // conquer in new thread 
right.run(); // conquer in this thread
left.join(); // wait
ans = left.ans + right.ans; //combine   

Exercise: What if the above code switched the order of the third and fourth lines to

right.run(); // conquer
left.start(); // conquer 

what goes wrong, if anything?

Java’s Fork-Join Framework

As it turns out, this divide-and-conquer approach is such a common paradigm for parallel computation that Java (and many other languages) actually provides an optimized library to do it for you! This is Java’s ForkJoin framework. All of the ideas are fundamentally the same – but some of the syntax and some minor details change. Here is a summary:

Let’s reproduce our divide-and-conquer code for summing a list in parallel one last time, this time with the ForkJoin library.

class SumTask extends  RecursiveTask<Integer>  {
      protected Integer compute(){ // override
        if(hi – lo < SEQUENTIAL_CUTOFF) // “base case” 
            for(int i=lo; i < hi; i++) ans += arr[i];
        else {
            SumTask left = new SumTask(arr,lo,(hi+lo)/2); // divide
            SumTask right= new SumTask(arr,(hi+lo)/2,hi); // divide
            left.fork(); // conquer in parallel
            right.compute() // conquer in this thread 
            left.join(); // wait
            ans = left.ans + right.ans; // combine 
        }
    }
}
static final ForkJoinPool POOL = new ForkJoinPool(); 
static int parallelSum(int[] arr){
    SumTask task = new SumTask(arr,0,arr.length)
    return POOL.invoke(task); // invoke returns the value compute returns
}

We have highlighted in red the key places the code syntax has changed above, but the overall structure is the same. It is preferable to use this library because it is the most optimized behind the scenes.

Reductions/Folds

Up to this point, we have discussed only the task of summing up the elements of a list in parallel. However, the techniques we have used are entirely general, and we could use the exact same divide-and-conquer in parallel approach to solve many other problems. For example, we could

1. Search for a particular element in our array, if it exists
2. count the number of items with a given property (say, how many of our elements are even?)
3. check if our array is sorted
4. find the first element that satisfies a property of interest,

just to name a few. If you’d like more practice, you could try modifying the pseudocode and/or java code we have for list summation to these examples.

All of these examples are examples of reductions or folds. These two interchangeable terms refer to any operation which “reduces” an array down to a single value, with the caveat that the underlying “reducing” operation that we are applying amongst the elements of the array be associative. For example (x+y) + z = x + (y+z), or if we were looking for the maximum element in our list, max (max(x,y),z) = max (x,max(y,z)).

Why is associativity important for being able to compute in parallel? Because of the combination step – in general, parallel computations can finish in any order (we may compute sums of sublists of size SEQUENTIAL_CUTOFF in an arbitrary order), so we need the combination operation to be invariant no matter which subtasks are completed first.

Maps

Another type of list function that is easy for us to compute is one that simply applies a function to each individual element of the input list(s). For example,

1. Given two lists arr1 and arr2, we can easily, using the same divide-and-conquer paradigm, compute the element-wise sum of these two lists quickly in parallel, or
2. in general, given arr, we can produce a list out such that out[i] = f(arr[i]) in parallel with the divide-and-conquer approach.

These types of operations are called maps and are actually simpler than reductions, since there is no need for any combination step.

Other Examples?

What are some operations that may be more difficult to parallelize? One such natural type of list operation is a filter operation, in which we “filter” out elements of our list that do not satisfy a certain predicate.

For example, given a list, we may want to produce a list where we have “filtered” out all of the odd numbers, and are left with only even numbers. This example is trickier – and it is perhaps not so clear how to achieve this in parallel. In fact, we encourage the reader to pause and try to come up with a parallel algorithm for this problem. One issue to note immediately is that the size of the output varies; in a reduction, we are always producing a single thing, whereas in a map, we are always producing a list that has the same length as the input(s), but here, the length of the output will depend on the particular input that is given. In the next set of notes, we will discuss some of the difficulties of doing these sorts of operations in parallel, and a perhaps surprsing parallel algorithm that achieves an optimal speedup.