CompletableFuture, Fork-Join Pool, ThreadLocal & Virtual Threads

This chapter explores advanced concurrency models in Java. It covers the evolution of asynchronous programming from the traditional blocking Future to the reactive, non-blocking CompletableFuture, the mechanics of the Fork-Join Pool and its work-stealing algorithm, thread-confined state management via ThreadLocal, and the high-concurrency paradigm shift introduced by Virtual Threads.

1. The Evolution of Asynchronous Java: Future and its Overloads

In Java, a Future represents the pending result of an asynchronous computation. When you submit a task to an ExecutorService, it returns a Future object acting as a placeholder. Because the task executes asynchronously within a thread pool, the result may not be immediately available.

Main Thread
   |
   | submits task to ExecutorService
   v
ExecutorService (assigns task to Worker Thread / Queue)
   |
   | returns immediately
   v
Future<T> (placeholder returned to Main Thread)

The submit() method of ExecutorService has three distinct overloaded variations:

I. Submit with Callable<T>

This is the most common pattern. It accepts a Callable (which returns a value and can throw checked exceptions) and returns a Future<T> containing the computed result.

ExecutorService executor = Executors.newFixedThreadPool(2);

// Submitting a Callable that returns an Integer
Future<Integer> futureResult = executor.submit(() -> {
    // Simulating computation
    return 10;
});

II. Submit with Runnable

This overload accepts a Runnable task. Since a Runnable does not return a value, the returned Future<?> uses a wildcard generic type.

Future<?> futureRunnable = executor.submit(() -> {
    System.out.println("Task executing asynchronously...");
});

Note: Calling .get() on a Future<?> returned from a Runnable submission will return null upon successful completion. This is still highly useful for tracking whether the task has completed, or for cancelling it.

III. Submit with Runnable and a Predefined Result

This overload accepts a Runnable along with a predefined result object. When the task completes successfully, the Future returns this pre-determined result. This is useful for signaling success messages or updating a shared state object.

String successMessage = "SUCCESS";
Future<String> futureWithResult = executor.submit(() -> {
    System.out.println("Processing background task...");
}, successMessage);

// Upon completion, futureWithResult.get() returns "SUCCESS"

2. Deep Dive into Future Methods & The Blocking Pitfall

The Future interface provides several methods to manage and query the lifecycle of an asynchronous task:

The Naive vs. Better Approach (The Blocking Pitfall)

A common anti-pattern is calling .get() immediately after submitting a task. Because .get() is a blocking operation, it halts the calling thread (often the main thread) until the worker thread finishes. This completely defeats the purpose of asynchronous execution, turning an asynchronous operation into a synchronous one.

Naive (Blocking) Approach

// Naive: Immediately blocking the main thread
Future<Integer> f1 = executor.submit(() -> {
    Thread.sleep(2000); // Simulating heavy work
    return 10;
});

// Main thread is blocked here for 2 seconds, unable to do other work
Integer result = f1.get(); 
System.out.println(result);

Better (Concurrent) Approach

To maximize concurrency, perform independent operations on the main thread before calling .get(), or check isDone() to verify completion without blocking.

Future<Integer> f1 = executor.submit(() -> {
    Thread.sleep(2000);
    return 10;
});

// Perform other independent tasks concurrently on the main thread
performMainThreadCalculations();

// Only block when the result is absolutely required
Integer result = f1.get();

Understanding cancel(boolean mayInterruptIfRunning)

The cancel method behaves differently depending on the state of the task:

1. Task Not Started (In Queue): The task is removed from the queue and cancelled successfully.
2. Task Already Running: The behavior depends on the mayInterruptIfRunning flag:
   • true: The JVM attempts to interrupt the thread executing the task (sending an interrupt signal, which throws an InterruptedException if the thread is in a blocking state like sleep or wait).
   • false: The JVM allows the running task to complete, but marks the Future as cancelled so that subsequent calls to get() throw a CancellationException.

3. Overcoming Limitations: The Rise of CompletableFuture

While Future was a step forward, it has significant architectural limitations:

• It is inherently blocking: To retrieve a value, you must eventually call the blocking get() method.
• No reactive chaining: You cannot specify a callback like "when this task finishes, automatically pass its output to this next task" without blocking a thread to wait for the first task.
• No combination: You cannot easily combine multiple independent futures (e.g., waiting for both Task A and Task B to finish, then combining their results).

