Using condition variable in a producer-consumer situation

Question

I'm trying to learn about condition variables and how to use it in a producer-consumer situation. I have a queue where one thread pushes numbers into the queue while another thread popping numbers from the queue. I want to use the condition variable to signal the consuming thread when there is some data placed by the producing thread. The problem is there are times (or most times) that it only pushes up to two items into the queue then hangs. I have indicated in the produce() function where it stops when running in debug mode. Can anyone help me point out why this is happening?

I have the following global variables:

boost::mutex mutexQ;               // mutex protecting the queue
boost::mutex mutexCond;            // mutex for the condition variable
boost::condition_variable condQ;

Below is my consumer thread:

void consume()
{
    while( !bStop )   // globally declared, stops when ESC key is pressed
    {
        boost::unique_lock lock( mutexCond );
        while( !bDataReady )
        {
            condQ.wait( lock );
        }

        // Process data
        if( !messageQ.empty() )
        {
            boost::mutex::scoped_lock lock( mutexQ );

            string s = messageQ.front();   
            messageQ.pop();
        }
    }
}

Below is my producer thread:

void produce()
{
    int i = 0;

    while(( !bStop ) && ( i < MESSAGE ))    // MESSAGE currently set to 10
    {
        stringstream out;
        out << i;
        string s = out.str();

        boost::mutex::scoped_lock lock( mutexQ );
        messageQ.push( s );

        i++;
        {
            boost::lock_guard lock( mutexCond );  // HANGS here
            bDataReady = true;
        }
        condQ.notify_one();
    }
}

Ramiroramjet

FWIW, this bug is a perfect example of "lock inversion". In one thread, you lock mutexQ first, and then without releasing it you lock mutexCond. In another thread, you lock mutexCond first and then mutexQ. This is almost always wrong. One thread gets one lock, and the other gets the other, simultaneously. Then they'll both wait to get the lock they haven't got. Which they can't get, ever. The simple solutions are either to use fewer locks (which is correct in this case anyway, even if not for the inversion), or to define a "hierarchy" of locks, so you can always take them in consistent order. – Poteat

Answer 1

You have to use the same mutex to guard the queue as you use in the condition variable.

This should be all you need:

void consume()
{
    while( !bStop )
    {
        boost::scoped_lock lock( mutexQ);
        // Process data
        while( messageQ.empty() ) // while - to guard agains spurious wakeups
        {
            condQ.wait( lock );

        }
        string s = messageQ.front();            
        messageQ.pop();
    }
}

void produce()
{
    int i = 0;

    while(( !bStop ) && ( i < MESSAGE ))
    {
        stringstream out;
        out << i;
        string s = out.str();

        boost::mutex::scoped_lock lock( mutexQ );
        messageQ.push( s );
        i++;
        condQ.notify_one();
    }
}

Misdemeanor

@nos: Thanks, your solution worked. This is the most common application/use for condition variables, right? – Ramiroramjet

"// while - to guard agains spurious wakeups" Could you please explain why a while loop is needed. Does that mean that the wait() can return without being signaled by the producer? – Backwardation

@Backwardation In some cases , yes - though probably not in this small example. They're somewhat explained here pubs.opengroup.org/onlinepubs/7908799/xsh/… , that's pthreads and not boost - but boost on *nixes uses pthreads and can't really take any measures against it. With the while() instead of an if() , it's easier to change the code to having multiple consumers, in which case you'll end up doing the right thing if another consumer grabbed the value before you did. (If you have multiple consumers, condQ.notify_one() can wake up more than one of them) – Misdemeanor

@nos: I know this is an old topic but I have a question. I assume the idea is that you can stop the process by setting bStop = true, but what if consume() is waiting at this point because the queue is empty? The producer stops producing, so the consumer doesn't get any more data and just waits. Therefore it doesn't get to check the bStop variable. Is my question clear? How would you resolve this? – Depalma

@PolyVox That's right, so you need to send a message to wake it up as well as setting bStop. It might even be a special message that the consumer recognizes as a "stop" message. – Misdemeanor

@nos: Ok, that does make sense. Thanks for getting back to me on such an old post :-) – Depalma