EDID: Thank you very much for your feedback. I updated the code and opened a new post for the updated version. See here.
This post is loosely based on A multi-threaded Producer Consumer with C++11.
I would like to implement producer consumer pattern, where a consumer consumes data from multiple producers. The idea is to share the data between each producer and the consumer via a buffer. The consumer has a list of these shared buffers.
The consumer is further encouraged to consume data as soon as it is available, no matter from which producer it comes. This is because in reality a producer might be delayed and it would not be ideal to wait for producer x, while producer y already produced something. The code below checks if this works by using a timer and delaying producer deliberately with different delays.
I would have liked to provide a ready-to-run example environment, but unfortunately, compiler-explorer does not allow multithreading. Please compile with -std=c++17 -pthread.
Code:
#include <atomic>
#include <chrono>
#include <iostream>
#include <math.h>
#include <memory>
#include <mutex>
#include <sstream>
#include <thread>
#include <vector>
/**
* RAII-style timer.
* Used only in main to measure performance
*/
class MyTimer
{
public:
MyTimer() : start(std::chrono::high_resolution_clock::now()) {}
~MyTimer()
{
std::cout << "elapsed time was " << std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::high_resolution_clock::now() - start).count() << " (us)\n";
}
private:
std::chrono::_V2::system_clock::time_point start;
};
class Buffer
{
public:
Buffer(){};
~Buffer() = default;
/**
* Add an element to the buffer
*/
void add(char c)
{
std::unique_lock<std::mutex> locker(mu);
buffer_ << c;
is_ready_ = true;
return;
}
/**
* pop/get an element from the buffer
*/
char pop()
{
std::lock_guard<std::mutex> locker(mu);
char c;
buffer_ >> c;
if (!production_ongoing_ && contains_input())
{
is_valid_.exchange(false);
this->print("is valid to false in pop \n");
}
return c;
}
/**
* getter for private is_valid_
*/
bool is_valid()
{
return is_valid_.load();
}
/**
* indicate to buffer that producer is finished/will not produce more data
*/
void no_more_production()
{
std::lock_guard<std::mutex> locker(mu);
production_ongoing_ = false;
if (!contains_input())
{
is_valid_ = false;
}
}
/**
* helper for synced printing
*/
void print(std::string msg)
{
std::lock_guard<std::mutex> lg(print_mu);
std::cout << msg;
}
/**
* getter for private is_ready_
*/
bool is_ready()
{
std::lock_guard<std::mutex> locker(mu);
return is_ready_;
}
/**
* getter for private production_ongoing_
*/
bool production_ongoing()
{
std::lock_guard<std::mutex> locker(mu);
return production_ongoing_;
}
private:
std::mutex mu; // sync all except print operation
std::mutex print_mu; // sync print operations
std::stringstream buffer_; // a stream for sharing data
bool production_ongoing_ = true; // false if production is finished
std::atomic_bool is_valid_ = true; // false, if producer is finished and buffer is empty
bool is_ready_ = false; // true after production initially began
bool contains_input() // check if there is input that can be retrieved from the buffer
{
// compare https://stackoverflow.com/questions/40608111/why-is-18446744073709551615-1-true
int tmp = buffer_.peek();
return tmp != -1 && tmp != std::pow(2, 64) - 1;
}
};
class Producer
{
public:
Producer(std::shared_ptr<Buffer> buffer, const int limit, const int id, const int delay) : buffer_(buffer), limit_(limit), id_(id), delay_(delay) {}
/**
* produces random data.
*/
void run()
{
// for simulating delay of the producer
std::this_thread::sleep_for(std::chrono::milliseconds(delay_));
for (int count = 0; count < limit_; ++count)
{
char upper_case_char = (char)((random() % 26) + int('A'));
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(50));
}
buffer_->no_more_production(); // indicate to buffer that production is done
}
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
const int delay_; // start delay of producer
};
class Consumer
{
public:
Consumer(std::vector<std::shared_ptr<Buffer>> &buffers, const int parallelism) : buffers_(buffers), parallelism_(parallelism){};
void run()
{
// Consumer responsible for multiple producer. Is any of them still producing?
bool any_valid = true;
do
{
// if not all producers joined yet. This is in case the consumer is created earlier than the prod
any_valid = buffers_.size() < parallelism_ ? true : false;
// iterate over all available buffers
for (size_t t = 0; t < buffers_.size(); ++t)
{
if (!buffers_.at(t)->is_ready())
{
// will skip this producer. Helpful if producer is slow (network delay)
any_valid = true;
continue;
}
if (buffers_.at(t)->is_valid())
{
// is_valid if we are expecting data from producer
any_valid = true;
char c = buffers_.at(t)->pop();
std::stringstream strs;
strs << "Consumed: " << c << '\n';
buffers_.at(t)->print(strs.str());
}
}
std::this_thread::sleep_for(std::chrono::milliseconds(50));
} while (any_valid);
buffers_.at(0)->print("consumer finished\n");
}
private:
std::vector<std::shared_ptr<Buffer>> &buffers_; // a vector of shared buffers
const int parallelism_;
};
int main()
{
{
// all numbers are based on measurements on my machine in debug mode
// Scenario 1: All producer threads have the same delay
// if we do not start with ready thread, this will take about 0.3s
// if we do start with ready thread, this will take about 0.25s
MyTimer mt;
const int parallelism = 3;
std::vector<std::shared_ptr<Buffer>> buffVec;
Consumer c{buffVec, parallelism};
std::thread consumer_thread(&Consumer::run, &c);
for (int i = 0; i < parallelism; ++i)
{
// each buffer is shared between a producer and the consumer
std::shared_ptr<Buffer> b = std::make_shared<Buffer>();
buffVec.push_back(b);
Producer *p = new Producer(b, 3, i, 30);
std::thread producer_thread(&Producer::run, &(*p));
producer_thread.detach();
}
consumer_thread.join();
}
{
// Scenario 2: First producer thread has long delay, others have none
// Total delay is equal to Scenario 1
// if we do not start with ready thread, this will take 0.5s
// if we do start with ready thread, this will take about 0.3s
MyTimer mt;
const int parallelism = 3;
std::vector<std::shared_ptr<Buffer>> buffVec;
Consumer c{buffVec, parallelism};
std::thread consumer_thread(&Consumer::run, &c);
for (int i = 0; i < parallelism; ++i)
{
const int delay = i == 0 ? 90 : 0;
// each buffer is shared between a producer and the consumer
std::shared_ptr<Buffer> b = std::make_shared<Buffer>();
buffVec.push_back(b);
Producer *p = new Producer(b, 3, i, delay);
std::thread producer_thread(&Producer::run, &(*p));
producer_thread.detach(); // start producers independent from each other and from consumer
}
consumer_thread.join();
}
return 0;
}
