src/dataqueue.cc

   1 ///
   2 /// \file       dataqueue.cc
   3 ///             FIFO queue of Data objects
   4 ///
   5
   6 /*
   7     Copyright (C) 2008, Net Direct Inc. (http://www.netdirect.ca/)
   8
   9     This program is free software; you can redistribute it and/or modify
  10     it under the terms of the GNU General Public License as published by
  11     the Free Software Foundation; either version 2 of the License, or
  12     (at your option) any later version.
  13
  14     This program is distributed in the hope that it will be useful,
  15     but WITHOUT ANY WARRANTY; without even the implied warranty of
  16     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  17
  18     See the GNU General Public License in the COPYING file at the
  19     root directory of this project for more details.
  20 */
  21
  22 #include "dataqueue.h"
  23 #include "scoped_lock.h"
  24 #include "data.h"
  25 #include <sys/time.h>
  26 #include <time.h>
  27
  28 namespace Barry {
  29
  30 //////////////////////////////////////////////////////////////////////////////
  31 // DataQueue class
  32
  33 DataQueue::DataQueue()
  34 {
  35         pthread_mutex_init(&m_waitMutex, NULL);
  36         pthread_cond_init(&m_waitCond, NULL);
  37
  38         pthread_mutex_init(&m_accessMutex, NULL);
  39 }
  40
  41 DataQueue::~DataQueue()
  42 {
  43         scoped_lock lock(m_accessMutex);        // FIXME - is this sane?
  44
  45         while( m_queue.size() ) {
  46                 delete m_queue.front();
  47                 m_queue.pop();
  48         }
  49 }
  50
  51 //
  52 // push
  53 //
  54 /// Pushes data into the end of the queue.
  55 ///
  56 /// The queue owns this pointer as soon as the function is
  57 /// called.  In the case of an exception, it will be freed.
  58 /// Performs a thread broadcast once new data has been added.
  59 ///
  60 void DataQueue::push(Data *data)
  61 {
  62         try {
  63
  64                 {
  65                         scoped_lock lock(m_accessMutex);
  66                         m_queue.push(data);
  67                 }
  68
  69                 scoped_lock wait(m_waitMutex);
  70                 pthread_cond_broadcast(&m_waitCond);
  71
  72         }
  73         catch(...) {
  74                 delete data;
  75                 throw;
  76         }
  77 }
  78
  79 //
  80 // pop
  81 //
  82 /// Pops the next element off the front of the queue.
  83 ///
  84 /// Returns 0 if empty.
  85 /// The queue no longer owns this pointer upon return.
  86 ///
  87 Data* DataQueue::pop()
  88 {
  89         scoped_lock lock(m_accessMutex);
  90
  91         if( m_queue.size() == 0 )
  92                 return 0;
  93
  94         Data *ret = m_queue.front();
  95         m_queue.pop();
  96         return ret;
  97 }
  98
  99 //
 100 // wait_pop
 101 //
 102 /// Pops the next element off the front of the queue, and
 103 /// waits until one exists if empty.  If still no data
 104 /// on timeout, returns null.
 105 /// (unlock the access mutex while waiting!)
 106 ///
 107 /// Timeout specified in milliseconds.  Default is wait forever.
 108 ///
 109 Data* DataQueue::wait_pop(int timeout)
 110 {
 111         Data *ret = 0;
 112
 113         // check if something's there already
 114         {
 115                 scoped_lock access(m_accessMutex);
 116                 if( m_queue.size() ) {
 117                         ret = m_queue.front();
 118                         m_queue.pop();
 119                         return ret;
 120                 }
 121         }
 122
 123         // nothing there, so wait...
 124
 125         if( timeout == -1 ) {
 126                 // no timeout
 127                 int size = 0;
 128                 do {
 129                         {
 130                                 scoped_lock wait(m_waitMutex);
 131                                 pthread_cond_wait(&m_waitCond, &m_waitMutex);
 132                         }
 133
 134                         // anything there?
 135                         scoped_lock access(m_accessMutex);
 136                         size = m_queue.size();
 137                         if( size != 0 ) {
 138                                 // already have the lock, return now
 139                                 ret = m_queue.front();
 140                                 m_queue.pop();
 141                                 return ret;
 142                         }
 143
 144                 } while( size == 0 );
 145         }
 146         else {
 147                 // timeout in conditional wait
 148                 struct timeval now;
 149                 struct timespec to;
 150
 151                 gettimeofday(&now, NULL);
 152                 to.tv_sec = now.tv_sec + timeout / 1000;
 153                 to.tv_nsec = (now.tv_usec + timeout % 1000 * 1000) * 1000;
 154
 155                 scoped_lock wait(m_waitMutex);
 156                 pthread_cond_timedwait(&m_waitCond, &m_waitMutex, &to);
 157         }
 158
 159         scoped_lock access(m_accessMutex);
 160         if( m_queue.size() == 0 )
 161                 return 0;
 162
 163         ret = m_queue.front();
 164         m_queue.pop();
 165         return ret;
 166 }
 167
 168 //
 169 // append_from
 170 //
 171 /// Pops all data from other and appends it to this.
 172 ///
 173 /// After calling this function, other will be empty, and
 174 /// this will contain all its data.
 175 ///
 176 /// In the case of an exception, any uncopied data will
 177 /// remain in other.
 178 ///
 179 /// This is a locking optimization, so all copying can happen
 180 /// inside one lock, instead of locking for each copy.
 181 ///
 182 void DataQueue::append_from(DataQueue &other)
 183 {
 184         scoped_lock us(m_accessMutex);
 185         scoped_lock them(other.m_accessMutex);
 186
 187         while( other.m_queue.size() ) {
 188                 m_queue.push( other.m_queue.front() );
 189
 190                 // only pop after the copy, since in the
 191                 // case of an exception we want to leave other intact
 192                 other.m_queue.pop();
 193         }
 194 }
 195
 196 //
 197 // empty
 198 //
 199 /// Returns true if the queue is empty.
 200 ///
 201 bool DataQueue::empty() const
 202 {
 203         scoped_lock access(m_accessMutex);
 204         return m_queue.size() == 0;
 205 }
 206
 207 //
 208 // size
 209 //
 210 /// Returns number of items in the queue.
 211 ///
 212 size_t DataQueue::size() const
 213 {
 214         scoped_lock access(m_accessMutex);
 215         return m_queue.size();
 216 }
 217
 218 } // namespace Barry
 219
 220
 221 #ifdef __DQ_TEST_MODE__
 222
 223 #include <iostream>
 224
 225 using namespace std;
 226 using namespace Barry;
 227
 228 void *WriteThread(void *userdata)
 229 {
 230         DataQueue *dq = (DataQueue*) userdata;
 231
 232         dq->push( new Data );
 233         dq->push( new Data );
 234         sleep(5);
 235         dq->push( new Data );
 236
 237         return 0;
 238 }
 239
 240 void *ReadThread(void *userdata)
 241 {
 242         DataQueue *dq = (DataQueue*) userdata;
 243
 244         sleep(1);
 245         if( Data *d = dq->pop() ) {
 246                 cout << "Received via pop: " << d << endl;
 247                 delete d;
 248         }
 249         else {
 250                 cout << "No data in the queue yet." << endl;
 251         }
 252
 253         while( Data *d = dq->wait_pop(5010) ) {
 254                 cout << "Received: " << d << endl;
 255                 delete d;
 256         }
 257         return 0;
 258 }
 259
 260 int main()
 261 {
 262         DataQueue from;
 263         from.push( new Data );
 264
 265         DataQueue dq;
 266         dq.append_from(from);
 267
 268         pthread_t t1, t2;
 269         pthread_create(&t1, NULL, &ReadThread, &dq);
 270         pthread_create(&t2, NULL, &WriteThread, &dq);
 271
 272         pthread_join(t2, NULL);
 273         pthread_join(t1, NULL);
 274 }
 275
 276 #endif
 277