Multithreading in Java for dummies (part 3)
Tags: java, concurrency, multithreading
Published: 2017-01-09 08:50:09

---
title: Multithreading in Java for dummies (part 3)
published: true
description: Thread pools and barriers
cover_image: http://raulavila.com/public/pictures/devto/ping-pong.jpg
tags: Java, concurrency, multithreading
---

This will be the last post of this series about multithreading. In the second post, we left our Ping Pong game in a good position, with the threads (players) being blocked while waiting for their turn, and with the possibility of scaling up the number of players as desired. However, there was something that wasn't very elegant, the class `Game` was in charge of creating, starting and finishing the threads, as well as synchronization with them to avoid the main thread from finishing before the players. Let's recap:

```java
//...
Thread thread2 = new Thread(player2);
thread2.start();
Thread thread1 = new Thread(player1);
thread1.start();

//Let the players play!
try {
   Thread.sleep(2);
} catch (InterruptedException e) {
   e.printStackTrace();
}

//Tell the players to stop
thread1.interrupt();
thread2.interrupt();

//Wait until players finish
try {
   thread1.join();
   thread2.join();
} catch (InterruptedException e) {
   e.printStackTrace();
}
//...
```

#### Thread management / Thread pools

Apart from being tedious, if we don't encapsulate the creation of threads conveniently, our code won't be very cohesive, because we're coupling the logic of the game itself to the concurrency management. In addition, creating threads is expensive, and in complex applications this will have an important cost in terms of performance.

The Java concurrency API exports a series of classes and interfaces allowing us to encapsulate the thread creation management with high flexibility: the `Executor` framework. Its three main elements are:

* `Executor`: it's an interface with a single method, `execute(Runnable)`. The idea with this framework is that we handle tasks instead of threads, so we're asking the instance of `Executor` to run the task (which is an instance of `Runnable`) when possible
* `ExecutorService`: interface extending `Executor`, which publishes a series of more advanced methods, to control the whole lifecycle of the work to be done (`shutdown`, `awaitTermination`), and to run tasks of the type `Callable`, which are just a `Runnable` returning a value ([more info here](http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Callable.html)). In the [official documentation](http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html) you can read more about all the possibilities of this interface
* `Executors`: the two previous components are interfaces, and we can create our own implementations if we want. However, the most common use cases are already implemented in the JDK. To use these implementations we need to request an instance using the static factory methods exposed by `Executors`

In general, the term "Thread Pool" is used to refer to the implementations of `Executor`/`ExecutorService` that we use to handle our threads. The most common types, which can be obtained invoking the static methods exposed by `Executors` are:

* Single Thread Executor (`newSingleThreadExecutor`): it contains a single thread, and it's not widely used
* Fixed Thread Pool (`newFixedThreadPool`): it contains a constant number of "alive" threads, which are waiting to receive tasks to run
* Cached Thread Pool (`newCachedThreadPoool`): it maintains a thread pool that can grow or shrink depending on the demand
* Scheduled Thread Pool (`newScheduledThreadPool`): it's used to scheduled the execution of tasks. It returns an instance of [ScheduledExecutorService](http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ScheduledExecutorService.html), given that `ExecutorService` doesn't expose the appropriate methods to schedule future tasks, only to execute them as soon as possible

### Ping Pong, version 5: Thread Pool

Without modifying the class `Player`, we can adapt our class `Game` to use a thread pool instead of taking charge itself of the tedious tasks of creating, starting and stopping the threads. Let's see how:

```java
public class Game {

    public static void main(String[] args) {
        Lock lock = new ReentrantLock();

        Player player1 = new Player("ping", lock);
        Player player2 = new Player("pong", lock);

        player1.setNextPlayer(player2);
        player2.setNextPlayer(player1);

        System.out.println("Game starting...!");

        player1.setPlay(true);

        ExecutorService executor = Executors.newFixedThreadPool(2);

        executor.execute(player1);
        executor.execute(player2);

        sleep(2);

        executor.shutdownNow();

        System.out.println("Game finished!");
    }

    public static void sleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
```

We're using a pool containing two threads (one per player), and we send to it the tasks to run. We sleep the main thread for 2ms, so both players can play for a while, and we invoke the method `shutdownNow()`, which is equivalent to interrupting the threads as we did in previous versions, but encapsulating all the logic in the pool. It's necessary to invoke `shutdownNow` instead of `shutdown`, because the latter waits until the running tasks finish, returning a list with the pending tasks. Our players play indefinitely until they're interrupted, so if we try to finish with `shutdown`, the application will never end!

Well, if we run this version several times, we'll see that it works as expected some times, while others it shows an output like this:

```java
Game starting...!
ping
pong
//...
Game finished!
pong
```

What happened? After requesting both threads to interrupt, it's possible that the main thread finishes before the rest of the threads. This is why the text "Game finished!" appears before the last "pong" turn. Exploring the `ExecutorService` API, we can see a method called `awaitTermination`. This method blocks the thread who invokes it until all the tasks in the pool have finished, or until a timeout passed by parameter expires:

```java
//...
ExecutorService executor = Executors.newFixedThreadPool(2);

executor.execute(player1);
executor.execute(player2);

sleep(2);

executor.shutdownNow();

try {
    executor.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
    System.out.println("Main thread interrupted while waiting for players to finish");
}

System.out.println("Game finished!");
//...
```

We finally get the expected output, and the game behaves as expected, with a cleaner and more readable main class. Have we finished? Not yet!

#### Barriers

Entry / Exit barriers are synchronization mechanisms facilitating the simultaneous execution of a thread group (entry barrier), or the waiting until a whole thread group finalizes.

