Practical Concurrency - Thread Pool

Ashok Dey

10th May, 2017

2.2k words • 13 min read
programming › java

Even before learning about concurrency and multithreading, I have heard about “thread pool” repeatedly. This used to feel heavy and my mind used to skip this topic. But since now I know about it, I must admit that it’s a beautiful concept which I should not have skipped earlier.

Back when I started exploring about “threads” and “concurrency” I was focusing more on the basic constructs of the multithreading universe and constantly learnt about the fundamentals like - threads, mutexes, semaphores, barriers and more. I also touched upon the basic data structures with concurrency like the Concurrent Queue in C++, yet I kept distance from Thread Pool.

Thread Pool

So what is a thread pool anyways?

Thread Pool simply means a collections of ready, reusable threads that are waiting to execute a task. Threads in the pool can take up task(s) and execute it(them) as and when they are assigned to the thread(s).

The core of a thread pool consists of the following:

Worker Threads: These are the actual threads that execute tasks.
Task Queue: Tasks waiting for an available worker are placed into a queue.
Pool Manager: The Thread Pool class itself which manage the lifecycle of creating, starting, executing and reusing the threads.

We will dive into the details of each of the above.

Benefits of Thread Pool

You may ask that why we are creating a pool of threads when we can spawn a thread as and when required during the runtime. But there are certain remarkable advantages of a thread pool.

System Performance

Creating threads on the go is expensive. The program has to request the Operating System for thread creation and if the load on the application is increasing then creating threads on the go can cause CPU and Memory overhead leading to system failure.

Limiting Resources

Thread Pool caps the max number of threads that can be created for a certain type of task. If we are not using a thread pool, the application can spin up unlimited threads causing the app to die a slow death due to extreme resource consumption.

Thread Reusability

Thread pool allows us to reuse the set of threads instead of killing them after task execution and re-creating a thread. This is particularly very helpful when the load on the app is increasing and it has to work on pre-defined resource limitations.

Thread Pool Internals

To create thread pool of our own, we will primarily need a Queue to store the tasks that are to be executed using threads and a collection of Worker Threads which will pick up the tasks as soon as they enter the task queue.

Task Queue

The queue data structure is a perfect fit for storing the tasks as they arrive to us. It will also help us in ordering the tasks but ordering is secondary or specific to the requirements of the Thread Pool. We can use an array or LinkedList as well to store the tasks as they come to us but logically queue makes more sense considering the FIFO ordering a queue provides, unless we also have a factor of priority for the tasks that can vary for each tasks.

We can use an array or a linked list to store that tasks but we have to adhere with the FIFO ordering a queue provides.

Now there are some restrictions to the queue that we are supposed to use for a Thread Pool, the constraints are:

Thread Safety

If the underlying data structure is not thread safe, it can be a chaos as multiple threads can jump on the same task and start executing it making the system unstable and volatile.

Fixed Capacity

The whole purpose of a Thread Pool is to bring a predictability to a system eliminating the exponential resource utilization. If we are not bounding the task queue with a fixed size, the tasks will be produced infinitely and the threads will never be able to consume them all. We will end up with the classic Producer-Consumer problem.

The queue we will use for the Thread Pool implementation should be a Bounded Concurrent Queue!

Let’s start with the implementation. Generally a queue has 2 methods called - enqueue() and dequeue(). But since we are taking about a thread pool, which will submit and take tasks, we will rename the methods to put() and take() which will be used to add a task and take out a task from the queue respectively.

public class ConcurrentQueue<T> {
    private final Object lock = new Object();
    private final LinkedList<T> list = new LinkedList<>();
    private final int maxSize;

    ConcurrentQueue(int maxSize) {
        this.maxSize = maxSize;
    }

    public void put(T task) throws InterruptedException {
      synchronized (lock)  {
          // if the queue is full, do not submit tasks
          while (list.size() >= this.maxSize) {
              lock.wait();
          }

          // add the task to the queue & notify the 
          // threads that a task has arrived 
          list.add(task);
          lock.notifyAll();
      }
    }

