I am writing a simple program with a producer and a few consumers: the producer pushes to a queue some integers, and the consumers pop elements from the queue and print them (order doesn't matter). The queue code can be found below (the implementation is based on this example).In addition to the regular pop and push operations, I want to add a kill method - this method will kill the queue but will let the consumers keep popping items until the queue is empty (hence the different implementation of pop). I am looking for an elegant way to do it. I want to know if this way is fine:void kill()    {        _isAlive = false;        _cv.notify_all();    }I tried it, and it seems to work, but I am not sure about it - I thought that I might need to lock the _mutex again, like this:void kill()    {        unique_lock<mutex> mlock(_mutex);        _isAlive = false;        mlock.unlock();        _cv.notify_all();    }and at last, my first version looked like this:void kill()    {        unique_lock<mutex> mlock(_mutex);        _isAlive = false;        mlock.unlock();        _cv.notify_all();        // wait until all elements are consumed        while (!_queue.empty()); // busy loop - I know it's bad but it worked...        // I tried the next line, but it got to a deadlock:        //_cv.wait(mlock, [this] {return _queue.empty(); });    }So I'd like to know which kill is the best for my purposes (reminder - all the ints in the queue must be printed!). A sample main is also shown below.Queue source codeMainly based on this:#include <queue>#include <thread>#include <mutex>#include <condition_variable>#include <iostream>using namespace std;template <class T>class SafeQueue{    queue<T> _queue;    mutex _mutex;    condition_variable _cv;    bool _isAlive;public:    struct IsDead {}; // for handling end of tournament    SafeQueue() : _isAlive(true) {}    // block copy & move ctors and assignments    SafeQueue(const SafeQueue& other) = delete;    SafeQueue& operator=(const SafeQueue& other) = delete;    SafeQueue(SafeQueue&& other) noexcept = delete;    SafeQueue& operator=(SafeQueue&& other) noexcept = delete;    /* returns the first item of the queue and deletes it*/    T pop()    {        unique_lock<mutex> mlock(_mutex);        _cv.wait(mlock, [this] {return !(_queue.empty() && _isAlive); });        // if the queue is not empty, we still want to consume the next game (even if _isAlive = false)        if (!_queue.empty())        {            T item = _queue.front();            _queue.pop();            return item;        }        // in this case the queue is empty - we notify all waiting threads (and especially the one that waits on kill())        mlock.unlock();        _cv.notify_all();        throw IsDead();    }    void push(const T& item)    {        unique_lock<mutex> mlock(_mutex);        _queue.push(item);        mlock.unlock();        _cv.notify_one();    }    void push(T&& item)    {        unique_lock<mutex> mlock(_mutex);        _queue.push(std::move(item));        mlock.unlock();        _cv.notify_one();    }};Sample mainSafeQueue<int> q;mutex cout_mutex;void consumerMethod(int id){    while (true)    {        try        {            auto a = q.pop();            lock_guard<mutex> mlock(cout_mutex);            cout << # << id << :  << a << endl;        }        catch(SafeQueue<int>::IsDead&)        {            lock_guard<mutex> mlock(cout_mutex);            cout << # << id << : catched IsDead << endl;            return;        }    }}int main(){    vector<thread> consumers(5);    for (auto i = 0; i < 5; ++i)    {        consumers[i] = thread(consumerMethod, i + 1);    }    for (auto j = 0; j < 30; ++j)    {        q.push(j + 1);    }    q.kill();    for (auto& t : consumers)    {        t.join();    }    return 0;}
Notifying consumers when a producer is done
c++;multithreading;thread safety;queue;producer consumer
Let's start with pop. As many have noted, a pop that returns the value being removed from the collection can (will) cause problems unless copying (or moving) the value is guaranteed to be exception free.Unfortunately, the design used by the standard containers (use front() to retrieve the item, then pop to remove it) has an equally serious problem in the face of multiple threads (specifically, multiple consumers): you have to make a mutex visible to all the consumers of the queue, and they all have to cooperate correctly in its use for synchronization to work correctly. I'd much rather hide the synchronization inside the queue itself if at all reasonable--and in this case, it is entirely possible, with the right signature/design. Specifically, a signature I've found to work well is something like this:bool pop(T &dest);With this, pop can lock the mutex, copy front() to dest, and pop_front() to remove the item from the queue--and (just as we want) the item is removed if and only if the value made it to its destination correctly.In practical use, that does often get embellished with one addition: a timeout specified:bool pop(T &dest, Duration const &dur);This lets us specify that if the queue is empty when it's called, it'll wait up to some maximum period of time attempting to retrieve a value.Either way, if popping an item fails for any reason (including the queue being empty) it simply returns false to indicate failure.The next point is how to kill the queue. In reality, this isn't about killing the queue itself though--it's about killing all the queue's consumers. Rather than using a notify_all to tell them that the queue has been killed, I'd set a flag inside the queue itself.Then, rather than just passing ints through the queue, I'd pass some type of object that can inform the target that it's to cease processing jobs from the queue. Calling the queue's kill just sets a flag. Then we add a bit to the queue's pop to check that flag, and if it's set it just gives the client a cease processing task:class task {    std::atomic_bool cease = false;    int value;public:    task() : cease(true) {}    task(int value) : value(value) {}    bool done() const { return cease; }    int get_value() const { return value; }};template <class T>class Q {    std::deque<T> tasks;    std::atomic_bool killed;public:    Q() : killed(false) { }    void kill() { killed = true; }    bool pop(T &dest) {         std::unique_lock // ...        if (tasks.empty() && killed) {            dest = task(); // default task has `cease` set to `true`            return true;        }        else           // normal return    }    // ...};This way, the parent does something like this:Q q;for (int i=0; i<n; i++)    q.push(i);q.kill();The queue itself will hand out tasks to its consumers as long as it has tasks for them to execute (regardless of whether kill has been called).After kill has been called and it's run out of tasks to give to clients, then it'll return a default constructed task to tell the client thread that processing is done. In a typical case, the client will look something like this:task t;while (q.pop(t) && !t.done())    process(t);As soon as it receives a task with cease set, it'll know it's done processing tasks (at least from this queue) so it'll drop out of the loop and can proceed appropriately (e.g., exit the thread, start trying to get tasks from some other queue, etc.)Another possibility is to keep the queue a pure queue--just something that queues up tasks and dispenses them to consumers. In this case, it has no kill at all, it just has push, pop and such. It's then up to the producer to queue up an appropriate number of kill tasks when it's produced all the real tasks:queue q;for (int i=0; i<thread_count; i++)     threads[i] = std::thread(i, q);for (int i=0; i<n; i++)    q.push(task(i));for (int i=0; i<thread_count; i++)    q.push(task());In this case, consumer threads retrieve tasks to execute as long as there are any. Each then receives a suicide task. In response to that, it quits attempting to retrieve tasks to execute. This ensures that one suicide task will be distributed to each consumer thread.