Producer - Consumer interaction

Question

I have an implementation of Producer-Consumer multi-threaded interaction. It works, but I feel that during the execution wait states happen too often between Consumer thread and Producer. In my case, Consumer accesses a queue at random intervals and takes data from it. Now, the producer thread runs during the whole process life time. The producer thread works as caching machine. It checks in a loop if the size of queue is smaller than the maximum cache size allowed, and if that's the case, it keeps pushing new data into that cache.

My concern is that when the consumer tries to access the queue, the last is still locked by the producer thread and the consumer should wait. I would like to get a feedback if what I am doing is optimal in terms of interaction flow between producer and consumer.

   //Called by consumer(main thead) to retrieve data from cache
 uint8_t* Worker::GetFrame() {

  boost::unique_lock<boost::mutex> lk(_frameCacheMutex);
  //If the cache is empty:
  while (0 == _frames_cache.size()) {
           //tell Producer to push data into cache
         _fill_cache_wait_cond.notify_one();
          //wait for the data to arrive(will be signaled by worker thread)
         _get_frame_wait_cond.wait(lk);

  }
  uint8_t * fr = _frames_cache.front();
  _frames_cache.pop();
  // notify worker thread to continue caching
  _fill_cache_wait_cond.notify_one();

  return fr;
}

Producer thread:

  void Worker::operator () () {
   //Some init here..
    while (isRunning) {
       boost::unique_lock<boost::mutex> lk(_frameCacheMutex);

       /// Here create the data for cache...

       _frames_cache.push(data);

       /// Notify waiting main thread to take data
       _get_frame_wait_cond.notify_one();

       /// If the cache is full ,wait
       while (_frames_cache.size() == cacheMaxSize ){

             _fill_cache_wait_cond.wait(lk);

       }

    }     

}

Answer 1

Definitely not optimal as you are holding locks longer than you need too.

   boost::unique_lock<boost::mutex> lk(_frameCacheMutex);

   /// Here create the data for cache...

   // You are creating data here.
   // But you should not be interacting with the cache here.
   // So why are you holding a lock?

   // You should only hold a lock while you are trying to update a shared resource.


   _frames_cache.push(data);

So I would expect it to look like this:

   /// Here create the data for cache...

   boost::unique_lock<boost::mutex> lk(_frameCacheMutex);
   _frames_cache.push(data);

As a side note.
Using _ as a prefix on identifiers in C++ user applications is frowned upon (as in _frames_cache). Do you know the exact rules on when identifiers are reserved? You did not break them; but I would still advise against it because you don't know the exact rules.

This looks wrong:

while (0 == _frames_cache.size()) {
       //tell Producer to push data into cache
     _fill_cache_wait_cond.notify_one();
      //wait for the data to arrive(will be signaled by worker thread)
     _get_frame_wait_cond.wait(lk);

}

It is in a while loop because the thread was woken early but another thread has grabbed the work. So I would not expect you notify_one at this point. Also You are notifying the producer when the cache is empty. I would rather notify the producer as soon as there is any space available. So each time you grab something notify the producer (because if you take something then there is space available.

  while (0 == _frames_cache.size()) {
      //wait for the data to arrive(will be signaled by worker thread)
     _get_frame_wait_cond.wait(lk);
  }
  //tell Producer to push data into cache
  // We are about to take one item so there will be space available,
  _fill_cache_wait_cond.notify_one();

Your usage of wait() is a bit cumbersome. With the new lambda expressions you can express the wait and loop in a single expression.

  _get_frame_wait_cond.wait(lk, [this](){return 0 != _frames_cache.size();});

You are using Yoda Conditionals.

  while( 0 == _frames_cache.size() ) {}

This is a bit dated. And does not provide any real protection but also imposes a mental tax on the reader of your code as it is not the natural way to express the condition.

You get much better protection by making the compiler detect the problems you are looking for.

 // You are trying to solve this problem
 while( _frames_cache.size() = 0 ) {}
                       //   ^^^  accidental assignment.

Turn you compiler warning level up (it should already be at the highest level). And treat all warnings as errors (as they are logical errors in your thinking of how the code works). Do that and this is no loger an issue as it is a compiler error detected for you and the reader of your code gets readable code.

Your indentation and blank lines in your code make it look real messy. Tidy it up.

Some (most) of your comments are usless and parat what the code says.

  // notify worker thread to continue caching
  _fill_cache_wait_cond.notify_one();

When you write a comment it should be a description of WHY or an explanation of some complex algorithm that you are trying to implement. Writing a comment like the above one is actually worse than not writing a comment because now you have to maintain the comment to be the same as the code (and I can already read the code and see what it is doing I don't need the comment to say the same thing).

You also seem to be using two different naming conventions for your identifiers:

 _frames_cache         http://en.wikipedia.org/wiki/Snake_case
 _frameCacheMutex      http://en.wikipedia.org/wiki/CamelCase

Personally I prefer CamelCase but am not that ground in. Follow the standards of the place you work (but pick one).

I use CamelCase. But a leading uppercase character indicates a User Type while a leading lower case character indicates a variable/method/function name (something I can take the address of). I don't use special prefixes to mark members (I just name the members very well so there usage is obvious).

Cleaned up code:

Consumer

// Consumer: Get one item from the cache.
//           If the cache is empty then wait until
//           data is available.
uint8_t* Worker::GetFrame() {

    std::unique_lock<std::mutex> lk(frameCacheMutex);
    getFrameWaitCond.wait(lk, [this](){return framesCache.size() != 0;});

    uint8_t* fr = framesCache.front();
    framesCache.pop();

    fillCacheWaitCond.notify_one();
    return fr;
}

Producer

// Producer: run in infinite loop
//           Generate the data.
//           Get the lock and wait for space to be available before
//           continuing. This way you pre generate the data and
//           can push it as soon as there is space.
void Worker::operator () () {

    while (isRunning) {

       /// Here create the data for cache...

       std::unique_lock<std::mutex> lk(frameCacheMutex);    
       fillCacheWaitCond.wait(lk, [this]{return framesCache.size() != cacheMaxSize;});

       framesCache.push(data);

       getFrameWaitCond.notify_one();
    }
}