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 #endif // SGQUEUE_HXX_INCLUDED