Coordinating Threads Using CountDownLatch

Since Java 5, the core Java APIs have been enhanced with more features for handling coordination between threads in concurrent programming. In this post, we discuss a class in the java.util.concurrent package that aids in this purpose: the CountDownLatch.

Introduction

The CountDownLatch class enables us to coordinate threads by introducing awareness of the number of threads that are carrying out related tasks and keeping track of the number of threads that have completed their task. 

This is done by initializing the CountDownLatch with the number of worker threads. Each worker thread should then invoke the countDown() method on the CountDownLatch. The thread that needs to wait for the worker threads to complete should invoke the await() method on the CountDownLatch. This will cause this thread to wait until all worker threads have invoked the countDown() method, in effect counting down from the number of worker threads to zero. Once the countdown reaches zero, the thread calling await() can proceed.

Examples

The examples below illustrate how CountDownLatch can be used in two different scenarios.

Scenario 1: Waiting for Threads to Indicate Completion of a Set of Tasks 

In this example, let us consider a scenario where a number of pizza makers make some pizzas. The pizzas made by the different pizza makers are then collected for delivery to a hungry customer.

To implement this, we use a thread to represent each pizza maker. The more of these we have, the more pizzas can be made concurrently. Here is the code for the PizzaMaker class.

interface FoodMaker {
    void make();
}

class PizzaMaker implements FoodMaker, Runnable {

    int numberOfPizzas;
    List<Pizza> pizzas;
    CountDownLatch countDownLatch;

    public PizzaMaker(int numberOfPizzas, CountDownLatch countDownLatch) {
        this.numberOfPizzas = numberOfPizzas;
        this.countDownLatch = countDownLatch;
    }

    public List<Pizza> getPizzas() {
        if(pizzas == null || pizzas.isEmpty())
            return null;

        return pizzas.stream().filter((pizza -> pizza.isPrepared())).collect(Collectors.toList());
    }

