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

Loki Astari · Accepted Answer · 2013-09-30 19:03:23Z

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.

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. — Mikhail
– Mikhail, Commented Sep 30, 2013 at 7:04
@Mikhail: You are correct. I was just trying to impart the pattern. But I have updated to make it more obvious. — Loki Astari
– Loki Astari, Commented Sep 30, 2013 at 19:03

Stack Exchange Network

Boost Threads - Producer Consumer threads with synchronization

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Boost Threads - Producer Consumer threads with synchronization

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions