C++ threadpool / executor implementation

Question

Yes, yes, I know, another threadpool implementation - but I didn't see any implementations that had quite what I wanted. That is, using modern features such as std::future and std::packaged_task to be able to simply enqueue tasks and wait for their result. I have been very unimpressed by std::async because (at least with gcc) it spawns a new thread for each function, which is very inefficient for small functions.

You create an instance of this somewhere, then call run with a function - run returns a future that will contain the function's return value (or exception thrown) once you wait for it.

#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <future>

class ThreadPool {
    public:
    ThreadPool(unsigned num_threads = std::thread::hardware_concurrency()) {
        while (num_threads--) {
            threads.emplace_back([this] {
                while(true) {
                    std::unique_lock<std::mutex> lock(mutex);
                    condvar.wait(lock, [this] {return !queue.empty();});
                    auto task = std::move(queue.front());
                    if (task.valid()) {
                        queue.pop();
                        lock.unlock();
                        // run the task - this cannot throw; any exception
                        // will be stored in the corresponding future
                        task();
                    } else {
                        // an empty task is used to signal end of stream
                        // don't pop it off the top; all threads need to see it
                        break;
                    }
                }
            });
        }
    }

    template<typename F, typename R = std::result_of_t<F&&()>>
    std::future<R> run(F&& f) const {
        auto task = std::packaged_task<R()>(std::forward<F>(f));
        auto future = task.get_future();
        {
            std::lock_guard<std::mutex> lock(mutex);
            // conversion to packaged_task<void()> erases the return type
            // so it can be stored in the queue. the future will still
            // contain the correct type
            queue.push(std::packaged_task<void()>(std::move(task)));
        }
        condvar.notify_one();
        return future;
    }

    ~ThreadPool() {
        // push a single empty task onto the queue and notify all threads,
        // then wait for them to terminate
        {
            std::lock_guard<std::mutex> lock(mutex);
            queue.push({});
        }
        condvar.notify_all();
        for (auto& thread : threads) {
            thread.join();
        }
    }

    private:
    std::vector<std::thread> threads;
    mutable std::queue<std::packaged_task<void()>> queue;
    mutable std::mutex mutex;
    mutable std::condition_variable condvar;
};

You would use this class like so:

#include <iostream>

#include <ThreadPool.h>

int main() {
    ThreadPool pool;
    std::vector<std::future<int>> futures;
    for (int i = 0; i < 1000; i++) {
        futures.emplace_back(pool.run([i] {return i;}));
    }
    for (int i = 990; i < 1000; i++) {
        // prints 990...999
        std::cout << futures[i].get() << std::endl;
    }
}

All the threads should get joined on destruction of the ThreadPool object and they should exit cleanly. It shouldn't matter if the futures are destructed first or not. It should work with non-copyable return types.

I am looking for feedback because I know this kind of concurrent code is easy to screw up - I think I have everything right (exception safe, data race-free etc.).

---

Preempting a particular comment that might come up:

I am using mutable fields because I want the const qualifier to apply to the run function. I think this because it should be safe to call run from 2 different threads concurrently, which I think of as each thread only having immutable access to the ThreadPool (const reference). The const tells me that no internals of the ThreadPool that are not locked by a mutex should change when I call the function. If this sounds loony, let me know (I'm coming from Rust where this is the standard way to do things).

Answer 1

I am using mutable fields because I want the const qualifier to apply to the run function...

Considering every single member is mutable save for the thread pool itself, I find that a very poor justification for injecting a pretty severe code smell.

Edit Specifically, it's not like your class has parts of its public interface that are not thread-safe, in which case, you could maybe have an argument.

That being said, a few quibbles:

• I find usage of unsigned pretty iffy. I would explicitely use unsigned int.

• MSVC refuses to compile your code because queue.push(std::packaged_task<void()>(std::move(task))); "somehow" invokes std::packaged_task<>'s copy constructor. GCC and clang are fine. Seems to be MSVC's fault, but something to keep in mind.

• On that subject, you should use queue.emplace() instead of queue.push()

• Not popping off the empty task seems a bit fragile to me. I'd rather push as many empty tasks as there are threads, and let the regular logic handle that instead of special-casing it.

Like this:

auto task = std::move(queue.front());
queue.pop();
lock.unlock();

// invalid task signals shutdown.
if(!task.valid()) {
  break;
}
// run the task - this cannot throw; any exception
// will be stored in the corresponding future
task();

Other than that, this looks reasonable to me.