FCam: include/FCam/TSQueue.h Source File

Go to the documentation of this file.
00001 #ifndef FCAM_TSQUEUE_H
00002 #define FCAM_TSQUEUE_H
00003 
00004 #include <deque>
00005 #include <iterator>
00006 #include <semaphore.h>
00007 #include <pthread.h>
00008 
00009 #include <errno.h>
00010 #include <string.h>
00011 
00012 #include "Base.h"
00013 
00014 //#include "../../src/Debug.h"
00015 
00025 namespace FCam {
00026 
00032 template<typename T>
00033 class TSQueue {
00034 public:
00035     TSQueue();
00036     ~TSQueue();
00037 
00040     class locking_iterator;
00041 
00043     void push(const T &val);
00045     void pop();
00046 
00048     void pushFront(const T &val);
00050     void popBack();
00051 
00054     T &front();
00055     const T &front() const;
00056 
00059     T &back();
00060     const T &back() const;
00061 
00063     bool empty() const;
00065     size_t size() const;
00066 
00069     bool wait(unsigned int timeout=0);
00070 
00078     T pull();
00079 
00081     T pullBack();
00082 
00085     bool tryPull(T *);
00086     bool tryPullBack(T *);
00087 
00091     locking_iterator begin();
00095     locking_iterator end();
00103     bool erase(TSQueue<T>::locking_iterator);
00104 
00105 private:
00106     std::deque<T> q;
00107     mutable pthread_mutex_t mutex;
00108     sem_t *sem;
00109 
00110     friend class locking_iterator;
00111 };
00112 
00113 template<typename T>
00114 class TSQueue<T>::locking_iterator: public std::iterator< std::random_access_iterator_tag, T> {
00115 public:
00116     locking_iterator();
00117     locking_iterator(TSQueue<T> *, typename std::deque<T>::iterator i);
00118     locking_iterator(const TSQueue<T>::locking_iterator &);
00119     ~locking_iterator();
00120     locking_iterator &operator=(const TSQueue<T>::locking_iterator &);
00121 
00122     locking_iterator &operator++();
00123     locking_iterator operator++(int);
00124     locking_iterator &operator--();
00125     locking_iterator operator--(int);
00126 
00127     locking_iterator operator+(int);
00128     locking_iterator operator-(int);
00129 
00130     locking_iterator &operator+=(int);
00131     locking_iterator &operator-=(int);
00132 
00133     int operator-(const TSQueue<T>::locking_iterator &);
00134 
00135     bool operator==(const TSQueue<T>::locking_iterator &other);
00136     bool operator!=(const TSQueue<T>::locking_iterator &other);
00137     bool operator<(const TSQueue<T>::locking_iterator &other);
00138     bool operator>(const TSQueue<T>::locking_iterator &other);
00139     bool operator<=(const TSQueue<T>::locking_iterator &other);
00140     bool operator>=(const TSQueue<T>::locking_iterator &other);
00141 
00142     T &operator*();
00143     T *operator->();
00144 private:
00145     TSQueue *parent;
00146     typename std::deque<T>::iterator qi;
00147 
00148     friend class TSQueue<T>;
00149 };
00150 
00151 template<typename T>
00152 TSQueue<T>::TSQueue() {
00153     pthread_mutexattr_t mutexAttr;
00154     pthread_mutexattr_init(&mutexAttr);
00155     pthread_mutexattr_settype(&mutexAttr, PTHREAD_MUTEX_RECURSIVE);
00156     pthread_mutex_init(&mutex, &mutexAttr);
00157 #ifdef FCAM_PLATFORM_OSX
00158     // unnamed semaphores not supported on OSX
00159     char semName[256];
00160     // Create a unique semaphore name for this TSQueue using its pointer value
00161     snprintf(semName, 256, "FCam::TSQueue::sema::%llx", (long long unsigned)this);
00162     sem = sem_open(semName, O_CREAT, 0600, 0);
00163     if (sem == SEM_FAILED) {
00164         fcamPanic("TSQueue::TSQueue: Unable to initialize semaphore %s: %s",
00165                   semName,
00166                   strerror(errno));
00167     }
00168 #else
00169     sem = new sem_t;
00170     int success = sem_init(sem, 0, 0);
00171     if (success == -1) {
00172         fcamPanic("TSQueue::TSQueue: Unable to initialize semaphore: %s",
00173                   strerror(errno));
00174     }
00175 #endif
00176     //dprintf(6,"TSQueue %llx initialized\n", (long long unsigned)this);
00177 }
00178 
00179 template<typename T>
00180 TSQueue<T>::~TSQueue() {
00181     pthread_mutex_destroy(&mutex);
00182 #ifdef FCAM_PLATFORM_OSX
00183     int success = sem_close(sem);
00184     char semName[256];
00185     // Recreate the unique semaphore name for this TSQueue using its pointer value
00186     snprintf(semName, 256, "FCam::TSQueue::sema::%llx", (long long unsigned)this);
00187     if (success == 0) { success = sem_unlink(semName); }
00188     if (success == -1) {
00189         fcamPanic("TSQueue::~TSQueue: Unable to destroy semaphore %s: %s\n",
00190                   semName,
00191                   strerror(errno));
00192     }
00193 
00194 #else
00195     int success = sem_destroy(sem);
00196     if (success == -1) {
00197         fcamPanic("TSQueue::~TSQueue: Unable to destroy semaphore: %s\n",
00198                   strerror(errno));
00199     }
00200     delete sem;
00201 #endif
00202 }
00203 
00204 template<typename T>
00205 void TSQueue<T>::push(const T &val) {
00206     // dprintf(6, "Pushing to queue %px\n",this);
00207     pthread_mutex_lock(&mutex);
00208     q.push_back(val);
00209     pthread_mutex_unlock(&mutex);
00210     sem_post(sem);
00211 }
00212 
00213 template<typename T>
00214 void TSQueue<T>::pushFront(const T &val) {
00215     pthread_mutex_lock(&mutex);
00216     q.push_front(val);
00217     pthread_mutex_unlock(&mutex);
00218     sem_post(sem);
00219 }
00220 
00221 template<typename T>
00222 void TSQueue<T>::pop() {
00223     // Only safe for a single consumer!!!
00224     sem_wait(sem);
00225     pthread_mutex_lock(&mutex);
00226     q.pop_front();
00227     pthread_mutex_unlock(&mutex);
00228 }
00229 
00230 template<typename T>
00231 void TSQueue<T>::popBack() {
00232     // Only safe for a single consumer!!!
00233     sem_wait(sem);
00234     pthread_mutex_lock(&mutex);
00235     q.pop_back();
00236     pthread_mutex_unlock(&mutex);
00237 }
00238 
00239 template<typename T>
00240 T &TSQueue<T>::front() {
00241     pthread_mutex_lock(&mutex);
00242     T &val = q.front();
00243     pthread_mutex_unlock(&mutex);
00244     return val;
00245 }
00246 
00247 template<typename T>
00248 const T &TSQueue<T>::front() const {
00249     const T &val;
00250     pthread_mutex_lock(&mutex);
00251     val = q.front();
00252     pthread_mutex_unlock(&mutex);
00253     return val;
00254 }
00255 
00256 template<typename T>
00257 T &TSQueue<T>::back() {
00258     T &val;
00259     pthread_mutex_lock(&mutex);
00260     val = q.back();
00261     pthread_mutex_unlock(&mutex);
00262     return val;
00263 }
00264 
00265 template<typename T>
00266 const T &TSQueue<T>::back() const {
00267     const T &val;
00268     pthread_mutex_lock(&mutex);
00269     val = q.back();
00270     pthread_mutex_unlock(&mutex);
00271     return val;
00272 }
00273 
00274 template<typename T>
00275 bool TSQueue<T>::empty() const {
00276     bool _empty;
00277     pthread_mutex_lock(&mutex);
00278     _empty = q.empty();
00279     pthread_mutex_unlock(&mutex);
00280     return _empty;
00281 }
00282 
00283 template<typename T>
00284 size_t TSQueue<T>::size() const {
00285     size_t _size;
00286     pthread_mutex_lock(&mutex);
00287     _size = q.size();
00288     pthread_mutex_unlock(&mutex);
00289     return _size;
00290 }
00291 
00292 template<typename T>
00293 bool TSQueue<T>::wait(unsigned int timeout) {
00294     bool res;
00295     int err;
00296     if (timeout == 0) {
00297         err=sem_wait(sem);
00298     } else {
00299 #ifndef FCAM_PLATFORM_OSX // No clock_gettime or sem_timedwait on OSX 
00300         timespec tv;
00301         // This will overflow around 1 hour or so of timeout
00302         clock_gettime(CLOCK_REALTIME, &tv);
00303         tv.tv_nsec += timeout*1000;
00304         tv.tv_sec += tv.tv_nsec / 1000000000;
00305         tv.tv_nsec = tv.tv_nsec % 1000000000;
00306         err=sem_timedwait(sem, &tv);
00307 #else
00308         err=sem_trywait(sem);
00309 #endif
00310     }
00311     if (err == -1) {
00312         switch (errno) {
00313         case EINTR:
00314         case ETIMEDOUT:
00315             res = false;
00316             break;
00317         default:
00318             //error(Event::InternalError, "TSQueue::wait: Unexpected error from wait on semaphore: %s", strerror(errno));
00319             res = false;
00320             break;
00321         }
00322     } else {
00323         res = true;
00324         sem_post(sem); // Put back the semaphore since we're not actually popping
00325     }
00326 
00327     return res;
00328 }
00329 
00330 template<typename T>
00331 T TSQueue<T>::pull() {
00332     //dprintf(6, "Pulling from queue %px\n",this);
00333     sem_wait(sem);
00334     pthread_mutex_lock(&mutex);
00335     T copyVal = q.front();
00336     q.pop_front();
00337     pthread_mutex_unlock(&mutex);
00338     //dprintf(6, "Done pull from queue %px\n",this);
00339     return copyVal;
00340 }
00341 
00342 template<typename T>
00343 T TSQueue<T>::pullBack() {
00344     sem_wait(sem);
00345     pthread_mutex_lock(&mutex);
00346     T copyVal = q.back();
00347     q.pop_back();
00348     pthread_mutex_unlock(&mutex);
00349     return copyVal;
00350 }
00351 
00352 template<typename T>
00353 bool TSQueue<T>::tryPull(T *ptr) {
00354     if (sem_trywait(sem)) { return false; }
00355     pthread_mutex_lock(&mutex);
00356     T copyVal = q.front();
00357     q.pop_front();
00358     pthread_mutex_unlock(&mutex);
00359     *ptr = copyVal;
00360     return true;
00361 }
00362 
00363 template<typename T>
00364 bool TSQueue<T>::tryPullBack(T *ptr) {
00365     if (sem_trywait(sem)) { return false; }
00366     pthread_mutex_lock(&mutex);
00367     T copyVal = q.back();
00368     q.pop_back();
00369     pthread_mutex_unlock(&mutex);
00370     *ptr = copyVal;
00371     return true;
00372 }
00373 
00374 template<typename T>
00375 typename TSQueue<T>::locking_iterator TSQueue<T>::begin() {
00376     return locking_iterator(this, q.begin());
00377 }
00378 
00379 template<typename T>
00380 typename TSQueue<T>::locking_iterator TSQueue<T>::end() {
00381     return locking_iterator(this, q.end());
00382 }
00383 
00384 template<typename T>
00385 bool TSQueue<T>::erase(TSQueue<T>::locking_iterator li) {
00386     /* Since we're erasing, decrement semaphore. */
00387     if (sem_trywait(sem)) {
00388         return false;
00389     }
00390     q.erase(li.qi);
00391     return true;
00392 }
00393 
00394 template<typename T>
00395 TSQueue<T>::locking_iterator::locking_iterator() : parent(NULL), qi()
00396 {}
00397 
00398 template<typename T>
00399 TSQueue<T>::locking_iterator::locking_iterator(TSQueue<T> *p,
00400                                                typename std::deque<T>::iterator i) :
00401     parent(p), qi(i) {
00402     if (parent) {
00403         pthread_mutex_lock(&(parent->mutex));
00404     }
00405 }
00406 
00407 template<typename T>
00408 TSQueue<T>::locking_iterator::locking_iterator(const TSQueue<T>::locking_iterator &other):
00409     parent(other.parent), qi(other.qi) {
00410     if (parent) {
00411         pthread_mutex_lock(&(parent->mutex));
00412     }
00413 }
00414 
00415 template<typename T>
00416 TSQueue<T>::locking_iterator::~locking_iterator() {
00417     if (parent) {
00418         pthread_mutex_unlock(&(parent->mutex));
00419     }
00420 }
00421 
00422 template<typename T>
00423 typename TSQueue<T>::locking_iterator &TSQueue<T>::locking_iterator::operator=(const TSQueue<T>::locking_iterator &other) {
00424     if (&other == this) { return (*this); }
00425     if (parent &&
00426         other.qi == qi) { return (*this); }
00427 
00428     if (parent) { pthread_mutex_unlock(&(parent->mutex)); }
00429     parent = other.parent;
00430     qi = other.qi;
00431     if (parent) { pthread_mutex_lock(&(parent->mutex)); }
00432 
00433 }
00434 
00435 template<typename T>
00436 typename TSQueue<T>::locking_iterator &TSQueue<T>::locking_iterator::operator++() {
00437     qi++;
00438     return (*this);
00439 }
00440 
00441 template<typename T>
00442 typename TSQueue<T>::locking_iterator TSQueue<T>::locking_iterator::operator++(int) {
00443     typename TSQueue<T>::locking_iterator temp(*this);
00444     qi++;
00445     return temp;
00446 }
00447 
00448 template<typename T>
00449 typename TSQueue<T>::locking_iterator &TSQueue<T>::locking_iterator::operator--() {
00450     qi--;
00451     return (*this);
00452 }
00453 
00454 template<typename T>
00455 typename TSQueue<T>::locking_iterator TSQueue<T>::locking_iterator::operator--(int) {
00456     typename TSQueue<T>::locking_iterator temp(*this);
00457     qi--;
00458     return temp;
00459 }
00460 
00461 template<typename T>
00462 typename TSQueue<T>::locking_iterator TSQueue<T>::locking_iterator::operator+(int n) {
00463     typename TSQueue<T>::locking_iterator temp(*this);
00464     temp+=n;
00465     return temp;
00466 }
00467 
00468 template<typename T>
00469 typename TSQueue<T>::locking_iterator operator+(int n,
00470                                                 typename TSQueue<T>::locking_iterator l) {
00471     return l+n;
00472 }
00473 
00474 template<typename T>
00475 typename TSQueue<T>::locking_iterator TSQueue<T>::locking_iterator::operator-(int n) {
00476     typename TSQueue<T>::locking_iterator temp(*this);
00477     temp-=n;
00478     return temp;
00479 }
00480 
00481 template<typename T>
00482 typename TSQueue<T>::locking_iterator &TSQueue<T>::locking_iterator::operator+=(int n) {
00483     qi += n;
00484     return *this;
00485 }
00486 
00487 template<typename T>
00488 typename TSQueue<T>::locking_iterator &TSQueue<T>::locking_iterator::operator-=(int n) {
00489     qi -= n;
00490     return *this;
00491 }
00492 
00493 template<typename T>
00494 int TSQueue<T>::locking_iterator::operator-(const TSQueue<T>::locking_iterator &other) {
00495     return qi - other.qi;
00496 }
00497 
00498 template<typename T>
00499 bool TSQueue<T>::locking_iterator::operator==(const TSQueue<T>::locking_iterator &other) {
00500     return qi == other.qi;
00501 }
00502 
00503 template<typename T>
00504 bool TSQueue<T>::locking_iterator::operator!=(const TSQueue<T>::locking_iterator &other) {
00505     return qi != other.qi;
00506 }
00507 
00508 template<typename T>
00509 bool TSQueue<T>::locking_iterator::operator<(const TSQueue<T>::locking_iterator &other) {
00510     return qi < other.qi;
00511 }
00512 
00513 template<typename T>
00514 bool TSQueue<T>::locking_iterator::operator>(const TSQueue<T>::locking_iterator &other) {
00515     return qi > other.qi;
00516 }
00517 
00518 template<typename T>
00519 bool TSQueue<T>::locking_iterator::operator<=(const TSQueue<T>::locking_iterator &other) {
00520     return qi <= other.qi;
00521 }
00522 
00523 template<typename T>
00524 bool TSQueue<T>::locking_iterator::operator>=(const TSQueue<T>::locking_iterator &other) {
00525     return qi >= other.qi;
00526 }
00527 
00528 template<typename T>
00529 T &TSQueue<T>::locking_iterator::operator*() {
00530     return *qi;
00531 }
00532 
00533 template<typename T>
00534 T *TSQueue<T>::locking_iterator::operator->() {
00535     return &(*qi);
00536 }
00537 
00538 }
00539 
00540 #endif