Topic: C++ Implementation of Lock-Free Queue and ThreadPool classes
In a project I'm currently working on, I need the implementation of ThreadPool along with a thread-safe queue. I've tried implementing the above structures and wonder about their safety and performance.
Reason for asking
I'm looking for your review to ensure it's bug-free and well-constructed. As far as I know, everything is working well, but I feel like there is much to improve, as I'm a beginner in this field, and probably I can miss something pretty easily.
Lock free queue
lock_free_queue.hpp:
#include <deque>
#include <mutex>
#include <condition_variable>
template<typename T>
class LockFreeQueue {
private:
std::deque<T> queue;
std::mutex mutex;
std::condition_variable element_available;
std::condition_variable space_available; // keep control of queue size
std::atomic_uint active_producers; // to throw poison pill when needed
size_t max_size{};
public:
LockFreeQueue() : max_size(std::numeric_limits<size_t>::max()), active_producers(1) {}
explicit LockFreeQueue(size_t max_size, unsigned active_producers = 1) :
max_size(max_size), active_producers(active_producers) {}
~LockFreeQueue() = default;
LockFreeQueue(const LockFreeQueue &) = delete;
LockFreeQueue &operator=(const LockFreeQueue &) = delete;
template<typename A>
void push(A &&item) {
{
std::unique_lock<std::mutex> lock(mutex);
space_available.wait(lock, [this] { return queue.size() < max_size; });
queue.push_back(std::forward<A>(item));
}
element_available.notify_one();
}
T pop() {
T item;
{
std::unique_lock<std::mutex> lock(mutex);
while (queue.empty()) {
if (active_producers.load(std::memory_order_acquire) == 0) {
return {}; // poison pill
}
element_available.wait(lock);
}
item = std::move(queue.front());
queue.pop_front();
}
space_available.notify_one();
return item;
}
bool try_pop(T &item) {
std::unique_lock<std::mutex> lock(mutex);
if (queue.empty()) {
return false;
}
item = std::move(queue.front());
queue.pop_front();
return true;
}
void shutdown() {
active_producers.fetch_sub(1, std::memory_order_release);
element_available.notify_all();
}
[[maybe_unused]] bool empty() {
std::unique_lock<std::mutex> lock(mutex);
return queue.empty();
}
[[maybe_unused]] size_t size() {
std::unique_lock<std::mutex> lock(mutex);
return queue.size();
}
[[maybe_unused]] void clear() {
std::unique_lock<std::mutex> lock(mutex);
queue.clear();
}
};
Thread pool
thread_pool.hpp:
#include "lock_free_queue.hpp"
#include <atomic>
#include <vector>
#include <thread>
#include <functional>
#include <future>
#include <cstddef>
using job_t = std::function<void()>;
class JoinThreads {
std::vector<std::thread> &threads;
public:
explicit JoinThreads(std::vector<std::thread> &to_join);
~JoinThreads();
};
class ThreadPool {
std::atomic_bool done;
LockFreeQueue<job_t> jobs;
std::vector<std::thread> threads;
JoinThreads joiner;
std::condition_variable cv;
std::mutex cv_m;
size_t pending_jobs = 0;
void worker_thread();
void job_done();
public:
explicit ThreadPool(size_t num_threads);
~ThreadPool();
template<typename func_T>
void submit(func_T &&f) {
jobs.push(std::forward<func_T>(f));
++pending_jobs;
}
void wait();
};
thread_pool.cpp:
#include "thread_pool.hpp"
#include <iostream>
using job_t = std::function<void()>;
JoinThreads::JoinThreads(std::vector<std::thread> &to_join) : threads(to_join) {}
JoinThreads::~JoinThreads() {
for (auto &thread: threads) {
if (thread.joinable()) {
thread.join();
}
}
}
void ThreadPool::worker_thread() {
while (!done) {
job_t job;
if (jobs.try_pop(job)) {
job();
job_done();
} else {
std::this_thread::yield();
}
}
}
void ThreadPool::job_done() {
{
std::unique_lock<std::mutex> lock(cv_m);
--pending_jobs;
}
cv.notify_all();
}
void ThreadPool::wait() {
std::unique_lock<std::mutex> lock(cv_m);
cv.wait(lock, [this] { return pending_jobs == 0; });
}
ThreadPool::ThreadPool(const size_t num_threads): done(false), joiner(threads) {
try {
for (size_t i = 0; i < num_threads; ++i) {
threads.emplace_back(&ThreadPool::worker_thread, this);
}
} catch (...) {
done = true;
throw;
}
}
ThreadPool::~ThreadPool() {
jobs.shutdown();
wait();
done = true;
}
Usage example
#include "thread_pool.hpp"
ThreadPool pool(num_threads);
pool.submit([this] { some_method(); });
pool.submit([this] { another_method(); });
Important
I'm also particularly curious if I've correctly implemented the wait() mechanism.
Thank you! I will be grateful for any advice and comments.
