MACRO NEPAL
Fork/Join Framework in Java
awiskar acharya
October 31, 2025
9:13 am

The Fork/Join Framework is a special implementation of the ExecutorService that helps take advantage of multiple processors for parallel processing. It's designed for work that can be broken down into smaller pieces recursively.

1. Introduction to Fork/Join

Key Concepts

  • Fork: Splitting a task into smaller subtasks
  • Join: Waiting for subtasks to complete and combining results
  • Work Stealing: Efficient thread utilization where idle threads steal work from busy threads

Core Components

  • ForkJoinPool - Special thread pool for fork/join tasks
  • ForkJoinTask - Abstract base class for tasks
  • RecursiveTask - Returns a result
  • RecursiveAction - No result returned

2. Basic Setup and Imports

import java.util.concurrent.ForkJoinPool; import java.util.concurrent.RecursiveTask; import java.util.concurrent.RecursiveAction; import java.util.concurrent.ForkJoinTask; import java.util.concurrent.TimeUnit;

3. RecursiveTask (Returns Result)

Example: Summing Array Elements

public class SumTask extends RecursiveTask<Long> { private static final int THRESHOLD = 10_000; private final int[] array; private final int start; private final int end; public SumTask(int[] array, int start, int end) { this.array = array; this.start = start; this.end = end; } @Override protected Long compute() { int length = end - start; // If below threshold, compute directly if (length <= THRESHOLD) { return computeDirectly(); } // Split task into subtasks int mid = start + (end - start) / 2; SumTask leftTask = new SumTask(array, start, mid); SumTask rightTask = new SumTask(array, mid, end); // Fork the left task (execute asynchronously) leftTask.fork(); // Compute right task directly and wait for left task Long rightResult = rightTask.compute(); Long leftResult = leftTask.join(); return leftResult + rightResult; } private Long computeDirectly() { long sum = 0; for (int i = start; i < end; i++) { sum += array[i]; } return sum; } }

Using the SumTask

public class SumTaskExample { public static void main(String[] args) { // Create a large array int[] numbers = new int[1_000_000]; for (int i = 0; i < numbers.length; i++) { numbers[i] = i + 1; } // Create ForkJoinPool ForkJoinPool pool = new ForkJoinPool(); // Create task SumTask task = new SumTask(numbers, 0, numbers.length); // Execute and get result long result = pool.invoke(task); System.out.println("Sum: " + result); System.out.println("Expected: " + ((long)numbers.length * (numbers.length + 1)) / 2); pool.shutdown(); } }

4. RecursiveAction (No Result)

Example: Parallel Array Processing

public class ArrayProcessor extends RecursiveAction { private static final int THRESHOLD = 10_000; private final int[] array; private final int start; private final int end; private final int multiplier; public ArrayProcessor(int[] array, int start, int end, int multiplier) { this.array = array; this.start = start; this.end = end; this.multiplier = multiplier; } @Override protected void compute() { int length = end - start; if (length <= THRESHOLD) { processDirectly(); } else { // Split the task int mid = start + (end - start) / 2; ArrayProcessor leftTask = new ArrayProcessor(array, start, mid, multiplier); ArrayProcessor rightTask = new ArrayProcessor(array, mid, end, multiplier); // Invoke all tasks - more efficient than separate fork/join invokeAll(leftTask, rightTask); } } private void processDirectly() { for (int i = start; i < end; i++) { array[i] = array[i] * multiplier; } } }

Using RecursiveAction

public class ArrayProcessorExample { public static void main(String[] args) { int[] data = new int[100_000]; Arrays.fill(data, 5); ForkJoinPool pool = new ForkJoinPool(); ArrayProcessor task = new ArrayProcessor(data, 0, data.length, 3); pool.invoke(task); // Verify results boolean allCorrect = true; for (int value : data) { if (value != 15) { allCorrect = false; break; } } System.out.println("All values correctly multiplied: " + allCorrect); System.out.println("First 10 values: " + Arrays.toString(Arrays.copyOf(data, 10))); pool.shutdown(); } }

5. Advanced Example: Parallel Merge Sort

public class ParallelMergeSort extends RecursiveAction { private static final int THRESHOLD = 10_000; private final int[] array; private final int start; private final int end; private final int[] temp; public ParallelMergeSort(int[] array, int start, int end, int[] temp) { this.array = array; this.start = start; this.end = end; this.temp = temp; } @Override protected void compute() { int length = end - start; if (length <= THRESHOLD) { // Use sequential sort for small arrays Arrays.sort(array, start, end); return; } int mid = start + (end - start) / 2; // Create subtasks ParallelMergeSort leftTask = new ParallelMergeSort(array, start, mid, temp); ParallelMergeSort rightTask = new ParallelMergeSort(array, mid, end, temp); // Execute subtasks in parallel invokeAll(leftTask, rightTask); // Merge results merge(array, start, mid, end, temp); } private void merge(int[] array, int start, int mid, int end, int[] temp) { System.arraycopy(array, start, temp, start, end - start); int i = start, j = mid, k = start; while (i < mid && j < end) { if (temp[i] <= temp[j]) { array[k++] = temp[i++]; } else { array[k++] = temp[j++]; } } while (i < mid) { array[k++] = temp[i++]; } while (j < end) { array[k++] = temp[j++]; } } public static void sort(int[] array) { int[] temp = new int[array.length]; ForkJoinPool pool = new ForkJoinPool(); pool.invoke(new ParallelMergeSort(array, 0, array.length, temp)); pool.shutdown(); } }

Testing Parallel Merge Sort

public class MergeSortExample { public static void main(String[] args) { int[] largeArray = new int[1_000_000]; Random random = new Random(); for (int i = 0; i < largeArray.length; i++) { largeArray[i] = random.nextInt(1_000_000); } int[] sequentialArray = largeArray.clone(); int[] parallelArray = largeArray.clone(); // Sequential sort long startTime = System.currentTimeMillis(); Arrays.sort(sequentialArray); long sequentialTime = System.currentTimeMillis() - startTime; // Parallel sort startTime = System.currentTimeMillis(); ParallelMergeSort.sort(parallelArray); long parallelTime = System.currentTimeMillis() - startTime; System.out.println("Sequential sort time: " + sequentialTime + "ms"); System.out.println("Parallel sort time: " + parallelTime + "ms"); System.out.println("Speedup: " + (double)sequentialTime/parallelTime + "x"); // Verify both arrays are sorted and equal System.out.println("Arrays equal: " + Arrays.equals(sequentialArray, parallelArray)); } }

6. Fibonacci Sequence with Fork/Join

public class FibonacciTask extends RecursiveTask<Long> { private final int n; public FibonacciTask(int n) { this.n = n; } @Override protected Long compute() { if (n <= 1) { return (long) n; } if (n <= 10) { // Threshold for sequential computation return computeSequentially(n); } FibonacciTask first = new FibonacciTask(n - 1); FibonacciTask second = new FibonacciTask(n - 2); first.fork(); Long secondResult = second.compute(); Long firstResult = first.join(); return firstResult + secondResult; } private Long computeSequentially(int n) { if (n <= 1) return (long) n; long a = 0, b = 1; for (int i = 2; i <= n; i++) { long temp = a + b; a = b; b = temp; } return b; } }

Fibonacci Example

public class FibonacciExample { public static void main(String[] args) { ForkJoinPool pool = new ForkJoinPool(); for (int i = 0; i <= 20; i++) { FibonacciTask task = new FibonacciTask(i); long result = pool.invoke(task); System.out.println("Fibonacci(" + i + ") = " + result); } // Compare performance int largeN = 40; long startTime = System.currentTimeMillis(); FibonacciTask parallelTask = new FibonacciTask(largeN); long parallelResult = pool.invoke(parallelTask); long parallelTime = System.currentTimeMillis() - startTime; startTime = System.currentTimeMillis(); long sequentialResult = new FibonacciTask(largeN).computeSequentially(largeN); long sequentialTime = System.currentTimeMillis() - startTime; System.out.println("\nFibonacci(" + largeN + ") = " + parallelResult); System.out.println("Parallel time: " + parallelTime + "ms"); System.out.println("Sequential time: " + sequentialTime + "ms"); System.out.println("Results match: " + (parallelResult == sequentialResult)); pool.shutdown(); } }

7. File Processing with Fork/Join

public class FileSearchTask extends RecursiveTask<List<File>> { private final File directory; private final String fileExtension; public FileSearchTask(File directory, String fileExtension) { this.directory = directory; this.fileExtension = fileExtension.toLowerCase(); } @Override protected List<File> compute() { List<File> result = new ArrayList<>(); List<FileSearchTask> subtasks = new ArrayList<>(); File[] files = directory.listFiles(); if (files == null) { return result; } for (File file : files) { if (file.isDirectory()) { // Create subtask for directory FileSearchTask subtask = new FileSearchTask(file, fileExtension); subtask.fork(); subtasks.add(subtask); } else { // Check file extension if (file.getName().toLowerCase().endsWith(fileExtension)) { result.add(file); } } } // Collect results from subtasks for (FileSearchTask subtask : subtasks) { result.addAll(subtask.join()); } return result; } }

File Search Example

public class FileSearchExample { public static void main(String[] args) { File startDirectory = new File("."); // Current directory ForkJoinPool pool = new ForkJoinPool(); FileSearchTask task = new FileSearchTask(startDirectory, ".java"); List<File> javaFiles = pool.invoke(task); System.out.println("Found " + javaFiles.size() + " Java files:"); javaFiles.stream() .map(File::getAbsolutePath) .sorted() .forEach(System.out::println); pool.shutdown(); } }

8. ForkJoinPool Configuration and Management

Creating Custom ForkJoinPool

public class ForkJoinPoolManagement { public static void main(String[] args) { // Create ForkJoinPool with custom parallelism level int parallelism = Runtime.getRuntime().availableProcessors(); ForkJoinPool customPool = new ForkJoinPool(parallelism); // Create common ForkJoinPool (usually preferred) ForkJoinPool commonPool = ForkJoinPool.commonPool(); System.out.println("Custom pool parallelism: " + customPool.getParallelism()); System.out.println("Common pool parallelism: " + commonPool.getParallelism()); System.out.println("Available processors: " + Runtime.getRuntime().availableProcessors()); // Monitor pool activity customPool.execute(() -> { SumTask task = new SumTask(new int[100_000], 0, 100_000); Long result = task.invoke(); System.out.println("Task result: " + result); }); // Wait for tasks to complete customPool.awaitQuiescence(1, TimeUnit.MINUTES); System.out.println("Active threads: " + customPool.getActiveThreadCount()); System.out.println("Pool size: " + customPool.getPoolSize()); System.out.println("Steal count: " + customPool.getStealCount()); customPool.shutdown(); } }

9. Best Practices and Performance Tips

1. Choose Appropriate Threshold

public class OptimizedSumTask extends RecursiveTask<Long> { // Dynamic threshold based on array size and available processors private static int calculateThreshold(int arraySize) { int processors = Runtime.getRuntime().availableProcessors(); return Math.max(arraySize / (processors * 4), 10_000); } // Rest of implementation... }

2. Avoid Too Many Small Tasks

@Override protected Long compute() { int length = end - start; int threshold = calculateThreshold(length); if (length <= threshold) { return computeDirectly(); } // Split and process... }

3. Use invokeAll for Multiple Subtasks

// More efficient than separate fork/join @Override protected void compute() { if (shouldComputeDirectly()) { computeDirectly(); } else { List<SubTask> subtasks = createSubtasks(); invokeAll(subtasks); // More efficient combineResults(subtasks); } }

4. Handle Exceptions Properly

public class ExceptionHandlingTask extends RecursiveTask<Long> { @Override protected Long compute() { try { // Task computation return computeInternal(); } catch (Exception e) { // Handle or rethrow completeExceptionally(e); return null; } } private Long computeInternal() { // Actual computation return 0L; } }

10. Common Pitfalls and Solutions

1. Stack Overflow with Deep Recursion

Solution: Use iterative approach for very deep recursion or increase threshold

2. Memory Overhead

Solution: Reuse objects where possible and avoid creating too many small tasks

3. Poor Load Balancing

Solution: Use work stealing effectively by creating reasonably sized tasks

4. Blocking Operations

Solution: Avoid blocking operations in compute() method

// Bad - blocking IO in compute method @Override protected Long compute() { // This blocks the worker thread String data = readFromNetwork(); return process(data); } // Better - use CompletableFuture for async IO @Override protected Long compute() { CompletableFuture<String> future = CompletableFuture.supplyAsync(this::readFromNetwork); return process(future.join()); }

Summary

The Fork/Join Framework is ideal for:

  • CPU-intensive tasks that can be recursively divided
  • Large data processing where work can be split
  • Recursive algorithms like sorting, searching, mathematical computations

Key Benefits:

  • Automatic work stealing for load balancing
  • Efficient use of multiple processors
  • Simplified parallel programming model

When to Use:

  • Tasks are CPU-bound
  • Work can be recursively divided
  • You have sufficient processors
  • Task granularity is appropriate (not too fine, not too coarse)
Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper