//
// A simple implementation of the well-known Producer/Consumer pattern
// demonstrating the C++0x thread library and certain other new features.
//
// Written by Johannes Dahlström
// This file is in the public domain.
//

#include <thread>
#include <condition_variable>
#include <queue>
#include <chrono>
#include <iostream>
#include <cstdlib>
#include <functional>

using namespace std;


// Represents a pool of data with thread-safe producer/consumer semantics
class work_pool {
 public:
  typedef unique_lock<std::mutex> lock_type;

  work_pool(int max_size) : max_size(max_size) { 
  }

  // Acquires a lock on the pool and returns it
  // A thread MUST hold a lock on the pool 
  // before calling ANY other member functions. 
  lock_type lock() {
    return lock_type(mutex);
  }

  // Pushes a datum to the work pool
  void produce(int i) {
    queue.push(i);
  }

  // Pops a datum from the work pool
  int consume() {
    int res = queue.front();
    queue.pop();
    return res;
  }

  // Returns whether the pool is full
  bool is_full() {
    return queue.size() >= max_size;
  }

  // Returns whether the pool has data
  bool has_data() {
    return !queue.empty();
  }

  // Returns the number of items in the pool
  int size() {
    return queue.size();
  }

  // Atomically releases lock, blocks until there
  // is new data in the pool, then reacquires the lock
  // lock MUST hold the same mutex as the one returned by lock()
  void wait(lock_type& lock) {
    // Wait until being notified AND there is data in the pool
    cvar.wait(lock, bind(&work_pool::has_data, this));
  }

  // Notifies the consumer threads waiting on this pool
  void notify() {
    cvar.notify_all();
  }

 private:
  int max_size;
  std::queue<int> queue;
  std::condition_variable cvar;
  std::mutex mutex;
};


// Produces data into the given work pool.
void produce(int id, work_pool& pool) {
  static int i = 0;
  while(true) {
    // Simulate some heavy work being done
    this_thread::sleep_for(chrono::milliseconds(rand() % 3000));

    // Acquire lock on the work pool
    work_pool::lock_type lock(pool.lock());

    // Push data to pool if not full
    if(!pool.is_full()) {
      pool.produce(i++);
      cerr << " producer " << id << ": " << i << " size=" << pool.size() << '\n';
    }

    // Notify consumers
    pool.notify();
  }
}

// Consumes data from the given work pool.
void consume(int id, work_pool& pool) {

  // Acquire lock on the work pool
  work_pool::lock_type lock(pool.lock());
  
  while(1) {

    // Release lock, wait for new data
    pool.wait(lock);
    // wait reacquires the lock before returning

    // Consume data from pool
    int i = pool.consume();
    cerr << " consumer " << id << ": " << i << " size=" << pool.size() << "\n\n";
    
    // rand is not thread-safe so call it before releasing the lock
    int delay = rand() % 3000;
    
    // Release lock on pool, simulate work being done
    lock.unlock();
    this_thread::sleep_for(chrono::milliseconds(delay));
    lock.lock();
    
  }
}


int main() {

  srand(time(0));

  work_pool pool(10);

  // Spawn producer and consumer threads
  vector<thread> producers, consumers;
  for(int i = 0; i < 4; ++i) {
    // emplace_back constructs the object in-place using perfect forwarding
    // equivalent to push_back(thread(...)) but avoids a potential copy/move
    consumers.emplace_back(consume, i, ref(pool));
    producers.emplace_back(produce, i, ref(pool));
  }

  while(true);
}