C++20 Thread Pool

Question

I've implemented a thread pool using C++20. I'm fairly new to concurrently/multi-threaded programming and wanted to work on a project that I could learn from while also getting to know some of the new features available in C++20.

To that end I've specifically made use of:

• std::binary_semaphore
• concepts
• std::jthread

I've also made use of some other language features and overall have tried to implement best practices for "modern" C++ standards.

The most up to date version of the code is available here. The library I created is header only. For brevity, I have omitted all documentation related comments.

thread_pool.h

#pragma once

#include <concepts>
#include <functional>
#include <future>
#include <memory>
#include <queue>
#include <semaphore>
#include <thread>
#include <type_traits>

#include "thread_pool/thread_safe_queue.h"

namespace dp {
    namespace detail {
        template <class T>
        std::decay_t<T> decay_copy(T &&v) {
            return std::forward<T>(v);
        }

        // Bind F and args... into a nullary one-shot lambda. Lambda captures by value.
        template <typename... Args, typename F>
        auto bind(F &&f, Args &&...args) {
            return [f = decay_copy(std::forward<F>(f)),
                    ... args = decay_copy(std::forward<Args>(args))]() mutable -> decltype(auto) {
                return std::invoke(std::move(f), std::move(args)...);
            };
        }

        template <class Queue, class U = typename Queue::value_type>
        concept is_valid_queue = requires(Queue q) {
            { q.empty() } -> std::convertible_to<bool>;
            { q.front() } -> std::convertible_to<U &>;
            { q.back() } -> std::convertible_to<U &>;
            q.pop();
        };

        static_assert(detail::is_valid_queue<std::queue<int>>);
        static_assert(detail::is_valid_queue<dp::thread_safe_queue<int>>);
    }  // namespace detail

    template <template <class T> class Queue, typename FunctionType = std::function<void()>>
    requires std::invocable<FunctionType> &&
        std::is_same_v<void, std::invoke_result_t<FunctionType>> &&
        detail::is_valid_queue<Queue<FunctionType>>
    class thread_pool_impl {
      public:
        thread_pool_impl(
            const unsigned int &number_of_threads = std::thread::hardware_concurrency()) {
            for (std::size_t i = 0; i < number_of_threads; ++i) {
                queues_.push_back(std::make_unique<task_pair>());
                threads_.emplace_back([&, id = i](std::stop_token stop_tok) {
                    do {
                        // check if we have task
                        if (queues_[id]->tasks.empty()) {
                            // no tasks, so we wait instead of spinning
                            queues_[id]->semaphore.acquire();
                        }

                        // ensure we have a task before getting task
                        // since the dtor releases the semaphore as well
                        if (!queues_[id]->tasks.empty()) {
                            // get the task
                            auto &task = queues_[id]->tasks.front();
                            // invoke the task
                            std::invoke(std::move(task));
                            // remove task from the queue
                            queues_[id]->tasks.pop();
                        }
                    } while (!stop_tok.stop_requested());
                });
            }
        }

        ~thread_pool_impl() {
            // stop all threads
            for (std::size_t i = 0; i < threads_.size(); ++i) {
                threads_[i].request_stop();
                queues_[i]->semaphore.release();
                threads_[i].join();
            }
        }

        /// thread pool is non-copyable
        thread_pool_impl(const thread_pool_impl &) = delete;
        thread_pool_impl &operator=(const thread_pool_impl &) = delete;

        template <typename Function, typename... Args,
                  typename ReturnType = std::invoke_result_t<Function &&, Args &&...>>
        requires std::invocable<Function, Args...>
        [[nodiscard]] std::future<ReturnType> enqueue(Function &&f, Args &&...args) {
            // use shared promise here so that we don't break the promise later
            auto shared_promise = std::make_shared<std::promise<ReturnType>>();
            auto task = [func = std::move(f), ... largs = std::move(args),
                         promise = shared_promise]() { promise->set_value(func(largs...)); };
            // get the future before enqueuing the task
            auto future = shared_promise->get_future();
            // enqueue the task
            enqueue_task(std::move(task));
            return future;
        }

