source: source/ariba/utility/system/SystemQueue.cpp@ 4483

Last change on this file since 4483 was 4483, checked in by Christoph Mayer, 15 years ago

-StartupWrapper bisschen umgeschrieben, erste Schritte für sauberes runterfahren

File size: 8.8 KB
Line 
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
38#include "SystemQueue.h"
39
40namespace ariba {
41namespace utility {
42
43use_logging_cpp(SystemQueue);
44
45SystemQueue::SystemQueue()
46#ifndef UNDERLAY_OMNET
47 : delayScheduler( &directScheduler ), systemQueueRunning( false )
48#endif
49{
50}
51
52SystemQueue::~SystemQueue() {
53}
54
55void SystemQueue::scheduleEvent( const SystemEvent& event, uint32_t delay ) {
56#ifndef UNDERLAY_OMNET
57 if( delay == 0 ) directScheduler.insert( event, delay );
58 else delayScheduler.insert ( event, delay );
59#else
60 Enter_Method_Silent();
61 cMessage* msg = new cMessage();
62 msg->setContextPointer( new SystemEvent(event) );
63
64 if( delay == 0 )
65 cSimpleModule::scheduleAt( cSimpleModule::simTime(), msg );
66 else
67 cSimpleModule::scheduleAt( cSimpleModule::simTime()+((double)delay/1000.0), msg );
68#endif
69}
70
71#ifdef UNDERLAY_OMNET
72void SystemQueue::handleMessage(cMessage* msg){
73 SystemEvent* event = (SystemEvent*)msg->contextPointer();
74
75 event->listener->handleSystemEvent( *event );
76
77 delete event; delete msg;
78}
79#endif
80
81void SystemQueue::run() {
82#ifndef UNDERLAY_OMNET
83 systemQueueRunning = true;
84 directScheduler.run();
85 delayScheduler.run();
86#endif
87}
88
89void SystemQueue::cancel() {
90#ifndef UNDERLAY_OMNET
91 systemQueueRunning = false;
92 directScheduler.cancel();
93 delayScheduler.cancel();
94#endif
95}
96
97bool SystemQueue::isEmpty() {
98#ifndef UNDERLAY_OMNET
99 return directScheduler.isEmpty() || delayScheduler.isEmpty();
100#else
101 return false;
102#endif
103}
104
105bool SystemQueue::isRunning() {
106#ifndef UNDERLAY_OMNET
107 return systemQueueRunning;
108#else
109 return true;
110#endif
111}
112
113//***************************************************************
114#ifndef UNDERLAY_OMNET
115
116SystemQueue::QueueThread::QueueThread(QueueThread* _transferQueue)
117 : running( false ), transferQueue( _transferQueue ) {
118}
119
120SystemQueue::QueueThread::~QueueThread(){
121}
122
123void SystemQueue::QueueThread::run(){
124 running = true;
125 queueThread = new boost::thread(
126 boost::bind(&QueueThread::threadFunc, this) );
127}
128
129void SystemQueue::QueueThread::cancel(){
130
131 // cause the thread to exit
132 {
133 boost::mutex::scoped_lock lock(queueMutex);
134 running = false;
135
136 while( eventsQueue.size() > 0 ){
137 eventsQueue.erase( eventsQueue.begin() );
138 }
139
140 itemsAvailable.notify_all();
141 }
142
143 // wait that the thread has exited
144 // and delete it, so that a subsuquent run() can be called
145
146 queueThread->join();
147 delete queueThread;
148 queueThread = NULL;
149}
150
151bool SystemQueue::QueueThread::isEmpty(){
152 boost::mutex::scoped_lock lock( queueMutex );
153 return eventsQueue.empty();
154}
155
156void SystemQueue::QueueThread::insert( const SystemEvent& event, uint32_t delay ){
157
158 // if this is called from a module that is currently handling
159 // a thread (called from SystemQueue::onNextQueueItem), the
160 // thread is the same anyway and the mutex will be already
161 // aquired, otherwise we aquire it now
162
163 boost::mutex::scoped_lock lock( queueMutex );
164
165 eventsQueue.push_back( event );
166 eventsQueue.back().scheduledTime = boost::posix_time::microsec_clock::local_time();
167 eventsQueue.back().delayTime = delay;
168 eventsQueue.back().remainingDelay = delay;
169
170 onItemInserted( event );
171 itemsAvailable.notify_all();
172}
173
174void SystemQueue::QueueThread::threadFunc( QueueThread* obj ) {
175
176 boost::mutex::scoped_lock lock( obj->queueMutex );
177
178 while( obj->running ) {
179
180 // wait until an item is in the queue or we are notified
181 // to quit the thread. in case the thread is about to
182 // quit, the queueThreadRunning variable will indicate
183 // this and cause the thread to exit
184
185 while ( obj->eventsQueue.empty() && obj->running ){
186
187// const boost::system_time duration =
188// boost::get_system_time() +
189// boost::posix_time::milliseconds(100);
190
191 obj->itemsAvailable.wait( lock );
192// obj->itemsAvailable.timed_wait( lock, duration );
193 }
194
195 //
196 // work all the items that are currently in the queue
197 //
198
199 while( !obj->eventsQueue.empty() && obj->running ) {
200
201 // fetch the first item in the queue
202 // and deliver it to the queue handler
203 SystemEvent ev = obj->eventsQueue.front();
204 obj->eventsQueue.erase( obj->eventsQueue.begin() );
205
206 // call the queue and this will
207 // call the actual event handler
208 obj->onNextQueueItem( ev );
209
210 } // !obj->eventsQueue.empty() )
211 } // while (obj->running)
212
213 logging_debug("system queue exited");
214}
215
216//***************************************************************
217
218SystemQueue::QueueThreadDirect::QueueThreadDirect(){
219}
220
221SystemQueue::QueueThreadDirect::~QueueThreadDirect(){
222}
223
224void SystemQueue::QueueThreadDirect::onItemInserted( const SystemEvent& event ){
225 // do nothing here
226}
227
228void SystemQueue::QueueThreadDirect::onNextQueueItem( const SystemEvent& event ){
229 // directly deliver the item to the
230 event.getListener()->handleSystemEvent( event );
231}
232
233//***************************************************************
234
235SystemQueue::QueueThreadDelay::QueueThreadDelay(QueueThread* _transferQueue)
236 : QueueThread( _transferQueue ), isSleeping( false ) {
237
238 assert( _transferQueue != NULL );
239}
240
241SystemQueue::QueueThreadDelay::~QueueThreadDelay(){
242}
243
244void SystemQueue::QueueThreadDelay::onItemInserted( const SystemEvent& event ){
245
246 if( !isSleeping) return;
247
248 // break an existing sleep and
249 // remember the time that was actually slept for
250 // and change it for every event in the queue
251
252 assert( !eventsQueue.empty());
253 sleepCond.notify_all();
254
255 ptime sleepEnd = boost::posix_time::microsec_clock::local_time();
256 boost::posix_time::time_duration duration = sleepEnd - sleepStart;
257 uint32_t sleptTime = duration.total_milliseconds();
258
259 EventQueue::iterator i = eventsQueue.begin();
260 EventQueue::iterator iend = eventsQueue.end();
261
262 for( ; i != iend; i++ ) {
263
264 if( sleptTime >= i->remainingDelay)
265 i->remainingDelay = 0;
266 else
267 i->remainingDelay -= sleptTime;
268
269 } // for( ; i != iend; i++ )
270
271 // now we have to reorder the events
272 // in the queue with respect to their remaining delay
273 // the SystemQueue::operator< takes care of the
274 // ordering with respect to the remaining delay
275
276 std::sort( eventsQueue.begin(), eventsQueue.end() );
277
278}
279
280void SystemQueue::QueueThreadDelay::onNextQueueItem( const SystemEvent& event ){
281
282 // sleeps will be cancelled in the
283 // onItemInserted function when a new
284 // event arrives during sleeping
285
286 assert( !isSleeping );
287
288 // the given item is the one with the least
289 // amount of sleep time left. because all
290 // items are reordered in onItemInserted
291
292 if( event.remainingDelay > 0 ) {
293
294 const boost::system_time duration =
295 boost::get_system_time() +
296 boost::posix_time::milliseconds(event.remainingDelay);
297
298 {
299 boost::unique_lock<boost::mutex> lock( sleepMutex );
300
301 sleepStart = boost::posix_time::microsec_clock::local_time();
302 isSleeping = true;
303
304 sleepCond.timed_wait( lock, duration );
305
306 isSleeping = false;
307 }
308
309 } // if( event.remainingDelay > 0 )
310
311 // if the sleep succeeded and was not
312 // interrupted by a new incoming item
313 // we can now deliver this event
314
315 ptime sleepEnd = boost::posix_time::microsec_clock::local_time();
316 boost::posix_time::time_duration duration = sleepEnd - sleepStart;
317 uint32_t sleptTime = duration.total_milliseconds();
318
319 if (event.remainingDelay <= sleptTime)
320 transferQueue->insert( event, 0 );
321}
322
323#endif // #ifndef UNDERLAY_OMNET
324
325//***************************************************************
326
327}} // spovnet, common
Note: See TracBrowser for help on using the repository browser.