tiny thread pool implementation

Question

I wanted to write a minimal thread pool implementation.

You can submit tasks and get a future. When the pool goes out of scope, all pending tasks will complete and join all workers. In an earlier revision, i made a separate join() member function that joins the workers but then it would keep the thread pool in an unusable state.

thread_pool.hpp

#ifndef RDX_THREAD_POOL_HPP
#define RDX_THREAD_POOL_HPP
#include <cstddef>
#include <future>
#include <queue>
#include <mutex>
#include <thread>
#include <condition_variable>
namespace rdx {

    /*
        /brief A thread pool that spawns a specified number of workers.

        Worker threads wait for the queue to be filled with tasks.
        Enqueueing tasks return futures so that they can return asynchronous values.
        Destructor will join all workers.
    */
    class thread_pool {
    private:
        std::queue<std::function<void()>> task_queue;
        std::mutex queue_mutex;
        std::condition_variable queue_notification;
        std::vector<std::thread> workers;
        bool should_stop;
        void worker_function();
    public:
        // constructs a thread pool with the given number of workers
        thread_pool(std::size_t num_workers = std::thread::hardware_concurrency());
        ~thread_pool();
        //enqueues a task to be performed and returns a future for that task
        template<class F, class... Args>
        std::future<typename std::result_of<F(Args...)>::type> enqueue(F&& f, Args&&... args);

        thread_pool(thread_pool&&) = delete;
        thread_pool(const thread_pool&) = delete;
        thread_pool& operator=(thread_pool&&) = delete;
        thread_pool& operator=(const thread_pool&) = delete;
    };

    template<class F, class ...Args>
    inline std::future<typename std::result_of<F(Args ...)>::type> thread_pool::enqueue(F&& f, Args && ...args) {
        using return_type = typename std::result_of<F(Args...)>::type;
        auto task = std::make_shared<std::packaged_task<return_type()>>(std::bind(std::forward<F>(f), std::forward<Args>(args)...));
        std::future<return_type> res = task->get_future();
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            task_queue.emplace([task]() { (*task)(); });
        }
        queue_notification.notify_one();
        return res;
    }

}
#endif // !RDX_THREAD_POOL_H

thread_pool.cpp

#include "thread_pool.hpp"

namespace rdx {
    void thread_pool::worker_function() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(queue_mutex);
                queue_notification.wait(lock, [this]() {
                    return should_stop || !task_queue.empty();
                });
                if (should_stop && task_queue.empty()) {
                    break;
                }
                if (task_queue.empty()) {
                    continue;
                }
                task = task_queue.front();
                task_queue.pop();
            }
            task();
        }
    }

    thread_pool::thread_pool(std::size_t num_workers) : should_stop(false) {
        for (auto i = 0; i < num_workers; i++) {
            workers.emplace_back(&thread_pool::worker_function,this);
        }
    }

    thread_pool::~thread_pool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            should_stop = true;
        }
        queue_notification.notify_all();
        for (auto& worker : workers) {
            worker.join();
        }
    }
}
 

repository can be found here

Answer 1

A solid approach to a simple threadpool

There are only a few minor things that I noticed which can be improved.

To start off, a single argument constructor should usually be marked explicit:

// constructs a thread pool with the given number of workers
explicit thread_pool(std::size_t num_workers = std::thread::hardware_concurrency());

You exclusively use std::unique_lock<std::mutex>, which is fine, but std::lock_guard<std::mutex> has slightly less overhead. Unless you need the extra features of a unique_lock, consider using a lock_guard instead. In fact, the only times you need a unique_lock in your current code is when passing the lock to the condition_variable for the wait call.

Speaking of condition_variable, I feel like your worker_function is slightly more complex than it needs to be. You also missed an optimization opportunity by not trying to move the task out of the queue:

void thread_pool::worker_function() {
    while (true) {
        std::function<void()> task;
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            // Loop here in case of a spurious wakeup
            while(task_queue.empty()) {
                if(should_stop) {
                    // Only return when queue is empty and should_stop
                    return;
                }
                // Wait until notified or woken up spuriously
                queue_notification.wait(lock);
            }
            task = std::move(task_queue.front());
            task_queue.pop();
        }
        task();
    }
}

In the destructor, consider checking whether a thread is joinable before joining it:

thread_pool::~thread_pool() {
    {
        std::lock_guard<std::mutex> lock(queue_mutex);
        should_stop = true;
    }
    queue_notification.notify_all();
    for (auto& worker : workers) {
        if (worker.joinable()) {
            worker.join();
        }
    }
}

In general, you currently have no exception handling at all. Especially when creating a class which accepts user given functions and executes them, you should consider what happens when someone "accidentally" tries to enqueue a task which raises an exception in one of your worker threads.

By default, the worker thread just silently dies in the background since exceptions are not propagated back to the main thread. You may (or not) get a broken_promise exception when calling get on the returned future, but you should try to handle things differently than that.

Additionally, your constructor doesn't handle possible exceptions raised during the creation of the threads. The problem is that, when an exception occurs inside a constructor, no destructor is called to stop and join any already created threads, leaving you in a problematic situation. In that case you should clean up the threads by yourself:

thread_pool::thread_pool(std::size_t num_workers) : should_stop(false) {
    // If you know the size beforehand, just reserve it directly
    workers.reserve(num_workers);
    try {
        for (auto i = 0; i < num_workers; i++) {
            workers.emplace_back(&thread_pool::worker_function,this);
        }
    }
    catch (...) { // catch all
        {
            std::lock_guard<std::mutex> lock(queue_mutex);
            should_stop = true;
        }
        queue_notification.notify_all();
        for (auto& worker : workers) {
            if (worker.joinable()) {
                worker.join();
            }
        }
        throw; // rethrow the caught exception
    }
}

Finally, speaking from personal experience, it is likely worth splitting the "queue logic" from the "threadpool logic". Having a separate threadsafe queue class, which supports the waiting functionality of the condition_variable, let's you easily reuse your code in different threadpools or even different projects. Most importantly though, it drastically improves the readability and maintainability since each part would be nicely encapsulated.