simgear/threads/SGQueue.hxx

   1 #ifndef SGQUEUE_HXX_INCLUDED
   2 #define SGQUEUE_HXX_INCLUDED 1
   3
   4 #include <simgear/compiler.h>
   5
   6 #if defined ( SG_HAVE_STD_INCLUDES )
   7 #  include <cassert>
   8 #else
   9 #  include <assert.h>
  10 #endif
  11
  12 #include <queue>
  13 #include "SGThread.hxx"
  14 #include "SGGuard.hxx"
  15
  16 /**
  17  * SGQueue defines an interface for a FIFO.
  18  * It can be implemented using different types of synchronization
  19  * and protection.
  20  */
  21 template<class T>
  22 class SGQueue
  23 {
  24 public:
  25
  26     /**
  27      * Create a new SGQueue object.
  28      */
  29     SGQueue() {}
  30
  31     /**
  32      * Destroy this object.
  33      */
  34     virtual ~SGQueue() {}
  35
  36     /**
  37      * Returns whether this queue is empty (contains no elements).
  38      *
  39      * @return bool True if queue is empty, otherwisr false.
  40      */
  41     virtual bool empty() = 0;
  42
  43     /**
  44      * Add an item to the end of the queue.
  45      *
  46      * @param T object to add.
  47      */
  48     virtual void push( const T& item ) = 0;
  49
  50     /**
  51      * Get an item from the head of the queue.
  52      *
  53      * @return T next available object.
  54      */
  55     virtual T pop() = 0;
  56
  57 protected:
  58     /**
  59      *
  60      */
  61     std::queue<T> fifo;
  62 };
  63
  64 /**
  65  * A simple thread safe queue.  All access functions are guarded with a mutex.
  66  */
  67 template<class T, class SGLOCK=SGMutex>
  68 class SGLockedQueue : public SGQueue<T>
  69 {
  70 public:
  71
  72     /**
  73      * Create a new SGLockedQueue object.
  74      */
  75     SGLockedQueue() {}
  76
  77     /**
  78      * Destroy this object.
  79      */
  80     ~SGLockedQueue() {}
  81
  82     /**
  83      * Returns whether this queue is empty (contains no elements).
  84      *
  85      * @return bool True if queue is empty, otherwisr false.
  86      */
  87     virtual bool empty() {
  88         SGGuard<SGLOCK> g(mutex);
  89         return fifo.empty();
  90     }
  91
  92     /**
  93      * Add an item to the end of the queue.
  94      *
  95      * @param T object to add.
  96      */
  97     virtual void push( const T& item ) {
  98         SGGuard<SGLOCK> g(mutex);
  99         fifo.push( item );
 100     }
 101
 102     /**
 103      * Get an item from the head of the queue.
 104      *
 105      * @return T next available object.
 106      */
 107     virtual T pop() {
 108         SGGuard<SGLOCK> g(mutex);
 109         //if (fifo.empty()) throw NoSuchElementException();
 110         assert( ! fifo.empty() );
 111 //      if (fifo.empty())
 112 //      {
 113 //          mutex.unlock();
 114 //          pthread_exit( PTHREAD_CANCELED );
 115 //      }
 116         T item = fifo.front();
 117         fifo.pop();
 118         return item;
 119     }
 120 private:
 121
 122     /**
 123      * Mutex to serialise access.
 124      */
 125     SGLOCK mutex;
 126
 127 private:
 128     // Prevent copying.
 129     SGLockedQueue(const SGLockedQueue&);
 130     SGLockedQueue& operator= (const SGLockedQueue&);
 131 };
 132
 133 /**
 134  * A guarded queue blocks threads trying to retrieve items
 135  * when none are available.
 136  */
 137 template<class T>
 138 class SGBlockingQueue : public SGQueue<T>
 139 {
 140 public:
 141     /**
 142      * Create a new SGBlockingQueue.
 143      */
 144     SGBlockingQueue() {}
 145
 146     /**
 147      * Destroy this queue.
 148      */
 149     ~SGBlockingQueue() { mutex.unlock(); }
 150
 151     /**
 152      *
 153      */
 154     virtual bool empty() {
 155         SGGuard<SGMutex> g(mutex);
 156         return fifo.empty();
 157     }
 158
 159     /**
 160      * Add an item to the end of the queue.
 161      *
 162      * @param T object to add.
 163      */
 164     virtual void push( const T& item ) {
 165         SGGuard<SGMutex> g(mutex);
 166         fifo.push( item );
 167         not_empty.signal();
 168     }
 169
 170     /**
 171      * Get an item from the head of the queue.
 172      * If no items are available then the calling thread is suspended
 173      *
 174      * @return T next available object.
 175      */
 176     virtual T pop() {
 177         SGGuard<SGMutex> g(mutex);
 178
 179         while (fifo.empty())
 180             not_empty.wait(mutex);
 181
 182         assert(fifo.empty() != true);
 183         //if (fifo.empty()) throw ??
 184
 185         T item = fifo.front();
 186         fifo.pop();
 187         return item;
 188     }
 189
 190 private:
 191
 192     /**
 193      * Mutex to serialise access.
 194      */
 195     SGMutex mutex;
 196
 197     /**
 198      * Condition to signal when queue not empty.
 199      */
 200     SGPthreadCond not_empty;
 201
 202 private:
 203     // Prevent copying.
 204     SGBlockingQueue( const SGBlockingQueue& );
 205     SGBlockingQueue& operator=( const SGBlockingQueue& );
 206 };
 207
 208 #endif // SGQUEUE_HXX_INCLUDED