# A multi-thread Producer Consumer, where a Consumer has multiple Producers (C++17)

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 <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
*/
{
std::unique_lock<std::mutex> locker(mu);
buffer_ << c;
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()
{
}
/**
* 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;
}
/**
*/
{
std::lock_guard<std::mutex> locker(mu);
}
/**
* 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
for (int count = 0; count < limit_; ++count)
{
char upper_case_char = (char)((random() % 26) + int('A'));
std::stringstream strs;
strs << "Produced: " << upper_case_char << ". Count at " << count << ". Producer was " << id_ << std::endl;
buffer_->print(strs.str());
}
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)
{
{
// 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());
}
}
} 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
MyTimer mt;
const int parallelism = 3;
std::vector<std::shared_ptr<Buffer>> buffVec;
Consumer c{buffVec, parallelism};

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);
}
}
{
// Scenario 2: First producer thread has long delay, others have none
// Total delay is equal to Scenario 1
MyTimer mt;
const int parallelism = 3;
std::vector<std::shared_ptr<Buffer>> buffVec;
Consumer c{buffVec, parallelism};

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);
producer_thread.detach(); // start producers independent from each other and from consumer
}

}
return 0;
}

• Your code doesn't compile for me on wandbox. What compiler version are you using? Jan 15 '20 at 11:29
• Builds fine for me (locally) with GCC 9.2. Just a couple of warnings. Jan 15 '20 at 11:47
• Don't roll your own. This is too hard to get right. Use Boost's: boost.org/doc/libs/1_72_0/doc/html/boost/lockfree/queue.html Facebook's Folly library also has one. Jan 15 '20 at 22:03

std::this_thread::sleep_for(std::chrono::milliseconds(50));

Don't do this. Use a condition_variable instead. This will require some minor revising. 50ms might be a long time. Remember that to the OS, it means "context switch out the thread and keep it idle for at least 50ms". A std::mutex may have all manner of fancy implementation dependent optimizations. For example, if the consumer exhausts its work and waits on the condition variable, it may not need to be context switched at all if new work is very quickly produced.

Furthermore, this is wasting precious CPU resources. If production is stalled, it will context switch up to 20 times per second for no reason.

buffVec needs to be synchronized

Avoid adding artificial delays to your code. I believe they're hiding potential race conditions.

In my opinion, consider removing shared_ptr and making the client manage the memory. I believe should be implicit to the client that the memory used by the producer/consumer needs to outlive both of them to function properly. It can be more efficient in some cases, but require more code in other cases to move the shared state around. If this were a library, it could potentially be a templated type and the client could choose their desired storage strategy.

I am very adverse to seeing atomics in code that isn't building other low-level synchronization primitives. Using a mutex with RAII guards is much safer, and without any benchmarking to compare, I would argue that there's no reason to expect them to be underperformant. It can cause cache invalidation issues and out of order problems that are hard to reason about.

• Thank you very much. I think you are right. The artificial delays to hide race conditions Jan 16 '20 at 11:01

I get a couple of warnings, which ought to be fixed:

g++ -std=c++2a -fPIC -g -Wall -Wextra -Wwrite-strings -Wno-parentheses -Wpedantic -Warray-bounds  -Weffc++    -pthread    235651.cpp    -o 235651
235651.cpp: In constructor ‘Buffer::Buffer()’:
235651.cpp:31:5: warning: ‘Buffer::mu’ should be initialized in the member initialization list [-Weffc++]
31 |     Buffer(){};
|     ^~~~~~
235651.cpp:31:5: warning: ‘Buffer::print_mu’ should be initialized in the member initialization list [-Weffc++]
235651.cpp:31:5: warning: ‘Buffer::buffer_’ should be initialized in the member initialization list [-Weffc++]
235651.cpp: In member function ‘void Consumer::run()’:
235651.cpp:159:41: warning: comparison of integer expressions of different signedness: ‘std::vector<std::shared_ptr<Buffer> >::size_type’ {aka ‘long unsigned int’} and ‘const int’ [-Wsign-compare]
159 |             any_valid = buffers_.size() < parallelism_ ? true : false;
|                         ~~~~~~~~~~~~~~~~^~~~~~~~~~~~~~


We include <math.h> but then use std::pow - we should be including <cmath> if we want the names to be in the std namespace (which we do).

The consumers don't block, but repeatedly get a null. That seems to be a failure of the whole purpose of the class:

Produced: N. Count at 0. Producer was 0
Produced: L. Count at 0. Producer was 2
Produced: W. Count at 0. Producer was 1
Consumed: N
Consumed: W
Consumed: L
Consumed: \0
Consumed: \0
Consumed: \0
Produced: B. Count at 1. Producer was 2
Produced: B. Count at 1. Producer was 0
Produced: R. Count at 1. Producer was 1
Consumed: \0
Consumed: \0
Consumed: \0
Produced: Q. Count at 2. Producer was 1
Produced: B. Count at 2. Producer was 2
Produced: M. Count at 2. Producer was 0
Consumed: \0


Other questionable bits:

• buffers_.size() < parallelism_ ? true : false


That should be written as just buffers_.size() < parallelism_.

• char upper_case_char = (char)((random() % 26) + int('A'));


We need <cstdlib> to define std::random(). And C++ doesn't guarantee that letters are contiguously encoded. Try

static char const alphabet[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
char upper_case_char = alphabet[(std::random() % (sizeof alphabet - 1))];

• Thank you very much for the comment. Would you change anything regarding the synchronization mechanisms? Jan 15 '20 at 14:36
• Probably - we probably want a semaphore to fix the consumers to wait for input. Fix that bug first; I think you'll want a new review when you've done that. Jan 15 '20 at 15:16
• Semaphores are definitely the right tool for the job, but standard C++ doesn't have them. You could use an OS implementation, or a cross platform library, or condition_variable.
– butt
Jan 15 '20 at 20:45

# Avoid mixing clock types

Why use std::chrono::_V2::system_clock::time_point start but initialize it with std::chrono::high_resolution_clock::now()? If there is a reason to use the non-standard _V2 clocks, then you should probably stick with it everywhere. But if there is no reason to, avoid the non-standard _V2 thing.

To make your code more consistent and to reduce the amount of code you have to type, define an alias for the clock namespace you want to use, like so:

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;
};


# Avoid useless definitions of default constructors and destructors

In class Buffer, the only constructor is not doing anything, and the destructor is set to the default. There is no need for this, just omit them completely.

# There is no need to lock in print()

Single calls to member functions of iostreams are atomic, see this post. So there is no need for print_mu.

There is almost never a good reason to detach threads. Doing so means losing control over the threads and the resources it uses. Threads can be easily managed by STL containers. So in your main() you could write:

std::vector<std::thread> producer_threads;

for (int i = 0; i < parallelism; ++i)
{
...
Producer *p = new Producer(b, 3, i, 30);

Note that you are still leaking Producer objects, since you never delete them. You could put those in a std::vector as well, or you could change class Producer to start a thread in its own constructor, so you just need the vector holding Producers.