Java 24: Structured Concurrency

With Java 24, Structured Concurrency moves closer to becoming a first-class feature in the Java platform. This is currently a preview language feature.

Traditional concurrency in Java often results in fragmented and error-prone code, where related threads are launched independently and can be hard to manage or coordinate. For example, to fetch a user and order in parallel, and then process the results, you would typically use an ExecutorService as shown below:

ExecutorService executor = Executors.newFixedThreadPool(2);
Future<String> userFuture = executor.submit(() -> fetchUser());
Future<String> orderFuture = executor.submit(() -> fetchOrder());
String user = userFuture.get();   // blocks until user is fetched
String order = orderFuture.get(); // blocks until order is fetched
String result = process(user, order);

The downsides of the above approach are:

• If one task fails, the other continues unless manually cancelled
• The executor and tasks outlive the method unless explicitly shut down
• You must manage the executor, handle exceptions, and ensure cleanup

Structured Concurrency abstracts much of this complexity, allowing you to focus on what your code is doing rather than how to coordinate threads. It enforces a hierarchical structure, in which tasks spawned together must complete together, much like local variables within a method.

StructuredTaskScope
Here is an example of using the StructuredTaskScope API:

try (var scope = new StructuredTaskScope<String>()) {
  Subtask<String> userTask = scope.fork(() -> fetchUser());
  Subtask<String> orderTask = scope.fork(() -> fetchOrder());

  scope.join(); // Wait for all subtasks to complete

  String user = userTask.get();
  String order = orderTask.get();

  System.out.println("user: " + user);
  System.out.println("order: " + order);
}

StructuredTaskScope has two subclasses, ShutdownOnSuccess and ShutdownOnFailure, to control how the scope reacts to task completion or failure.

StructuredTaskScope.ShutdownOnFailure
With this policy, if any task fails, the scope cancels the remaining tasks, and propagates the exception when throwIfFailed() is called.

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
  Subtask<String> userTask = scope.fork(() -> fetchUser());
  Subtask<String> orderTask = scope.fork(() -> fetchOrder());

  // wait for all subtasks to complete, or one to fail
  scope.join();
  
  // throw if any subtask failed 
  scope.throwIfFailed();

  String user = userTask.get();
  String order = orderTask.get();

  System.out.println("user: " + user);
  System.out.println("order: " + order);
}

StructuredTaskScope.ShutdownOnSuccess
This policy is the opposite — it stops once one task succeeds, cancelling the others. It's great when you want the first successful result and don't care about the rest.

try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
  scope.fork(() -> fetchFromPrimary());
  scope.fork(() -> fetchFromBackup());

  // wait for any subtask to complete, or all to fail
  scope.join();

  // get the result of the first task that completed successfully,
  // or throw an exception if none did
  System.out.println(scope.result()); 
}

Java 19: Virtual Threads

Java 19 introduces Virtual Threads, which are lightweight threads designed to improve application throughput. This is a preview language feature so must be enabled using --enable-preview.

As you know, the JDK implements platform threads (java.lang.Thread) as thin wrappers around operating system (OS) threads. OS threads are expensive to create, and the number of threads is limited to the number of OS threads that the underlying hardware can support. A platform thread captures the OS thread for the code's entire lifetime.

On the other hand, a virtual thread is an instance of java.lang.Thread that is not tied to a particular OS thread and does not capture the OS thread for the code's entire lifetime. A virtual thread consumes an OS thread only while it performs calculations on the CPU. This means that several virtual threads can run their Java code on the same OS thread, effectively sharing it. When code running in a virtual thread calls a blocking I/O operation, the JVM performs a non-blocking OS call and automatically suspends the virtual thread until it can be resumed later.

Virtual threads help to improve the throughput of thread-per-request style server applications in particular because such applications consist of a great number of concurrent tasks that are not CPU bound and spend much of their time waiting.

Here is an example which creates 10,000 virtual threads; however, the JDK runs the code on perhaps only one OS thread. If we were using 10,000 platform threads (and thus 10,000 OS threads) instead, the program might crash, depending on the hardware available. Virtual threads are cheap and plentiful, and there is no need to pool them.

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
  IntStream.range(0, 10_000).forEach(i -> {
    executor.submit(() -> {
      Thread.sleep(Duration.ofSeconds(1));
      return i;
    });
  });
}

