I've been reading about thread pools in C++ for a few days now and decided to roll out my own. I mainly intend to use it to learn how to implement parallel algorithms at some point in the future but before that I have to know if there's something I can do to make it more efficient.
These are all the variables I'm using. I decided to put everything inside its own namespace and make the std::condition_variable
(responsible for pausing the main thread) static because there's really no need for every thread_pool
object to have a copy of it.
namespace async {
static std::condition_variable main_thread_cv;
template <size_t Size>
class thread_pool {
private:
std::mutex mutex_m;
std::atomic<int> busy_m;
std::condition_variable pool_cv_m;
std::array<std::thread, Size> workers_m;
std::queue<std::function<void()>> task_queue_m;
bool should_terminate_m;
void thread_loop();
public:
thread_pool();
~thread_pool();
thread_pool(const thread_pool& other) = delete;
thread_pool(const thread_pool&& other) = delete;
thread_pool& operator=(const thread_pool& other) = delete;
thread_pool& operator=(const thread_pool&& other) = delete;
void wait();
template <typename Fn, typename... Args> void enqueue(Fn&& function, Args&&... args);
};
...
}
This is the thread loop executed by all of the worker threads.
template<size_t Size>
void thread_pool<Size>::thread_loop() {
thread_local std::function<void()> task;
for (;;) {
{ std::unique_lock<std::mutex> lock(mutex_m);
pool_cv_m.wait(lock, [this]() { return !task_queue_m.empty() || should_terminate_m; });
if (should_terminate_m) {
break;
}
task = task_queue_m.front();
task_queue_m.pop();
}
busy_m++;
task();
busy_m--;
main_thread_cv.notify_one();
}
}
ctor and dtor:
template<size_t Size>
thread_pool<Size>::thread_pool() {
busy_m = 0;
should_terminate_m = false;
for (auto& thread : workers_m) {
thread = std::thread([this]() { thread_loop(); });
}
}
template<size_t Size>
thread_pool<Size>::~thread_pool() {
busy_m = 0;
should_terminate_m = true;
pool_cv_m.notify_all();
for (auto& thread : workers_m) {
thread.join();
}
}
The wait
and enqueue
functions:
template<size_t Size>
void thread_pool<Size>::wait() {
{ std::unique_lock<std::mutex> lock(mutex_m);
main_thread_cv.wait(lock, [this]() { return busy_m == 0 && task_queue_m.empty(); });
}
}
template<size_t Size>
template<typename Fn, typename ...Args>
void thread_pool<Size>::enqueue(Fn&& function, Args&& ...args) {
{ std::scoped_lock<std::mutex> lock(mutex_m);
task_queue_m.push(std::bind(std::forward<Fn>(function), std::forward<Args>(args)...));
pool_cv_m.notify_one();
}
}