    @Override
    public void make() {
        pizzas = new ArrayList<>();

        for(int i=0 ; i < numberOfPizzas; i++) {
            try {
                Pizza pizza = new Pizza();
                Thread.sleep(1000);
                pizza.setPrepared(true);
                pizzas.add(pizza);
                System.out.println(Thread.currentThread().getName() + " - Pizza ready!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        countDownLatch.countDown();
    }

    @Override
    public void run() {
        this.make();
    }
}

Note that the constructor takes in a CountDownLatch. This will be passed in by the invoking thread that will await completion. Each PizzaMaker instance will need to invoke the countDown() method to indicate the completion of their work. As seen in the above code, this is done in the make() method once all the pizzas have been made.

The code below illustrates a working example of how multiple pizza makers can be invoked using a CountDownLatch and how the main thread can be blocked until all pizza makers have made their pizzas.

public class CountDownLatchDemo {

 public static void main(String[] args) {
   int pizzaMakerCount = 4;

   CountDownLatch countDownLatch = new CountDownLatch(pizzaMakerCount);

   List<PizzaMaker> pizzaMakers = Stream.generate(() -> new PizzaMaker(3, countDownLatch))
   .limit(pizzaMakerCount)
   .collect(Collectors.toList());

   pizzaMakers.stream()
   .map(pizzaMaker -> new Thread(pizzaMaker))
   .forEach(Thread::start);

   try {
     countDownLatch.await();
     List<Pizza> allPizzas = new ArrayList<>();
     pizzaMakers.forEach(pizzaMaker -> allPizzas.addAll(pizzaMaker.getPizzas()));
     System.out.println(allPizzas.size() + " pizzas ready to go!");
   } catch (InterruptedException e) {
      System.out.println("Interrupted!");
   }

 }
}

The code creates four pizza makers and passes the same CountDownLatch into each of them. Note that the CountDownLatch is instantiated with the number of pizza makers. Each pizza maker, responsible for making three pizzas, runs in a separate thread.

The main thread then invokes await() on the CountDownLatch, thus waiting until all four pizza makers have invoked countDown(). Once this has happened, the main thread proceeds to collect the pizzas from all four pizza makers into a single collection.

The output of the code looks like this:

Thread-2 - Pizza ready!
Thread-3 - Pizza ready!
Thread-0 - Pizza ready!
Thread-1 - Pizza ready!
Thread-1 - Pizza ready!
Thread-0 - Pizza ready!
Thread-2 - Pizza ready!
Thread-3 - Pizza ready!
Thread-3 - Pizza ready!
Thread-1 - Pizza ready!
Thread-2 - Pizza ready!
Thread-0 - Pizza ready!
12 pizzas ready to go!

We can see that the final message 12 pizzas ready to go! is displayed after all the pizzas have been made by the four different pizza makers. This is how a CountDownLatch can be used to block a thread until multiple threads have completed their tasks.

Scenario 2: Threads Waiting to Begin Based on a Pre-Condition

In this example, we will stretch the pizza dough — sorry, scenario — to illustrate how a CountDownLatch can be used to make a pool of threads wait until a condition is met.

In this example, let us consider a scenario where a number of pizza makers are waiting for their ingredients to be fetched so that they can start making pizza.

To implement this, each pizza maker will run in a separate thread, just like in the previous example. However, these threads will need to wait until a condition is met (ingredients delivery) before they can start executing their task.

We modify the PizzaMaker class to take in an additional CountDownLatch, which the pizza maker has to wait on before making pizzas.

class PizzaMaker implements FoodMaker, Runnable {

    int numberOfPizzas;
    List<Pizza> pizzas;
    CountDownLatch ingredientsCountDownLatch;
    CountDownLatch countDownLatch;

    public PizzaMaker(int numberOfPizzas, CountDownLatch countDownLatch, CountDownLatch ingredientsCountDownLatch) {
        this.numberOfPizzas = numberOfPizzas;
        this.countDownLatch = countDownLatch;
        this.ingredientsCountDownLatch = ingredientsCountDownLatch;
    }
...
}

We then modify the make() method to call await() on the ingredientsCountDownLatch so that each pizza maker waits for the ingredients to be fetched before attempting to make pizzas:

@Override
public void make() {
    try {
        System.out.println(Thread.currentThread().getName() + " - Waiting for ingredients..");
        ingredientsCountDownLatch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    pizzas = new ArrayList<>();

    for(int i=0 ; i < numberOfPizzas; i++) {
        try {
            Pizza pizza = new Pizza();
            Thread.sleep(1000);
            pizza.setPrepared(true);
            pizzas.add(pizza);
            System.out.println(Thread.currentThread().getName() + " - Pizza ready!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    countDownLatch.countDown();
}

The demo code below shows how the coordination for this scenario is done:

public class CountDownLatchDemo {

 public static void main(String[] args) {
   int pizzaMakerCount = 4;

   CountDownLatch ingredientsCountDownLatch = new CountDownLatch(1);
   CountDownLatch countDownLatch = new CountDownLatch(pizzaMakerCount);

   List<PizzaMaker> pizzaMakers = Stream.generate(() -> new PizzaMaker(3, countDownLatch, ingredientsCountDownLatch))
   .limit(pizzaMakerCount)
   .collect(Collectors.toList());

   pizzaMakers.stream()
   .map(pizzaMaker -> new Thread(pizzaMaker))
   .forEach(Thread::start);

   try {
     fetchIngredients();
     System.out.println("Ingredients ready!");
     ingredientsCountDownLatch.countDown();
     countDownLatch.await();
     List<Pizza> allPizzas = new ArrayList<>();
     pizzaMakers.forEach(pizzaMaker -> allPizzas.addAll(pizzaMaker.getPizzas()));
     System.out.println(allPizzas.size() + " pizzas ready to go!");
   } catch (InterruptedException e) {
   	System.out.println("Interrupted!");
   }

 }

 public static void fetchIngredients() throws InterruptedException {
   Thread.sleep(5000);
 }

}

Note the new CountDownLatch — ingredientsCountDownLatch — is initialized to a count of one because only the main thread will be fetching the ingredients. Once the fetchIngredients() method finishes, the main thread calls countDown() on ingredientsCountDownLatch, which causes each PizzaMaker thread to stop waiting and proceed to make pizzas.

As before, the main thread calls await() on the CountDownLatch associated with the pizza makers, causing it to wait until all pizzas are ready.

The output now looks like this:

Thread-1 - Waiting for ingredients..
Thread-0 - Waiting for ingredients..
Thread-2 - Waiting for ingredients..
Thread-3 - Waiting for ingredients..
Ingredients ready!
Thread-2 - Pizza ready!
Thread-0 - Pizza ready!
Thread-3 - Pizza ready!
Thread-1 - Pizza ready!
Thread-0 - Pizza ready!
Thread-2 - Pizza ready!
Thread-3 - Pizza ready!
Thread-1 - Pizza ready!
Thread-0 - Pizza ready!
Thread-2 - Pizza ready!
Thread-3 - Pizza ready!
Thread-1 - Pizza ready!
12 pizzas ready to go!

We can see that the four pizza makers are waiting for the ingredients to be fetched before proceeding to make pizzas. The final message 12 pizzas ready to go! is displayed after all pizzas are ready. Thus, we have full coordination between the main thread and the PizzaMaker threads.

It is worth noting that there is a variation of the await() method that takes in a timeout parameter so that the calling thread can avoid waiting indefinitely.

Conclusion

We have seen a simple, yet useful, example of how a CountDownLatch can be used to coordinate concurrent tasks in a flexible manner.