Type-agnostic abortion condition in Producer-Consumer model queue

Question

This is a follow-up post to the topic discussed here.

Description

I am trying to implement a robust, multi-threaded producer-consumer model with a single producer in the main thread and several consumers in other threads.

After researching the topic, I also added a copy constructor and an assignment operator (or better, deleted it), to satisfy the rule of three. I did so, as being able to copy a thread managing class instance does not make sense.

The operating mode is the same as before, e.g. the generator thread loops over data, recognizes through some condition which sort of data it is dealing with, and proceeds to delegate the sample to the respective worker thread (via pushBack(data)). When reaching the end of the source data, it queues a special value to the worker threads and waits for them to finish. Afterwards, it cleans up and exits.

The queue implementation (found here) provides a blocking dequeue function, e.g. no polling is necessary. The downside is that the consumers do not necessarily register that the main thread has set m_running (a flag indicating if execution should be stopped) because they are stuck waiting. As mentioned before, this can be bypassed by queueing a special value. In the provided example, I quietly assume that the queue only manages pointer types (e.g. int*). In that case, the special value was determined to be nullptr.

Credit goes to @G.Sliepen, who provided many suggestions in the previous topic.

Code

#include <chrono>
#include <iostream>
#include <thread>
#include <functional>
#include "readerwriterqueue.h"

template<typename T>
class Consumer
{
public:
    Consumer(int id, std::function<void(int, T&)> func) : m_id(id), m_func(func) {}

    ~Consumer(){
        m_running = false;
        this->pushBack(nullptr);
        m_thread.join();
    }

    Consumer(const Consumer&) = delete;
    Consumer& operator=(const Consumer&) = delete;
    Consumer() : m_id(0) {};

    void pushBack(const T& t){
        m_BufferQueue.enqueue(t);
    }

private:

    void work() {
        m_running = true;

        while(m_running || m_BufferQueue.peek())
        {
            T t;
            m_BufferQueue.wait_dequeue(t);

            if (t == nullptr)
                break;

            m_func(m_id, t);
        }

        std::cout << "EXIT thread " << m_id << std::endl;
    }

    int m_id;
    std::function<void(int, T&)> m_func;
    moodycamel::BlockingReaderWriterQueue<T> m_BufferQueue{64};
    std::thread m_thread{&Consumer::work, this};
    std::atomic_bool m_running;
};

int main() {

    auto func = [](int id, int* val){
        std::cout << "Thread " << id << " received value " << *val << std::endl;
        delete val;
    };

    Consumer<int*> c1(1, func);
    Consumer<int*> c2(2, func);

    // data generator
    for(int i = 0; i < 10; i++)
    {
        int* val = new int(i);
        if (i % 2 == 0)
            c1.pushBack(val);
        else
            c2.pushBack(val);
    }

    std::cout << "EXIT" << std::endl;
    return 0;
}

Question

This implementation breaks when using a non-pointer datatype in the template. As far as I know, references always have to reference non-null objects, so checking for "null-references" does not make sense. And checking for every possible type seems to be very error-prone. I thought of adding a parametrized member variable that holds the abortion value, but I am not sure if this is the best way of doing it.

In short: is there an elegant way to define an abortion value that is type-agnostic?

Answer 1

Avoid having T be a pointer

I strongly recommend avoiding template parameters be pointers, instead let them be value types. You can easily turn them into pointers where needed. First of all, it is not very idiomatic to pass pointers as template arguments to container-like types, and a Consumer is just a container (the queue) and an associated thread. So as a user of your class, my first instinct would be to write:

Consumer<int> c1(1, func);

But this would fail to compile. Also consider that if you leave it up to the user to pass the pointer type, they could try to do something like:

Consumer<const int*> c1(1, func);

What would the semantics of that be? Another issue is that bad things will happen is a caller does something like:

c1.pushBack(nullptr);
c1.pushBack(new int(42));

The first call to pushBack() will cause the worker to stop processing the queue any further, which means the second call will have caused a memory leak. Of course it is unlikely that someone would intentionally push a nullptr, but consider that you might have code that looks like:

c1.pushBack(get_the_thing(...));

Where the function you are calling might sometimes return a nullptr. This would lead to hard to debug problems. Finally, this design encourages the use of raw calls to new and delete, which should be avoided:

Let the Consumer manage the lifetime of queue items

In the previous version of your code, a Consumer took care of managing storage for the items in the queue. You can still do that, and avoid raw pointers, by having the queue store items of type std::unique_ptr<T>. And you can still have an empty std::unique_ptr, which you can use as a signal that the worker should stop:

template<typename T>
class Consumer {
public:
    ...
    ~Consumer() {
         ...
         m_BufferQueue.enqueue(nullptr);
         ...
    }
    ...   
    void pushBack(const T& t) {
        m_BufferQueue.enqueue(std::make_unique<T>(t));
    }

private:
    void work() {
        ...
        while(...) {
            std::unique_ptr<T> ptr;
            m_BufferQueue.wait_dequeue(ptr);

            if (!ptr)
                break;

            m_func(m_id, *ptr);
        }
        ...
    }
    ...
    moodycamel::BlockingReaderWriterQueue<std::unique_ptr<T>> m_BufferQueue{64};
    ...
};

The drawback of using pointers, whether raw or smart ones, is that memory is allocated for every element you push to the queue, which is not great for performance. Alternatively, consider using std::optional<T> instead. You can use it as a drop-in replacement for std::unique_ptr<T> in the above code.

Consider adding an emplaceBack() function as well

Just like STL containers have a push_back() and emplace_back(), consider adding the latter as well. This avoids first having to construct an object of type T, which then has to be move- or copy-constructed into the queue. Of course, this requires your BlockingReaderWriterQueue() to also have an emplace() function. Having an emplace-like function is especially important if you want to be able to add uncopyable and unmovable object to the queue.

m_running is no longer necessary

Since you guarantee that a special value will always be pushed to the queue when the destructor of Consumer is running, you don't need m_running at all. You can safely remove it.

Note that setting m_running to true at the start of the worker thread is also a bit dangerous; consider that a Consumer object might be destroyed immediately after it was created, then a race condition could happen between the m_running = false in the destructor and m_running = true in work().

Alternative solutions

All approaches with pointers or optionals have the drawback that they increase the amount of storage necessary for every element in the queue. For very large queues with small elements that might be an issue. The only thing you want to know is whether you should break out of the loop in work() or not, and ideally just a single (atomic) boolean would be necessary, like m_running. But the problem is ensuring work() gets woken up if it's waiting on an empty queue. You could instead push this problem down to BlockingReadeWriterQueue, so that you signal the queue itself that it should terminate, and then wait_dequeue() should return false once the queue is terminated and no items are left. Your Consumer could then just do:

~Consumer() {
    m_BUfferQueue.terminate();
    m_thread.join();
}

void work() {
    for (T t; m_BufferQueue.wait_dequeue(t);)
        m_func(m_id, t);
}