28
\$\begingroup\$

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;
}
\$\endgroup\$
  • \$\begingroup\$ Is Buffer::add's body wrapped in the while for the sake of consistency with the other methods, or is there some reason behind it? It doesn't seem like it's being used for control flow, and it's not introducing a meaningful scope. I'm just wondering if I'm missing some trick or something. \$\endgroup\$ – Corbin Mar 15 '15 at 19:54
  • \$\begingroup\$ There is nothing special about it. It just illustrates that the production is infinite. However, if the buffer_ is full, that thread will put itself into wait or suspended condition (production stops until something is consumed) without wasting the CPU cycles. \$\endgroup\$ – Robomatt Mar 15 '15 at 22:31
  • \$\begingroup\$ Alright, I was just confused since a while(true) with an unconditional return is a bit pointless other than to introduce a scope (which plain brackets would be better for). Still not sure I get the point (other than consistency?), but it seems to be non-technical, so my curiosity is sated. \$\endgroup\$ – Corbin Mar 16 '15 at 1:09
  • \$\begingroup\$ is the while in the Buffer::remove necessary? Also, if there were no std::cout in the Producer and Consumer run methods, the cout_mu lock would not be necessary, right? \$\endgroup\$ – dev_nut Aug 24 '15 at 0:37
25
\$\begingroup\$

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;
};
\$\endgroup\$
  • \$\begingroup\$ Thanks a lot for your comment. Prefer Reference over Pointer I am confused about using reference over a pointer though. I mean whether you pass by reference or pointer, the Producer and Consumer objects do not have a copy of a buffer allocated to them. Prefer to avoid this I am just learning lambda notation, what is the proper way to capture buffer_ inside the lambda function? if I type '[&buffer_](){return buffer_.size() < size_;});` I am getting compile error. \$\endgroup\$ – Robomatt Mar 15 '15 at 12:19
  • \$\begingroup\$ __Member variables over globals __ How would you suggest that I use member mutexes? \$\endgroup\$ – Robomatt Mar 15 '15 at 12:31
  • 1
    \$\begingroup\$ __Member variables over globals __ How would you suggest that I use member mutexes? See updated answer. \$\endgroup\$ – Martin York Mar 15 '15 at 19:26
  • 1
    \$\begingroup\$ can some one explain use of while loop in this program? in size (add, remove). ? \$\endgroup\$ – Rupesh Yadav. Nov 15 '16 at 5:31
  • 1
    \$\begingroup\$ @LokiAstari yaa no need of argument just, was confused so asked, thank for your reply. \$\endgroup\$ – Rupesh Yadav. Nov 15 '16 at 15:17
14
\$\begingroup\$

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)
    {}
    
\$\endgroup\$

Not the answer you're looking for? Browse other questions tagged or ask your own question.