- Main Page
- Namespaces
- Classes
- Files
- Related Pages

# /Users/fpy/titan-1-5-MBARI/Mers/Executives/DeductiveController/Rtapi/ThreadSafeObjectQueue.cpp

    00001
    00002 //
    00003 // Copyright (c) 2004           Model-based Embedded and Robotic Systems Group
    00004 // All Rights Reserved          Massachusetts Institute of Technology
    00005 //
    00006 // This software is provided as is. Model-based Embedded and Robotic Systems
    00007 // (MERS) group does not assume any responsibility.
    00008 //
    00009 // Organization:        Model-based Embedded and Robotic Systems Group
    00010 //                                      Massachusetts Institute of Technology
    00011 //
    00013
    00014 #include "ThreadSafeObjectQueue.h"
    00015 #include <stdio.h>
    00016 #include <Mers/Utils/Exceptions/Exception.h>
    00017
    00019
    00020 using namespace Mers::Executives::DeductiveController::Rtapi;
    00021 using Mers::Utils::Lang::BaseModelObject;
    00022 using Mers::Utils::Containers::ObjectQueue;
    00023 using Mers::Utils::Lang::ReferencedObject;
    00024 using std::ostream;
    00025
    00027 // Construction/Destruction
    00029
    00030 ThreadSafeObjectQueue::ThreadSafeObjectQueue(int myMaxQueueSize)
    00031         : ObjectQueue(true, myMaxQueueSize), ReferencedObject()
    00032 {
    00033         int error = pthread_mutex_init(&queueMutex, NULL);
    00034         if(error){
    00035                 printf("Unable to initialize queue mutex\n");
    00036                 return;
    00037         }
    00038         error = pthread_cond_init(&queueNotEmpty, NULL);
    00039         if(error){
    00040                 printf("Unable to initialize not empty condition\n");
    00041                 return;
    00042         }
    00043         error = pthread_cond_init(&queueIsEmpty, NULL);
    00044         if(error){
    00045                 printf("Unable to initialize empty condition\n");
    00046                 return;
    00047         }
    00048 }
    00049
    00051
    00052
    00053 ThreadSafeObjectQueue::~ThreadSafeObjectQueue()
    00054 {
    00055         pthread_mutex_destroy(&queueMutex);
    00056
    00057         pthread_cond_destroy(&queueNotEmpty);
    00058         pthread_cond_destroy(&queueIsEmpty);
    00059
    00060         // printf("Deleting Queue\n");
    00061 }
    00062
    00064
    00065 namespace Mers {
    00066         namespace Executives {
    00067                 namespace DeductiveController {
    00068                         namespace Rtapi {
    00069                                 class NULLObjectException : public Mers::Utils::Exceptions::Exception
    00070                                 {
    00071                                 public:
    00076                                         NULLObjectException(char const * str) {
    00077                                                 mMessage = str;
    00078                                         }
    00079
    00080 #ifndef EMBEDDED
    00081
    00086                                         virtual void print(ostream &out) const {
    00087                                                 out << mMessage;
    00088                                         }
    00089 #endif
    00090
    00091                                 protected:
    00093                                         char const * mMessage;
    00094                                 };
    00095                         }
    00096                 }
    00097         }
    00098 }
    00099
    00101
    00102
    00103 int ThreadSafeObjectQueue::enqueue(BaseModelObject * element)
    00104 {
    00105         int ret;
    00106
    00107         if(element == NULL) {
    00108                 fprintf(stderr, "Error: Enqueue of a NULL object\n");
    00109                 throw new NULLObjectException("Error: Enqueue of a NULL object\n");
    00110         }
    00111
    00112         //cerr << "ObjectQueue: -WLen " << getLength();
    00113         lock();
    00114         ret = ObjectQueue::enqueue(element);
    00115         if(!FAILED(ret)){
    00116                 // if we didn't fail to enqueue an extra element, then we're not
    00117                 // empty for sure
    00118                 pthread_cond_broadcast(&queueNotEmpty);
    00119         }
    00120         unlock();
    00121         //cerr << " -> " << getLength() << " (" << (void *) element << ")" << endl;
    00122         //cerr << "Queue [";
    00123         //printArray(cerr, ", ");
    00124         //cerr << ']' << endl;
    00125         return ret;
    00126 }
    00127
    00128
    00130
    00131
    00132 int ThreadSafeObjectQueue::dequeue(BaseModelObject * &element, bool wait)
    00133 {
    00134         int ret;
    00135
    00136         if(wait){
    00137                 // We're going to want to wait till the queue is non-empty so we
    00138                 // can remove an element before we return
    00139                 //cerr << "ObjectQueue: -WLen " << getLength();
    00140                 lock();
    00141                 do{
    00142                         while(ObjectQueue::isEmpty()) {
    00143                                 pthread_cond_wait(&queueNotEmpty, &queueMutex);
    00144                         }
    00145                         ret = ObjectQueue::dequeue(element);
    00146                 }while(FAILED(ret));
    00147                 unlock();
    00148                 //cerr << " -> " << getLength() << " (" << (void *) element << ")" << endl;
    00149                 //cerr << "Queue [";
    00150                 //printArray(cerr, ", ");
    00151                 //cerr << ']' << endl;
    00152         }else{
    00153                 //cerr << "ObjectQueue: --Len " << getLength();
    00154                 lock();
    00155                 ret = ObjectQueue::dequeue(element);
    00156                 unlock();
    00157                 //cerr << " -> " << getLength() << " (" << (void *) element << ")" << endl;
    00158                 //cerr << "Queue [";
    00159                 //printArray(cerr, ", ");
    00160                 //cerr << ']' << endl;
    00161         }
    00162
    00163         if(isEmpty())
    00164                 pthread_cond_broadcast(&queueIsEmpty);
    00165
    00166         return ret;
    00167 }
    00168
    00170
    00171
    00172 int ThreadSafeObjectQueue::peekQueue(BaseModelObject * &element)
    00173 {
    00174         int ret;
    00175
    00176         lock();
    00177         ret = ObjectQueue::peekQueue(element);
    00178         unlock();
    00179         return ret;
    00180 }
    00181
    00183
    00184
    00185 unsigned int ThreadSafeObjectQueue::getLength()
    00186 {
    00187         unsigned int ret;
    00188         lock();
    00189         ret = ObjectQueue::getLength();
    00190         unlock();
    00191         return ret;
    00192 }
    00193
    00195
    00196
    00197 bool ThreadSafeObjectQueue::isEmpty()
    00198 {
    00199         bool ret;
    00200         lock();
    00201         ret = ObjectQueue::isEmpty();
    00202         unlock();
    00203         return ret;
    00204 }
    00205
    00207
    00208
    00209 void ThreadSafeObjectQueue::waitForEmpty()
    00210 {
    00211         lock();
    00212         while(!ObjectQueue::isEmpty()){
    00213                 pthread_cond_wait(&queueIsEmpty, &queueMutex);
    00214         }
    00215         unlock();
    00216 }
    00217
    00219
    00220
    00221 #ifndef EMBEDDED
    00222
    00223 // this may have some problems.  If your object is not actually const but you
    00224 // treat it as a const pointer, it won't call this function and won't protect the queue...
    00225 void ThreadSafeObjectQueue::printArray(ostream &out, char const * seperator)
    00226 {
    00227         lock();
    00228         ObjectQueue::printArray(out, seperator);
    00229         unlock();
    00230 }
    00231
    00232 #endif
    00233
    00235