00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037
00038 #include "SystemQueue.h"
00039
00040 namespace ariba {
00041 namespace utility {
00042
00043 use_logging_cpp(SystemQueue);
00044
00045 SystemQueue::SystemQueue()
00046 #ifndef UNDERLAY_OMNET
00047 : delayScheduler( &directScheduler ), systemQueueRunning( false )
00048 #endif
00049 {
00050 }
00051
00052 SystemQueue::~SystemQueue() {
00053 }
00054
00055 void SystemQueue::scheduleEvent( const SystemEvent& event, uint32_t delay ) {
00056 #ifndef UNDERLAY_OMNET
00057 if( delay == 0 ) directScheduler.insert( event, delay );
00058 else delayScheduler.insert ( event, delay );
00059 #else
00060 Enter_Method_Silent();
00061 cMessage* msg = new cMessage();
00062 msg->setContextPointer( new SystemEvent(event) );
00063
00064 if( delay == 0 )
00065 cSimpleModule::scheduleAt( cSimpleModule::simTime(), msg );
00066 else
00067 cSimpleModule::scheduleAt( cSimpleModule::simTime()+((double)delay/1000.0), msg );
00068 #endif
00069 }
00070
00071 #ifdef UNDERLAY_OMNET
00072 void SystemQueue::handleMessage(cMessage* msg){
00073 SystemEvent* event = (SystemEvent*)msg->contextPointer();
00074
00075 event->listener->handleSystemEvent( *event );
00076
00077 delete event; delete msg;
00078 }
00079 #endif
00080
00081 void SystemQueue::run() {
00082 #ifndef UNDERLAY_OMNET
00083 systemQueueRunning = true;
00084 directScheduler.run();
00085 delayScheduler.run();
00086 #endif
00087 }
00088
00089 void SystemQueue::cancel() {
00090 #ifndef UNDERLAY_OMNET
00091 systemQueueRunning = false;
00092 directScheduler.cancel();
00093 delayScheduler.cancel();
00094 #endif
00095 }
00096
00097 bool SystemQueue::isEmpty() {
00098 #ifndef UNDERLAY_OMNET
00099 return directScheduler.isEmpty() || delayScheduler.isEmpty();
00100 #else
00101 return false;
00102 #endif
00103 }
00104
00105 bool SystemQueue::isRunning() {
00106 #ifndef UNDERLAY_OMNET
00107 return systemQueueRunning;
00108 #else
00109 return true;
00110 #endif
00111 }
00112
00113 void SystemQueue::enterMethod(){
00114
00115 directScheduler.enter();
00116 }
00117
00118 void SystemQueue::leaveMethod(){
00119
00120 directScheduler.leave();
00121 }
00122
00123
00124 #ifndef UNDERLAY_OMNET
00125
00126 SystemQueue::QueueThread::QueueThread(QueueThread* _transferQueue)
00127 : running( false ), transferQueue( _transferQueue ) {
00128 }
00129
00130 SystemQueue::QueueThread::~QueueThread(){
00131 }
00132
00133 void SystemQueue::QueueThread::run(){
00134 running = true;
00135
00136 queueThread = new boost::thread(
00137 boost::bind(&QueueThread::threadFunc, this) );
00138 }
00139
00140 void SystemQueue::QueueThread::cancel(){
00141
00142 logging_debug("cancelling system queue");
00143
00144
00145 {
00146
00147
00148 boost::mutex::scoped_lock lock(queueMutex);
00149
00150
00151
00152 running = false;
00153 itemsAvailable.notify_all();
00154 }
00155
00156
00157 logging_debug("joining system queue thread");
00158 queueThread->join();
00159
00160
00161 logging_debug("deleting pending system queue events");
00162 while( eventsQueue.size() > 0 ){
00163 eventsQueue.erase( eventsQueue.begin() );
00164 }
00165
00166
00167 delete queueThread;
00168 queueThread = NULL;
00169 }
00170
00171 bool SystemQueue::QueueThread::isEmpty(){
00172 boost::mutex::scoped_lock lock( queueMutex );
00173 return eventsQueue.empty();
00174 }
00175
00176 void SystemQueue::QueueThread::insert( const SystemEvent& event, uint32_t delay ){
00177
00178
00179
00180
00181
00182
00183 boost::mutex::scoped_lock lock( queueMutex );
00184
00185 eventsQueue.push_back( event );
00186 eventsQueue.back().scheduledTime = boost::posix_time::microsec_clock::local_time();
00187 eventsQueue.back().delayTime = delay;
00188 eventsQueue.back().remainingDelay = delay;
00189
00190 onItemInserted( event );
00191 itemsAvailable.notify_all();
00192 }
00193
00194 void SystemQueue::QueueThread::threadFunc( QueueThread* obj ) {
00195
00196 boost::mutex::scoped_lock lock( obj->queueMutex );
00197
00198 while( obj->running ) {
00199
00200
00201
00202
00203
00204
00205 while ( obj->running && obj->eventsQueue.empty() ){
00206
00207
00208
00209
00210
00211
00212 obj->itemsAvailable.wait( lock );
00213 }
00214
00215
00216
00217
00218
00219 while( obj->running && (!obj->eventsQueue.empty()) ) {
00220
00221
00222
00223 SystemEvent ev = obj->eventsQueue.front();
00224 obj->eventsQueue.erase( obj->eventsQueue.begin() );
00225
00226
00227
00228 obj->queueMutex.unlock();
00229 obj->onNextQueueItem( ev );
00230 obj->queueMutex.lock();
00231
00232 }
00233 }
00234
00235 logging_debug("system queue exited");
00236 }
00237
00238 void SystemQueue::QueueThread::enter(){
00239 queueMutex.lock();
00240 }
00241
00242 void SystemQueue::QueueThread::leave(){
00243 queueMutex.unlock();
00244 }
00245
00246
00247
00248
00249 SystemQueue::QueueThreadDirect::QueueThreadDirect(){
00250 }
00251
00252 SystemQueue::QueueThreadDirect::~QueueThreadDirect(){
00253 }
00254
00255 void SystemQueue::QueueThreadDirect::onItemInserted( const SystemEvent& event ){
00256
00257 }
00258
00259 void SystemQueue::QueueThreadDirect::onNextQueueItem( const SystemEvent& event ){
00260
00261 event.getListener()->handleSystemEvent( event );
00262 }
00263
00264
00265
00266 SystemQueue::QueueThreadDelay::QueueThreadDelay(QueueThread* _transferQueue)
00267 : QueueThread( _transferQueue ), isSleeping( false ) {
00268
00269 assert( _transferQueue != NULL );
00270 }
00271
00272 SystemQueue::QueueThreadDelay::~QueueThreadDelay(){
00273 }
00274
00275 void SystemQueue::QueueThreadDelay::onItemInserted( const SystemEvent& event ){
00276
00277 if( !isSleeping) return;
00278
00279
00280
00281
00282
00283 assert( !eventsQueue.empty());
00284 sleepCond.notify_all();
00285
00286 ptime sleepEnd = boost::posix_time::microsec_clock::local_time();
00287 boost::posix_time::time_duration duration = sleepEnd - sleepStart;
00288 uint32_t sleptTime = duration.total_milliseconds();
00289
00290 EventQueue::iterator i = eventsQueue.begin();
00291 EventQueue::iterator iend = eventsQueue.end();
00292
00293 for( ; i != iend; i++ ) {
00294
00295 if( sleptTime >= i->remainingDelay)
00296 i->remainingDelay = 0;
00297 else
00298 i->remainingDelay -= sleptTime;
00299
00300 }
00301
00302
00303
00304
00305
00306
00307 std::sort( eventsQueue.begin(), eventsQueue.end() );
00308
00309 }
00310
00311 void SystemQueue::QueueThreadDelay::onNextQueueItem( const SystemEvent& event ){
00312
00313
00314
00315
00316
00317 assert( !isSleeping );
00318
00319
00320
00321
00322
00323 if( event.remainingDelay > 0 ) {
00324
00325 const boost::system_time duration =
00326 boost::get_system_time() +
00327 boost::posix_time::milliseconds(event.remainingDelay);
00328
00329 {
00330 boost::unique_lock<boost::mutex> lock( sleepMutex );
00331
00332 sleepStart = boost::posix_time::microsec_clock::local_time();
00333 isSleeping = true;
00334
00335 sleepCond.timed_wait( lock, duration );
00336
00337 isSleeping = false;
00338 }
00339
00340 }
00341
00342
00343
00344
00345
00346 ptime sleepEnd = boost::posix_time::microsec_clock::local_time();
00347 boost::posix_time::time_duration duration = sleepEnd - sleepStart;
00348 uint32_t sleptTime = duration.total_milliseconds();
00349
00350 if (event.remainingDelay <= sleptTime)
00351 transferQueue->insert( event, 0 );
00352 }
00353
00354 #endif // #ifndef UNDERLAY_OMNET
00355
00356
00357
00358 }}