/*
生產者/消費者通用模板
特點：
高效能：採用多執行緒，多佇列平衡的訊號量等待模型，有效減少鎖等待
可調節：可以根據實際應用環境調整佇列數，最多可支援64個佇列
使用簡單，一個建構函式，一個生產函式，一個消費函式。
*/
#ifndef PANDC_H
#define PANDC_H

#include <vector>
#include <deque>
#include <Windows.h>
#include <limits.h>


using namespace std;

enum QueueType
{
  qtFIFO,
  qtFILO
};
//T生產的物件
template<typename T>
class Pandc
{
public:
  //建構函式, QUEUE_COUNT是佇列數 QT:先進先出還是先進後出
  Pandc(unsigned long QUEUE_COUNT,QueueType QT);
  ~Pandc();
  void P(T obj); //生產
  bool C(/*out*/T &ret); //消費
  long ItemCount(){return m_item_count;}
  long QueueItemCount(unsigned long queue_id);
private:
  Pandc(const Pandc&);
  Pandc& operator=(const Pandc&);
private:
  volatile LONG m_queue_id; //當前佇列ID;
  vector<deque<T>* > m_queues; //佇列
  vector<RTL_CRITICAL_SECTION*> m_queuelocks;   //佇列鎖
  HANDLE *m_semaphores; //佇列訊號量
  unsigned long m_queue_count; //佇列數
  QueueType m_queue_type; //進出佇列方式 FIFO/FILO
  volatile LONG m_item_count; //佇列中的總條目數
};

template<typename T>
long Pandc<T>::QueueItemCount( unsigned long queue_id )
{
  if (queue_id >= m_queue_count)
    return 0;
  return (m_queues[queue_id])->size();
}

template<typename T>
bool Pandc<T>::C(/*out*/T &ret)
{
  DWORD wait_ret = WaitForMultipleObjects(m_queue_count,m_semaphores,false,INFINITE);
  if (WAIT_FAILED == wait_ret)
    return false;

  size_t i = wait_ret - WAIT_OBJECT_0;
  EnterCriticalSection(m_queuelocks[i]);
  if (qtFIFO == m_queue_type)
  {  
    ret = m_queues[i]->front();
    m_queues[i]->pop_front();
  }
  else
  {
    ret = m_queues[i]->back();
    m_queues[i]->pop_back();
  }
  LeaveCriticalSection(m_queuelocks[i]);
  InterlockedDecrement(&m_item_count);
  return true;
}

template<typename T>
void Pandc<T>::P( T obj )
{  
  if (InterlockedIncrement(&m_queue_id) > 1024 * 1024)
    InterlockedExchange(&m_queue_id,0);
  size_t i = (m_queue_id % m_queue_count);  
  EnterCriticalSection(m_queuelocks[i]);  
  m_queues[i]->push_back(obj);
  ReleaseSemaphore(m_semaphores[i],1,NULL);
  LeaveCriticalSection(m_queuelocks[i]);
  InterlockedIncrement(&m_item_count);
}

template<typename T>
Pandc<T>::~Pandc()
{
  for(vector<deque<T> * >::iterator it = m_queues.begin(); it!=m_queues.end();++it)
  { 
    delete (*it);    
  }
  for(vector<RTL_CRITICAL_SECTION*>::iterator it = m_queuelocks.begin(); it!

=m_queuelocks.end();++it)
  {
    DeleteCriticalSection(*it);
    delete (*it);
  }
  for(size_t i = 0; i<m_queue_count; ++i)
  {
    CloseHandle(m_semaphores[i]);
  }  
}

template<typename T>
Pandc<T>::Pandc(unsigned long QUEUE_COUNT,QueueType QT)
{
  m_queue_id = 0;
  m_queue_type = QT;
  m_queue_count = QUEUE_COUNT;
  m_item_count = 0;
  m_semaphores = (HANDLE*)malloc(sizeof(HANDLE)*QUEUE_COUNT);
  memset(m_semaphores,0,sizeof(HANDLE)*QUEUE_COUNT);
  for(size_t i = 0; i< m_queue_count; ++i)
  {
    deque<T> *q = new deque<T>;
    m_queues.push_back(q);
    RTL_CRITICAL_SECTION *lock = new RTL_CRITICAL_SECTION;
    InitializeCriticalSection(lock);
    m_queuelocks.push_back(lock);
    HANDLE sp = CreateSemaphoreA(NULL,0,LONG_MAX,"");
    m_semaphores[i] = sp;
  }
}

#endif