Index: source/ariba/utility/system/SystemEvent.h =================================================================== --- source/ariba/utility/system/SystemEvent.h (revision 12762) +++ source/ariba/utility/system/SystemEvent.h (revision 12763) @@ -86,4 +86,5 @@ } + // FIXME AKTUELL MARIO: copy deadline (etc?) !!! inline SystemEvent(const SystemEvent& copy) { this->scheduledTime = copy.scheduledTime; @@ -133,7 +134,13 @@ } - inline bool operator<(const SystemEvent& right) const { - return remainingDelay < right.remainingDelay; + bool operator<(const SystemEvent& right) const + { + return this->deadline < right.deadline; } + + bool operator>(const SystemEvent& right) const + { + return this->deadline > right.deadline; + } }; Index: source/ariba/utility/system/SystemQueue.cpp =================================================================== --- source/ariba/utility/system/SystemQueue.cpp (revision 12762) +++ source/ariba/utility/system/SystemQueue.cpp (revision 12763) @@ -110,6 +110,9 @@ // wait till actually completes // (should be fast, but the current event is allowed to finish) - logging_debug("/// ... joining SysQ thread"); - sysq_thread->join(); + if ( sysq_thread ) + { + logging_debug("/// ... joining SysQ thread"); + sysq_thread->join(); + } // delete thread object @@ -154,6 +157,6 @@ SystemQueue::QueueThread::QueueThread() : - now( not_a_date_time ), - next_deadline( not_a_date_time ), +// now( not_a_date_time ), +// next_deadline( not_a_date_time ), processing_event( false ), running( false ), @@ -198,7 +201,4 @@ void SystemQueue::QueueThread::run_immediate_event() { - // measure execution time - ptime start_time = get_clock(); - // get next event and process it if ( ! immediateEventsQ.empty() ) @@ -206,4 +206,5 @@ scoped_ptr currently_processed_event; + /* dequeue event */ // SYNCHRONIZED { @@ -212,26 +213,39 @@ this->processing_event = true; - // dequeue first event + // * dequeue first event * currently_processed_event.reset( new SystemEvent(immediateEventsQ.front()) ); // copy immediateEventsQ.pop_front(); } + /* dispatch event */ logging_debug("/// SysQ: dispatching event"); + // measure execution time (1/2) + ptime start_time = get_clock(); + + // * dispatch event * currently_processed_event->getListener()->handleSystemEvent( *currently_processed_event ); + + // measure execution time (2/2) + time_duration execution_time = get_clock() - start_time; + + // DEBUG OUTPUT: warning when execution takes too much time + // [ TODO how long is "too much time"? ] + if ( execution_time.total_milliseconds() > 50 ) + { + logging_info("WARNING: Last event took " << execution_time.total_milliseconds() << " to complete."); + } + + /* [ TODO ] + * + * - we could also measure how long an event has been waiting in the queue before it's dispatched + * (in order to detect overload) + * + * - especially for timed events, the displacement could be calculated + * (and, e.g., put in relation with the actual intended sleep time) + */ } - // measure execution time - now = get_clock(); // NOTE: this is reused in check_timed_queue(); - time_duration execution_time = now - start_time; - this->processing_event = false; - - // DEBUG OUTPUT: warning when execution takes too much time - // [ TODO how long is "too much time"? ] - if ( execution_time.total_milliseconds() > 50 ) - { - logging_info("WARNING: Last event took " << execution_time.total_milliseconds() << " to complete."); - } } @@ -241,10 +255,5 @@ scoped_lock lock( queue_mutex ); - - // NOTE: this->now was just set by this->run_immediate_event() - - // reset next deadline - this->next_deadline = boost::date_time::not_a_date_time; - + ptime now = get_clock(); bool not_expired_events_reached = false; @@ -252,7 +261,7 @@ while ( ! timedEventsQ.empty() && ! not_expired_events_reached ) { - SystemEvent& ev = timedEventsQ.front(); - - time_duration remaining_sleep_time = ev.deadline - this->now; + const SystemEvent& ev = timedEventsQ.top(); + + time_duration remaining_sleep_time = ev.deadline - now; // BRANCH: deadline reached @@ -261,12 +270,9 @@ // move to immediateEventsQ immediateEventsQ.push_back(ev); - timedEventsQ.pop_front(); + timedEventsQ.pop(); } // BRANCH: deadline not reached else { - // store time for next sleep - this->next_deadline = ev.deadline; - // okay, that's all for now. not_expired_events_reached = true; @@ -277,8 +283,9 @@ void SystemQueue::QueueThread::wait_for_next_deadline() { + // SYNCHRONIZED + boost::mutex::scoped_lock lock(queue_mutex); + if ( immediateEventsQ.empty() ) { - boost::mutex::scoped_lock lock(queue_mutex); - // don't sleep when the SystemQueue is not already canceled if ( aborted ) @@ -287,5 +294,5 @@ // BRANCH: no timed events: sleep "forever" (until new events are scheduled) - if ( this->next_deadline.is_not_a_date_time() ) + if ( timedEventsQ.empty() ) { logging_debug("/// SysQ is going to sleep."); @@ -296,9 +303,12 @@ else { - logging_debug("/// SysQ is going to sleep for " - << ( next_deadline - get_clock() ).total_milliseconds() - << "ms."); - - this->system_queue_idle.timed_wait( lock, next_deadline ); + logging_debug( "/// SysQ is going to sleep for " + << ( timedEventsQ.top().deadline - get_clock() ).total_milliseconds() + << "ms. Deadline: " + << timedEventsQ.top().deadline + << ", Clock: " + << get_clock() ); + + this->system_queue_idle.timed_wait( lock, timedEventsQ.top().deadline ); } } @@ -309,5 +319,6 @@ ptime SystemQueue::QueueThread::get_clock() { - return microsec_clock::universal_time(); +// return microsec_clock::universal_time(); + return microsec_clock::local_time(); } @@ -352,4 +363,26 @@ immediateEventsQ.push_back(event); } + // BRANCH: timed event + else + { + event.deadline = event.scheduledTime + boost::posix_time::milliseconds(delay); + event.delayTime = delay; // XXX I think this is no longer needed.. + + // XXX debug + 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/ + */ + } } @@ -414,6 +447,6 @@ // wait until the thread has exited - logging_debug("joining system queue thread"); - queueThread->join(); + logging_debug("joining system queue thread"); + queueThread->join(); // delete pending events Index: source/ariba/utility/system/SystemQueue.h =================================================================== --- source/ariba/utility/system/SystemQueue.h (revision 12762) +++ source/ariba/utility/system/SystemQueue.h (revision 12763) @@ -40,10 +40,14 @@ #define SYSTEMQUEUE_H_ -// #include -#include -#include #include "SystemEvent.h" #include "SystemEventListener.h" #include "ariba/utility/logging/Logging.h" + +#include +#include +#include +#include // std::priority_queue +#include // std::greater + #include #include @@ -169,4 +173,10 @@ typedef list EventQueue; +typedef std::priority_queue, + std::greater > PriorityEventQueue; +// typedef std::priority_queue PriorityEventQueue; +// TODO is vector the best underlay? + //******************************************************** @@ -205,9 +215,6 @@ private: EventQueue immediateEventsQ; - EventQueue timedEventsQ; - - ptime now; - ptime next_deadline; - + PriorityEventQueue timedEventsQ; + boost::condition_variable system_queue_idle; boost::mutex queue_mutex;