2
\$\begingroup\$

Please review this for concurrency correctness:

#include <iostream>
#include <queue>

#include "boost\thread.hpp"
#include "boost\timer.hpp"

std::queue<int> itemQ;

boost::mutex m;
boost::condition_variable qFull, qEmpty;

const int max_size_q = 5;


void producer()
{
    int i = 0;
    while (1)
    {
        boost::this_thread::sleep(boost::posix_time::millisec(1000));
        boost::mutex::scoped_lock lock(m);      
        if (itemQ.size() <= max_size_q)
        {
            itemQ.push(++i);
            qEmpty.notify_one();
        }   
        else 
        {
            std::cout << "Q Full.notify_one Producer Waiting" << std::endl;
            qFull.wait(lock);
            std::cout << "Producer Notified to Continue" << std::endl;
        }
    }

}

void consumer()
{
    while (1)
    {
        boost::this_thread::sleep(boost::posix_time::millisec(4000));

        boost::mutex::scoped_lock lock(m);

        if (itemQ.size() == 0)
        {
            std::cout << "Q Empty. Consumer " << boost::this_thread::get_id() <<" Waiting" << std::endl;
            qEmpty.wait(lock);
            std::cout << "Consumer Notified to Continue" << std::endl;
        }
        else
        {
            std::cout << itemQ.front() << std::endl;
            itemQ.pop();
            qFull.notify_one();
        }

    }
}

int main()
{
    boost::thread producerthread(producer);

    boost::thread consumerthread1(consumer);
    boost::thread consumerthread2(consumer);
    boost::thread consumerthread3(consumer);
    boost::thread consumerthread4(consumer);
    boost::thread consumerthread5(consumer);

    consumerthread1.join();
    consumerthread2.join();
    consumerthread3.join();
    consumerthread4.join();
    consumerthread5.join();
}
\$\endgroup\$
0

1 Answer 1

1
\$\begingroup\$

Your usage of the condition variable will probably work but is a-typical.

This is what you basically have:

while (1)
{
    boost::mutex::scoped_lock lock(m);      
    if (<Test OK>)
    {
        <Do Work>
        <Notify Consumer>
    }   
    else 
    {
        <Wait for consumer to signal conditional>
    }
}

Notice that here you lock the mutex m. If the test is not OK then you wait for the consumer to signal the condition variable. This releases the lock on m. Once the condition variable is signaled then your thread must wait to re-aquire the lock to continue. Once it does it releases the lock and then re-starts the loop which immediately try to re-aquire the lock.

A more typical pattern would be:

// pseudo code.
std::unique_ptr<WORK> getWork()
{
    boost::mutex::scoped_lock lock(m);      
    while(! <Test OK> )
    {
       <Wait for consumer to signal conditional>
    }
    return <getWorkObjectFromQueue>; 
}
.....
while(1)
{
    work = getWork();

    <Do Work>
    <Notify Consumer>
}

This way when you are waiting on the conditional variable and are signaled you do not have multiple attempts to re-acquire the lock before you do the work. As soon as you are signaled and have acquired the lock you can do a bit of work.

\$\endgroup\$
2
  • \$\begingroup\$ In your case the scopes make the two operations serial, which is often not desired. For example I might want to queue up 1000 work units. \$\endgroup\$
    – Mikhail
    Sep 30, 2013 at 7:04
  • \$\begingroup\$ @Mikhail: You are correct. I was just trying to impart the pattern. But I have updated to make it more obvious. \$\endgroup\$ Sep 30, 2013 at 19:03

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

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