As an exercise in multithreaded programming, I am trying to implement a basic FIFO task queue. For this, I also implement simplified version of my lock_guard, from STL. Here is the code:
#include <mutex>
#include <iostream>
#include <thread>
#include <chrono>
#include <optional>
namespace cho {
template <typename M>
struct lock_guard {
lock_guard() noexcept = default;
lock_guard(M& m) noexcept {
mtx = &m;
mtx->lock();
}
lock_guard(lock_guard const& rhs) = delete;
lock_guard(lock_guard && rhs) = delete;
lock_guard& operator=(lock_guard const& rhs) = delete;
lock_guard& operator=(lock_guard && rhs) = delete;
~lock_guard() {
mtx->unlock();
}
private:
M *mtx = nullptr;
};
template <typename T>
struct queue {
queue() noexcept = default;
~queue() {
queue_node *cur = head;
while(cur) {
queue_node *tmp = cur;
cur = cur->next;
delete tmp;
}
}
void push(T const& elem) {
queue_node *node = new queue_node{};
T *new_elem = new T(elem);
node->data = *new_elem;
node->next = nullptr;
if(head == tail)
one_elem_mtx.lock();
lock_guard<std::mutex> grd(push_mtx);
if(head) {
tail->next = node;
} else {
head = node;
}
tail = node;
one_elem_mtx.unlock();
}
auto pop() {
if(!head) {
return std::optional<T>{std::nullopt};
}
if(head == tail)
one_elem_mtx.lock();
lock_guard<std::mutex> grd(pop_mtx);
std::optional<T> ret{head->data};
queue_node *c = head;
head = head->next;
delete c;
one_elem_mtx.unlock();
return ret;
}
private:
struct queue_node {
T data = T{};
queue_node *next = nullptr;
};
queue_node *head = nullptr;
queue_node *tail = nullptr;
std::mutex push_mtx = std::mutex{};
std::mutex pop_mtx = std::mutex{};
std::mutex one_elem_mtx = std::mutex{};
};
}
struct log_task {
void operator()(char const* msg) {
std::cout << "Log: " << msg << "#" << log_id++<< "\n" << std::flush;
}
private:
static std::size_t log_id;
};
std::size_t log_task::log_id = 0;
int main() {
cho::queue<log_task> log_queue;
std::thread task_creator1([&]() {
while(true) {
using namespace std::chrono_literals;
std::this_thread::sleep_for(300ms);
log_queue.push(log_task());
}
});
std::thread task_creator2([&]() {
while(true) {
using namespace std::chrono_literals;
std::this_thread::sleep_for(500ms);
log_queue.push(log_task());
}
});
std::thread task_consumer1([&]() {
while(true) {
using namespace std::chrono_literals;
std::this_thread::sleep_for(300ms);
std::optional<log_task> t = log_queue.pop();
if(t)
t.value()("log_message");
}
});
std::thread task_consumer2([&]() {
while(true) {
using namespace std::chrono_literals;
std::this_thread::sleep_for(500ms);
std::optional<log_task> t = log_queue.pop();
if(t)
t.value()("log_message");
}
});
task_creator1.join();
task_creator2.join();
task_consumer1.join();
task_consumer2.join();
return 0;
}
lock_guard and queue implementation lives in my cho namespace. For queue, I only provide basic push and pop, not a production-ready implementation by any means. Then, I implement a logger to be used as test functor in my task queue, called log_task. My queue push and pop operations are guarded by mutexes to avoid messing queue pointers. I want to be able to push tasks to the queue from multiple threads and consume those tasks from multiple threads as well. For this reason, I hold two different mutexes: push_mtx and pop_mtx. On the other hand, I realized that if there is only one element, they can still mess the values since head and tail points to the same node. I added another one_elem_mtx that needs to be locked just on this occasion.
What are code smells you can point? I am, I don't know why, not happy with using three mutexes and feel like there is something wrong there.
I played with delays in
sleep_for()to try different values and my tests are working fine. But still, can you see any chance of deadlock or livelock?I've seen many implementations on other questions that use only one mutex. So when one thread pushes, popping threads need to wait as well and vice versa. This seemed like a performance penalty to me but I am not sure. Does my approach indicate a bad design and should I abandon concurrent push/pop?