    public T take() throws InterruptedException {
      synchronized (lock) {
          // wait coz the queue is empty, let tasks come in
          while (list.isEmpty()) {
              lock.wait();
          }

          T task = list.removeFirst();
          lock.notifyAll();
          return task;
      }
  }

  public int size() {
      // workers use this to check if there is more work left in the queue 
      synchronized (lock) {
          return list.size();
      }
  }
}

synchronized block is used to make the put() method thread safe. it’s similar to using a mutex where in the start of the put() will lock the operation and, unlock() it once we are about to return from it.

Also we created a separate monitoring lock, a java object instead of using the list as the monitor just to have a distinct difference between the data holder (list) and the dedicated monitor lock.

Now that our bounded concurrent queue is ready, we can move onto the next part which is to create the Worker Threads!

Worker Threads

The worker threads are the implementation of the Runnable interface which will be responsible to pull out task from the ConcurrentQueue, execute the task and again go back to take new tasks or wait for new tasks to be pushed into the queue.

public class Worker implements Runnable{
    private final ConcurrentQueue<Runnable> queue;
    private final ThreadPool pool;

    Worker(ConcurrentQueue<Runnable> queue, ThreadPool pool) {
        this.queue = queue;
        this.pool = pool;
    }

    @Override
    public void run() {
        while (true) {
            try {
                if (pool.isShutdown() && queue.size() == 0) {
                    break;
                }

                Runnable task = queue.take();
                task.run();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (RuntimeException e) {
                System.out.println("Something broke" + Arrays.toString(e.getStackTrace()));
            }
        }
    }
}

The worker will be initialized with the ThreadPool and the ConcurrentQueue, it will continuously pull out tasks from the queue and execute them until the ThreadPool is shutdown and the queue becomes empty. The thread will also interrupt() itself and break out of the loop during runtime exceptions.

Point to be noted here, we are not using any dedicated policies for exception or graceful shutdown to keep things simple. Production ThreadPool supports multiple related policies.

In case of exception, the Thread will break out of the loop and terminate itself and we will drain the queue and exit once the pool.shutdown() is invoked.

Pool Manager

The pool manager is the final piece which ties up the Workers with the TaskQueue and manages the lifecycle of the ThreadPool and it is the ThreadPool class itself.

The responsibilities of the ThreadPool class are:

Initialize and start the workers
Manage the task lifecycle from submit to completion
Manage the thread reusability
Exception handling and graceful shutdown

public class ThreadPool {
    private enum State { NEW, RUNNING, SHUTDOWN }

    private final ConcurrentQueue<Runnable> taskQueue;
    private final List<Thread> workers = new ArrayList<>();
    private final Object poolLock = new Object();

    // set the default state to 'NEW'
    private volatile State state = State.NEW;

    public ThreadPool(int totalWorkers, int queueCapacity) {
        this.taskQueue = new ConcurrentQueue<>(queueCapacity);
        for (int i = 0; i < totalWorkers; i++) {
            // create workers
            workers.add(new Thread(new Worker(taskQueue, this)));
        }
    }

    public void start() {
        if (state != State.NEW) {
            return;
        }

        synchronized (poolLock) {
            if (state != State.NEW) {
                return;
            }
            // set the state to 'RUNNING'
            state = State.RUNNING;
        }

        // start all the workers
        for (Thread t : workers) {
            t.start();
        }
    }

    public void submit(Runnable task) throws InterruptedException {
        if (state == State.NEW) {
            this.start();
        }
        if (state == State.SHUTDOWN) {
            throw new RejectedExecutionException("Pool is shutting down");
        }

        // Add task to the TaskQueue
        taskQueue.put(task);
    }

    public void shutdown() {
        synchronized (poolLock) {
            if (state == State.SHUTDOWN) {
                return;
            }
            // set the state to 'SHUTDOWN'
            state = State.SHUTDOWN;
        }

        // Interrupt workers so any thread stuck
        // in an empty queue wait() can
        // re-check the exit condition
        for (Thread t : workers) {
            t.interrupt();
        }
    }

