Fork Join Framework

In the ever-evolving landscape of software development, leveraging the full power of multi-core processors is no longer a luxury but a necessity. While traditional threading mechanisms offer a path to concurrency, managing them effectively can quickly become a complex and error-prone endeavour. This is where Java’s Fork/Join Framework steps in, offering a robust and efficient solution for parallelising tasks, especially those that can be broken down into smaller, independent subtasks.

What is the Fork/Join Framework?

At its core, the Fork/Join Framework, introduced in Java 7, is designed to efficiently execute tasks that follow the “divide and conquer” paradigm. Imagine a large problem that’s too big for one thread to handle quickly. The Fork/Join Framework allows you to recursively break this problem down into smaller, manageable subproblems (forking), solve these subproblems independently, and then combine their results to get the final solution (joining).

It’s built upon the java.util.concurrent.ForkJoinPool and ForkJoinTask classes, providing a powerful abstraction over the complexities of thread management.

The Core Components: ForkJoinPool and ForkJoinTask

ForkJoinPool: This is the heart of the framework. It's an intelligent thread pool specifically designed to work with ForkJoinTasks. Unlike traditional thread pools, ForkJoinPool implements work-stealing algorithm. This means that if a worker thread in the pool finishes its tasks and has nothing else to do, it can "steal" tasks from the queues of other busy worker threads. This dynamic load balancing greatly improves resource utilisation and overall efficiency. Each worker thread maintains its own deque (double-ended queue) of tasks. Threads push and pop tasks from the head of their own deque. When idle, a thread steals tasks from the tail of another thread’s deque — minimizing contention. This enables load balancing and parallelism without centralized scheduling

A convenient way to get access to the instance of the ForkJoinPool is to use its static method commonPool(). This will provide a reference to the common pool, which is a default thread pool for every ForkJoinTask.

ForkJoinPool commonPool = ForkJoinPool.commonPool();

According to Oracle’s documentation, using the predefined common pool reduces resource consumption since this discourages the creation of a separate thread pool per task. The common pool is used by any ForkJoinTask that is not explicitly submitted to a specified pool. This pool is statically constructed; its run state is unaffected by attempts to shutdown() or shutdownNow(). However, this pool and any ongoing processing are automatically terminated uponSystem.exit(int). Any program that relies on asynchronous task processing to complete before program termination should invokecommonPool().awaitQuiescence, before exit.

For applications that require separate or custom pools, a ForkJoinPool may be constructed with a given target parallelism level; by default, equal to the number of available processors. The pool attempts to maintain enough active (or available) threads by dynamically adding, suspending, or resuming internal worker threads, even if some tasks are stalled waiting to join others. However, no such adjustments are guaranteed in the face of blocked I/O or other unmanaged synchronization.

int parallelism = Runtime.getRuntime().availableProcessors();
  ForkJoinPool pool = new ForkJoinPool(parallelism);  // default constructor uses paralle

ForkJoinTask: This is an abstract class that represents a task that can be executed in a ForkJoinPool. It has two main concrete implementations:

• RecursiveAction: Used for tasks that don't return a result (e.g., modifying an array in place).
• RecursiveTask<V>: Used for tasks that return a result of type V (e.g., summing elements in an array).

Both RecursiveAction and RecursiveTask require you to implement the compute() method, which contains the logic for your task.

Let’s illustrate with a common example: summing all elements in a large array.

1. Initial Task: You start with a RecursiveTask that needs to sum the entire array.
2. Threshold Check: Inside the compute() method, you first check if the task is small enough to be solved directly (e.g., if the array segment is below a certain threshold). If it is, you perform the direct computation.
3. Forking: If the task is too large, you divide it into two smaller subtasks (e.g., sum the first half and sum the second half of the array). You then fork these subtasks using the fork() method. This submits them to the ForkJoinPool for asynchronous execution.
4. Joining: Once the subtasks are forked, the parent task doesn’t necessarily wait idly. It might process another subtask or even get stolen by another thread. When it needs the results of its forked subtasks, it calls the join() method. join() blocks until the subtask completes and returns its result.

