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