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 <OpenThreads/Mutex>
   9 #include <OpenThreads/ScopedLock>
  10 #include <OpenThreads/Condition>
  11
  12 /**
  13  * SGQueue defines an interface for a FIFO.
  14  * It can be implemented using different types of synchronization
  15  * and protection.
  16  */
  17 template<class T>
  18 class SGQueue
  19 {
  20 public:
  21
  22     /**
  23      * Create a new SGQueue object.
  24      */
  25     SGQueue() {}
  26
  27     /**
  28      * Destroy this object.
  29      */
  30     virtual ~SGQueue() {}
  31
  32     /**
  33      * Returns whether this queue is empty (contains no elements).
  34      *
  35      * @return bool True if queue is empty, otherwisr false.
  36      */
  37     virtual bool empty() = 0;
  38
  39     /**
  40      * Add an item to the end of the queue.
  41      *
  42      * @param T object to add.
  43      */
  44     virtual void push( const T& item ) = 0;
  45
  46     /**
  47      * View the item from the head of the queue.
  48      *
  49      * @return T next available object.
  50      */
  51     virtual T front() = 0;
  52
  53     /**
  54      * Get an item from the head of the queue.
  55      *
  56      * @return T next available object.
  57      */
  58     virtual T pop() = 0;
  59
  60     /**
  61      * Query the size of the queue
  62      *
  63      * @return size_t size of queue.
  64      */
  65     virtual size_t size() = 0;
  66
  67 protected:
  68     /**
  69      *
  70      */
  71     std::queue<T> fifo;
  72 };
  73
  74 /**
  75  * A simple thread safe queue.  All access functions are guarded with a mutex.
  76  */
  77 template<class T, class SGLOCK=OpenThreads::Mutex>
  78 class SGLockedQueue : public SGQueue<T>
  79 {
  80 public:
  81
  82     /**
  83      * Create a new SGLockedQueue object.
  84      */
  85     SGLockedQueue() {}
  86
  87     /**
  88      * Destroy this object.
  89      */
  90     ~SGLockedQueue() {}
  91
  92     /**
  93      * Returns whether this queue is empty (contains no elements).
  94      *
  95      * @return bool True if queue is empty, otherwisr false.
  96      */
  97     virtual bool empty() {
  98         OpenThreads::ScopedLock<SGLOCK> g(mutex);
  99         return this->fifo.empty();
 100     }
 101
 102     /**
 103      * Add an item to the end of the queue.
 104      *
 105      * @param T object to add.
 106      */
 107     virtual void push( const T& item ) {
 108         OpenThreads::ScopedLock<SGLOCK> g(mutex);
 109         this->fifo.push( item );
 110     }
 111
 112     /**
 113      * View the item from the head of the queue.
 114      *
 115      * @return T next available object.
 116      */
 117     virtual T front() {
 118         OpenThreads::ScopedLock<SGLOCK> g(mutex);
 119         assert( ! this->fifo.empty() );
 120         T item = this->fifo.front();
 121         return item;
 122     }
 123
 124     /**
 125      * Get an item from the head of the queue.
 126      *
 127      * @return T next available object.
 128      */
 129     virtual T pop() {
 130         OpenThreads::ScopedLock<SGLOCK> g(mutex);
 131         //if (fifo.empty()) throw NoSuchElementException();
 132         assert( ! this->fifo.empty() );
 133 //      if (fifo.empty())
 134 //      {
 135 //          mutex.unlock();
 136 //          pthread_exit( PTHREAD_CANCELED );
 137 //      }
 138         T item = this->fifo.front();
 139         this->fifo.pop();
 140         return item;
 141     }
 142
 143     /**
 144      * Query the size of the queue
 145      *
 146      * @return size_t size of queue.
 147      */
 148     virtual size_t size() {
 149         OpenThreads::ScopedLock<SGLOCK> g(mutex);
 150         return this->fifo.size();
 151     }
 152
 153 private:
 154
 155     /**
 156      * Mutex to serialise access.
 157      */
 158     SGLOCK mutex;
 159
 160 private:
 161     // Prevent copying.
 162     SGLockedQueue(const SGLockedQueue&);
 163     SGLockedQueue& operator= (const SGLockedQueue&);
 164 };
 165
 166 /**
 167  * A guarded queue blocks threads trying to retrieve items
 168  * when none are available.
 169  */
 170 template<class T>
 171 class SGBlockingQueue : public SGQueue<T>
 172 {
 173 public:
 174     /**
 175      * Create a new SGBlockingQueue.
 176      */
 177     SGBlockingQueue() {}
 178
 179     /**
 180      * Destroy this queue.
 181      */
 182     ~SGBlockingQueue() {}
 183
 184     /**
 185      *
 186      */
 187     virtual bool empty() {
 188         OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 189         return this->fifo.empty();
 190     }
 191
 192     /**
 193      * Add an item to the end of the queue.
 194      *
 195      * @param T object to add.
 196      */
 197     virtual void push( const T& item ) {
 198         OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 199         this->fifo.push( item );
 200         not_empty.signal();
 201     }
 202
 203     /**
 204      * View the item from the head of the queue.
 205      * Calling thread is not suspended
 206      *
 207      * @return T next available object.
 208      */
 209     virtual T front() {
 210         OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 211
 212         assert(this->fifo.empty() != true);
 213         //if (fifo.empty()) throw ??
 214
 215         T item = this->fifo.front();
 216         return item;
 217     }
 218
 219     /**
 220      * Get an item from the head of the queue.
 221      * If no items are available then the calling thread is suspended
 222      *
 223      * @return T next available object.
 224      */
 225     virtual T pop() {
 226         OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 227
 228         while (this->fifo.empty())
 229             not_empty.wait(&mutex);
 230
 231         assert(this->fifo.empty() != true);
 232         //if (fifo.empty()) throw ??
 233
 234         T item = this->fifo.front();
 235         this->fifo.pop();
 236         return item;
 237     }
 238
 239     /**
 240      * Query the size of the queue
 241      *
 242      * @return size_t size of queue.
 243      */
 244     virtual size_t size() {
 245         OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 246         return this->fifo.size();
 247     }
 248
 249 private:
 250
 251     /**
 252      * Mutex to serialise access.
 253      */
 254     OpenThreads::Mutex mutex;
 255
 256     /**
 257      * Condition to signal when queue not empty.
 258      */
 259     OpenThreads::Condition not_empty;
 260
 261 private:
 262     // Prevent copying.
 263     SGBlockingQueue( const SGBlockingQueue& );
 264     SGBlockingQueue& operator=( const SGBlockingQueue& );
 265 };
 266
 267
 268 /**
 269  * A guarded deque blocks threads trying to retrieve items
 270  * when none are available.
 271  */
 272 template<class T>
 273 class SGBlockingDeque
 274 {
 275 public:
 276     /**
 277      * Create a new SGBlockingDequeue.
 278      */
 279     SGBlockingDeque() {}
 280
 281     /**
 282      * Destroy this dequeue.
 283      */
 284     ~SGBlockingDeque() {}
 285
 286     /**
 287      *
 288      */
 289     virtual void clear() {
 290     OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 291     this->queue.clear();
 292     }
 293
 294     /**
 295      *
 296      */
 297     virtual bool empty() {
 298     OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 299     return this->queue.empty();
 300     }
 301
 302     /**
 303      * Add an item to the front of the queue.
 304      *
 305      * @param T object to add.
 306      */
 307     virtual void push_front( const T& item ) {
 308     OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 309     this->queue.push_front( item );
 310     not_empty.signal();
 311     }
 312
 313     /**
 314      * Add an item to the back of the queue.
 315      *
 316      * @param T object to add.
 317      */
 318     virtual void push_back( const T& item ) {
 319     OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 320     this->queue.push_back( item );
 321     not_empty.signal();
 322     }
 323
 324     /**
 325      * View the item from the head of the queue.
 326      * Calling thread is not suspended
 327      *
 328      * @return T next available object.
 329      */
 330     virtual T front() {
 331     OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 332
 333     assert(this->queue.empty() != true);
 334     //if (queue.empty()) throw ??
 335
 336     T item = this->queue.front();
 337     return item;
 338     }
 339
 340     /**
 341      * Get an item from the head of the queue.
 342      * If no items are available then the calling thread is suspended
 343      *
 344      * @return T next available object.
 345      */
 346     virtual T pop_front() {
 347     OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 348
 349     while (this->queue.empty())
 350         not_empty.wait(&mutex);
 351
 352     assert(this->queue.empty() != true);
 353     //if (queue.empty()) throw ??
 354
 355     T item = this->queue.front();
 356     this->queue.pop_front();
 357     return item;
 358     }
 359
 360     /**
 361      * Get an item from the tail of the queue.
 362      * If no items are available then the calling thread is suspended
 363      *
 364      * @return T next available object.
 365      */
 366     virtual T pop_back() {
 367     OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 368
 369     while (this->queue.empty())
 370         not_empty.wait(&mutex);
 371
 372     assert(this->queue.empty() != true);
 373     //if (queue.empty()) throw ??
 374
 375     T item = this->queue.back();
 376     this->queue.pop_back();
 377     return item;
 378     }
 379
 380     /**
 381      * Query the size of the queue
 382      *
 383      * @return size_t size of queue.
 384      */
 385     virtual size_t size() {
 386     OpenThreads::ScopedLock<OpenThreads::Mutex> g(mutex);
 387         return this->queue.size();
 388     }
 389
 390 private:
 391
 392     /**
 393      * Mutex to serialise access.
 394      */
 395     OpenThreads::Mutex mutex;
 396
 397     /**
 398      * Condition to signal when queue not empty.
 399      */
 400     OpenThreads::Condition not_empty;
 401
 402 private:
 403     // Prevent copying.
 404     SGBlockingDeque( const SGBlockingDeque& );
 405     SGBlockingDeque& operator=( const SGBlockingDeque& );
 406
 407 protected:
 408     std::deque<T> queue;
 409 };
 410
 411 #endif // SGQUEUE_HXX_INCLUDED