Boost Threads - Producer Consumer threads with synchronization

Question

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

Answer 1

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.

• Mikhail: 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.
• Loki Astari: @Mikhail: You are correct. I was just trying to impart the pattern. But I have updated to make it more obvious.