Changeset 12763

Timestamp:

Mar 13, 2014, 7:30:17 PM (11 years ago)

Author:

hock@…

Message:

priority queue (but not tested)

--> FIXME in SystemEvent.h (has to be fixed first!)

Files:

• source/ariba/utility/system/SystemEvent.h (modified) (2 diffs)
• source/ariba/utility/system/SystemQueue.cpp (modified) (14 diffs)
• source/ariba/utility/system/SystemQueue.h (modified) (3 diffs)
• tests/SystemQueue-tests.cc (modified) (2 diffs)

source/ariba/utility/system/SystemEvent.h

@@ -86 +86 @@
         }
 
+        // FIXME AKTUELL MARIO: copy deadline (etc?) !!!
         inline SystemEvent(const SystemEvent& copy) {
                 this->scheduledTime = copy.scheduledTime;
@@ -133 +134 @@
         }
 
-        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;
+    }
 
 };

source/ariba/utility/system/SystemQueue.cpp

@@ -110 +110 @@
     // 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 +157 @@
 
 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 +201 @@
 void SystemQueue::QueueThread::run_immediate_event()
 {
-    // measure execution time
-    ptime start_time = get_clock();
-    
     // get next event and process it
     if ( ! immediateEventsQ.empty() )
@@ -206 +206 @@
         scoped_ptr<SystemEvent> currently_processed_event;
         
+        /* dequeue event */
         // SYNCHRONIZED
         {
@@ -212 +213 @@
             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 +255 @@
     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 +261 @@
     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 +270 @@
             // 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 +283 @@
 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 +294 @@
         
         // 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 +303 @@
         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 +319 @@
 ptime SystemQueue::QueueThread::get_clock()
 {
-    return microsec_clock::universal_time();
+//     return microsec_clock::universal_time();
+    return microsec_clock::local_time();
 }
 
@@ -352 +363 @@
             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 +447 @@
 
         // 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

source/ariba/utility/system/SystemQueue.h

@@ -40 +40 @@
 #define SYSTEMQUEUE_H_
 
-// #include <vector>
-#include <list>
-#include <cassert>
 #include "SystemEvent.h"
 #include "SystemEventListener.h"
 #include "ariba/utility/logging/Logging.h"
+
+#include <cassert>
+#include <list>
+#include <vector>
+#include <queue>          // std::priority_queue
+#include <functional>     // std::greater
+
 #include <boost/date_time.hpp>
 #include <boost/cstdint.hpp>
@@ -169 +173 @@
 
 typedef list<SystemEvent> EventQueue;
+typedef std::priority_queue<SystemEvent, 
+                            std::vector<SystemEvent>, 
+                            std::greater<SystemEvent> > PriorityEventQueue;
+// typedef std::priority_queue<SystemEvent> PriorityEventQueue;
+// TODO is vector the best underlay?
+                            
 
         //********************************************************
@@ -205 +215 @@
         private:
         EventQueue immediateEventsQ;
-        EventQueue timedEventsQ;
-        
-        ptime now;
-        ptime next_deadline;
-
+        PriorityEventQueue timedEventsQ;
+        
         boost::condition_variable system_queue_idle;
         boost::mutex queue_mutex;

tests/SystemQueue-tests.cc

@@ -92 +92 @@
     void Check()
     {
-        // XXX
         cout << "### Check ### "<< endl;
 
         checkmark = true;
     }
-    
+
+    void Cancel()
+    {
+        cout << "### Cancel ### "<< endl;
+
+        SystemQueue::instance().cancel();
+        checkmark = true;
+    }
+
     void LongRunner()
     {
@@ -163 +170 @@
 
 /**
+ *  Enqueues an event but then cancels the SystemQueue without running
+ */
+TEST_F(SystemQueueTest, EmptyAfterCancel)
+{
+    SystemQueue& sysq = SystemQueue::instance();
+
+    EXPECT_TRUE( sysq.isEmpty() );
+    
+    // enqueue event
+    sysq.scheduleCall(
+        boost::bind(&SystemQueueTest::Check, this)
+    );
+    
+    EXPECT_FALSE( sysq.isEmpty() );
+    
+    // cancel
+    sysq.cancel();
+    
+    EXPECT_TRUE( sysq.isEmpty() );
+}
+
+
+/**
+ *  cancels the SystemQueue from inside a scheduled event
+ */
+TEST_F(SystemQueueTest, CancelFromInsideEvent)
+{
+    SystemQueue& sysq = SystemQueue::instance();
+    checkmark = false;  // just to be sure..
+    
+    // start
+    sysq.run();
+    
+    // scheduleCall
+    sysq.scheduleCall(
+        boost::bind(&SystemQueueTest::Cancel, this)
+    );
+
+    // wait for the event to happen
+    wait_for_checkmark(MAX_WAIT);
+
+    EXPECT_FALSE( sysq.isRunning() ) << "SystemQueue has not stopped properly.";
+    EXPECT_TRUE( sysq.isEmpty() ) << "SystemQueue has not stopped properly.";
+}
+
+
+/**
  *  schedule a call and test whether it is actually performed by the SystemQueue
  */