Index: source/ariba/utility/system/SystemQueue.cpp
===================================================================
--- source/ariba/utility/system/SystemQueue.cpp	(revision 12775)
+++ source/ariba/utility/system/SystemQueue.cpp	(revision 2378)
@@ -1,3 +1,3 @@
-// [License]
+// [Licence]
 // The Ariba-Underlay Copyright
 //
@@ -37,165 +37,83 @@
 
 #include "SystemQueue.h"
-#include <ariba/utility/misc/sha1.h>
-#include <stdexcept>
-
-// 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;
-    
 
 use_logging_cpp(SystemQueue);
 
-SystemQueue::SystemQueue() :
-    SysQ( new QueueThread() )
+SystemQueue::SystemQueue()
+#ifndef UNDERLAY_OMNET
+ : delayScheduler( &directScheduler ), systemQueueRunning( false )
+#endif
 {
-	logging_debug("Creating SystemQueue at: " << this);
-}
-
-SystemQueue::~SystemQueue()
-{
-}
-
-
-SystemQueue& SystemQueue::instance()
-{
-	static SystemQueue _inst;
-	return _inst;
-}
-
-
-void SystemQueue::scheduleEvent( const SystemEvent& event, uint32_t delay )
-{
-//     assert ( SysQ->running );  // should we really enforce this?
-    if ( ! SysQ->running )
-    {
-        logging_debug("/// WARNING: The SystemQueue is NOT RUNNING!");
-    }
-    
-    // copy
-    SystemEvent ev(event);
-    
-    SysQ->insert(ev, delay);
-}
-
-// maps to function call internally to the Event-system
-void SystemQueue::scheduleCall( const boost::function0<void>& function, uint32_t delay)
-{
-    // copy function object
-    boost::function0<void>* function_ptr = new boost::function0<void>();
-    (*function_ptr) = function; 
-
-    // schedule special call-event
-    scheduleEvent( SystemEvent(&internal_function_caller, SystemEventType::DEFAULT, function_ptr), delay );
-}
-
-
-
-
-void SystemQueue::run()
-{
-    // TODO should these be assertion or exceptions..? (exceptions => unit test)
-    assert ( ! SysQ->running );
-    assert ( ! SysQ->unclean );
-    
-    SysQ->running = true;
-    
-    // start thread
-    sysq_thread.reset( new boost::thread(boost::ref(*SysQ)) );
-}
-
-void SystemQueue::cancel()
-{
-    // CHECK: this function must not be called from within a SystemQueue-Event
-    if ( am_I_in_the_SysQ_thread() )
-//     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();
-    
-    // wait till actually completes
-    //   (should be fast, but the current event is allowed to finish)
-    if ( sysq_thread )
-    {
-        logging_debug("/// ... joining SysQ thread");
-        sysq_thread->join();
-    }
-    
-    // delete thread object
-    sysq_thread.reset();
-    
-    assert ( ! SysQ->isRunning() );
-
-    
-    // clean up and respawn
-    logging_debug("/// respawning SysQ");
-    SysQ.reset( new QueueThread() );
-}
-
-
-void SystemQueue::leave()
-{
-    // signal SysQ to quit (and abort queued events)
-    SysQ->cancel();
-}
-
-void SystemQueue::join()
-{
-    if ( sysq_thread )
-    {
-        logging_debug("/// ... joining SysQ thread");
-        sysq_thread->join();
-    }
-}
-
-
-
-void SystemQueue::dropAll( const SystemEventListener* mlistener)
-{
-//  TODO
-// 	directScheduler.dropAll(mlistener);
-// 	delayScheduler.dropAll(mlistener);
-}
-
-bool SystemQueue::isEmpty()
-{
-    return SysQ->isEmpty();
-}
-
-bool SystemQueue::isRunning()
-{
-    return SysQ->isRunning();
-}
-
-bool SystemQueue::am_I_in_the_SysQ_thread()
-{
-    return sysq_thread && boost::this_thread::get_id() == sysq_thread->get_id();
-}
-
-
-//********************************************************
-
-
-/// constructor
-SystemQueue::QueueThread::QueueThread() : 
-    processing_event( false ),
-    running( false ),
-    aborted( false ),
-    unclean( false )
-{
+}
+
+SystemQueue::~SystemQueue() {
+}
+
+void SystemQueue::scheduleEvent( const SystemEvent& event, uint32_t delay ) {
+#ifndef UNDERLAY_OMNET
+	if( delay == 0 ) directScheduler.insert( event, delay );
+	else delayScheduler.insert ( event, delay );
+#else
+	Enter_Method_Silent();
+	cMessage* msg = new cMessage();
+	msg->setContextPointer( new SystemEvent(event) );
+
+	if( delay == 0 )
+		cSimpleModule::scheduleAt( cSimpleModule::simTime(), msg );
+	else
+		cSimpleModule::scheduleAt( cSimpleModule::simTime()+((double)delay/1000.0), msg );
+#endif
+}
+
+#ifdef UNDERLAY_OMNET
+void SystemQueue::handleMessage(cMessage* msg){
+	SystemEvent* event = (SystemEvent*)msg->contextPointer();
+
+	event->listener->handleSystemEvent( *event );
+
+	delete event; delete msg;
+}
+#endif
+
+void SystemQueue::run() {
+#ifndef UNDERLAY_OMNET
+	systemQueueRunning = true;
+	directScheduler.run();
+	delayScheduler.run();
+#endif
+}
+
+void SystemQueue::cancel() {
+#ifndef UNDERLAY_OMNET
+	systemQueueRunning = false;
+	directScheduler.cancel();
+	delayScheduler.cancel();
+#endif
+}
+
+bool SystemQueue::isEmpty() {
+#ifndef UNDERLAY_OMNET
+	return directScheduler.isEmpty() || delayScheduler.isEmpty();
+#else
+	return false;
+#endif
+}
+
+bool SystemQueue::isRunning() {
+#ifndef UNDERLAY_OMNET
+	return systemQueueRunning;
+#else
+	return true;
+#endif
+}
+
+//***************************************************************
+#ifndef UNDERLAY_OMNET
+
+SystemQueue::QueueThread::QueueThread(QueueThread* _transferQueue)
+	: running( false ), transferQueue( _transferQueue ) {
 }
 
