Creating a new thread for every short lived piece of work is inefficient, with the operating system doing work that is expensive in time and resources: preparing the stack, registering the thread with the scheduler, and eventually cleaning up the stack. In applications that constantly spawn and kill threads for tasks of limited duration, thread creation and destruction overhead can dwarf work costs, and an unbounded stream of workers can starve resources (including memory, and CPUs through excessive context switching).
A thread pool offers both efficient usage of workers and bounded resource usage.
A number of worker threads are created once at the beginning and live for the duration of the pool.
Work items are placed into a queue which the idle workers service. Tasks can be submitted to the pool and received from the queue by idle workers.
The pool provides an easy "run me this work asynchronously" interface while managing the efficient usage of a small number of worker threads and limiting thread creation overhead to only a single time.
A thread pool typically requires four components:
A container for the worker threads.
A thread safe queue for submitted tasks.
An efficient way for workers to be suspended when the queue is empty.
A defined shutdown protocol which either processes outstanding work or discards it, and waits for the workers to terminate.
A task may simply be a std::function<void()>, however, when needing a return value and potentially error detection, tasks should be represented by a std::packaged_task which is wrapped inside a callable submitted to the pool, and the std::future returned by that task used to retrieve the result (and/or detect exceptions).
The only synchronisation required are exactly the ones previously used: a mutex to protect the task queue, a condition variable to signal that a task is available or the pool should stop, and std::jthread to manage the lifecycle of the worker threads cleanly.
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <stdexcept>
class ThreadPool {
public:
explicit ThreadPool(
std::size_t thread_count = std::thread::hardware_concurrency())
{
if (thread_count == 0)
thread_count = 1;
for (std::size_t i = 0; i < thread_count; ++i) {
workers_.emplace_back([this](std::stop_token stoken) {
worker_loop(stoken);
});
}
}
~ThreadPool()
{
// jthreads request stop and join automatically
}
template <typename F, typename... Args>
auto submit(F&& f, Args&&... args)
-> std::future<std::invoke_result_t<F, Args...>>
{
using return_type = std::invoke_result_t<F, Args...>;
auto task =
std::make_shared<std::packaged_task<return_type()>>(
std::bind(
std::forward<F>(f),
std::forward<Args>(args)...
)
);
std::future<return_type> result = task->get_future();
{
std::lock_guard<std::mutex> lock(mutex_);
if (stop_requested_)
throw std::runtime_error(
"submit on stopped ThreadPool"
);
tasks_.emplace([task]() {
(*task)();
});
}
cv_.notify_one();
return result;
}
private:
void worker_loop(std::stop_token stoken)
{
while (true) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, stoken, [this] {
return !tasks_.empty();
});
if (stoken.stop_requested() && tasks_.empty())
return;
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
}
std::vector<std::jthread> workers_;
std::queue<std::function<void()>> tasks_;
std::mutex mutex_;
std::condition_variable_any cv_;
bool stop_requested_ = false;
};
On construction, the pool creates std::jthread workers (by default the same number of workers as hardware threads). The task is provided automatically through each worker thread's worker_loop method and receives an automatic std::stop_token.
The destructor is simple and empty, however, each std::jthread requests a stop and joins automatically when the ThreadPool object is destroyed. The stop token signals blocked worker threads to stop waiting on the condition variable, allowing them to exit.
Once they exit their loop, they are automatically joined by the std::jthread destructor.
A simple pool that accepts any callable, runs it on a worker thread, and returns a std::future that allows the caller to retrieve the result later is shown below.
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <stdexcept>
class ThreadPool {
public:
explicit ThreadPool(
// Start this function on a separate thread
std::size_t thread_count = std::thread::hardware_concurrency())
{
if (thread_count == 0)
thread_count = 1;
for (std::size_t i = 0; i < thread_count; ++i) {
workers_.emplace_back([this](std::stop_token stoken) {
worker_loop(stoken);
});
}
}
How many worker threads? The number of worker threads is determined in the constructor. We default the argument to std::thread::hardware_concurrency(), which requests that the implementation should give us the number of available hardware threads. If the call returns zero for some reason, we force it to be at least one so that the pool is never launched with zero threads running immediately.
Now we construct enough std::jthreads for the calculated number of worker threads and save them in the workers_ member. We pass each jthread a lambda that captures this and accepts a std::stop_token, which simply calls the private worker_loop member function, passing the token to it. Because they are jthreads, the threads will eventually stop, or should, request stop and join. They do so automatically because they'll get destroyed when the ThreadPool object is.
~ThreadPool()
{
// jthreads request stop and join automatically
}
Notice the destructor is intentionally empty. When a std::jthread is destroyed, it automatically requests stop from the associated std::stop_source and then joins the thread. Because the worker thread's worker_loop receives and responds to the stop token, it is guaranteed to finish before other members of the ThreadPool are destroyed. There's no need to make any manual calls to stop or join.
// Accept a parameter pack with any number of template arguments
template <typename F, typename... Args>
auto submit(F&& f, Args&&... args)
-> std::future<std::invoke_result_t<F, Args...>>
{
using return_type = std::invoke_result_t<F, Args...>;
auto task =
std::make_shared<std::packaged_task<return_type()>>(
std::bind(
std::forward<F>(f),
std::forward<Args>(args)...
)
);
std::future<return_type> result = task->get_future();
{
std::lock_guard<std::mutex> lock(mutex_);
if (stop_requested_)
throw std::runtime_error(
"submit on stopped ThreadPool"
);
tasks_.emplace([task]() {
(*task)();
});
}
cv_.notify_one();
return result;
}
The function submit accepts any callable along with any required arguments. We use std::invoke_result_t to figure out the return type of the callable and use it as the argument for the std::future.
First, we dynamically allocate a std::packaged_task on the heap, managing it with shared_ptr. This is because the packaged_task must live longer than the submit function call. We use std::bind to perfect forward arguments into a zero-argument callable that we'll run inside worker_loop. Then, we extract the future associated with the packaged_task to give to the caller.
Inside the critical section protected by the mutex, we first check if stop_requested_ is already true, which prevents new tasks from being submitted. Then, we put a simple lambda on the tasks_ queue; when executed by a worker, this lambda will just run the packaged task. We unlock the mutex and notify one waiting worker. Finally, we return the future associated with the packaged task to the caller, who will use it to retrieve the result, or any exceptions that occurred.
private:
void worker_loop(std::stop_token stoken)
{
while (true) {
std::function<void()> task;
{
// Declare lock before code needs to call it
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, stoken, [this] {
return !tasks_.empty();
});
if (stoken.stop_requested() && tasks_.empty())
return;
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
}
Each worker thread runs the loop below for as long as it is alive. We construct an empty std::function variable to hold the work we pull out of the task queue for each iteration.
Here we acquire a unique lock and wait on the condition variable. The wait has a reference to the stop token, so that it will awaken if ThreadPool is being destroyed, and a predicate that returns true when there is something on the task queue. Once released from wait, we check if we are shutting down and the queue is empty, in which case we are done and the thread terminates by breaking the loop. Otherwise, we pull the first task out of the queue, unlock, and execute it, before looping back to wait for another task.
// Use dynamic storage for a resizable sequence
std::vector<std::jthread> workers_;
std::queue<std::function<void()>> tasks_;
std::mutex mutex_;
std::condition_variable_any cv_;
bool stop_requested_ = false;
};
These are the remaining members that hold the state of the thread pool.
workers_: The vector of worker threads.
tasks_: The queue of pending work items, which are represented as zero-argument std::functions.
mutex_: The lock protecting access to the task queue and to the stop_requested_ flag.
cv_: A condition_variable_any that can be awakened based on a stop token or the standard condition for a change in state.
stop_requested_: Intended to signal that submit should refuse new work; as provided, this flag is never set. The workers respond to stop token notifications directly.
This all sums up to a fairly standard implementation of the producer/consumer pattern: submit enqueues new work items with a notification for waiting consumers, the workers, while workers remove items from the queue, perform the computation, and continue looping until notified to shut down, again, using stop tokens which are inherently part of jthreads, thus naturally handling termination for us. std::packaged_task and std::future facilitate synchronous communication of results between the producer and consumers.
Each worker thread will continuously perform these actions:
Obtain a unique lock for the condition variable/mutex.
Wait on the condition variable for the presence of a task or for the stop request from the pool. The predicate form of the wait combined with the stop token effectively ensures that the worker does not wake spuriously (
i.e., not for either of the above reasons) and terminates safely on shutdown.If the condition to stop has been requested and no tasks remain in the queue, the worker thread finishes its execution.
Otherwise, extract a task from the queue, release the lock on the mutex, and execute the task.
Releasing the mutex is a must step, preventing users code (that may have been called by a worker thread) from serialising work or causing deadlocks by, for example, calling submit on the same pool.
Each worker thread will die on shutdown thanks to using std::jthread. Any blocked worker will receive the stop request from the pool, terminate, and return when it has completed its current task and determined no tasks remain in the queue.
Any submitted tasks waiting in the queue will be discarded by the default implementation; a production quality thread pool may provide a mechanism to drain the queue instead of throwing it away.
Submitting new work after the thread pool's destruction has begun will result in an exception and prevent new items from being added to the queue.
The presented thread pool implementation is minimal and doesn't feature work stealing, priorities, or dynamic thread count adjustment.
A simple
std::queueis used here, whichrequiresa single lock. A more performance critical thread pool might use a lock free queue or a more complex synchronisation strategy to minimise contention.While
std::function<void()>might incur memory allocation for tasks themselves (especially for very small objects, though modern compilers have some clever optimisation), alternatively a custom object for tasks or using a specific allocator may yield performance benefits in some cases.Exception safety is ensured through the use of RAII. Locks are guaranteed to be released, packaged tasks capture exceptions, and
std::jthreads guarantee that their thread is joined automatically.Deadlocks could occur in some cases, specifically when a thread submits a new task to the same pool it is running on and no other worker thread is immediately available to execute it. This may lead to all workers subsequently blocking and a condition where a submitted task will not execute.
Limiting the maximum number of concurrently executing tasks or creating a recursive task structure is one approach to avoiding these deadlocks.
In summary, a thread pool transforms long lived worker threads into a reusable resource pool that can execute short lived tasks.
Combined with a thread safe queue, a condition variable for pausing, and std::jthreads, a basic, yet robust and understandable thread pool can be created.
0 comments
No comments yet. Be the first.