rpc/fifo.h

   1 #ifndef fifo_h
   2 #define fifo_h
   3
   4 // fifo template
   5 // blocks enq() and deq() when queue is FULL or EMPTY
   6
   7 #include <errno.h>
   8 #include <list>
   9 #include <sys/time.h>
  10 #include <time.h>
  11 #include <errno.h>
  12 #include "slock.h"
  13 #include "lang/verify.h"
  14
  15 template<class T>
  16 class fifo {
  17         public:
  18                 fifo(int m=0);
  19                 ~fifo();
  20                 bool enq(T, bool blocking=true);
  21                 void deq(T *);
  22                 bool size();
  23
  24         private:
  25                 std::list<T> q_;
  26                 pthread_mutex_t m_;
  27                 pthread_cond_t non_empty_c_; // q went non-empty
  28                 pthread_cond_t has_space_c_; // q is not longer overfull
  29                 unsigned int max_; //maximum capacity of the queue, block enq threads if exceeds this limit
  30 };
  31
  32 template<class T>
  33 fifo<T>::fifo(int limit) : max_(limit)
  34 {
  35         VERIFY(pthread_mutex_init(&m_, 0) == 0);
  36         VERIFY(pthread_cond_init(&non_empty_c_, 0) == 0);
  37         VERIFY(pthread_cond_init(&has_space_c_, 0) == 0);
  38 }
  39
  40 template<class T>
  41 fifo<T>::~fifo()
  42 {
  43         //fifo is to be deleted only when no threads are using it!
  44         VERIFY(pthread_mutex_destroy(&m_)==0);
  45         VERIFY(pthread_cond_destroy(&non_empty_c_) == 0);
  46         VERIFY(pthread_cond_destroy(&has_space_c_) == 0);
  47 }
  48
  49 template<class T> bool
  50 fifo<T>::size()
  51 {
  52         ScopedLock ml(&m_);
  53         return q_.size();
  54 }
  55
  56 template<class T> bool
  57 fifo<T>::enq(T e, bool blocking)
  58 {
  59         ScopedLock ml(&m_);
  60         while (1) {
  61                 if (!max_ || q_.size() < max_) {
  62                         q_.push_back(e);
  63                         break;
  64                 }
  65                 if (blocking)
  66                         VERIFY(pthread_cond_wait(&has_space_c_, &m_) == 0);
  67                 else
  68                         return false;
  69         }
  70         VERIFY(pthread_cond_signal(&non_empty_c_) == 0);
  71         return true;
  72 }
  73
  74 template<class T> void
  75 fifo<T>::deq(T *e)
  76 {
  77         ScopedLock ml(&m_);
  78
  79         while(1) {
  80                 if(q_.empty()){
  81                         VERIFY (pthread_cond_wait(&non_empty_c_, &m_) == 0);
  82                 } else {
  83                         *e = q_.front();
  84                         q_.pop_front();
  85                         if (max_ && q_.size() < max_) {
  86                                 VERIFY(pthread_cond_signal(&has_space_c_)==0);
  87                         }
  88                         break;
  89                 }
  90         }
  91         return;
  92 }
  93
  94 #endif