To solve these issues, Java 8 introduced CompletableFuture, which implements both Future and CompletionStage. It enables non-blocking, event-driven, and reactive programming pipelines.

Supply vs. Run Asynchronously

Unlike traditional futures, CompletableFuture does not require you to manually instantiate an ExecutorService (though you can pass a custom one). By default, it runs tasks asynchronously using the global ForkJoinPool.commonPool().

• supplyAsync(Supplier<U>): Used for tasks that return a value.
• runAsync(Runnable): Used for tasks that do not return a value.

// supplyAsync returns a value
CompletableFuture<Integer> futureVal = CompletableFuture.supplyAsync(() -> {
    return 10;
});

// runAsync does not return a value
CompletableFuture<Void> futureVoid = CompletableFuture.runAsync(() -> {
    System.out.println("Running task asynchronously...");
});

4. Reactive Chaining with CompletableFuture: Non-Blocking Workflows

The power of CompletableFuture lies in its ability to chain asynchronous operations together without blocking the calling thread.

Core Chaining Methods

1. thenApply(Function<T, R>): Transforms the result of the previous stage. It accepts a function that takes the output of the previous stage and returns a new value. (Equivalent to a map operation).
2. thenAccept(Consumer<T>): Consumes the result of the previous stage. It receives the output but returns nothing (CompletableFuture<Void>). This acts as a terminal stage.
3. thenRun(Runnable): Executes a Runnable task after the previous stage completes. It does not receive the output of the previous stage and returns nothing.

Code Implementation: Chaining and Combining

The following code demonstrates how to chain multiple transformations and how to combine two independent futures using thenCombine().

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        
        // Example 1: Chaining transformations using thenApply and thenAccept
        CompletableFuture<Void> pipeline = CompletableFuture.supplyAsync(() -> {
            return 10;
        }).thenApply(result -> {
            // Stage 2: Multiply by 2
            return result * 2;
        }).thenApply(result -> {
            // Stage 3: Multiply by 3
            return result * 3;
        }).thenAccept(finalResult -> {
            // Stage 4: Consume and print (Non-blocking)
            System.out.println("Pipeline Result: " + finalResult); // Prints 60
        });

        // Example 2: Combining two independent futures using thenCombine
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10);
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 20);

        CompletableFuture<Void> combinedPipeline = future1.thenCombine(future2, (a, b) -> {
            // Combine results by adding them
            return a + b;
        }).thenAccept(sum -> {
            System.out.println("Combined Sum: " + sum); // Prints 30
        });

        // Block main thread briefly to allow async tasks to complete in the background pool
        pipeline.get();
        combinedPipeline.get();
    }
}

5. The Fork-Join Pool Framework: Divide-and-Conquer Parallelism

The ForkJoinPool is an implementation of ExecutorService designed specifically for divide-and-conquer algorithms. It optimizes parallel processing by recursively breaking a large task down into smaller, independent subtasks until they are small enough to be solved sequentially.

The Work-Stealing Algorithm

To maximize CPU utilization, ForkJoinPool implements a Work-Stealing Algorithm:

1. Global Submission Queue: External tasks submitted to the pool enter a shared submission queue.
2. Private Deques (Double-Ended Queues): Every worker thread in the pool owns a private double-ended queue (deque) to store its subtasks.
3. LIFO Execution (Own Tasks): A worker thread processes its own subtasks in a Last-In, First-Out (LIFO) order (pushing and popping from the head of its own deque). This keeps the most recently created subtasks (which are hot in CPU cache) local to the thread.
4. FIFO Stealing (Other Tasks): If a worker thread runs out of tasks in its own deque and the global queue is empty, it becomes idle. To avoid remaining idle, it attempts to "steal" a task from the tail of another busy thread's deque in a First-In, First-Out (FIFO) order.

This dual-ended queue design minimizes contention between the owning thread (working at the head) and stealing threads (stealing from the tail).

6. Implementing Fork-Join: Parallel Array Summation

To submit tasks to a ForkJoinPool, tasks must be wrapped in a subclass of ForkJoinTask. Java provides two primary classes for this:

• RecursiveTask<V>: Represents a task that returns a value of type V.
• RecursiveAction: Represents a task that does not return a value (void).

Parallel Summation Code Example

The following program implements a parallel array summation using RecursiveTask<Integer>. It recursively splits the array in half until the subsegment size is less than or equal to a threshold of 2 elements, at which point it computes the sum sequentially.

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

// Task class to compute sum of array elements in parallel
class SumTask extends RecursiveTask<Integer> {
    private final int[] arr;
    private final int start;
    private final int end;
    private static final int THRESHOLD = 2; // Base case threshold

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

    @Override
    protected Integer compute() {
        // Base Case: If sub-array size is <= threshold, compute sequentially
        if (end - start <= THRESHOLD) {
            int sum = 0;
            for (int i = start; i <= end; i++) {
                sum += arr[i];
            }
            return sum;
        }

        // Recursive Case: Divide and Conquer
        int mid = start + (end - start) / 2;

        SumTask leftTask = new SumTask(arr, start, mid);
        SumTask rightTask = new SumTask(arr, mid + 1, end);

        // Fork the left task: Push it to the current thread's work-stealing queue
        leftTask.fork();

        // Compute the right task synchronously in the current thread
        int rightResult = rightTask.compute();

        // Join the left task: Wait for its result (may be stolen and executed by another thread)
        int leftResult = leftTask.join();

        // Combine results
        return leftResult + rightResult;
    }
}

public class ForkJoinDemo {
    public static void main(String[] args) {
        int[] arr = {1, 2, 3, 4, 5, 6, 7, 8};

        // Create a ForkJoinPool
        ForkJoinPool pool = new ForkJoinPool();

        // Instantiate the root task
        // Note: The end index is arr.length - 1 to avoid ArrayIndexOutOfBoundsException
        SumTask rootTask = new SumTask(arr, 0, arr.length - 1);

        // Invoke the task and get the result
        int totalSum = pool.invoke(rootTask);

        System.out.println("Total Parallel Sum: " + totalSum); // Prints 36
        
        // Shut down the pool
        pool.shutdown();
    }
}

Note: In the lecture, passing arr.length as the end index initially caused an ArrayIndexOutOfBoundsException during sequential execution. This was corrected by passing arr.length - 1 as the upper bound of the array indices.

7. ThreadLocal: Thread-Confined Variables Without Locks

In a multi-threaded Java application, all threads share the heap memory. While sharing objects is efficient, it introduces race conditions when multiple threads attempt to modify the same variable simultaneously. Typically, developers use synchronization or locks to prevent this, which introduces performance overhead and thread contention.

ThreadLocal provides an alternative approach: Thread Confinement. It allows you to create variables that can only be read and written by the same thread. Even if multiple threads access a single static ThreadLocal instance, each thread maintains its own isolated, independent copy of the variable.

Code Implementation: Isolating State

public class ThreadLocalDemo {
    // Define a ThreadLocal variable to hold a String value per thread
    private static final ThreadLocal<String> threadUser = new ThreadLocal<>();

    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            threadUser.set("Aditya");
            try {
                Thread.sleep(1000); // Simulate processing
                // Retrieve thread-confined value
                System.out.println(Thread.currentThread().getName() + " User: " + threadUser.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // Always clean up ThreadLocal to prevent memory leaks in thread pools
                threadUser.remove();
            }
        }, "Thread-1");

        Thread t2 = new Thread(() -> {
            threadUser.set("Rohit");
            try {
                Thread.sleep(500);
                System.out.println(Thread.currentThread().getName() + " User: " + threadUser.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                threadUser.remove();
            }
        }, "Thread-2");

        t1.start();
        t2.start();
    }
}

ThreadLocal vs. Local Variables

A common question is: Why use ThreadLocal instead of simple local variables inside a method, which are already thread-safe because they reside on the thread's private stack?

• Scope and Lifecycle: Local variables are confined to the method they are declared in. Once the method returns, the variable is popped off the stack.
• Context Propagation: ThreadLocal variables persist across multiple method calls, classes, and layers (e.g., Controller -> Service -> Repository) as long as the execution remains within the same thread. This is highly useful for managing transaction contexts, security tokens, or user session IDs without passing them explicitly as parameters through every single method signature.