        template <typename Function, typename... Args>
        requires std::invocable<Function, Args...> &&
            std::is_same_v<void, std::invoke_result_t<Function &&, Args &&...>>
        void enqueue_detach(Function &&func, Args &&...args) {
            enqueue_task(detail::bind(std::forward<Function>(func), std::forward<Args>(args)...));
        }

      private:
        using semaphore_type = std::binary_semaphore;
        using task_type = FunctionType;
        struct task_pair {
            semaphore_type semaphore{0};
            Queue<task_type> tasks{};
        };

        template <typename Function>
        void enqueue_task(Function &&f) {
            const std::size_t i = count_++ % queues_.size();
            queues_[i]->tasks.push(std::forward<Function>(f));
            queues_[i]->semaphore.release();
        }

        std::vector<std::jthread> threads_;
        // have to use unique_ptr here because std::binary_semaphore is not move/copy
        // assignable/constructible
        std::vector<std::unique_ptr<task_pair>> queues_;
        std::size_t count_ = 0;
    };

    using thread_pool = thread_pool_impl<dp::thread_safe_queue>;

}  // namespace dp

For the sake of completeness, I will also include my safe queue implementation though it is not necessary to review that here. I will more than likely open a new code review post for that class specifically.

thread_safe_queue.h

#pragma once

#include <condition_variable>
#include <mutex>
#include <queue>

namespace dp {
    template <typename T>
    class thread_safe_queue {
      public:
        using value_type = T;
        using size_type = typename std::queue<T>::size_type;

        thread_safe_queue() = default;
        void push(T&& value) {
            std::lock_guard lock(mutex_);
            data_.push(std::forward<T>(value));
            condition_variable_.notify_all();
        }
        bool empty() {
            std::lock_guard lock(mutex_);
            return data_.empty();
        }

        [[nodiscard]] size_type size() {
            std::lock_guard lock(mutex_);
            return data_.size();
        }

        [[nodiscard]] T& front() {
            std::unique_lock lock(mutex_);
            condition_variable_.wait(lock, [this] { return !data_.empty(); });
            return data_.front();
        }

        [[nodiscard]] T& back() {
            std::unique_lock lock(mutex_);
            condition_variable_.wait(lock, [this] { return !data_.empty(); });
            return data_.back();
        }

        void pop() {
            std::unique_lock lock(mutex_);
            condition_variable_.wait(lock, [this] { return !data_.empty(); });
            data_.pop();
        }

      private:
        using mutex_type = std::mutex;
        std::queue<T> data_;
        mutable mutex_type mutex_{};
        std::condition_variable condition_variable_{};
    };
}  // namespace dp

Example driver code illustration how to use the thread pool:

dp::thread_pool pool(4);
const auto total_tasks = 30;
std::vector<std::future<int>> futures;

for (auto i = 0; i < total_tasks; i++) {
    auto task = [index = i]() { return index; };
    futures.push_back(pool.enqueue(task));
}

Any feedback anyone has would be much appreciated. One thing in particular I'm not too thrilled about is the use of a std::shared_ptr in conjunction with a std::promise in the enqueue function.

auto shared_promise = std::make_shared<std::promise<ReturnType>>();

I could not find a good way around this. I tried pretty extensively to get std::packaged_task to work, but I ran into issues that I wasn't able to solve since std::packaged_task is move only. This, however, may just be due to my lack of familiarity with this part of the standard.

Answer 1

Unnecessary abstractions

Why is there a class thread_pool_impl, and then a single type alias to that class:

using thread_pool = thread_pool_impl<dp::thread_safe_queue>;

This indirection doesn't do anything useful as far as I can see, why not just make dp::thread_safe_queue a default value for the first template argument?

template <template <class T> class Queue = thread_safe_queue,
          typename FunctionType = std::function<void()>>
requires std::invocable<FunctionType> &&
         std::is_same_v<void, std::invoke_result_t<FunctionType>> &&
         detail::is_valid_queue<Queue<FunctionType>>
class thread_pool {
    ...
};

However, I would not even make the queue type a template parameter, but just hardcode it to thread_safe_queue. Why would you want the user of a thread_pool to be able to change the queue type? The YAGNI principle applies here.

