Fast producer and slow consumer

concurrency, multithreading, producer-consumer, pthreads

Solution

Code snippet of "bounded queue" with pthread :

#include <queue>
#include <pthread.h>

template <class T, size_t UpperLimit>
class BoundedQueue {
  std::queue<T> q_;
  pthread_mutex_t mtx_;
  pthread_cond_t cv_not_empry_;
  pthread_cond_t cv_not_full_;

  // lock/unlock helper
  struct auto_locker {
    auto_locker(pthread_mutex_t* pm) : pm_(pm)
      { pthread_mutex_lock(pm_); }
    ~auto_locker()
      { pthread_mutex_unlock(pm_);}
    pthread_mutex_t *pm_;
  };

public:
  BoundedQueue() { /* initialize member... */ }
  ~BoundedQueue() { /* uninitialize member...*/ }
  // for Producer
  void push(T x) {
    auto_locker lk(&mtx_);
    while (UpperLimit <= q_.size()) {
      pthread_cond_wait(&cv_not_full_, &mtx_);
    }
    q_.push(x);
    pthread_cond_broadcast(&cv_not_empry_);
    return ret;
  }
  // for Consumer
  T pop() {
    auto_locker lk(&mtx_);
    while (q_.empty()) {
      pthread_cond_wait(&cv_not_empry_, &mtx_);
    }
    T ret = q_.front();
    q_.pop();
    pthread_cond_broadcast(&cv_not_full_);
    return ret;
  }
}

Problem

I have a use case where a fast producer inserts data in a queue and a slow consumer consumes data. The problem I'm facing is continuous increase in queue size over time. I have a class implementation where a std::queue is protected by mutex and condition variable for concurrent reads and writes. How can this adapted to case where the producer after reaching a MAX_THRESHOLD till stop inserting data into the queue and the consumer has consumed some amount of data signals the producer to insert data into the queue. Could someone provide a sample implementation ? Also with out changing the class implementation is it possible to solve this problem by adding another layer of syncronization in the producer and consumer ?

Original source