We have seen the idea of exit barrier before in this post, with `awaitTermination`. This method allows us to create an exit barrier, but it forces us to set up a timeout (it can be hours, but it's a timeout!). We would like to create an exit barrier without a timeout.

In order to understand what an entry barrier is, we're going to add an instruction to `Game`:

```java
//...
executor.execute(player1);
sleep(1000);
executor.execute(player2);
//...
```

We send the main thread to sleep for a second before sending the second player to play. Even though the result is hard to reproduce here, because it's related with the timing of the application, something like this happens:

```java
Game starting...!
ping
// Waiting 1 second
pong
```

That is, the player "ping" plays, but during a second there isn't anybody to play with! So the game is "suspended" for a second, which could be minutes (the time the main thread takes to start the second thread/player). This situation is not ideal, because we're starting a concurrent process requiring the presence of several threads before all of them are ready. To avoid this we need an entry barrier.

There are several classes in the Concurrency API that can be used as barriers, but the simplest one, and the one we'll use as entry and exit barrier is `CountdownLatch`. The usage of this class can be summarized in three points:

1. We create a barrier with a counter initialized to **N**
2. The threads that must continue depending on the barrier will invoke `await()`, and will be blocked until the counter reaches zero. There's an `await` method with timeout too
3. The actors that can influence the opening of the barrier will invoke `countDown()` when necessary conditions are met. In general, **N** conditions must be met for the barrier to be opened

### Version 6: Entry / Exit barriers

In this new version we'll have to modify both `Game` and `Player`. Let's see how they would look like:

```java
public class Player implements Runnable {

    private final String text;

    private final Lock lock;
    private final Condition myTurn;

    private final CountDownLatch entryBarrier;
    private final CountDownLatch exitBarrier;

    private Condition nextTurn;

    private Player nextPlayer;

    private volatile boolean play = false;

    public Player(String text,
                  Lock lock,
                  CountDownLatch entryBarrier,
                  CountDownLatch exitBarrier) {
        this.text = text;
        this.lock = lock;
        this.myTurn = lock.newCondition();

        this.entryBarrier = entryBarrier;
        this.exitBarrier = exitBarrier;
    }

    @Override
    public void run() {
        if(entryBarrierOpen())
            play();
    }

    public boolean entryBarrierOpen() {
        try {
            entryBarrier.await();
            return true;
        } catch (InterruptedException e) {
            System.out.println("Player "+text+
                                " was interrupted before starting Game!");
            return false;
        }
    }

    private void play() {
        while (!Thread.interrupted()) {
            lock.lock();

            try {
                while (!play)
                    myTurn.awaitUninterruptibly();

                System.out.println(text);

                this.play = false;
                nextPlayer.play = true;

                nextTurn.signal();
            } finally {
                lock.unlock();
            }
        }

        exitBarrier.countDown();
    }

    public void setNextPlayer(Player nextPlayer) {
        this.nextPlayer = nextPlayer;
        this.nextTurn = nextPlayer.myTurn;
    }

    public void setPlay(boolean play) {
        this.play = play;
    }
}
```

The class doesn't start until the entry barrier is open, and when it's interrupted to finish, it invokes `countDown` on the exit barrier, which will be how `Game` knows that both players are done.

Think for a second the values we have to initialize the counters in `entryBarrier` and `exitBarrier` before you keep reading...

```java
public class Game {

    public static void main(String[] args) {
        CountDownLatch entryBarrier = new CountDownLatch(1);
        CountDownLatch exitBarrier = new CountDownLatch(2);

        Lock lock = new ReentrantLock();

        Player player1 = new Player("ping", lock, entryBarrier, exitBarrier);
        Player player2 = new Player("pong", lock, entryBarrier, exitBarrier);

        player1.setNextPlayer(player2);
        player2.setNextPlayer(player1);

        System.out.println("Game starting...!");

        player1.setPlay(true);

        ExecutorService executor = Executors.newFixedThreadPool(2);

        executor.execute(player1);

        sleep(1000);

        executor.execute(player2);

        entryBarrier.countDown();

        sleep(2);

        executor.shutdownNow();

        try {
            exitBarrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("Game finished!");
    }

    public static void sleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
```

Indeed, the entry barrier has a counter set to 1, because it will be opened as soon as the main thread has passed all the tasks to the thread pool, while the exit barrier, which here it's used as an alternative to `awaitTermination`, has a counter set to 2, which is the number of actors that must finish the execution before the main thread can progress.

This way, the timing in our application is the desired one, although we have complicated the code a bit. The thing is that concurrency itself is quite complex, so it's hard to encapsulate perfectly all the mechanisms used.

Before finishing the post, I would like to mention that the exit barrier has been added to this version with didactic goals. The best mechanism to wait for the finalization of a group of threads is through `awaitTermination`, using a reasonable timeout, so if we reach this timeout it will be because a failure has happened in one of the tasks we're waiting for its termination. I have added a [version 7](https://github.com/raulavila/concurrent-pingpong/tree/master/src/main/java/com/raulavila/pingpong/version7) to GitHub, which uses an entry barrier, and `awaitTermination` as exit barrier. This version could be considered the optimal version of the application.

I hope this series of posts has been helpful to clarify many of the concepts used in concurrent applications implemented with Java. If you want to learn more, the best book ever bout this topic is [Java Concurrency In Practice](https://www.amazon.com/Java-Concurrency-Practice-Brian-Goetz/dp/0321349601/ref=sr_1_1) by Brian Goetz.

(All the code can be found in this [GitHub repository](https://github.com/raulavila/concurrent-pingpong/))