close Warning: Can't use blame annotator: 
No changeset 2259 in the repository

source: source/ariba/utility/system/SystemQueue.h@ 12750

Last change on this file since 12750 was 12745, checked in by hock@…, 11 years ago

Debug output for faulty SystemQueue
File size: 7.6 KB
RevLine 
1// [License]
2// The Ariba-Underlay Copyright
3//
4// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH)
5//
6// Institute of Telematics
7// UniversitÀt Karlsruhe (TH)
8// Zirkel 2, 76128 Karlsruhe
9// Germany
10//
11// Redistribution and use in source and binary forms, with or without
12// modification, are permitted provided that the following conditions are
13// met:
14//
15// 1. Redistributions of source code must retain the above copyright
16// notice, this list of conditions and the following disclaimer.
17// 2. Redistributions in binary form must reproduce the above copyright
18// notice, this list of conditions and the following disclaimer in the
19// documentation and/or other materials provided with the distribution.
20//
21// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND
22// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
23// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
24// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR
25// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
26// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
27// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
28// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
29// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
30// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
31// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32//
33// The views and conclusions contained in the software and documentation
34// are those of the authors and should not be interpreted as representing
35// official policies, either expressed or implied, of the Institute of
36// Telematics.
37// [License]
38
39#ifndef SYSTEMQUEUE_H_
40#define SYSTEMQUEUE_H_
41
42#include <vector>
43#include <cassert>
44#include "SystemEvent.h"
45#include "SystemEventListener.h"
46#include "ariba/utility/logging/Logging.h"
47#include <boost/date_time.hpp>
48#include <boost/cstdint.hpp>
49
50#ifdef UNDERLAY_OMNET
51 #include <csimplemodule.h>
52 #include <cmessage.h>
53 #include <macros.h>
54#else
55 #include <boost/thread/mutex.hpp>
56 #include <boost/thread/thread.hpp>
57 #include <boost/thread/condition_variable.hpp>
58 #include <boost/utility.hpp>
59 #include <boost/bind.hpp>
60#endif
61
62#include <boost/function.hpp>
63
64
65using std::vector;
66using boost::posix_time::ptime;
67
68namespace ariba {
69namespace utility {
70
71/**
72 * This class implements a simple system event queue to allow
73 * a simulation of cooperative multitasking. It also allows
74 * events to be scheduled from other tasks. This allows
75 * dispatching asynchronous tasks.
76 *
77 * @author Christoph Mayer, Sebastian Mies
78 */
79
80class SystemQueue : private boost::noncopyable
81{
82 use_logging_h(SystemQueue);
83 friend class EnterMethod;
84public:
85 /**
86 * Get the SystemQueue singleton instance.
87 */
88 static SystemQueue& instance();
89
90 /**
91 * This methods schedules a given event.
92 *
93 * @param The event to be scheduled
94 * @param The delay in milli-seconds
95 */
96 void scheduleEvent( const SystemEvent& event, uint32_t delay = 0 );
97
98 /**
99 * This method schedules a function call in the SystemQueue.
100 * (Like scheduleEvent, but to be used with boost::bind.)
101 *
102 * @param function: The function to be called [void function()]
103 * @param The delay in milli-seconds
104 */
105 void scheduleCall( const boost::function0<void>& function, uint32_t delay = 0 );
106
107 /**
108 * Starts the processing and waiting for events.
109 * Use <code>cancel()</code> to end system queue processing and
110 * <code>isEmpty()</code> to check wheter the queue is empty.
111 */
112 void run();
113
114 /**
115 * Cancels the system queue and ends the processing after the
116 * currently processed event is processed.
117 *
118 * This method is thread-safe.
119 */
120 void cancel();
121
122 /**
123 * Drop all queued events for that listener
124 */
125 void dropAll( const SystemEventListener* mlistener);
126
127 /**
128 * Check wheter this queue has items or not.
129 *
130 * @return True, if this queue is empty.
131 */
132 bool isEmpty();
133
134 /**
135 * Is the system queue already started and running?
136 *
137 * @return True, if the system queue is running.
138 */
139 bool isRunning();
140
141protected:
142
143 /**
144 * Aqcuire the mutex
145 */
146 void enterMethod();
147
148 /**
149 * Leave the mutex
150 */
151 void leaveMethod();
152
153 /**
154 * Constructs a system queue.
155 */
156 SystemQueue();
157
158 /**
159 * Destroys the system queue. Beware that all events
160 * are canceled
161 */
162 ~SystemQueue();
163
164
165private:
166
167#ifndef UNDERLAY_OMNET
168 typedef vector<SystemEvent> EventQueue;
169
170 //********************************************************
171
172 class QueueThread {
173 public:
174 QueueThread(QueueThread* _transferQueue = NULL);
175 virtual ~QueueThread();
176 void run();
177 void cancel();
178 bool isEmpty();
179 void insert( const SystemEvent& event, uint32_t delay );
180 void enter();
181 void leave();
182 void dropAll( const SystemEventListener* mlistener);
183
184 protected:
185 virtual void onItemInserted( const SystemEvent& event ) = 0;
186 virtual void onNextQueueItem( const SystemEvent& event ) = 0;
187 QueueThread* transferQueue;
188 EventQueue eventsQueue;
189 boost::mutex queueMutex;
190 private:
191 boost::thread* queueThread;
192 static void threadFunc( QueueThread* obj );
193 boost::condition_variable itemsAvailable;
194 volatile bool running;
195 }; // class QueueThread
196
197 //********************************************************
198
199 class QueueThreadDirect : public QueueThread {
200 public:
201 QueueThreadDirect();
202 ~QueueThreadDirect();
203 protected:
204 virtual void onItemInserted( const SystemEvent& event );
205 virtual void onNextQueueItem( const SystemEvent& event );
206 }; // class QueueThreadDirect
207
208 //********************************************************
209
210 class QueueThreadDelay : public QueueThread {
211 public:
212 QueueThreadDelay(QueueThread* _transferQueue = NULL);
213 ~QueueThreadDelay();
214 protected:
215 virtual void onItemInserted( const SystemEvent& event );
216 virtual void onNextQueueItem( const SystemEvent& event );
217 private:
218 volatile bool isSleeping;
219 ptime sleepStart;
220 boost::mutex sleepMutex;
221 boost::condition_variable sleepCond;
222 }; // class QueueThreadDelay
223
224 //********************************************************
225
226
227// /// Singleton
228// static SystemQueue* _inst;
229
230 QueueThreadDirect directScheduler;
231 QueueThreadDelay delayScheduler;
232 volatile bool systemQueueRunning;
233#endif
234
235
236private:
237 /**
238 * This inner class handles the function-call events.
239 * @see SystemQueue::scheduleCall
240 */
241 class FunctionCaller : public SystemEventListener
242 {
243 void handleSystemEvent(const SystemEvent& event)
244 {
245 boost::function0<void>* function_ptr = event.getData< boost::function0<void> >();
246 (*function_ptr)();
247 delete function_ptr;
248 }
249 };
250
251 FunctionCaller internal_function_caller;
252}; // class SystemQueue
253
254#ifdef UNDERLAY_OMNET
255
256#if 0
257 //
258 // the system queue must be a singleton in simulations, too.
259 // and to include it in the simulation the module is defined
260 // as submodule in every SpoVNet host. Therefore we hack the
261 // Define_Module (see omnet/includes/macros.h) the way we need
262 // it with our singleton ...
263 //
264 // this is the macro definition from macros.h
265 //
266 // #define Define_Module(CLASSNAME) /backslash
267 // static cModule *CLASSNAME##__create() {return new CLASSNAME();} /backslash
268 // EXECUTE_ON_STARTUP(CLASSNAME##__mod, modtypes.instance()->add(new cModuleType(#CLASSNAME,#CLASSNAME,(ModuleCreateFunc)CLASSNAME##__create));)
269 //
270 // and this is how we do it :)
271 //
272#endif
273
274 static cModule* SystemQueue__create() {
275 return &SystemQueue::instance();
276 }
277
278 EXECUTE_ON_STARTUP(SystemQueue__mod, modtypes.instance()->add(new cModuleType("SystemQueue","SystemQueue",(ModuleCreateFunc)SystemQueue__create));)
279
280#endif
281
282}} // spovnet, common
283
284#endif /* SYSTEMQUEUE_H_ */
Note: See TracBrowser for help on using the repository browser.