This post is based on A multi-thread Producer Consumer, where a Consumer has multiple Producers (C++17). I am trying to build a Consumer that consumes data from multiple Producers in a thread-safe manner. I extended the code in such a way that it is now possible to have an n:m relationship (many Producers and many Consumers). I would appreciate your thoughts and criticism. I also want to note that I will probably use a boost version in the and, as suggested in the previous post. I still would like to know if I did this correctly.
Some notes:
A Producer will not live indefinitely. At some point, it is done and will signal this to the Buffer. If there is no more Producer producing, the Consumer will stop consuming and the program will exit. This synchronization is handled by the producer_sem.
I am assuming a buffer that can grow indefinitely. This is why I do not have an emptyCount sempathore (compare wiki).
I am using only a single Buffer this time, instead of one Buffer per Producer. I believe this scales better with an increasing number of Consumers and Producers.
The random delay in the threads is there to simulate delay in the real world and to see if I run into synchronization issues.
Some questions:
For the Semaphore I am not using atomics, but lock_guards, as advised in the previous post. Is this smart? Why should I not use atomics?
When calling Buffer::add and Buffer::pop, does it make a difference if I first do lock.unlock() and then cond_var.notify_all() vs. the other way around?
#include <memory>
#include <optional>
#include <atomic>
#include <chrono>
#include <cmath>
#include <cstdlib>
#include <iostream>
#include <memory>
#include <mutex>
#include <sstream>
#include <thread>
#include <vector>
#include <shared_mutex>
/**
* RAII-style timer.
* Used only in main to measure performance
*/
class MyTimer
{
public:
using clock = std::chrono::high_resolution_clock;
MyTimer() : start(clock::now()) {}
~MyTimer()
{
auto duration = clock::now() - start;
std::cout << "elapsed time was " << std::chrono::duration_cast<std::chrono::microseconds>(duration).count() << " (us)\n";
}
private:
clock::time_point start;
};
/**
* Semaphore for coordination. Should I use lock_gards or atomics here?
*/
class Semaphore
{
public:
Semaphore() = delete;
Semaphore(int n) : m_(), n_(n) {}
void up()
{
std::lock_guard<std::mutex> lg(m_);
++n_;
}
void down()
{
std::lock_guard<std::mutex> lg(m_);
--n_;
}
bool greater_zero() const
{
std::lock_guard<std::mutex> lg(m_);
return n_ > 0;
}
private:
mutable std::mutex m_;
int n_;
};
class Buffer
{
public:
Buffer(int producer_parallelism) : buff_sem(0), producer_sem(producer_parallelism), mu(), print_mu(), cond_var(), buffer_(){};
Buffer() = delete;
/**
* Add an element to the buffer
*/
void add(char c)
{
std::unique_lock<std::mutex> lock(mu);
buffer_ << c;
buff_sem.up();
lock.unlock();
cond_var.notify_all();
}
/**
* Pop/get an element from the buffer. Return empty optional, if no value in queue
*/
std::optional<char> pop()
{
std::unique_lock<std::mutex> lock(mu);
// continue if there is data, or all producers are done
cond_var.wait(lock, [this]() -> bool { return buff_sem.greater_zero() || !producer_sem.greater_zero(); });
if (!producer_sem.greater_zero()) // return empty if all producers are done
{
return std::nullopt;
}
char c;
buffer_ >> c;
buff_sem.down();
lock.unlock();
cond_var.notify_all();
return c;
}
/**
* Indicate that one producer is finished
*/
void production_ended()
{
producer_sem.down();
cond_var.notify_all(); // if we do not notify here, the consumer will get stuck
}
/**
* Helper for synced printing
*/
template <typename... Args>
void print(Args... args) const
{
const std::lock_guard<std::mutex> lg(print_mu);
(std::cout << ... << args);
}
private:
Semaphore buff_sem;
Semaphore producer_sem;
mutable std::mutex mu; // sync all except print operation
mutable std::mutex print_mu; // sync print operations
mutable std::condition_variable cond_var; // sync access to underlying buffer
std::stringstream buffer_; // a stream for sharing data
};
/**
* A producer that produces a given number of items and shuts down afterwards.
*/
class Producer
{
public:
Producer(std::shared_ptr<Buffer> buffer, const int limit, const int id) : buffer_(buffer), limit_(limit), id_(id) {}
Producer() = delete;
/**
* produces random data.
*/
void run()
{
// for simulating delay of the producer
for (int count = 0; count < limit_; ++count)
{
static char const alphabet[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
char upper_case_char = alphabet[(random() % (sizeof alphabet - 1))];
buffer_->add(upper_case_char);
std::stringstream strs;
strs << "Produced: " << upper_case_char << ". Count at " << count << ". Producer was " << id_ << std::endl;
buffer_->print(strs.str());
std::this_thread::sleep_for(std::chrono::milliseconds(random() % 3));
}
buffer_->production_ended(); // signal to buffer that this producer is done
return;
}
private:
std::shared_ptr<Buffer> buffer_; // buffer is shared between producer and consumer
const int limit_; // number of elements to produce
const int id_; // id of producer
};
/**
* A consumer that consumes as long as something is produced.
*/
class Consumer
{
public:
Consumer(std::shared_ptr<Buffer> &buffer, const int parallelism, const int id) : buffer_(buffer), parallelism_(parallelism), id_(id){};
Consumer() = delete;
void run()
{
std::this_thread::sleep_for(std::chrono::milliseconds(random() % 3));
while (true)
{
auto c = buffer_->pop();
if (!c)
{
break;
}
buffer_->print("Consumer ", id_, " consumed ", c.value(), '\n');
}
}
private:
std::shared_ptr<Buffer> &buffer_; // a vector of shared buffers
const unsigned int parallelism_;
const int id_;
};
/**
* A simple thread pool. You can add threads here and join the all.
*/
class ThreadPool
{
public:
ThreadPool() : threads_(new std::vector<std::thread *>()), is_finished_(false){};
void add_thread(std::thread *t)
{
threads_->push_back(t);
}
void join_all()
{
for (auto it = threads_->begin(); it != threads_->end(); ++it)
{
(*it)->join();
}
}
private:
std::vector<std::thread *> *threads_;
bool is_finished_;
};
int main()
{
{
MyTimer mt;
// constants for this "experiment"
const int producer_parallelism = 5;
const int consumer_parallelism = 3;
const int produced_preaces_per_producer = 5;
// one buffer and one threadPool for all threads
std::shared_ptr<Buffer> buff = std::make_shared<Buffer>(producer_parallelism);
ThreadPool tp;
for (int i = 0; i < consumer_parallelism; ++i)
{
Consumer *c = new Consumer{buff, producer_parallelism, i};
std::thread *consumer_thread = new std::thread(&Consumer::run, c);
tp.add_thread(consumer_thread);
}
for (int i = 0; i < producer_parallelism; ++i)
{
Producer *p = new Producer{buff, produced_preaces_per_producer, i};
std::thread *producer_thread = new std::thread(&Producer::run, p);
tp.add_thread(producer_thread);
}
tp.join_all();
}
return 0;
}
#includecommands... \$\endgroup\$newin modern C++ is a definite code smell. \$\endgroup\$newthere must also be a matching call todelete. You have a lot of calls tonewbut zero calls todelete. Now you can wrap dynamic allocation in shared pointers. The one shared pointer you use (the buffer) does not even need to be allocated dynamically. \$\endgroup\$deleteandnew). Prefer automatic objects, pass by reference to allow other objects to use; pointers can be used but should be a last resort when non owning nullable references are needed (ie wait until you are an expert). \$\endgroup\$