Generic Producer-Consumer in C++11

Question

I wrote the following simple 1 producer - 1 consumer problem, in attempt to learn some C++11 threading / generics.

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

struct data {
    int x, y;
    std::string s;

    data() : s("test") {
        x = std::rand() % 1000000;
        y = std::rand() % 1000000;
    }

    friend std::ostream& operator<<(std::ostream& os, const struct data d) {
        os << d.x << ", " << d.y << ", " << d.s << "\n";
    }
};


typedef struct data elem_t;

template <typename T>
class Buffer 
{
public:
    Buffer(): m_size(1) {}
    Buffer(int s): m_size(s) {}

    void add(T x) {
        while (true) {
            std::unique_lock<std::mutex> locker(m_mtx);
            m_cond.wait(locker, [this](){return m_buffer.size() < m_size;});

            m_buffer.push_back(x);

            locker.unlock();
            m_cond.notify_all();

            return;
        }
    }

    T remove() {
        while (true) {
            std::unique_lock<std::mutex> locker(m_mtx);
            m_cond.wait(locker, [this](){return m_buffer.size() > 0;});

            T x = m_buffer.back();
            m_buffer.pop_back();

            locker.unlock();
            m_cond.notify_all();

            return x;
        }
    }

private:
    std::mutex m_mtx;
    std::condition_variable m_cond;

    std::deque<T> m_buffer;
    unsigned int m_size;
};



template <typename T>
class Producer
{
public:
    Producer() : m_buf(nullptr) {}
    Producer(std::string&& id, Buffer<T> *buf) : m_id(id), m_buf(buf) {}

    void produce() {
        while (true) {
            T x = T();

            m_buf->add(x);
            std::cout << x;

            int sleep = std::rand() % 1000 + 10;
            std::this_thread::sleep_for(std::chrono::milliseconds(sleep));
        }
    }

private:
    Buffer<T> *m_buf;
    std::string m_id;
};



template <typename T = int>
class Consumer
{
public:
    Consumer() : m_buf(nullptr) {}
    Consumer(std::string&& id, Buffer<T> *buf) : m_id(id), m_buf(buf) {}

    void consume() {
        while(true) {
            T x = m_buf->remove();
            std::cout << x;

            int sleep = std::rand() % 1000 + 10;
            std::this_thread::sleep_for(std::chrono::milliseconds(sleep));
        }
    }

private:
    Buffer<T> *m_buf;
    std::string m_id;
};

int main(int argc, char const *argv[])
{
    std::srand(std::time(nullptr));

    Buffer<elem_t> *buffer = new Buffer<elem_t>(std::stoi(argv[1]));
    Producer<elem_t> *p = new Producer<elem_t>("prod", buffer);
    Consumer<elem_t> *c = new Consumer<elem_t>("cons", buffer);

    std::thread pt(&Producer<elem_t>::produce, p);
    std::thread ct(&Consumer<elem_t>::consume, c);

    pt.join();
    ct.join();

    return 0;
}

The main concern I am having with this code is regarding genericity. I have a generic buffer Buffer<T> (which is basically a std::deque<T>) and the produced value: T x = T();

For now, T = struct data, and everything works because of data(), but if T = int, for example, things start to get ugly.

One idea I have is to define a struct data packing all the needed fields, and then remove templates from classes (i.e. std::deque<struct data> only).

So, are templates an overkill here, and if not what is the / a correct way of expressing this code with them? Also, are there any other problems / code smells / things to improve?

Answer 1

Buffer<T>

This class seems fine for the most part. Some small issues:

• The while(true) loops in insert(T) and remove() only ever run one iteration. Those loops can be replaced by their body.

• add calls m_buffer.push_back(x), which might not compile for types without a copy constructor. Try m_buffer.push_back(std::move(x)) instead: That will move if possible, and fall back to copy if not.

• m_buffer is treated like a stack (push_back and pop_back are used only), so newest insertions get removed first. Maybe a queue was intended (using e.g. push_front and pop_back)?

• The constructors initialize Buffer<T>::m_size (an unsigned int) with a signed value (1 or int s, respectively).

• notify_all() is called on the condition variable. This would wake up every thread waiting on it - but very likely only one would actually be able to do anything. notify_one() would likely be a better fit.

• The notify_all calls on the condition variable aren't needed every time: remove only needs to notify if other threads might be waiting to add elements, which is only when m_buffer.size() == m_size - 1 after the call to pop_back. Similarly, add only needs to notify if m_buffer.size() == 1 after push_front.

• Buffer<T>::m_size doesn't represent the current size of the buffer (= number of elements contained), but the maximum capacity. Maybe m_capacity would be a better name.

• Buffer itself isn't very descriptive of what it actually does. ConcurrentQueue, MessageQueue or similar would better represent its behavior.

• While add and remove by themselves are fine names, common expectation for queue-like types are push and pop.

• This actually is a working queue for n producers and m consumers. It seems a bit overkill for a 1 producer + 1 consumer scenario.

Furthermore, some improvements could be made:

• bool try_push(T&&)/bool try_push(const T&)/bool try_pop(T&) member functions could be added that return early if the operation would otherwise have to wait.

• push could be overloaded for const T& and T&& to enable better performance for moveable types.

Producer<T>/Consumer<T>

Both of these don't really seem necessary. Generally, its unlikely that someone will ever need generic producers/consumers (other than maybe for testing purposes), so I won't go into too much detail

• Default constructors initialize m_buf to nullptr, but this is never checked afterwards. Either add checks for nullptr in the other member functions, or simply remove the default constructor.

• Similarly, the buf parameter of the other constructor can be nullptr. If m_buf should never be nullptr (as the member functions seem to assume), simply make buf a Buffer<T>& instead. m_buf could then be initialized to &buf - but it might be more reasonable to change m_buf to Buffer<T>&, too.

• Access to std::cout isn't synchronized. In some cases, this means the consumer might print before the producer does.

• There is no way to stop a producer/consumer (other than forcefully terminating the program).

• The member m_id is never used.

• The second constructor takes a std::string&& id parameter - but initializes m_id without moving - which creates a copy. Generally, it would be expected that the object id refers to would be moved from after construction.

General stuff

• main never checks if argv[1] is actually valid.

• Using struct data when referencing the type isn't necessary in C++, just data suffices. So typedef struct data elem_t; could be simplified to typedef data elem_t;, or with a more modern approach, using elem_t = data;. (The struct data would be necessary in C, but this code will never interact with C as C doesn't support templates anyways.)

• Many places use parentheses to initialize objects, like std::unique_lock<std::mutex> locker(m_mtx);. Prefer braces {} instead: In the example, if m_mtx could be interpreted as a type, then the compiler would be forced to make locker a function declaration. Using braces would prevent that from happening.

• Prefer the utilities in the <random> header for generating random numbers, as std::rand is kinda terrible.

• There's no need to create buffer, p and c on the heap (inside main). Using local variables would be just fine. (Also, they never get deleted properly. If allocation on the heap is absolutely needed, prefer using smart pointers like std::unique_ptr or std::shared_ptr.)

• I cannot think of a 1 producer/1 consumer scenario in which threads are necessary: Most could be simplified to this pseudocode:

  int main() {
      while(keep_running()) {
          consume_one(produce_one());
      }
  }