The java.lang.Thread class has been updated with new methods to create virtual and platform threads, such as:

// create a new unstarted virtual thread named "foo".
Thread.ofVirtual().name("foo").unstarted(runnable);

// create and start a virtual thread
Thread.startVirtualThread(runnable)

Java 8 Concurrency: LongAdder and LongAccumulator

Java 8 introduces LongAdder, LongAccumulator, DoubleAdder and DoubleAccumulator, which are recommended instead of the Atomic classes when multiple threads update frequently but read less frequently (for example, in the context of gathering statistics). Under high thread contention, these new classes are designed to grow dynamically and, therefore, there is higher throughput at the expense of higher space consumption.

The "Adder" classes support operations for additions, whereas the "Accumulator" clases are given a function to combine values.

To understand how these classes work, you need to take a look at the implementation of the base class, Striped64. It has a "base" field, plus a table of "cells", each of which is a padded variant of AtomicLong to reduce cache contention. It first tries to update the base field using the usual AtomicLong CAS operation. If there is no contention, CAS will succeed. However, if there is contention (i.e. the CAS update of "base" failed), it will use the "probe" value (hash value) of the thread to map to a cell in the table. If a cell does not exist, it will be created. New cells are created upon contention until the number of cells reaches the nearest power of two greater than or equal to the number of CPUs. If the cell exists, CAS will be tried to update the value of the cell. If CAS fails, the probe value will be updated using a secondary hash until an uncontended cell is found. Finally, the sum method simply adds the base field and the elements in the cells array.

The code below shows how you can use LongAdder to calculate the sum of several values:

LongAdder adder = new LongAdder();

// do some addition in different threads
adder.add(10);
adder.increment();
// ...

// get the current sum
long sum = adder.sum();

Or you can use LongAccumulator as follows:

LongAccumulator acc = new LongAccumulator(Long::sum, 0);

acc.accumulate(10);
adder.increment();
// ...

// get the current sum
long sum = acc.get();

You can also use LongAdder in conjuction with a ConcurrentHashMap to maintain a frequency map:

ConcurrentMap<String, LongAdder> freqs = new ConcurrentHashMap<>();
freqs.computeIfAbsent("foo", k -> new LongAdder()).increment();

Throttling Task Submission with a BlockingExecutor

The JDK's java.util.concurrent.ThreadPoolExecutor allows you to submit tasks to a thread pool and uses a BlockingQueue to hold submitted tasks. If you have thousands of tasks to submit, you specify a "bounded" queue (i.e. one with a maximum capacity) otherwise your JVM may run out of memory. You can set a RejectedExecutionHandler to handle what happens when the queue is full, but there are still outstanding tasks to submit.

Here is a simple example showing how you would use a ThreadPoolExecutor with a BlockingQueue with capacity 1000. The CallerRunsPolicy ensures that, when the queue is full, additional tasks will be processed by the submitting thread.

int numThreads = 5;
ExecutorService exec = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
                                 new ArrayBlockingQueue<Runnable>(1000),
                                 new ThreadPoolExecutor.CallerRunsPolicy());

The problem with this approach is that, when the queue is full, the thread submitting the tasks to the pool becomes busy executing a task itself and during this time, the queue could become empty and the threads in the pool could become idle. This is not very efficient. We want to keep the thread pool busy and the work queue saturated at all the times.

There are various solutions to this problem. One of them is to use a custom Executor which blocks (and thus prevents further tasks from being submitted to the pool) when the queue is full. The code for BlockingExecutor is shown below. It is based on the BoundedExecutor example from Brian Goetz, 2006. Java Concurrency in Practice. 1 Edition. Addison-Wesley Professional. (Section 8.3.3).

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An executor which blocks and prevents further tasks from
 * being submitted to the pool when the queue is full.
 * <p>
 * Based on the BoundedExecutor example in:
 * Brian Goetz, 2006. Java Concurrency in Practice. (Listing 8.4)
 */