@@ -203,248 +121,199 @@
 }
 
-
-void SystemQueue::QueueThread::operator()()
-{
-    logging_debug( "/// 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();
-    }
-    
-    logging_debug( "/// SysQ thread is quitting." );
-    
-    unclean = true;
-    running = false;
-}
-
-
-
-
-
-/// main loop functions
-
-void SystemQueue::QueueThread::run_immediate_event()
-{
-    // get next event and process it
-    if ( ! immediateEventsQ.empty() )
-    {
-        scoped_ptr<SystemEvent> currently_processed_event;
-        
-        /* dequeue event */
-        // SYNCHRONIZED
-        {
-            scoped_lock lock( queue_mutex );
-            
-            this->processing_event = true;
-            
-            // * 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
-        //   [ TODOx how long is "too much time"? ]
-        if ( execution_time.total_milliseconds() > 50 )
-        {
-            logging_info("WARNING: Last event took " << execution_time.total_milliseconds() << " ms to complete.");
-        }
-
-        /*  [ TODOx ]
-        * 
-        *  - 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)
-        */
-    }
-    
-    this->processing_event = false;
-}
-
-void SystemQueue::QueueThread::check_timed_queue()
-{
-    // this whole function is SYNCHRONIZED
-    scoped_lock lock( queue_mutex );
-
-    ptime now = get_clock();
-    bool not_expired_events_reached = false;
-    
-    // move all expired events into the immediateEventsQ
-    while ( ! timedEventsQ.empty() && ! not_expired_events_reached )
-    {
-        const SystemEvent& ev = timedEventsQ.top();
-        
-        time_duration remaining_sleep_time = ev.deadline - now;
-        
-        // BRANCH: deadline reached
-        if ( remaining_sleep_time.is_negative() )
-        {
-            // move to immediateEventsQ
-            immediateEventsQ.push_back(ev);
-            timedEventsQ.pop();
-        }
-        // BRANCH: deadline not reached
-        else
-        {
-            // okay, that's all for now.
-            not_expired_events_reached = true;
-        }
-    } // while
-}
-
-void SystemQueue::QueueThread::wait_for_next_deadline()
-{
-    // SYNCHRONIZED
-    boost::mutex::scoped_lock lock(queue_mutex);
-
-    if ( immediateEventsQ.empty() )
-    {
-         // 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 ( timedEventsQ.empty() )
-        {
-//             logging_debug("/// SysQ is going to sleep.");
-            
-            this->system_queue_idle.wait( lock );
-        }
-        // BRANCH: sleep till next timed event
-        else
-        {
-//             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 );
-        }
-    }
-}
-
-
-/// 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;
-    }
-    
-    logging_debug("/// 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);
-        }
-        // BRANCH: timed event
-        else
-        {
-            event.deadline = event.scheduledTime + boost::posix_time::milliseconds(delay);
-            event.delayTime = delay;  // ( I think this is no longer needed.. )
-            
-//             // debug output
-//             logging_debug("/// inserting timed event, due at: " << event.deadline << " (in " << delay << " ms)");
-            
-            timedEventsQ.push(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)
-}
-
-
-bool SystemQueue::QueueThread::isEmpty()
-{
-    // SYNCHRONIZED
-    scoped_lock lock( queue_mutex );
-    
-    return immediateEventsQ.empty() && timedEventsQ.empty() && ! processing_event;
-}
-
-
-
-// FIXME
-void SystemQueue::enterMethod()
-{
-    assert( false );
-}
-void SystemQueue::leaveMethod()
-{
-    assert( false );
-}
-
-
-
-// XXX code from old system queue
-// void SystemQueue::QueueThread::enter(){
-// 	queueMutex.lock();
-// }
-// 
-// void SystemQueue::QueueThread::leave(){
-// 	queueMutex.unlock();
-// }
+void SystemQueue::QueueThread::run(){
+	running = true;
+	queueThread = new boost::thread(
+		boost::bind(&QueueThread::threadFunc, this) );
+}
+
+void SystemQueue::QueueThread::cancel(){
+
+	// cause the thread to exit
+	running = false;
+	itemsAvailable.notify_all();
+
+	// wait that the thread has exited
+	// and delete it, so that a subsuquent run() can be called
+
+	queueThread->join();
+	delete queueThread;
+	queueThread = NULL;
+
+	while( eventsQueue.size() > 0 ){
+		eventsQueue.erase( eventsQueue.begin() );
+	}
+}
+
+bool SystemQueue::QueueThread::isEmpty(){
+	boost::mutex::scoped_lock lock( queueMutex );
+	return eventsQueue.empty();
+}
+
+void SystemQueue::QueueThread::insert( const SystemEvent& event, uint32_t delay ){
+	{
+		boost::mutex::scoped_lock lock( queueMutex );
+
+		eventsQueue.push_back( event );
+		eventsQueue.back().scheduledTime = boost::posix_time::microsec_clock::local_time();
+		eventsQueue.back().delayTime = delay;
+		eventsQueue.back().remainingDelay = delay;
+
+		onItemInserted( event );
+		itemsAvailable.notify_all();
+	}
+}
+
+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->eventsQueue.empty() && obj->running ){
+
+			const boost::system_time duration =
+					boost::get_system_time() +
+					boost::posix_time::milliseconds(40);
+
+			obj->itemsAvailable.wait( lock );
+//			obj->itemsAvailable.timed_wait( lock, duration );
+		}
+
+		// work all the items that are currently in the queue
+
+		while( !obj->eventsQueue.empty() ) {
+
+			// fetch the first item in the queue and deliver it to the
+			// queue handler
+			SystemEvent ev = obj->eventsQueue.front();
+			obj->eventsQueue.erase( obj->eventsQueue.begin() );
+
+			{
+				// unlock the queue to that an event handler
+				// can insert new items into the queue
+//				obj->queueMutex.unlock();
+				obj->onNextQueueItem( ev );
+//				obj->queueMutex.lock();
+			}
+
+		} // !obj->eventsQueue.empty() )
+	} // while (obj->running)
+}
+
+//***************************************************************
+
+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.listener->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) return;
+
+	// 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();
+
+	for( ; i != iend; i++ ) {
+
+		if( sleptTime >= i->remainingDelay)
+			i->remainingDelay = 0;
+		else
+			i->remainingDelay -= sleptTime;
+
+	} // 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);
+
+		{
+			boost::unique_lock<boost::mutex> 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();
+
+	if (event.remainingDelay <= sleptTime)
+		transferQueue->insert( event, 0 );
+}
+
+#endif // #ifndef UNDERLAY_OMNET
+
+//***************************************************************
 
 }} // spovnet, common
