Using condition variable in a producer-consumer situation

25,387

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();
    }
}
Share:
25,387
jasonline
Author by

jasonline

C++ Developer

Updated on March 05, 2020

Comments

  • jasonline
    jasonline about 4 years

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