This recursive forking and joining continues until all subtasks are small enough to be computed directly. The results then propagate back up the hierarchy, eventually yielding the final solution.

Get Kaustubh Saha’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

In this example, we used an array stored in the arr field of the SumTask class to represent the work ( which is to find the sum of all elements in the array arr between indices start and end). The task is then recursively divided into smaller pieces of work until each piece is smaller than the threshold. Then the invokeAll() method submits the subtasks to the common pool.

class SumTask extends RecursiveTask<Long>{
    final int start;
    final int end;
    final int arr[];

    static final int THRESHOLD = 100;

    @Override
    protected Long compute() {
        if ((end - start) <= THRESHOLD){
            long sum = 0;
            for (int i=start; i<=end; i++){
                sum += arr[i];
            }
            return sum;
        }
        else {
            int mid = (start + end) / 2;
            SumTask t1 = new SumTask(start, mid, arr);
            SumTask t2 = new SumTask(mid+1, end, arr);
            // Executes the given tasks, returning a list of Futures holding their status and results when all complete
            invokeAll(t1, t2);
            // To trigger execution, the join() method is called for each subtask.
            return t1.join() + t2.join(); // we can get the results from the futures returned by invokeAll as well
        }
    }

    SumTask(int start, int end, int[] arr){
        this.start = start;
        this.end = end;
        this.arr = arr;
    }

Now we need to submit the task to the pool:

   int parallelism = Runtime.getRuntime().availableProcessors();
   ForkJoinPool pool = new ForkJoinPool(parallelism);

   Random random = new Random(1000);
   int start = 0;
   int end = 100_000_000;
   int[] arr = new int[end+1];
   for (int i=0; i<end; i++){
        arr[i] = random.nextInt(end);
   }

   RecursiveTask<Long> task = new SumTask(start, end, arr);
   pool.invoke(task);
   
   System.out.println("Sum : " + task.join());

Now, let's create a similar example using RecursiveAction. Instead of finding the sum, we’ll print all the values in the array between two given indices:

class PrintAction extends RecursiveAction {
    final int start;
    final int end;
    final int arr[];

    static final int THRESHOLD = 100;

    PrintAction(int start, int end, int[] arr){
        this.start = start;
        this.end = end;
        this.arr = arr;
    }

    @Override
    protected void compute() {
        if ((end - start) <= THRESHOLD){
            StringBuilder builder = new StringBuilder();
            for (int i=start; i<=end; i++){
                builder.append(arr[i]).append(" ");
            }
            System.out.println(builder.toString());
        }
        else {
            int mid = (start + end) / 2;
            PrintAction t1 = new PrintAction(start, mid, arr);
            PrintAction t2 = new PrintAction(mid+1, end, arr);
            invokeAll(t1, t2);
            t1.join();
            t2.join();
        }
    }
}

We can submit the task to the pool in the same way as before:

   PrintAction action = new PrintAction(start, end, arr);
   pool.invoke(action);

Benefits of the Fork/Join Framework

• Efficient Parallelism: Leverages multi-core processors effectively with its work-stealing algorithm, minimizing idle threads.
• Simplified Concurrency: Abstracts away much of the low-level thread management, allowing developers to focus on task decomposition.
• Scalability: Scales well with the number of available processor cores.
• Reduced Overhead: Compared to creating and managing a large number of individual threads manually, the ForkJoinPool is optimized for performance.

When to Use the Fork/Join Framework

The Fork/Join Framework shines in scenarios where:

• Tasks can be naturally broken down into independent subtasks (e.g., recursive algorithms like mergesort, quicksort, tree traversals, image processing, and financial simulations).
• The results of subtasks need to be combined to form a final result.
• You have CPU-bound tasks, as it’s designed to make efficient use of processor time.

The Java Fork/Join Framework provides a powerful and elegant solution for tackling complex, concurrent problems using the divide and conquer strategy. By understanding its core components and principles, you can write more efficient, scalable, and robust applications that fully harness the power of modern multi-core architectures. So, the next time you encounter a problem that begs for parallelization, consider forking your way to a faster solution !!