8. Virtual Threads: High-Concurrency Java at Scale

Historically, Java threads (Platform Threads) had a 1-to-1 mapping with Operating System (OS) kernel threads.

This model has severe scaling limitations:

• Memory Overhead: Each platform thread reserves a large chunk of memory for its call stack (typically 512 KB to 1 MB). Creating 100,000 platform threads would require 100 GB of RAM just for stack space.
• Context Switching Cost: When the OS switches execution between platform threads, it must save and restore CPU registers, which is highly expensive.
• Blocking I/O Waste: When a platform thread performs blocking I/O (e.g., waiting for a database query or an API response), the underlying OS thread is also blocked and remains idle, wasting valuable system resources.

The Virtual Thread Solution

Introduced as a production-ready feature in Java 21, Virtual Threads are lightweight threads managed entirely by the JVM rather than the OS. They are mapped to OS threads in a Many-to-Many (M:N) relationship.

• Carrier Threads: The JVM maintains a small pool of platform threads (usually matching the number of available CPU cores) called Carrier Threads.
• Mounting and Unmounting: When a Virtual Thread is executing CPU-bound work, the JVM mounts it onto an available Carrier Thread.
• Non-Blocking I/O Handling: When a Virtual Thread encounters a blocking operation (e.g., Thread.sleep(), database call, network I/O):
  1. The JVM intercepts the blocking call.
  2. The Virtual Thread is unmounted from the Carrier Thread. Its execution state (stack frame) is copied to the JVM heap.
  3. The Carrier Thread is now free to execute other Virtual Threads.
  4. Once the blocking I/O operation completes, the JVM remounts the Virtual Thread onto any available Carrier Thread to resume execution.

This allows applications to run millions of concurrent virtual threads on a handful of physical OS threads, drastically increasing throughput for I/O-bound applications.

Creating Virtual Threads

You can create virtual threads using the Thread builder API or via a specialized ExecutorService.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class VirtualThreadsDemo {
    public static void main(String[] args) throws InterruptedException {
        
        // Approach 1: Using Thread.startVirtualThread
        Thread vThread = Thread.startVirtualThread(() -> {
            System.out.println("Running inside virtual thread: " + Thread.currentThread());
        });
        
        // Wait for the virtual thread to complete (Virtual threads are daemon threads by default)
        vThread.join();

        // Approach 2: Using Executors.newVirtualThreadPerTaskExecutor()
        // This executor creates a new, lightweight virtual thread for every submitted task.
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 1; i <= 5; i++) {
                final int taskId = i;
                executor.submit(() -> {
                    System.out.println("Task " + taskId + " executed by: " + Thread.currentThread());
                });
            }
        } // Executor auto-closes and waits for all virtual threads to finish
    }
}

9. Summary and Key Takeaways

• Future Limitations: Traditional Future objects are passive and blocking. Calling .get() immediately after submission blocks the calling thread, turning asynchronous execution into synchronous execution.
• CompletableFuture Power: CompletableFuture enables non-blocking, event-driven pipelines using reactive chaining methods like thenApply() (transforming), thenAccept() (consuming), and thenCombine() (merging).
• Fork-Join Pool Design: Optimized for recursive divide-and-conquer algorithms. It uses a Work-Stealing Algorithm where idle threads steal tasks from the tail of busy threads' private deques to maximize CPU core utilization.
• RecursiveTask vs. Action: Use RecursiveTask<V> when your parallel subtasks need to return a value, and RecursiveAction when they perform void operations.
• ThreadLocal Isolation: Provides thread-confined storage, giving each thread its own isolated copy of a variable. This prevents race conditions without requiring locks or synchronization.
• Platform Thread Limits: Traditional platform threads have a 1:1 mapping with OS threads, making them memory-intensive (512KB-1MB stack) and expensive to context-switch.
• Virtual Thread Paradigm: Virtual threads are lightweight, JVM-managed threads mapped M:N to OS carrier threads. They are unmounted during blocking I/O operations, allowing other tasks to run on the carrier thread.
• High-Throughput Concurrency: Virtual threads allow applications to scale to millions of concurrent threads, making them ideal for I/O-heavy workloads like web servers, database operations, and API integrations.