1
\$\begingroup\$

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 code

Mainly 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 main

SafeQueue<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;
}
\$\endgroup\$
1
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ thanks! I liked the last idea, I think I will use that "poisoning". \$\endgroup\$ – noamgot May 30 '17 at 6:10
  • \$\begingroup\$ there's still a small problem though - you need to be able to "poison" your task, regardless of their type. if the task is some struct then it's possible to add a boolean flag, but if it's a string, or int etc. you should think of a proper way to "poison"... \$\endgroup\$ – noamgot May 30 '17 at 6:31
1
\$\begingroup\$

The thing to remember is that C++ (as of C++11) has an "official" memory model, which defines what operations are legal and illegal according to the official spec. In particular, a program which contains a data race is not legal according to the official memory model. A data race is any occasion on which two different threads access the same data, at the same time, at least one of which is doing a write, and where there's no special-case synchronization primitive involved (for example, concurrent accesses to a std::atomic or a std::mutex are fine).


void kill()
{
    _isAlive = false;
    _cv.notify_all();
}

This function will likely be fine in practice... until it matters! The problem is that you're modifying bool _isAlive in thread A while, simultaneously, you might be reading _isAlive from thread B (within pop). This is a data race and invalidates your program.


void kill()
{
    unique_lock<mutex> mlock(_mutex);
    _isAlive = false;
    mlock.unlock();
    _cv.notify_all();
}

This is better. You're no longer modifying _isAlive while another thread might be reading it, because both kill and pop touch _isAlive only under the mutex lock. So this code is data-race-free.

However, your next modification breaks it again:

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(); });
}

Here you're reading _queue not-under-a-mutex-lock, while simultaneously pop might be writing to _queue (that is, it might be popping from it). So you have a data race and your program is invalid.

Also, your "busy loop"

while (!_queue.empty());

is highly likely to be optimized into an infinite loop by any modern compiler, since the compiler can see that _queue.empty() is a loop invariant — there is no code in the body of the loop that could conceivably change the status of the queue, so there's no point in testing the condition every time through the loop. If you want the compiler to know that there's some other thread participating in this loop, you'll have to explicitly tell the compiler that, via e.g. waiting on a condition variable.


How many producers and how many consumers are you planning to have, by the way? If it's just 1 and 1 respectively (a Single-Producer-Single-Consumer or SPSC queue), then Loki's suggestion to omit some locks might be reasonable — although I still wouldn't recommend it.


The simplest way to "wait for" this one-time condition is to use a one-shot std::future<void>:

template <class T>
class SafeQueue
{
    queue<T> _queue;
    mutex _mutex;
    condition_variable _cv;
    bool _isAlive;
    std::promise<void> _pAllDone; // NEW
    std::future<void> _fAllDone;  // NEW

    T pop() {
        unique_lock<mutex> mlock(_mutex);
        while (_queue.empty() && _isAlive) {
            _cv.wait(mlock);
        }
        if (!_queue.empty()) {
            T item = std::move(_queue.front());
            _queue.pop();
            return item;
        }
        // in this case the queue is empty
        mlock.unlock();
        try { _pAllDone.set_value(); } catch (...) {} // NEW
        _cv.notify_all();
        throw IsDead();
    }
    void kill() {
        unique_lock<mutex> mlock(_mutex);
        _isAlive = false;
        mlock.unlock();
        _cv.notify_all();
        _fAllDone.wait();  // NEW
    }
};

The try-catch is necessary if there might be more than one "consumer" thread trying to set the value of the promise. Also notice that we're assuming that there is at least one "consumer" thread; if there's only the "killer" left, then it'll wait forever since there's no "consumer" left to finish emptying the queue.


Really, the logic for cleaning up the queue should probably be external to the queue class itself. The person who knows how best to deal with the lifetime and ownership of the queue itself is, by definition, outside of the queue class.

For example, if I were killing a queue, I wouldn't want all of its consumers to "finish up" emptying the queue; I'd actually want the consumers to stop ASAP, and just discard whatever tasks were left in the queue. I might do that by having the "killer" push a "killer task" onto the queue, and then making sure that any consumer who dequeued that task (A) decremented some external "count of consumers still living", (B) requeued the killer task if that count were >=1, and then (C) killed itself.

But if you only have a single consumer thread in your program, then the above idea might be overkill. It really depends on your use-case... which is why this logic shouldn't be part of the reusable queue class itself, if you can help it.

\$\endgroup\$
  • \$\begingroup\$ thanks! as you can see in the sample main I published, for now I have one producer and many consumers. about kill - maybe it's a bad name choice, but I really need the consumers to finish what's in the queue. the real program will be a multi-thread tournament of some game, where the elements of the queue would be "game details" - each thread will pop out a game and run it. then the producer produces all the "game detail" objects, and it wants the consumers to play those games until all games were played (that's why I had to add the kill method) \$\endgroup\$ – noamgot May 30 '17 at 6:00
0
\$\begingroup\$

There is no need to manually unlock.

void push(const T& item)
{
    unique_lock<mutex> mlock(_mutex);
    _queue.push(item);
    mlock.unlock();             // This is not needed.
    _cv.notify_one();
}

Your pop is fine (apart from the unlock() as before).

There are reason that most c++ queues separate front/pop

Why doesn't std::queue::pop return value.?

So we may need to re-think pop() and not use the traditional of pop. Why not pass pop() a function that is called for each value.

// Don't return a value 
// Pass a function to be called from each value.
template<typename F>
void pop(F f)
{
    {
        // Make sure there is a value in the queue.
        unique_lock<mutex> mlock(_mutex);
        _cv.wait(mlock, [this] {return !(_queue.empty() && _isAlive); });
    }

    if (!_queue.empty())
    {
        // We can read the value without a lock?
        f(_queue.front());

        // Just lock for modification.
        unique_lock<mutex> mlock(_mutex);
        _queue.pop();
        return;
    }

    // Bad things happen.
    _cv.notify_all();
    throw IsDead();
}
\$\endgroup\$
  • \$\begingroup\$ thanks for the answer. as for push - I followed this: stackoverflow.com/questions/17101922/…. as for pop - well, I prefer to keep it that way (it should work fine with my real program). however, I think that I must lock to read the queue.front() - there might be a case where I want to read this value and another thread changes it, right? If you could say something about the kill function I'd appreciate it very much, it's the main reason why I asked this question.. thanks \$\endgroup\$ – noamgot May 29 '17 at 21:13

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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