public class BlockingExecutor extends ThreadPoolExecutor {

  private static final Logger LOGGER = LoggerFactory.
                                          getLogger(BlockingExecutor.class);
  private final Semaphore semaphore;

  /**
   * Creates a BlockingExecutor which will block and prevent further
   * submission to the pool when the specified queue size has been reached.
   *
   * @param poolSize the number of the threads in the pool
   * @param queueSize the size of the queue
   */
  public BlockingExecutor(final int poolSize, final int queueSize) {
    super(poolSize, poolSize, 0L, TimeUnit.MILLISECONDS,
          new LinkedBlockingQueue<Runnable>());

    // the semaphore is bounding both the number of tasks currently executing
    // and those queued up
    semaphore = new Semaphore(poolSize + queueSize);
  }

  /**
   * Executes the given task.
   * This method will block when the semaphore has no permits
   * i.e. when the queue has reached its capacity.
   */
  @Override
  public void execute(final Runnable task) {
    boolean acquired = false;
    do {
        try {
            semaphore.acquire();
            acquired = true;
        } catch (final InterruptedException e) {
            LOGGER.warn("InterruptedException whilst aquiring semaphore", e);
        }
    } while (!acquired);

    try {
        super.execute(task);
    } catch (final RejectedExecutionException e) {
        semaphore.release();
        throw e;
    }
  }

  /**
   * Method invoked upon completion of execution of the given Runnable,
   * by the thread that executed the task.
   * Releases a semaphore permit.
   */
  @Override
  protected void afterExecute(final Runnable r, final Throwable t) {
    super.afterExecute(r, t);
    semaphore.release();
  }
}

Java 7: ThreadLocalRandom for Concurrent Random Numbers

The ThreadLocalRandom class in JDK 7, allows you to generate random numbers from multiple threads. It is more efficient than using shared Random objects and will result in better performance as there is less overhead and contention.

In addition, this class also provides "bounded" generation methods.

For example, the statement below generates a random number between 1 (inclusive) and 100 (exclusive).

int random = ThreadLocalRandom.current().nextInt(1,100);

Wait, there's a BUG!
While trying out this class, I noticed that the SAME random numbers were being produced across all my threads. I then discovered this bug which reported the same issue I was having. It appears that the seed is never initialised so the same random numbers are produced every time. I wouldn't recommend using this class until the bug is fixed. The following code illustrates the issue:

//3 threads
for(int i = 0; i < 3; i++) {
    final Thread thread = new Thread() {
        @Override
        public void run() {
            System.out.print(Thread.currentThread().getName() + ":");

            //each thread prints 3 random numbers
            for(int j = 0; j < 3; j++) {
                final int random = ThreadLocalRandom.current().nextInt(1, 50);
                System.out.print(random + ",");
            }
            System.out.println();
        }
    };
    thread.start();
    thread.join();
}

prints:

Thread-0:1,5,24,
Thread-1:1,5,24,
Thread-2:1,5,24,

Lock-Ordering Deadlock

The code snippet below shows an example of a lock-ordering deadlock.

/**
 * Swaps the balances of two accounts.
 * Deadlock-prone!
 *
 * @param a
 * @param b
 */
public void swap(Account a, Account b) {
 synchronized (a) {
  synchronized (b) {
   double aBal = a.getBalance();
   double bBal = b.getBalance();
   a.setBalance(bBal);
   b.setBalance(aBal);
  }
 }
}

It acquires locks on both Account objects and then swaps their balances. It may look harmless at first, but if you look closely you will see that a deadlock can occur if two threads call swap at the same time, for the same accounts, but in the opposite order:

Thread_1: swap(a,b)
Thread_2: swap(b,a)

It is possible that Thread_1 will acquire the lock on account a and wait for the lock on account b, while Thread_2 is holding the lock on b and waiting for the lock on a.

In order to fix this, we need to make sure that the locks are always acquired in the same order. Here are a few different approaches that can be taken to resolve this:

1. Synchronizing the method
Remove the nested lock acquisitions and synchronize the method instead.

/**
 * Swaps the balances of two accounts.
 *
 * @param a
 * @param b
 */
