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
00039 #ifndef SYSTEMQUEUE_H_
00040 #define SYSTEMQUEUE_H_
00041
00042 #include <vector>
00043 #include <cassert>
00044 #include "SystemEvent.h"
00045 #include "SystemEventListener.h"
00046 #include "ariba/utility/logging/Logging.h"
00047 #include <boost/date_time.hpp>
00048 #include <boost/cstdint.hpp>
00049
00050 #ifdef UNDERLAY_OMNET
00051 #include <csimplemodule.h>
00052 #include <cmessage.h>
00053 #include <macros.h>
00054 #else
00055 #include <boost/thread/mutex.hpp>
00056 #include <boost/thread/thread.hpp>
00057 #include <boost/thread/condition_variable.hpp>
00058 #include <boost/utility.hpp>
00059 #include <boost/bind.hpp>
00060 #endif
00061
00062 using std::vector;
00063 using boost::posix_time::ptime;
00064
00065 namespace ariba {
00066 namespace utility {
00067
00077 #ifndef UNDERLAY_OMNET
00078 class SystemQueue : private boost::noncopyable {
00079 #else
00080 class SystemQueue : public cSimpleModule {
00081 #endif
00082
00083 use_logging_h(SystemQueue);
00084 friend class EnterMethod;
00085 public:
00089 static SystemQueue& instance() {
00090 static SystemQueue _inst;
00091 return _inst;
00092 }
00093
00094 #ifdef UNDERLAY_OMNET
00095
00100 virtual void deleteModule(){}
00101 #endif
00102
00109 void scheduleEvent( const SystemEvent& event, uint32_t delay = 0 );
00110
00116 void run();
00117
00124 void cancel();
00125
00129 void dropAll( const SystemEventListener* mlistener);
00130
00136 bool isEmpty();
00137
00143 bool isRunning();
00144
00145 protected:
00146
00150 void enterMethod();
00151
00155 void leaveMethod();
00156
00160 SystemQueue();
00161
00166 ~SystemQueue();
00167
00168 #ifdef UNDERLAY_OMNET
00169 virtual void handleMessage( cMessage* msg );
00170 #endif
00171
00172 private:
00173
00174 #ifndef UNDERLAY_OMNET
00175 typedef vector<SystemEvent> EventQueue;
00176
00177
00178
00179 class QueueThread {
00180 public:
00181 QueueThread(QueueThread* _transferQueue = NULL);
00182 virtual ~QueueThread();
00183 void run();
00184 void cancel();
00185 bool isEmpty();
00186 void insert( const SystemEvent& event, uint32_t delay );
00187 void enter();
00188 void leave();
00189 void dropAll( const SystemEventListener* mlistener);
00190
00191 protected:
00192 virtual void onItemInserted( const SystemEvent& event ) = 0;
00193 virtual void onNextQueueItem( const SystemEvent& event ) = 0;
00194 QueueThread* transferQueue;
00195 EventQueue eventsQueue;
00196 boost::mutex queueMutex;
00197 private:
00198 boost::thread* queueThread;
00199 static void threadFunc( QueueThread* obj );
00200 boost::condition_variable itemsAvailable;
00201 volatile bool running;
00202 };
00203
00204
00205
00206 class QueueThreadDirect : public QueueThread {
00207 public:
00208 QueueThreadDirect();
00209 ~QueueThreadDirect();
00210 protected:
00211 virtual void onItemInserted( const SystemEvent& event );
00212 virtual void onNextQueueItem( const SystemEvent& event );
00213 };
00214
00215
00216
00217 class QueueThreadDelay : public QueueThread {
00218 public:
00219 QueueThreadDelay(QueueThread* _transferQueue = NULL);
00220 ~QueueThreadDelay();
00221 protected:
00222 virtual void onItemInserted( const SystemEvent& event );
00223 virtual void onNextQueueItem( const SystemEvent& event );
00224 private:
00225 volatile bool isSleeping;
00226 ptime sleepStart;
00227 boost::mutex sleepMutex;
00228 boost::condition_variable sleepCond;
00229 };
00230
00231
00232
00233 QueueThreadDirect directScheduler;
00234 QueueThreadDelay delayScheduler;
00235 volatile bool systemQueueRunning;
00236 #endif
00237
00238 };
00239
00240 #ifdef UNDERLAY_OMNET
00241
00242 #if 0
00243
00244
00245
00246
00247
00248
00249
00250
00251
00252
00253
00254
00255
00256
00257
00258 #endif
00259
00260 static cModule* SystemQueue__create() {
00261 return &SystemQueue::instance();
00262 }
00263
00264 EXECUTE_ON_STARTUP(SystemQueue__mod, modtypes.instance()->add(new cModuleType("SystemQueue","SystemQueue",(ModuleCreateFunc)SystemQueue__create));)
00265
00266 #endif
00267
00268 }}
00269
00270 #endif