Index: source/ariba/utility/system/SystemEvent.h =================================================================== --- source/ariba/utility/system/SystemEvent.h (revision 12763) +++ source/ariba/utility/system/SystemEvent.h (revision 12764) @@ -71,8 +71,11 @@ inline SystemEvent(SystemEventListener* mlistener, SystemEventType mtype = SystemEventType::DEFAULT, void* mdata = NULL) : - listener(mlistener), type(mtype), data(mdata), scheduledTime( - boost::posix_time::not_a_date_time), delayTime(0), - remainingDelay(0) - + listener(mlistener), + type(mtype), + data(mdata), + scheduledTime(boost::posix_time::not_a_date_time), + deadline(boost::posix_time::not_a_date_time), + delayTime(0), + remainingDelay(0) { } @@ -81,12 +84,16 @@ inline SystemEvent(SystemEventListener* mlistener, SystemEventType mtype = SystemEventType::DEFAULT, T* mdata = NULL) : - listener(mlistener), type(mtype), data((void*) mdata), scheduledTime( - boost::posix_time::not_a_date_time), delayTime(0), - remainingDelay(0) { + listener(mlistener), + type(mtype), + data((void*) mdata), + scheduledTime(boost::posix_time::not_a_date_time), + deadline(boost::posix_time::not_a_date_time), + delayTime(0), + remainingDelay(0) { } - // FIXME AKTUELL MARIO: copy deadline (etc?) !!! inline SystemEvent(const SystemEvent& copy) { this->scheduledTime = copy.scheduledTime; + this->deadline = copy.deadline; this->delayTime = copy.delayTime; this->remainingDelay = copy.remainingDelay; @@ -98,4 +105,5 @@ inline void operator=(const SystemEvent& right) { this->scheduledTime = right.scheduledTime; + this->deadline = right.deadline; this->delayTime = right.delayTime; this->remainingDelay = right.remainingDelay; Index: source/ariba/utility/system/SystemQueue.cpp =================================================================== --- source/ariba/utility/system/SystemQueue.cpp (revision 12763) +++ source/ariba/utility/system/SystemQueue.cpp (revision 12764) @@ -40,13 +40,16 @@ #include +// TODO Mario: +// check if there is any debug out left to remove + namespace ariba { namespace utility { - typedef boost::mutex::scoped_lock scoped_lock; - - using boost::posix_time::microsec_clock; - using boost::posix_time::time_duration; - using boost::date_time::not_a_date_time; - using boost::scoped_ptr; +typedef boost::mutex::scoped_lock scoped_lock; + +using boost::posix_time::microsec_clock; +using boost::posix_time::time_duration; +using boost::date_time::not_a_date_time; +using boost::scoped_ptr; @@ -73,4 +76,6 @@ void SystemQueue::scheduleEvent( const SystemEvent& event, uint32_t delay ) { + assert ( SysQ->running ); // should we really enforce this? + // copy SystemEvent ev(event); @@ -96,4 +101,5 @@ { assert ( ! SysQ->running ); + assert ( ! SysQ->unclean ); SysQ->running = true; @@ -105,4 +111,12 @@ void SystemQueue::cancel() { + // CHECK: this function must not be called from within a SystemQueue-Event + if ( sysq_thread && boost::this_thread::get_id() == sysq_thread->get_id() ) + { + logging_warn("SystemQueue::cancel() was called from within a SystemQueue-Event. This is not allowed!"); + throw std::logic_error("SystemQueue::cancel() was called from within a SystemQueue-Event. This is not allowed!"); + } + + // signal SysQ to quit (and abort queued events) SysQ->cancel(); @@ -129,5 +143,5 @@ void SystemQueue::dropAll( const SystemEventListener* mlistener) { -// XXX +// TODO // directScheduler.dropAll(mlistener); // delayScheduler.dropAll(mlistener); @@ -144,22 +158,14 @@ } -// XXX -// void SystemQueue::enterMethod(){ -// // TODO: omnet case and delay scheduler -// directScheduler.enter(); -// } -// -// void SystemQueue::leaveMethod(){ -// // TODO: omnet case and delay scheduler -// directScheduler.leave(); -// } - - + +//******************************************************** + + +/// constructor SystemQueue::QueueThread::QueueThread() : -// now( not_a_date_time ), -// next_deadline( not_a_date_time ), processing_event( false ), running( false ), - aborted( false ) + aborted( false ), + unclean( false ) { } @@ -190,4 +196,5 @@ logging_debug( "/// SysQ thread is quitting." ); + unclean = true; running = false; } @@ -219,5 +226,5 @@ /* dispatch event */ - logging_debug("/// SysQ: dispatching event"); +// logging_debug("/// SysQ: dispatching event"); // measure execution time (1/2) @@ -319,6 +326,5 @@ ptime SystemQueue::QueueThread::get_clock() { -// return microsec_clock::universal_time(); - return microsec_clock::local_time(); + return microsec_clock::universal_time(); } @@ -367,21 +373,10 @@ { event.deadline = event.scheduledTime + boost::posix_time::milliseconds(delay); - event.delayTime = delay; // XXX I think this is no longer needed.. + event.delayTime = delay; // ( I think this is no longer needed.. ) - // XXX debug - logging_debug("/// inserting timed event, due at: " << event.deadline << " (in " << delay << " ms)"); +// // debug output +// logging_debug("/// inserting timed event, due at: " << event.deadline << " (in " << delay << " ms)"); timedEventsQ.push(event); - - // TODO push sorted.. (use sorted queue..) -// timedEventsQ.pu - /* - * std::priority_queue - * - * but it orders high-to-low (must reverse order..) - * - * ... ah cool, da ist direkt eine anleitung fÃŒr reverse order: - * http://www.cplusplus.com/reference/queue/priority_queue/priority_queue/ - */ } } @@ -415,304 +410,12 @@ -//*************************************************************** -//// XXX old SystemQueue subclasses -// (needed as reference during development of the replacement) - -// #ifndef UNDERLAY_OMNET -#define NOOLDSYSQ -#ifndef NOOLDSYSQ - -void SystemQueue::QueueThread::run(){ - running = true; - - queueThread = new boost::thread( - boost::bind(&QueueThread::threadFunc, this) ); -} - -void SystemQueue::QueueThread::cancel(){ - - logging_debug("cancelling system queue"); - - // cause the thread to exit - { - // get the lock, when we got the lock the - // queue thread must be in itemsAvailable.wait() - boost::mutex::scoped_lock lock(queueMutex); - - // set the running indicator and signal to run on - // this will run the thread and quit it - running = false; - itemsAvailable.notify_all(); - } - - // wait until the thread has exited - logging_debug("joining system queue thread"); - queueThread->join(); - - // delete pending events - logging_debug("deleting pending system queue events"); - while( eventsQueue.size() > 0 ){ - eventsQueue.erase( eventsQueue.begin() ); - } - - // delete the thread, so that a subsuquent run() can be called - delete queueThread; - queueThread = NULL; -} - -bool SystemQueue::QueueThread::isEmpty(){ - boost::mutex::scoped_lock lock( queueMutex ); - return eventsQueue.empty(); -} - -void SystemQueue::QueueThread::insert( const SystemEvent& event, uint32_t delay ){ - - // if this is called from a module that is currently handling - // a thread (called from SystemQueue::onNextQueueItem), the - // thread is the same anyway and the mutex will be already - // aquired, otherwise we aquire it now - - boost::mutex::scoped_lock lock( queueMutex ); - - if ( delay > 0 ) - { - logging_debug("SystemQueue(" << this << ") : Schedule event in: " << delay << " ms; Events in queue (before insert): " << eventsQueue.size() ); - } - - eventsQueue.push_back( event ); - eventsQueue.back().scheduledTime = boost::posix_time::microsec_clock::local_time(); - eventsQueue.back().delayTime = delay; - eventsQueue.back().remainingDelay = delay; - - if ( delay > 0 ) - { - logging_debug("SystemQueue(" << this << ") : Events in queue (after insert): " << eventsQueue.size() ); - } - - onItemInserted( event ); - itemsAvailable.notify_all(); -} - -void SystemQueue::QueueThread::dropAll( const SystemEventListener* mlistener) { - boost::mutex::scoped_lock lock( queueMutex ); - - bool deleted; - do{ - deleted = false; - EventQueue::iterator i = eventsQueue.begin(); - EventQueue::iterator iend = eventsQueue.end(); - - for( ; i != iend; i++){ - if((*i).getListener() == mlistener){ - eventsQueue.erase(i); - deleted = true; - break; - } - } - }while(deleted); -} - -void SystemQueue::QueueThread::threadFunc( QueueThread* obj ) { - - boost::mutex::scoped_lock lock( obj->queueMutex ); - - while( obj->running ) { - - // wait until an item is in the queue or we are notified - // to quit the thread. in case the thread is about to - // quit, the queueThreadRunning variable will indicate - // this and cause the thread to exit - - while ( obj->running && obj->eventsQueue.empty() ){ - -// const boost::system_time duration = -// boost::get_system_time() + -// boost::posix_time::milliseconds(100); -// obj->itemsAvailable.timed_wait( lock, duration ); - - obj->itemsAvailable.wait( lock ); - } - - // - // work all the items that are currently in the queue - // - - while( obj->running && (!obj->eventsQueue.empty()) ) { - - // fetch the first item in the queue - // and deliver it to the queue handler - SystemEvent ev = obj->eventsQueue.front(); - - // XXX debugging the delay-scheduler.. - if ( ev.delayTime > 0 ) - logging_debug("SystemQueue(" << obj << ") : Events in queue (before execution): " << obj->eventsQueue.size()); - - obj->eventsQueue.erase( obj->eventsQueue.begin() ); - - // call the queue and this will - // call the actual event handler - obj->queueMutex.unlock(); - obj->onNextQueueItem( ev ); - obj->queueMutex.lock(); - - // XXX debugging the delay-scheduler.. - if ( ev.delayTime > 0 ) - logging_debug("SystemQueue(" << obj << ") : Remaining events in queue (after execution): " << obj->eventsQueue.size()); - - } // !obj->eventsQueue.empty() ) - } // while (obj->running) - - logging_debug("system queue exited"); -} - -void SystemQueue::QueueThread::enter(){ - queueMutex.lock(); -} - -void SystemQueue::QueueThread::leave(){ - queueMutex.unlock(); -} - - -//*************************************************************** - -SystemQueue::QueueThreadDirect::QueueThreadDirect(){ -} - -SystemQueue::QueueThreadDirect::~QueueThreadDirect(){ -} - -void SystemQueue::QueueThreadDirect::onItemInserted( const SystemEvent& event ){ - // do nothing here -} - -void SystemQueue::QueueThreadDirect::onNextQueueItem( const SystemEvent& event ){ - // directly deliver the item to the - event.getListener()->handleSystemEvent( event ); -} - -//*************************************************************** - -SystemQueue::QueueThreadDelay::QueueThreadDelay(QueueThread* _transferQueue) - : QueueThread( _transferQueue ), isSleeping( false ) { - - assert( _transferQueue != NULL ); -} - -SystemQueue::QueueThreadDelay::~QueueThreadDelay(){ -} - -void SystemQueue::QueueThreadDelay::onItemInserted( const SystemEvent& event ){ - - if( !isSleeping) - { - logging_warn("SystemQueue(" << this << ") : No, I'm not asleep!! New item inserted."); - return; // TODO Mario: shouldn't we sort anyway..? - } - - // break an existing sleep and - // remember the time that was actually slept for - // and change it for every event in the queue - - assert( !eventsQueue.empty()); - sleepCond.notify_all(); - - ptime sleepEnd = boost::posix_time::microsec_clock::local_time(); - boost::posix_time::time_duration duration = sleepEnd - sleepStart; - uint32_t sleptTime = duration.total_milliseconds(); - - EventQueue::iterator i = eventsQueue.begin(); - EventQueue::iterator iend = eventsQueue.end(); - - logging_debug("SystemQueue(" << this << ") : Adjusting remaining delays:"); - - // TODO Mario: What about the just inserted event..? - for( ; i != iend; i++ ) { - - if( sleptTime >= i->remainingDelay) - i->remainingDelay = 0; - else - { - i->remainingDelay -= sleptTime; - - // XXX Mario: Testcode, just to find a bug... - boost::posix_time::time_duration time_passed = sleepEnd - i->getScheduledTime(); - logging_debug("SystemQueue(" << this << ") : Total: " << - i->delayTime << ", remainingDelay: " << i->remainingDelay << - ", time already passed: " << time_passed.total_milliseconds() ); - } - - } // for( ; i != iend; i++ ) - - // now we have to reorder the events - // in the queue with respect to their remaining delay - // the SystemQueue::operator< takes care of the - // ordering with respect to the remaining delay - - std::sort( eventsQueue.begin(), eventsQueue.end() ); - -} - -void SystemQueue::QueueThreadDelay::onNextQueueItem( const SystemEvent& event ){ - - // sleeps will be cancelled in the - // onItemInserted function when a new - // event arrives during sleeping - - assert( !isSleeping ); - - // the given item is the one with the least - // amount of sleep time left. because all - // items are reordered in onItemInserted - - if( event.remainingDelay > 0 ) { - - const boost::system_time duration = - boost::get_system_time() + - boost::posix_time::milliseconds(event.remainingDelay); - - logging_debug("SystemQueue(" << this << ") : Sleeping for: " << event.remainingDelay << " ms"); - - { - boost::unique_lock lock( sleepMutex ); - - sleepStart = boost::posix_time::microsec_clock::local_time(); - isSleeping = true; - - sleepCond.timed_wait( lock, duration ); - - isSleeping = false; - } - - } // if( event.remainingDelay > 0 ) - - // if the sleep succeeded and was not - // interrupted by a new incoming item - // we can now deliver this event - - ptime sleepEnd = boost::posix_time::microsec_clock::local_time(); - boost::posix_time::time_duration duration = sleepEnd - sleepStart; - uint32_t sleptTime = duration.total_milliseconds(); - - logging_debug("SystemQueue(" << this << ") : Slept for: " << sleptTime << " ms; until: " << sleepEnd); - - // TODO MARIO: find the bug that loses events... - if (event.remainingDelay <= sleptTime) - { - logging_debug("SystemQueue(" << this << ") : Transferring scheduled event into the direct queue. Scheduled time: " << event.getScheduledTime() ); - transferQueue->insert( event, 0 ); - } - else - { - logging_warn("SystemQueue(" << this << ") : Scheduled event lost!! :-( (Sleep should have been " << event.remainingDelay - sleptTime << " ms longer..)"); - logging_debug("SystemQueue(" << this << ") : Total delay: " << event.delayTime << "; remaining delay: " << event.remainingDelay); - -// throw std::logic_error("Scheduled event lost!! :-("); - } -} - -#endif // #ifndef UNDERLAY_OMNET - -//*************************************************************** +// XXX code from old system queue +// void SystemQueue::QueueThread::enter(){ +// queueMutex.lock(); +// } +// +// void SystemQueue::QueueThread::leave(){ +// queueMutex.unlock(); +// } }} // spovnet, common Index: source/ariba/utility/system/SystemQueue.h =================================================================== --- source/ariba/utility/system/SystemQueue.h (revision 12763) +++ source/ariba/utility/system/SystemQueue.h (revision 12764) @@ -174,9 +174,6 @@ typedef list EventQueue; typedef std::priority_queue, + std::vector, // [ TODO is vector the best underlay? ] std::greater > PriorityEventQueue; -// typedef std::priority_queue PriorityEventQueue; -// TODO is vector the best underlay? - //******************************************************** @@ -224,4 +221,5 @@ volatile bool running; volatile bool aborted; + volatile bool unclean; }; // class QueueThread