public synchronized void swap(Account a, Account b) {
 double aBal = a.getBalance();
 double bBal = b.getBalance();
 a.setBalance(bBal);
 b.setBalance(aBal);
}

2. Inducing a lock ordering
To induce a lock ordering, you can compare the two accounts based on a unique, immutable key such as an account number. If your accounts are not Comparable, you can use System.identityHashCode instead. In case, for some reason, the two accounts being passed in are the same, you need to aquire a tie-breaking lock before aquiring the account locks.

private static final Object tieBreaker = new Object();

public void swap(Account a, Account b) {
 final int c = a.compareTo(b);
 if (c > 0) {
  synchronized (a) {
   synchronized (b) {
    double aBal = a.getBalance();
    double bBal = b.getBalance();
    a.setBalance(bBal);
    b.setBalance(aBal);
   }
  }
 } else if (c < 0) {
  synchronized (b) {
   synchronized (a) {
    double aBal = a.getBalance();
    double bBal = b.getBalance();
    a.setBalance(bBal);
    b.setBalance(aBal);
   }
  }
 } else {
  synchronized (tieBreaker) {
   synchronized (a) {
    synchronized (b) {
     double aBal = a.getBalance();
     double bBal = b.getBalance();
     a.setBalance(bBal);
     b.setBalance(aBal);
    }
   }
  }
 }
}

3. Using tryLock
Use tryLock to acquire both locks, but backoff and retry if they cannot both be acquired.

public void swap(Account a, Account b)
                        throws InterruptedException {
 Random random = new Random();
 while(true){
  if(a.getLock().tryLock()){
   try{
    if(b.getLock().tryLock()){
     try{
      double aBal = a.getBalance();
      double bBal = b.getBalance();
      a.setBalance(bBal);
      b.setBalance(aBal);
      return;
     }
     finally{
      b.getLock().unlock();
     }
    }
   }
   finally{
    a.getLock().unlock();
   }
  }
  Thread.sleep(random.nextInt(1000));
 }
}

DateFormat with Multiple Threads

The DateFormat class is not thread-safe. The javadocs state that "Date formats are not synchronized. It is recommended to create separate format instances for each thread. If multiple threads access a format concurrently, it must be synchronized externally."

The following code shows how you would typically use DateFormat to convert a String to a Date in a single-threaded environment. It is more efficient to get the format as an instance variable and use it multiple times so that the system doesn't have to fetch the information about the local language and country conventions multiple times.

public class DateFormatTest {

  private final DateFormat format =
            new SimpleDateFormat("yyyyMMdd");

  public Date convert(String source)
                      throws ParseException{
    Date d = format.parse(source);
    return d;
  }
}

This code is not thread-safe. We can test it out by invoking the method using multiple threads. In the calling code below, I create a thread pool with 2 threads and submit 5 date conversion tasks to it. I then examine the results.

final DateFormatTest t = new DateFormatTest();
Callable<Date> task = new Callable<Date>(){
    public Date call() throws Exception {
        return t.convert("20100811");
    }
};

//lets try 2 threads only
ExecutorService exec = Executors.newFixedThreadPool(2);
List<Future<Date>> results =
             new ArrayList<Future<Date>>();

//perform 5 date conversions
for(int i = 0 ; i < 5 ; i++){
    results.add(exec.submit(task));
}
exec.shutdown();

//look at the results
for(Future<Date> result : results){
    System.out.println(result.get());
}

When the code is run the output is unpredictable - sometimes it prints out the correct dates, sometimes the WRONG ones (e.g. Sat Jul 31 00:00:00 BST 2012!) and at other times it throws a NumberFormatException!

How can you use DateFormat concurrently?
There are different approaches you can take to use DateFormat in a thread-safe manner:

1. Synchronization
The easiest way of making this code thread-safe is to obtain a lock on the DateFormat object before parsing the date string. This way only one thread can access the object at a time, and the other threads must wait.

public Date convert(String source)
                    throws ParseException{
  synchronized (format) {
    Date d = format.parse(source);
    return d;
  }
}

2. ThreadLocals
Another approach is to use a ThreadLocal variable to hold the DateFormat object, which means that each thread will have its own copy and doesn't need to wait for other threads to release it. This is generally more efficient than synchronising sa in the previous approach.

