include/PVBuffer.hpp

   1 #ifndef SCX_PVBUFFER_HPP
   2 #define SCX_PVBUFFER_HPP
   3
   4 #include <deque>
   5 #include "Mutex.hpp"
   6 #include "SemVar.hpp"
   7
   8 namespace scx {
   9
  10 template <typename item_t>
  11 class PVBuffer
  12 {
  13 public:
  14     PVBuffer():
  15         m_pFreeListSemVar(new SemVar(0, 0)),
  16         m_pDataListSemVar(new SemVar(0, 0))
  17     {
  18     }
  19
  20     ~PVBuffer()
  21     {
  22         if (m_pFreeListSemVar != NULL)
  23             delete m_pFreeListSemVar;
  24
  25         if (m_pDataListSemVar != NULL)
  26             delete m_pDataListSemVar;
  27     }
  28
  29     void AllocBuffer(size_t bufCount)
  30     {
  31         m_BufferQueue.resize(bufCount);
  32
  33         m_FreeQueue.resize(bufCount);
  34         for (size_t i = 0; i < bufCount; ++i) {
  35             m_BufferQueue[i] = new item_t;
  36             m_FreeQueue[i] = m_BufferQueue[i];
  37             m_pFreeListSemVar->Post();
  38         }
  39
  40         while (m_pDataListSemVar->TryWait() != -1);
  41     }
  42
  43     void ClearBuffer()
  44     {
  45         for (size_t i = 0; i < m_BufferQueue.size(); ++i) {
  46             delete m_BufferQueue[i];
  47         }
  48         m_BufferQueue.clear();
  49
  50         m_FreeQueue.clear();
  51         m_DataQueue.clear();
  52
  53         while (m_pFreeListSemVar->TryWait() != -1);
  54         while (m_pDataListSemVar->TryWait() != -1);
  55     }
  56
  57     /**
  58      * This method should be called after both customer and producer
  59      * has been suspended by TakeFree()/TakeData().
  60      * After calling this method,
  61      * the producer will begin to work again, then the customer.
  62      */
  63     void ResetPV()
  64     {
  65         const size_t bufCount = m_BufferQueue.size();
  66
  67         // No mutex here,
  68         // because we assume both thread has been suspended.
  69         m_DataQueue.clear();
  70         while (m_pDataListSemVar->TryWait() != -1);
  71
  72         // Lock the FreeQueue first,
  73         // because after SemVar::Post() the producer will begin to work,
  74         // and it will take the first item in FreeQueue.
  75         m_FreeListMutex.Lock();
  76         m_FreeQueue.resize(bufCount);
  77         while (m_pFreeListSemVar->TryWait() != -1);
  78         for (size_t i = 0; i < bufCount; ++i) {
  79             m_FreeQueue[i] = m_BufferQueue[i];
  80             m_pFreeListSemVar->Post();
  81         }
  82         m_FreeListMutex.Unlock();
  83     }
  84
  85     size_t GetBufferCount() const
  86     {
  87         return m_BufferQueue.size();
  88     }
  89
  90     size_t GetFreeCount() const
  91     {
  92         return m_pFreeListSemVar->GetValue();
  93     }
  94
  95     size_t GetDataCount() const
  96     {
  97         return m_pDataListSemVar->GetValue();
  98     }
  99
 100     /**
 101      * This method can be used for initialize buffer.
 102      */
 103     item_t* GetRawItem(size_t i)
 104     {
 105         return m_BufferQueue[i];
 106     }
 107
 108     item_t* TakeFree()
 109     {
 110         m_pFreeListSemVar->Wait();
 111         m_FreeListMutex.Lock();
 112         item_t* pItem = m_FreeQueue.front();
 113         m_FreeQueue.pop_front();
 114         m_FreeListMutex.Unlock();
 115         return pItem;
 116     }
 117
 118     void RecycleFree(item_t* pItem)
 119     {
 120         m_DataListMutex.Lock();
 121         m_DataQueue.push_back(pItem);
 122         m_DataListMutex.Unlock();
 123         m_pDataListSemVar->Post();
 124     }
 125
 126     item_t* TakeData()
 127     {
 128         m_pDataListSemVar->Wait();
 129         m_DataListMutex.Lock();
 130         item_t* pItem = m_DataQueue.front();
 131         m_DataQueue.pop_front();
 132         m_DataListMutex.Unlock();
 133         return pItem;
 134     }
 135
 136     void RecycleData(item_t* pItem)
 137     {
 138         m_FreeListMutex.Lock();
 139         m_FreeQueue.push_back(pItem);
 140         m_FreeListMutex.Unlock();
 141         m_pFreeListSemVar->Post();
 142     }
 143
 144     /**
 145      * eg.
 146      * 1.Call ClearFree() in customer thread, then call ClearData() for customer itself.
 147      * Or Call ClearData() in producer thread, then call ClearFree() for producer it self.
 148      * 2.Finally, after prepared, we can call ResetPV() to wake the previous PV-loop again.
 149      */
 150
 151     /**
 152      * Clear the empty-buffer, so that the producer can be blocked.
 153      */
 154     void ClearFree()
 155     {
 156         while (m_pFreeListSemVar->TryWait() != -1);
 157         m_FreeListMutex.Lock();
 158         m_FreeQueue.clear();
 159         m_FreeListMutex.Unlock();
 160     }
 161
 162     /**
 163      * Clear the filled-data, so that the customer can be blocked.
 164      */
 165     void ClearData()
 166     {
 167         while (m_pDataListSemVar->TryWait() != -1);
 168         m_DataListMutex.Lock();
 169         m_DataQueue.clear();
 170         m_DataListMutex.Unlock();
 171     }
 172
 173 private:
 174     std::deque<item_t*> m_BufferQueue;
 175
 176     std::deque<item_t*> m_FreeQueue;
 177     Mutex m_FreeListMutex;
 178     SemVar* m_pFreeListSemVar;
 179
 180     std::deque<item_t*> m_DataQueue;
 181     Mutex m_DataListMutex;
 182     SemVar* m_pDataListSemVar;
 183 };
 184
 185 }
 186
 187 #endif