simgear/threads/SGQueue.hxx

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