public class DateFormatTest {

  private static final ThreadLocal<DateFormat> df
                 = new ThreadLocal<DateFormat>(){
    @Override
    protected DateFormat initialValue() {
        return new SimpleDateFormat("yyyyMMdd");
    }
  };

  public Date convert(String source)
                     throws ParseException{
    Date d = df.get().parse(source);
    return d;
  }
}

3. Joda-Time
Joda-Time is a great, open-source alternative to the JDK's Date and Calendar API. It's DateTimeFormat is "thread-safe and immutable".

import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import java.util.Date;

public class DateFormatTest {

  private final DateTimeFormatter fmt =
       DateTimeFormat.forPattern("yyyyMMdd");

  public Date convert(String source){
    DateTime d = fmt.parseDateTime(source);
    return d.toDate();
  }
}

JDK 7 - Fork-join and ParallelArrays

A few weeks ago, I had the opportunity of attending an interesting presentation by Sun's Brian Goetz on the new concurrency features in JDK 7 which will allow applications to take advantage of idle CPU cores. Here is a summary of how this works:

In the recent past we have stopped seeing increases in CPU clock rates. We have started getting more cores per chip rather than faster cores. This brings new challenges in programming. We need to learn the art of decomposing our applications into units of work that can be executed in parallel over multiple cores.

Enter the fork-join framework - a library to provide better support for fine-grained parallelism. The most promising candidates for parallelism are sorting and searching. Sorting can be parallelised with merge sort. Searching can be parallelised by searching sub-regions of the data in parallel and then merging the results. This is an example of divide-and-conquer, in which a problem is broken down into sub-problems which are solved and the results combined. In merge sort, we divide the data set into pieces, sort the pieces and merge the results. Result is still O(n log n), but sub-problems can be solved in parallel. We can apply this process recursively, until sub problems are small enough that a sequential solution would be faster.

Divide-and-conquer algorithms take this general form

// PSEUDOCODE
Result solve(Problem problem) {
    if (problem.size < SEQUENTIAL_THRESHOLD)
        return solveSequentially(problem);
    else {
        Result left, right;
        INVOKE-IN-PARALLEL {
            left = solve(extractLeftHalf(problem));
            right = solve(extractRightHalf(problem));
        }
        return combine(left, right);
    }
}

The INVOKE-IN-PARALLEL step creates two or more new tasks (fork) and suspends the current task until the new tasks complete (join). Naive implementations would create a new thread for each task and then invoke Thread.join() for the join operation. But thread creation is expensive and requires O(log n) idle threads. A non-naive implementation is available in package java.util.concurrent.forkjoin in JDK 7 or you can download jsr166y.

The fork-join framework is implemented using work stealing. You create a limited number of worker threads (best choice is Runtime.availableProcessors()). Each worker maintains a double-ended work queue (deque). When forking, worker pushes the new task at the head of its deque. When waiting or idle, worker pops a task off the head of its deque and executes it instead of sleeping. If worker's deque is empty, steals an element off the tail of the deque of another randomly chosen worker. This means that there is never any contention for head and almost never contention for tail. Stealing is infrequent - because order is LIFO, when a worker steals, it generally steals a big chunk which will keep it from having to steal again for a while.

Then pool.invoke() is called, task is placed on a random deque. That worker executes the task and usually just pushes two more tasks onto its deque. It then starts on one of the subtasks. Soon some other worker steals the other top-level subtask. Pretty soon, most of the forking is done and the tasks are distributed among the various work queues. Now the workers start on the sequential subtasks. If work is unequally distributed, corrected via stealing. The result is that there is reasonable load balancing, with no central coordination, little scheduling overhead and minimal synchronisation costs.

We can reduce the amount of boilerplate code in fork-join tasks, by using the ParallelArray classes which let you declaratively specify aggregate operations on data arrays. ParallelArrays use fork-join to efficiently execute on the available hardware.

Further reading:
http://www.ibm.com/developerworks/java/library/j-jtp11137.html
http://www.ibm.com/developerworks/java/library/j-jtp03048.html