You are also defining type aliases like semaphore_type and mutex_type, but those are all made private. I don't see a good reason for that. Those type aliases are normally meant to be part of the public API, so that users of a class can easily use the right types.

Use a std::deque for queues_

You can avoid having queues_ be a vector of std::unique_ptrs by using a std::deque<task_pair> instead.

About shared_promise

You might have already tried this:

std::promise<ReturnType> promise;
auto future = promise.get_future();
auto task = [func = std::move(f), ... largs = std::move(args),
             promise = std::move(promise)]() {
    promise.set_value(func(largs...));
};
enqueue_task(std::move(task));

By getting the future before creating the task, we can then std::move() the promise into the lambda. However, that doesn't work, because set_value() is not const, and by default lambda captures are const. So we could make it mutable:

auto task = [func = std::move(f), ... largs = std::move(args),
             promise = std::move(promise)]() mutable {...};

But then the problem is that pushing the task into queues_[i].tasks doesn't work, because std::function unfortunately cannot hold move-only functions. You will have to wait for C++23 for std::move_only_function, or implement it yourself, or perhaps store the task in a std::unique_ptr (but that would be no better than storing the promise in one).

Naming things

Functions and variables have very good names, except that I would rename task_pair to task_queue.

Unlock the mutex before notifying a condition variable

When notifying a condition variable, the idea is that this wakes up another thread waiting on that variable. However, if the corresponding mutex is still locked, the thread that's just woken up then has to first wait for the mutex to be unlocked. It is more efficient to first unlock the mutex, then notify the condition variable, like so:

void push(T&& value) {
    {
        std::lock_guard lock(mutex_);
        data_.push(std::forward<T>(value));
    }
    condition_variable_.notify_all();`
}

Unsafe access to front and back element of thread_safe_queue

Returning a reference to an element of a thread-safe queue is dangerous. What if another thread pops that element? What if you wanted the back element, and then another thread pushes another element to the queue, then the first thread thinks it has a reference to the back element, but it is no longer the back.

Also, are you sure references to elements are still valid if another thread pushes or pops elements? A std::queue is a container adapter; it uses a regular container to store its elements. Luckily, the default container it uses is a std::deque, and that one does provide stable references for its elements. However, you could have made a std::queue<T, std::vector<T>>, and then the references would be invalidated as soon as something is pushed to the queue.

While the pop() functions of STL containers don't return a value (mostly with good reason), it would be better for a thread-safe queue to provide a pop() function that atomically removes the front element and returns it, and remove the front() and back() functions.

Consider using a single queue

You have one queue for each thread in the thread pool. However, that means if the tasks don't all take the same time, then it can be that one queue has much more work to do than the other queues. At some point the thread belonging to that one queue is the only one still running tasks, while the other threads are idle. If you instead use a single queue for all threads, you will not have this issue.

The problem with futures

Just for your interest: while using promises and futures is a very good idea to be able to enqueue arbitrary tasks and get back the results, there is a problem with std::future: you can't wait on multiple futures at once. So asynchronously processing the results is hard, unless something like when_any() gets implemented.

Answer 2

It looks like your code is not working as expected. If I try it with the following :

std::atomic<int> counter = 0;

void test_pools(const int taskCount)
{
    dp::thread_pool pool(4);
    const auto total_tasks = taskCount;
    std::vector<std::future<int>> futures;
    for (auto i = 0; i < total_tasks; i++) {
        auto task = [index = i]() { counter++; return index;  };
        futures.push_back(pool.enqueue(task));
    }
}

int main()
{
    test_pools(20);
    std::cout << counter;
}

Then I would have expected counter to be 20. Changing test_pools to the following partially solves the problem:

void test_pools(const int taskCount)
{
    dp::thread_pool pool(4);
    const auto total_tasks = taskCount;
    std::vector<std::future<int>> futures;
    for (auto i = 0; i < total_tasks; i++) {
        auto task = [index = i]() { counter++; return index;  };
        futures.push_back(pool.enqueue(task));
    }
    while (counter < taskCount)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
}

To fix this, you will have to create some way to determine if all tasks have been created and completed before joining the threads.