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 "SGGuard.hxx"
   9 #include "SGThread.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>
  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<SGMutex> 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<SGMutex> 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<SGMutex> 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<SGMutex> 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<SGMutex> g(mutex);
 149         return this->fifo.size();
 150     }
 151
 152 private:
 153
 154     /**
 155      * Mutex to serialise access.
 156      */
 157     SGMutex 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     SGWaitCondition not_empty;
 259
 260 private:
 261     // Prevent copying.
 262     SGBlockingQueue( const SGBlockingQueue& );
 263     SGBlockingQueue& operator=( const SGBlockingQueue& );
 264 };
 265
 266
 267 /**
 268  * A guarded deque blocks threads trying to retrieve items
 269  * when none are available.
 270  */
 271 template<class T>
 272 class SGBlockingDeque
 273 {
 274 public:
 275     /**
 276      * Create a new SGBlockingDequeue.
 277      */
 278     SGBlockingDeque() {}
 279
 280     /**
 281      * Destroy this dequeue.
 282      */
 283     ~SGBlockingDeque() {}
 284
 285     /**
 286      *
 287      */
 288     virtual void clear() {
 289     SGGuard<SGMutex> g(mutex);
 290     this->queue.clear();
 291     }
 292
 293     /**
 294      *
 295      */
 296     virtual bool empty() {
 297     SGGuard<SGMutex> g(mutex);
 298     return this->queue.empty();
 299     }
 300
 301     /**
 302      * Add an item to the front of the queue.
 303      *
 304      * @param T object to add.
 305      */
 306     virtual void push_front( const T& item ) {
 307     SGGuard<SGMutex> g(mutex);
 308     this->queue.push_front( item );
 309     not_empty.signal();
 310     }
 311
 312     /**
 313      * Add an item to the back of the queue.
 314      *
 315      * @param T object to add.
 316      */
 317     virtual void push_back( const T& item ) {
 318     SGGuard<SGMutex> g(mutex);
 319     this->queue.push_back( item );
 320     not_empty.signal();
 321     }
 322
 323     /**
 324      * View the item from the head of the queue.
 325      * Calling thread is not suspended
 326      *
 327      * @return T next available object.
 328      */
 329     virtual T front() {
 330     SGGuard<SGMutex> g(mutex);
 331
 332     assert(this->queue.empty() != true);
 333     //if (queue.empty()) throw ??
 334
 335     T item = this->queue.front();
 336     return item;
 337     }
 338
 339     /**
 340      * Get an item from the head of the queue.
 341      * If no items are available then the calling thread is suspended
 342      *
 343      * @return T next available object.
 344      */
 345     virtual T pop_front() {
 346     SGGuard<SGMutex> g(mutex);
 347
 348     while (this->queue.empty())
 349         not_empty.wait(mutex);
 350
 351     assert(this->queue.empty() != true);
 352     //if (queue.empty()) throw ??
 353
 354     T item = this->queue.front();
 355     this->queue.pop_front();
 356     return item;
 357     }
 358
 359     /**
 360      * Get an item from the tail of the queue.
 361      * If no items are available then the calling thread is suspended
 362      *
 363      * @return T next available object.
 364      */
 365     virtual T pop_back() {
 366     SGGuard<SGMutex> g(mutex);
 367
 368     while (this->queue.empty())
 369         not_empty.wait(mutex);
 370
 371     assert(this->queue.empty() != true);
 372     //if (queue.empty()) throw ??
 373
 374     T item = this->queue.back();
 375     this->queue.pop_back();
 376     return item;
 377     }
 378
 379     /**
 380      * Query the size of the queue
 381      *
 382      * @return size_t size of queue.
 383      */
 384     virtual size_t size() {
 385     SGGuard<SGMutex> g(mutex);
 386         return this->queue.size();
 387     }
 388
 389 private:
 390
 391     /**
 392      * Mutex to serialise access.
 393      */
 394     SGMutex mutex;
 395
 396     /**
 397      * Condition to signal when queue not empty.
 398      */
 399     SGWaitCondition not_empty;
 400
 401 private:
 402     // Prevent copying.
 403     SGBlockingDeque( const SGBlockingDeque& );
 404     SGBlockingDeque& operator=( const SGBlockingDeque& );
 405
 406 protected:
 407     std::deque<T> queue;
 408 };
 409
 410 #endif // SGQUEUE_HXX_INCLUDED