Changeset 12762

Timestamp:

Mar 12, 2014, 6:58:54 PM (11 years ago)

Author:

hock@…

Message:

New System Queue

Threading test (aka "isEmpty() test") is now passed, too.

Files:

• source/ariba/utility/system/SystemQueue.cpp (modified) (19 diffs)
• source/ariba/utility/system/SystemQueue.h (modified) (3 diffs)
• tests/SystemQueue-tests.cc (modified) (1 diff)

source/ariba/utility/system/SystemQueue.cpp

@@ -43 +43 @@
 namespace utility {
     
+    typedef boost::mutex::scoped_lock scoped_lock;
     
     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;
@@ -54 +53 @@
 use_logging_cpp(SystemQueue);
 
-SystemQueue::SystemQueue()
-{
-        logging_info("Creating SystemQueue at: " << this);
+SystemQueue::SystemQueue() :
+    SysQ( new QueueThread() )
+{
+        logging_debug("Creating SystemQueue at: " << this);
 }
 
@@ -76 +76 @@
     SystemEvent ev(event);
     
-    SysQ.insert(ev, delay);
+    SysQ->insert(ev, delay);
 }
 
@@ -95 +95 @@
 void SystemQueue::run()
 {
-    assert ( ! SysQ.running );
-    
-    SysQ.running = true;
+    assert ( ! SysQ->running );
+    
+    SysQ->running = true;
     
     // start thread
-    sysq_thread.reset( new boost::thread(boost::ref(SysQ)) );
+    sysq_thread.reset( new boost::thread(boost::ref(*SysQ)) );
 }
 
@@ -106 +106 @@
 {
     // signal SysQ to quit (and abort queued events)
-    SysQ.cancel();
+    SysQ->cancel();
     
     // wait till actually completes
     //   (should be fast, but the current event is allowed to finish)
-    logging_info("/// joining system queue thread");
+    logging_debug("/// ... joining SysQ thread");
     sysq_thread->join();
     
@@ -116 +116 @@
     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..
+    assert ( ! SysQ->isRunning() );
+
+    
+    // clean up and respawn
+    logging_debug("/// respawning SysQ");
+    SysQ.reset( new QueueThread() );
 }
 
@@ -133 +133 @@
 bool SystemQueue::isEmpty()
 {
-//  XXX
-//      return directScheduler.isEmpty() || delayScheduler.isEmpty();
-
-    return true;
+    return SysQ->isEmpty();
 }
 
 bool SystemQueue::isRunning()
 {
-    return SysQ.isRunning();
+    return SysQ->isRunning();
 }
 
@@ -159 +156 @@
     now( not_a_date_time ),
     next_deadline( not_a_date_time ),
+    processing_event( false ),
     running( false ),
     aborted( false )
@@ -170 +168 @@
 void SystemQueue::QueueThread::operator()()
 {
-    // XXX debug
-    logging_info( "/// SysQ thread is alive." );
+    logging_debug( "/// SysQ thread is alive." );
     
     assert( running );  // this is set before the thread starts
@@ -188 +185 @@
     }
     
-    // XXX debug
-    logging_info( "/// SysQ thread is quitting." );
+    logging_debug( "/// SysQ thread is quitting." );
     
     running = false;
@@ -213 +209 @@
         {
             scoped_lock lock( queue_mutex );
+            
+            this->processing_event = true;
             
             // dequeue first event
@@ -219 +217 @@
         }
         
-        // XXX debug
-        logging_info("/// SysQ: dispatching event");
+        logging_debug("/// SysQ: dispatching event");
         
         currently_processed_event->getListener()->handleSystemEvent( *currently_processed_event );
@@ -228 +225 @@
     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
@@ -233 +232 @@
     if ( execution_time.total_milliseconds() > 50 )
     {
-        logging_debug("WARNING: Last event took " << execution_time.total_milliseconds() << " to complete.");
+        logging_info("WARNING: Last event took " << execution_time.total_milliseconds() << " to complete.");
     }
 }
@@ -290 +289 @@
         if ( this->next_deadline.is_not_a_date_time() )
         {
-            // XXX debug output
-            logging_info("/// SysQ is going to sleep (forever..).");
+            logging_debug("/// SysQ is going to sleep.");
             
             this->system_queue_idle.wait( lock );
@@ -298 +296 @@
         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.");
+            logging_debug("/// SysQ is going to sleep for "
+                        << ( next_deadline - get_clock() ).total_milliseconds()
+                        << "ms.");
 
             this->system_queue_idle.timed_wait( lock, next_deadline );
@@ -326 +324 @@
 void SystemQueue::QueueThread::cancel()
 {
-    logging_debug("/// cancelling system queue");
+    logging_debug("/// Cancelling system queue... ");
 
     // SYNCHRONIZED
@@ -334 +332 @@
     }
     
-    // XXX debug output
-    logging_info("/// SysQ: " << immediateEventsQ.size() << " immediate event(s) + "
+    logging_debug("/// SysQ: " << immediateEventsQ.size() << " immediate event(s) + "
                               << timedEventsQ.size() << " timed event(s) left.");
     
@@ -360 +357 @@
     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;
 }
 

source/ariba/utility/system/SystemQueue.h

@@ -206 +206 @@
         EventQueue immediateEventsQ;
         EventQueue timedEventsQ;
-//      boost::mutex queueMutex;
-        
-//         SystemEvent& currently_processed_event;
         
         ptime now;
@@ -215 +212 @@
         boost::condition_variable system_queue_idle;
         boost::mutex queue_mutex;
-        
-//              boost::condition_variable itemsAvailable;  DEPRECATED
+
+        bool processing_event;
+        
                 volatile bool running;
         volatile bool aborted;
@@ -244 +242 @@
     /// member variables of class SystemQueue
 private:
-    QueueThread SysQ;
+    boost::scoped_ptr<QueueThread> SysQ;
     boost::scoped_ptr<boost::thread> sysq_thread;
     

tests/SystemQueue-tests.cc

@@ -201 +201 @@
  *    instead a wait_until_empty function.
  */
-TEST_F(SystemQueueTest, DISABLED_Threading)
+TEST_F(SystemQueueTest, Threading)
 {
     SystemQueue& sysq = SystemQueue::instance();