A multi-threaded Producer Consumer with C++11

Question

I am trying to learn concurrent programming in C++11. I tried to write code for a classic producer consumer concurrency problem. Would you please review and make any comments about it?

#include <iostream>
#include <thread>
#include <deque>
#include <mutex>
#include <chrono>
#include <condition_variable>

using std::deque;
std::mutex mu,cout_mu;
std::condition_variable cond;

class Buffer
{
public:
    void add(int num) {
        while (true) {
            std::unique_lock<std::mutex> locker(mu);
            cond.wait(locker, [this](){return buffer_.size() < size_;});
            buffer_.push_back(num);
            locker.unlock();
            cond.notify_all();
            return;
        }
    }
    int remove() {
        while (true)
        {
            std::unique_lock<std::mutex> locker(mu);
            cond.wait(locker, [this](){return buffer_.size() > 0;});
            int back = buffer_.back();
            buffer_.pop_back(); 
            locker.unlock();
            cond.notify_all();
            return back;
        }
    }
    Buffer() {}
private:
    deque<int> buffer_;
    const unsigned int size_ = 10;
};

class Producer
{
public:
    Producer(Buffer* buffer)
    {
        this->buffer_ = buffer;
    }
    void run() {
        while (true) {
            int num = std::rand() % 100;
            buffer_->add(num);
            cout_mu.lock();
            std::cout << "Produced: " << num << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
            cout_mu.unlock();
        }
    }
private:
    Buffer *buffer_;
};

class Consumer
{
public:
    Consumer(Buffer* buffer)
    {
        this->buffer_ = buffer;
    }
    void run() {
        while (true) {
            int num = buffer_->remove();
            cout_mu.lock();
            std::cout << "Consumed: " << num << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
            cout_mu.unlock();
        }
    }
private:
    Buffer *buffer_;
};

int main() {
    Buffer b;
    Producer p(&b);
    Consumer c(&b);

    std::thread producer_thread(&Producer::run, &p);
    std::thread consumer_thread(&Consumer::run, &c);

    producer_thread.join();
    consumer_thread.join();
    getchar();
    return 0;
}

Answer 1

Prefer Reference over Pointer

Since the producer and consumer must have a buffer you should pass it by reference (rather than pointer). This also makes sure there is no confusion over ownership of the buffer (the owner of a pointer is responsible for deleting it). By using a RAW pointer you can not tell the owner but by using a reference you are explicitly stating that you are not passing ownership.

class Producer
{
    Buffer&  buffer_;
public:
    Producer(Buffer& buffer)
        : buffer_(buffer)
    {}

Prefer to avoid this->

When you use this-> it means you have scoping issues with your variables (which is a code smell for bad design). Use accurate variable names so that there is no confusion on where the variables belongs.

Member variables over globals.

Looks like your encapsulation of the mu and cond is wrong.

std::mutex mu,cout_mu;

Because they are global, you use a single mutex/condition for all buffer objects. This looks like a design flaw. It seems like the mutex/condition belongs to the class so that you are just locking the buffer you are manipulating (so you can use multiple buffers in the same application).

Sleeping inside lock

        cout_mu.lock();
        std::cout << "Produced: " << num << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
        cout_mu.unlock();

@Morwenn already suggested using scoped lock to make sure the mutex were correctly locked and unlocked. But I would also move the sleep outside the lock. This causes the current thread to stall but the other thread can not continue as the lock is still held while it is sleeping.

        {
            std::unique_lock<std::mutex> locker(cout_mu);
            std::cout << "Produced: " << num << std::endl;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(50));

Prefer '\n' over std::endl

std::cout << "Produced: " << num << std::endl;

The difference between the two is a flush. Usually you do not want to manually flush (the stream have good flushing techniques built in). So it is usually best to let the streams flush themselves when appropriate.

---

In answer to comments:

class Buffer
{
public:
    void add(int num) {
        while (true) {
            std::unique_lock<std::mutex> locker(mu);
            cond.wait(locker, [this](){return buffer_.size() < size_;});
            buffer_.push_back(num);
            locker.unlock();
            cond.notify_all();
            return;
        }
    }
    int remove() {
        while (true)
        {
            std::unique_lock<std::mutex> locker(mu);
            cond.wait(locker, [this](){return buffer_.size() > 0;});
            int back = buffer_.back();
            buffer_.pop_back();
            locker.unlock();
            cond.notify_all();
            return back;
        }
    }
    Buffer() {}
private:
   // Add them as member variables here
    std::mutex mu;
    std::condition_variable cond;

   // Your normal variables here
    deque<int> buffer_;
    const unsigned int size_ = 10;
};

Answer 2

I have a few comments about your code:

• Using std::lock_guard is great to handle mutexes since it automatically unlocks the acquired mutex when leaving the scope. That's a real great tool. You should really use it everywhere you can. Using it consistently will make sure that you can't forget to unlock any mutex. Moreover, it makes sure that mutexes are unlocked even when an exception is thrown.

  while (true) {
      int num = buffer_->remove();
      std::lock_guard<std::mutex> lock(cout_mu);
      std::cout << "Consumed: " << num << std::endl;
      std::this_thread::sleep_for(std::chrono::milliseconds(50));
  }
  

• std::rand is not guaranteed to be thread-safe by the standard and many platforms don't bother to make it thread-safe because locking and unlocking mutexes has a price. Therefore you should use the new pseudo-random number generators from <random> instead to make sure that you can use several producers at once:

  void run() {
      // Non-deterministic pseudo-random numbers generator
      thread_local std::random_device rd;
      // Pseudo-random engine
      thread_local std::mt19937 engine(rd());
      // Linear distribution in [0, 100[
      thread_local std::uniform_int_distribution<int> dist(0, 99);
  
      while (true) {
          int num = dist(engine);
          buffer_->add(num);
          std::lock_guard<std::mutex> lock(cout_mu);
          std::cout << "Produced: " << num << std::endl;
          std::this_thread::sleep_for(std::chrono::milliseconds(50));
      }
  }
  

  This module has a higher learning curve than std::rand but it is more powerful, more flexible and you can thread_local the objects to make the pseudo-random number generation thread-safe without losing the time needed to lock/unlock mutexes keep the thread safety.

• Use constructor initialization lists to construct your objects when you can instead of assigning to the member variables in the body of the constructor:

  Producer(Buffer* buffer):
      buffer_(buffer)
  {}