A thread pool
Starting a thread per task costs a stack and a system call, and a program that does it a thousand times a second spends its time in the kernel. A thread pool starts a few workers once and hands each new task to the first one that is free.
The pieces are all from the previous lessons: a mutex and a condition variable for the queue, and a packaged_task so that every task carries the future its caller reads. This lesson assembles them into the smallest pool that shuts down cleanly.
A pool is a policy as much as a structure. The number of workers, the size of the queue and what happens when the queue is full are decisions a real implementation makes configurable, and the constants of this lesson are the simplest choice of each.
A queue and two workers
The pool owns a queue of std::packaged_task<int()>, a mutex, a condition variable and a vector of threads. submit wraps the work in a task, pushes it and wakes one worker; the worker pops it under the lock and runs it outside the lock.
The program submits four values to two workers and adds the futures. The result is 30, and the two threads run four tasks between them, one after the other.
The task runs outside the lock on purpose: a task can submit another task, and holding the queue mutex while a task runs would deadlock the pool. The lock protects the queue, not the work.
#include <condition_variable>
#include <future>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>
class thread_pool
{
public:
explicit thread_pool(int count)
{
for (int i = 0; i < count; ++i) {
workers_.emplace_back([this] { work(); });
}
}
~thread_pool()
{
{
std::lock_guard<std::mutex> lock{guard_};
stop_ = true;
}
ready_.notify_all();
for (auto& worker : workers_) {
worker.join();
}
}
std::future<int> submit(int value)
{
std::packaged_task<int()> task{[value] { return value * value; }};
std::future<int> result = task.get_future();
{
std::lock_guard<std::mutex> lock{guard_};
tasks_.push(std::move(task));
}
ready_.notify_one();
return result;
}
private:
void work()
{
while (true) {
std::packaged_task<int()> task;
{
std::unique_lock<std::mutex> lock{guard_};
ready_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
if (stop_ && tasks_.empty()) {
return;
}
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
}
std::mutex guard_;
std::condition_variable ready_;
std::queue<std::packaged_task<int()>> tasks_;
std::vector<std::thread> workers_;
bool stop_ = false;
};
int main()
{
thread_pool pool{2};
std::future<int> a = pool.submit(1);
std::future<int> b = pool.submit(2);
std::future<int> c = pool.submit(3);
std::future<int> d = pool.submit(4);
std::cout << a.get() + b.get() + c.get() + d.get() << '\n';
}
g++ -std=c++23 -Wall -Wextra -pthread pool.cpp -o pool
./pool
30
The destructor shuts it down
Shutdown is a state of the queue, not a signal: the destructor sets stop_ under the lock, notifies every worker and joins them. A worker that wakes with the flag set and an empty queue returns, and one that finds work still runs it.
The program submits two tasks, reads both results and leaves the block, which destroys the pool. The print after the block happens once every worker has returned.
The order is the one that makes the pool safe to destroy: no thread is left waiting on a condition variable that will never be notified, and no task is dropped. A queue that is emptied before the workers stop needs the same flag and the same empty check.
#include <condition_variable>
#include <future>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>
class thread_pool
{
public:
explicit thread_pool(int count)
{
for (int i = 0; i < count; ++i) {
workers_.emplace_back([this] { work(); });
}
}
~thread_pool()
{
{
std::lock_guard<std::mutex> lock{guard_};
stop_ = true;
}
ready_.notify_all();
for (auto& worker : workers_) {
worker.join();
}
}
std::future<int> submit(int value)
{
std::packaged_task<int()> task{[value] { return value * value; }};
std::future<int> result = task.get_future();
{
std::lock_guard<std::mutex> lock{guard_};
tasks_.push(std::move(task));
}
ready_.notify_one();
return result;
}
private:
void work()
{
while (true) {
std::packaged_task<int()> task;
{
std::unique_lock<std::mutex> lock{guard_};
ready_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
if (stop_ && tasks_.empty()) {
return;
}
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
}
std::mutex guard_;
std::condition_variable ready_;
std::queue<std::packaged_task<int()>> tasks_;
std::vector<std::thread> workers_;
bool stop_ = false;
};
int main()
{
int total = 0;
{
thread_pool pool{2};
std::future<int> a = pool.submit(3);
std::future<int> b = pool.submit(4);
total = a.get() + b.get();
}
std::cout << total << " pool stopped\n";
}
g++ -std=c++23 -Wall -Wextra -pthread shutdown.cpp -o shutdown
./shutdown
25 pool stopped
More tasks than workers
The queue is what makes the pool useful: tasks wait in it when every worker is busy, and the number of threads stays the same. The program submits 100 tasks to two workers and adds the results.
The sum of the squares of 1 to 100 is 338350, and it arrives whatever order the workers pick. The futures are read in the loop that submits, so the program does not need to keep the list.
A pool with a queue can also grow and shrink, and a real one decides the number of workers from hardware_concurrency of lesson 105. The queue is bounded in the next lesson, where back pressure is the subject.
#include <condition_variable>
#include <future>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>
class thread_pool
{
public:
explicit thread_pool(int count)
{
for (int i = 0; i < count; ++i) {
workers_.emplace_back([this] { work(); });
}
}
~thread_pool()
{
{
std::lock_guard<std::mutex> lock{guard_};
stop_ = true;
}
ready_.notify_all();
for (auto& worker : workers_) {
worker.join();
}
}
std::future<int> submit(int value)
{
std::packaged_task<int()> task{[value] { return value * value; }};
std::future<int> result = task.get_future();
{
std::lock_guard<std::mutex> lock{guard_};
tasks_.push(std::move(task));
}
ready_.notify_one();
return result;
}
private:
void work()
{
while (true) {
std::packaged_task<int()> task;
{
std::unique_lock<std::mutex> lock{guard_};
ready_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
if (stop_ && tasks_.empty()) {
return;
}
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
}
std::mutex guard_;
std::condition_variable ready_;
std::queue<std::packaged_task<int()>> tasks_;
std::vector<std::thread> workers_;
bool stop_ = false;
};
int main()
{
thread_pool pool{2};
long total = 0;
for (int i = 1; i <= 100; ++i) {
total += pool.submit(i).get();
}
std::cout << total << '\n';
}
g++ -std=c++23 -Wall -Wextra -pthread many.cpp -o many
./many
338350
A move-only task in std::function
A queue of std::function<void()> cannot hold a task that captures a unique_ptr, because a std::function requires its target to be copyable. The compiler refuses the conversion and the message names the deleted copy constructor.
The program captures a unique_ptr in a lambda and stores the lambda in a std::function. The error lists the candidate constructors of std::function, and the note at the end says the lambda is not copyable.
A pool that has to accept such a task stores a move-only wrapper instead: std::packaged_task, or a small class with a virtual call, which is the type erasure of lesson 125. std::function is for callables that can be copied, and that is a real limit.
#include <functional>
#include <memory>
int main()
{
auto handle = std::make_unique<int>(1);
std::function<void()> task = [value = std::move(handle)] { (void)value; };
task();
}
g++ -std=c++23 -Wall -Wextra move_only.cpp -o move_only
In file included from /usr/include/c++/15.2.0/functional:61,
from move_only.cpp:1:
/usr/include/c++/15.2.0/bits/std_function.h: In instantiation of 'std::function<_Res(_ArgTypes ...)>::function(_Functor&&) [with _Functor = main()::<lambda()>; _Constraints = void; _Res = void; _ArgTypes = {}]':
move_only.cpp:8:34: required from here
8 | std::function<void()> task = [value = std::move(handle)] { (void)value; };
| ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
/usr/include/c++/15.2.0/bits/std_function.h:441:69: error: static assertion failed: std::function target must be copy-constructible
441 | static_assert(is_copy_constructible<__decay_t<_Functor>>::value,
| ^~~~~
/usr/include/c++/15.2.0/bits/std_function.h:441:69: note: 'std::integral_constant<bool, false>::value' evaluates to false
Exercises
- Add a method that returns the number of tasks still waiting in the queue and print it after the submits.
-
Change the pool to take `std::function
- Make the pool use four workers and check that the result of the 100-task program does not change.
Questions
A new set of questions in a new order every time. Choose one answer for each, then check.
-
What does the destructor of the pool do first?
Answer
It sets the stop flag under the lock and notifies every worker
The flag plus notify_all wakes the sleeping workers, and only then can the joins return.
-
When does a worker of the pool return?
Answer
When the stop flag is set and the queue is empty
The empty check is what lets a worker finish the tasks that are already queued.
-
Why does the pool store packaged_tasks instead of plain functions?
Answer
Each task carries the future its caller reads
The promise inside the task is what connects the worker's result to the caller's future.
-
What does this program print?
#include <future> #include <iostream> #include <thread> int square(int value) { return value * value; } int main() { auto one = std::async(std::launch::async, square, 1); auto two = std::async(std::launch::async, square, 2); auto three = std::async(std::launch::async, square, 3); auto four = std::async(std::launch::async, square, 4); std::cout << one.get() + two.get() + three.get() + four.get(); }Answer
30
The four tasks return 1, 4, 9 and 16, and the futures add them after the reads.
-
What does this program print?
#include <future> #include <iostream> #include <thread> #include <vector> int main() { int total = 0; { auto a = std::async(std::launch::async, [] { return 3 * 3; }); auto b = std::async(std::launch::async, [] { return 4 * 4; }); total = a.get() + b.get(); } std::cout << total << " pool stopped"; }Answer
25 pool stopped
The two results are read before the block ends, and the message follows the cleanup.
-
What does the 100-task program print?
Answer
338350
It adds the squares of 1 to 100, and the sum is fixed regardless of the order.
-
Why does the worker run the task outside the lock?
Answer
So a task can submit another task without deadlocking the queue
Holding the queue mutex during a task would block the submit that the task itself makes.
-
Which container does the pool use for the waiting tasks?
Answer
std::queueThe queue gives the push at one end and the pop at the other that a task list needs.
-
Why can a lambda that captures a unique_ptr not be stored in a
std::function<void()>?Answer
std::function requires a copyable target, and the lambda is move-only
The lambda's copy constructor is deleted, so the std::function has nothing to copy into its storage.
-
How many tasks run at the same time in a pool constructed with 2?
Answer
At most two
Each worker runs one task at a time, so the concurrency is the number of workers.
The code on this page was built with g++ 15.2.0 and run on Alpine Linux 3.24.3 (x86_64) in a Docker container. The programs whose section shows a compiler error were compiled until that error and were not run.
Article author: Arthur Isaev