    public boolean isShutdown() {
        return state == State.SHUTDOWN;
    }
}

The above is the implementation of the ThreadPool, which happens to be the manager of the Task lifecycle as well. Let’s decode the parts of it one by one.

Pool Lifecycle

The ThreadPool lifecycle can have different state at a given time. When the TheadPool is created it will be in the NEW state. As soon as it is started, it will go to the RUNNING state and then when shutdown is triggered, the state will change to SHUTDOWN.

These states are important to take suitable actions. For example submit() should not allow tasks when the state is SHUTDOWN, also when the ThreadPool is created, the state should not be running unless explicitly started.

We are using volatile variable to hold the state so that the change is reflected immediately across threads in execution.

Starting the Pool

We are providing a start() method which will change the ThreadPool’s state from NEW to RUNNING. This is important because it can cause side effects.

Assume we are starting the threads inside the constructor and the Thread is holding a Worker which is holding this, now before the constructor is completely executed, a worker has taken away partially created object with it.

Adding Tasks

When we submit a task to the pool, it performs a check to see if the state of the ThreadPool is NEW, if so then we explicitly call the start() which changes the state of the ThreadPool to RUNNING. Now when the pool is running, we can add task to the TaskQueue of the ThreadPool which the workers can pull and execute.

The submit() method will throw RejectedExecutionException if the pool is shutting down and it also throws InterruptedException if the calling thread is interrupted while waiting for space in the queue, which the caller should handle.

Graceful Shutdown

Once the job of the ThreadPool is complete, we should be able to gracefully shut it down. For this we will be using the shutdown() method. This method will change the state of the pool to SHUTDOWN and if it was already invoked then it will return from the check.

Once the state is set to SHUTDOWN, we will loop over the workers and call t.interrupt() explicitly to shut them down.

Thread Pool in Action

Now it’s time to run the ThreadPool we created and see it in action. Here’s a small driver program to test the thread pool along with the workers and the task queue.

public class Main {
    public static void main(String[] args) throws InterruptedException {
        // 3 workers, queue capacity 5
        ThreadPool pool = new ThreadPool(3, 5);
        
        for (int i = 1; i <= 15; i++) {
            final int taskId = i;
            pool.submit(() -> {
                System.out.println("Task " + taskId + " running on "
                        + Thread.currentThread().getName());
                try {
                    Thread.sleep(200); // simulate work
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("Task " + taskId + " done");
            });
        }

        // Allow draining before we shut down
        Thread.sleep(500);

        System.out.println("Calling shutdown()");
        pool.shutdown();

        // Try submitting after shutdown and it should be rejected
        try {
            pool.submit(() -> System.out.println("This should never run"));
        } catch (RejectedExecutionException e) {
            System.out.println("Rejected after shutdown call: " + e.getMessage());
        }

        // Let remaining tasks drain
        Thread.sleep(2000);
        System.out.println(">>> Main finished");
    }
}

Conclusion

Building a thread pool from scratch looks simple until you sit down and write it up with a queue, a set of worker threads, and a state switch. But what happens when two things happen at the exact same moment? Like a submit() and a shutdown() invocation racing each other? This is when we start appreciating how much thought goes into getting this right.

The version we built here handles the common path correctly. The tasks queue up, workers pick them up and execute them, and a shutdown drains whatever is left before the pool goes down. But it isn’t bulletproof, there’s a narrow window where a task can slip in right as the pool is shutting down and never get picked up by any worker.

Java has in-built a production grade version of ThreadPool since Java 5, and it’s fully available in Java 8 inside java.util.concurrent.ThreadPoolExecutor. It solves the same problems we did here but it goes further with separate core and maximum pool sizes, configurable RejectedExecutionHandler policies instead of throwing directly, idle-thread timeouts, much more.