Index: source/ariba/utility/logging/Logging.h =================================================================== --- source/ariba/utility/logging/Logging.h (revision 12760) +++ source/ariba/utility/logging/Logging.h (revision 12761) @@ -96,6 +96,6 @@ #define logging_stdout(x) std::cout << x << std::endl; - static int __loglevel__ = 2; //default is info -// static int __loglevel__ = 1; // XXX use higher log level +// static int __loglevel__ = 2; //default is info + static int __loglevel__ = 1; // XXX use higher log level #define logging_trace(x) { logging_stdout(x); } Index: source/ariba/utility/system/SystemEvent.h =================================================================== --- source/ariba/utility/system/SystemEvent.h (revision 12760) +++ source/ariba/utility/system/SystemEvent.h (revision 12761) @@ -64,4 +64,5 @@ // private class that accesses these variables. Therefore they are public for now. ptime scheduledTime; //< time the event was originally given to the queue + ptime deadline; //< time when the event should be executed TODO Mario uint32_t delayTime; //< time the event is scheduled at, or 0 if it is to be fired immediately uint32_t remainingDelay; //< remaining delay time for sleeping Index: source/ariba/utility/system/SystemQueue.cpp =================================================================== --- source/ariba/utility/system/SystemQueue.cpp (revision 12761) +++ source/ariba/utility/system/SystemQueue.cpp (revision 12761) @@ -0,0 +1,679 @@ +// [License] +// The Ariba-Underlay Copyright +// +// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH) +// +// Institute of Telematics +// UniversitÀt Karlsruhe (TH) +// Zirkel 2, 76128 Karlsruhe +// Germany +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// 1. Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// 2. Redistributions in binary form must reproduce the above copyright +// notice, this list of conditions and the following disclaimer in the +// documentation and/or other materials provided with the distribution. +// +// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR +// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// +// The views and conclusions contained in the software and documentation +// are those of the authors and should not be interpreted as representing +// official policies, either expressed or implied, of the Institute of +// Telematics. + +#include "SystemQueue.h" +#include +#include + +namespace ariba { +namespace utility { + + + using boost::posix_time::microsec_clock; + using boost::posix_time::time_duration; +// using boost::mutex::scoped_lock; + typedef boost::mutex::scoped_lock scoped_lock; + using boost::date_time::not_a_date_time; + using boost::scoped_ptr; + + +use_logging_cpp(SystemQueue); + +SystemQueue::SystemQueue() +{ + logging_info("Creating SystemQueue at: " << this); +} + +SystemQueue::~SystemQueue() +{ +} + + +SystemQueue& SystemQueue::instance() +{ + static SystemQueue _inst; + return _inst; +} + + +void SystemQueue::scheduleEvent( const SystemEvent& event, uint32_t delay ) +{ + // copy + SystemEvent ev(event); + + SysQ.insert(ev, delay); +} + +// maps to function call internally to the Event-system +void SystemQueue::scheduleCall( const boost::function0& function, uint32_t delay) +{ + // copy function object + boost::function0* function_ptr = new boost::function0(); + (*function_ptr) = function; + + // schedule special call-event + scheduleEvent( SystemEvent(&internal_function_caller, SystemEventType::DEFAULT, function_ptr), delay ); +} + + + + +void SystemQueue::run() +{ + assert ( ! SysQ.running ); + + SysQ.running = true; + + // start thread + sysq_thread.reset( new boost::thread(boost::ref(SysQ)) ); +} + +void SystemQueue::cancel() +{ + // signal SysQ to quit (and abort queued events) + SysQ.cancel(); + + // wait till actually completes + // (should be fast, but the current event is allowed to finish) + logging_info("/// joining system queue thread"); + sysq_thread->join(); + + // delete thread object + sysq_thread.reset(); + + assert ( ! SysQ.isRunning() ); + + + // TODO FIXME gtest reuses the singleton... :-/ + // maybe delete the SysQ object here..? (scoped_ptr...) + SysQ.aborted = false; // XXX hotfix.. +} + +void SystemQueue::dropAll( const SystemEventListener* mlistener) +{ +// XXX +// directScheduler.dropAll(mlistener); +// delayScheduler.dropAll(mlistener); +} + +bool SystemQueue::isEmpty() +{ +// XXX +// return directScheduler.isEmpty() || delayScheduler.isEmpty(); + + return true; +} + +bool SystemQueue::isRunning() +{ + return SysQ.isRunning(); +} + +// XXX +// void SystemQueue::enterMethod(){ +// // TODO: omnet case and delay scheduler +// directScheduler.enter(); +// } +// +// void SystemQueue::leaveMethod(){ +// // TODO: omnet case and delay scheduler +// directScheduler.leave(); +// } + + +SystemQueue::QueueThread::QueueThread() : + now( not_a_date_time ), + next_deadline( not_a_date_time ), + running( false ), + aborted( false ) +{ +} + +SystemQueue::QueueThread::~QueueThread(){ +} + + +void SystemQueue::QueueThread::operator()() +{ + // XXX debug + logging_info( "/// SysQ thread is alive." ); + + assert( running ); // this is set before the thread starts + + // main loop + while ( ! aborted ) + { + // run next immediate event (only one) + run_immediate_event(); + + // maintain timed events (move to immediateEventsQ, when deadline expired) + check_timed_queue(); + + // wait for next deadline (if no immediate events pending) + wait_for_next_deadline(); + } + + // XXX debug + logging_info( "/// SysQ thread is quitting." ); + + running = false; +} + + + + + +/// main loop functions + +void SystemQueue::QueueThread::run_immediate_event() +{ + // measure execution time + ptime start_time = get_clock(); + + // get next event and process it + if ( ! immediateEventsQ.empty() ) + { + scoped_ptr currently_processed_event; + + // SYNCHRONIZED + { + scoped_lock lock( queue_mutex ); + + // dequeue first event + currently_processed_event.reset( new SystemEvent(immediateEventsQ.front()) ); // copy + immediateEventsQ.pop_front(); + } + + // XXX debug + logging_info("/// SysQ: dispatching event"); + + currently_processed_event->getListener()->handleSystemEvent( *currently_processed_event ); + } + + // measure execution time + now = get_clock(); // NOTE: this is reused in check_timed_queue(); + time_duration execution_time = now - 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_debug("WARNING: Last event took " << execution_time.total_milliseconds() << " to complete."); + } +} + +void SystemQueue::QueueThread::check_timed_queue() +{ + // this whole function is SYNCHRONIZED + 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; + + bool not_expired_events_reached = false; + + // move all expired events into the immediateEventsQ + while ( ! timedEventsQ.empty() && ! not_expired_events_reached ) + { + SystemEvent& ev = timedEventsQ.front(); + + time_duration remaining_sleep_time = ev.deadline - this->now; + + // BRANCH: deadline reached + if ( remaining_sleep_time.is_negative() ) + { + // move to immediateEventsQ + immediateEventsQ.push_back(ev); + timedEventsQ.pop_front(); + } + // 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; + } + } // while +} + +void SystemQueue::QueueThread::wait_for_next_deadline() +{ + if ( immediateEventsQ.empty() ) + { + boost::mutex::scoped_lock lock(queue_mutex); + + // don't sleep when the SystemQueue is not already canceled + if ( aborted ) + return; + + + // BRANCH: no timed events: sleep "forever" (until new events are scheduled) + if ( this->next_deadline.is_not_a_date_time() ) + { + // XXX debug output + logging_info("/// SysQ is going to sleep (forever..)."); + + this->system_queue_idle.wait( lock ); + } + // BRANCH: sleep till next timed event + else + { + // XXX debug output + time_duration sleep_time = next_deadline - get_clock(); + logging_info("/// SysQ is going to sleep for " << sleep_time.total_milliseconds() << "ms."); + + this->system_queue_idle.timed_wait( lock, next_deadline ); + } + } +} + + +/// uniform clock interface +ptime SystemQueue::QueueThread::get_clock() +{ + return microsec_clock::universal_time(); +} + + + +/// external interface + +bool SystemQueue::QueueThread::isRunning() +{ + return running; +} + + +void SystemQueue::QueueThread::cancel() +{ + logging_debug("/// cancelling system queue"); + + // SYNCHRONIZED + { + scoped_lock lock(queue_mutex); + aborted = true; + } + + // XXX debug output + logging_info("/// SysQ: " << immediateEventsQ.size() << " immediate event(s) + " + << timedEventsQ.size() << " timed event(s) left."); + + system_queue_idle.notify_all(); +} + + +void SystemQueue::QueueThread::insert( SystemEvent& event, uint32_t delay ) +{ + event.scheduledTime = get_clock(); + + // SYNCHRONIZED + { + scoped_lock lock( queue_mutex ); + + // BRANCH: immediate event + if ( delay == 0 ) + { + immediateEventsQ.push_back(event); + } + } + + // wake SysQ thread + system_queue_idle.notify_one(); // NOTE: there is only one thread + // (so it doesn't matter whether to call notify_one, or notify_all) +} + + + +// FIXME +void SystemQueue::enterMethod() +{ + assert( false ); +} +void SystemQueue::leaveMethod() +{ + assert( false ); +} + + + +//*************************************************************** +//// 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 + +//*************************************************************** + +}} // spovnet, common Index: source/ariba/utility/system/SystemQueue.h =================================================================== --- source/ariba/utility/system/SystemQueue.h (revision 12761) +++ source/ariba/utility/system/SystemQueue.h (revision 12761) @@ -0,0 +1,255 @@ +// [License] +// The Ariba-Underlay Copyright +// +// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH) +// +// Institute of Telematics +// UniversitÀt Karlsruhe (TH) +// Zirkel 2, 76128 Karlsruhe +// Germany +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// 1. Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// 2. Redistributions in binary form must reproduce the above copyright +// notice, this list of conditions and the following disclaimer in the +// documentation and/or other materials provided with the distribution. +// +// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR +// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// +// The views and conclusions contained in the software and documentation +// are those of the authors and should not be interpreted as representing +// official policies, either expressed or implied, of the Institute of +// Telematics. +// [License] + +#ifndef SYSTEMQUEUE_H_ +#define SYSTEMQUEUE_H_ + +// #include +#include +#include +#include "SystemEvent.h" +#include "SystemEventListener.h" +#include "ariba/utility/logging/Logging.h" +#include +#include +#include + +#include +#include +#include +#include +#include + +#include + +namespace ariba { +namespace utility { + +using std::list; +using boost::posix_time::ptime; + + +/** + * This class implements a simple system event queue to allow + * a simulation of cooperative multitasking. It also allows + * events to be scheduled from other tasks. This allows + * dispatching asynchronous tasks. + * + * @author Christoph Mayer, Sebastian Mies + */ +/** XXX Mario Hock -- reworking the entire module **/ + + +class SystemQueue : private boost::noncopyable +{ + use_logging_h(SystemQueue); + friend class EnterMethod; +public: + /** + * Get the SystemQueue singleton instance. + */ + static SystemQueue& instance(); + + /** + * This methods schedules a given event. + * + * @param The event to be scheduled + * @param The delay in milli-seconds + */ + void scheduleEvent( const SystemEvent& event, uint32_t delay = 0 ); + + /** + * This method schedules a function call in the SystemQueue. + * (Like scheduleEvent, but to be used with boost::bind.) + * + * @param function: The function to be called [void function()] + * @param The delay in milli-seconds + */ + void scheduleCall( const boost::function0& function, uint32_t delay = 0 ); + + /** + * Starts the processing and waiting for events. + * Use cancel() to end system queue processing and + * isEmpty() to check wheter the queue is empty. + */ + void run(); + + /** + * Cancels the system queue and ends the processing after the + * currently processed event is processed. + * + * This method is thread-safe. + */ + void cancel(); + + /** + * Drop all queued events for that listener + */ + void dropAll( const SystemEventListener* mlistener); + + /** + * Check wheter this queue has items or not. + * + * @return True, if this queue is empty. + */ + bool isEmpty(); + + /** + * Is the system queue already started and running? + * + * @return True, if the system queue is running. + */ + bool isRunning(); + +protected: + + /** + * Aqcuire the mutex + */ + void enterMethod(); + + /** + * Leave the mutex + */ + void leaveMethod(); + + /** + * Constructs a system queue. + */ + SystemQueue(); + + /** + * Destroys the system queue. Beware that all events + * are canceled + */ + ~SystemQueue(); + + +/** + * inner class of class SystemQueue: + * + * QueueThread -- the class the does the actual work + */ +private: + +typedef list EventQueue; + + //******************************************************** + + class QueueThread + { + friend class SystemQueue; + + public: + QueueThread(); + virtual ~QueueThread(); + + /// main loop -- called from boost::thread + void operator()(); + +// void run(); + void cancel(); + bool isEmpty(); + void insert( SystemEvent& event, uint32_t delay ); + void enter(); + void leave(); + void dropAll( const SystemEventListener* mlistener); + bool isRunning(); + + private: + + /// main loop functions + void run_immediate_event(); + void check_timed_queue(); + void wait_for_next_deadline(); + + /// makes sure that always the same clock is used + ptime get_clock(); + + + private: + EventQueue immediateEventsQ; + EventQueue timedEventsQ; +// boost::mutex queueMutex; + +// SystemEvent& currently_processed_event; + + ptime now; + ptime next_deadline; + + boost::condition_variable system_queue_idle; + boost::mutex queue_mutex; + +// boost::condition_variable itemsAvailable; DEPRECATED + volatile bool running; + volatile bool aborted; + }; // class QueueThread + + +/// inner class of class SystemQueue +private: + /** + * This inner class handles the function-call events. + * @see SystemQueue::scheduleCall + */ + class FunctionCaller : public SystemEventListener + { + void handleSystemEvent(const SystemEvent& event) + { + boost::function0* function_ptr = event.getData< boost::function0 >(); + (*function_ptr)(); + delete function_ptr; + } + }; + + FunctionCaller internal_function_caller; + + + + /// member variables of class SystemQueue +private: + QueueThread SysQ; + boost::scoped_ptr sysq_thread; + +// volatile bool systemQueueRunning; + +}; // class SystemQueue + +}} // spovnet, common + +#endif /* SYSTEMQUEUE_H_ */