Changeset 7e6c0d for src/Actions/ActionQueue.cpp

Timestamp:

Jan 30, 2015, 10:34:16 AM (10 years ago)

Author:

Frederik Heber <heber@…>

Children:

419fa2

Parents:

eea0bb (diff), fff8fc (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

git-author:

Frederik Heber <heber@…> (01/29/15 07:42:59)

git-committer:

Frederik Heber <heber@…> (01/30/15 10:34:16)

Message:

Merge branch 'ThreadFixes' into Candidate_v1.4.10

Conflicts:

src/Actions/ActionQueue.cpp
src/Actions/ActionQueue.hpp

• many conflicts due to last commit in ThreadFixes. Combined stuff from ThreadFixes and preventing failing actions from clearing queue entirely.

File:

• src/Actions/ActionQueue.cpp (modified) (12 diffs)

src/Actions/ActionQueue.cpp

@@ -65 +65 @@
     lastActionOk(true)
 #else
+    lastActionOk(true),
     CurrentAction(0),
-    lastActionOk(true),
-    run_thread(boost::bind(&ActionQueue::run, this)),
-    run_thread_isIdle(true)
+    run_thread_isIdle(true),
+    run_thread_running(false),
+    run_thread(boost::bind(&ActionQueue::run, this))
 #endif
 {
@@ -104 +105 @@
   newaction->prepare(state);
 #ifdef HAVE_ACTION_THREAD
-  mtx_queue.lock();
+  mtx_actionqueue.lock();
 #endif
   actionqueue.push_back( newaction );
@@ -114 +115 @@
     std::cerr << "Action " << *boost::get_error_info<ActionNameString>(e) << " has failed." << std::endl;
     World::getInstance().setExitFlag(5);
-    clearQueue(actionqueue.size()-1);
+//    clearQueue(actionqueue.size()-1);
     lastActionOk = false;
-    std::cerr << "Remaining Actions cleared from queue." << std::endl;
+//    std::cerr << "Remaining Actions cleared from queue." << std::endl;
   } catch (std::exception &e) {
     pushStatus("FAIL: General exception caught, aborting.");
     World::getInstance().setExitFlag(134);
-    clearQueue(actionqueue.size()-1);
+//    clearQueue(actionqueue.size()-1);
     lastActionOk = false;
-    std::cerr << "Remaining Actions cleared from queue." << std::endl;
+//    std::cerr << "Remaining Actions cleared from queue." << std::endl;
   }
   if (lastActionOk) {
@@ -130 +131 @@
   }
 #else
-  setRunThreadIdle(CurrentAction == actionqueue.size());
-  mtx_queue.unlock();
-#endif
-}
+  mtx_actionqueue.unlock();
+  const bool new_run_thread_isIdle = isActionQueueDone();
+  setrun_thread_isIdle(new_run_thread_isIdle);
+#endif
+}
+
+#ifdef HAVE_ACTION_THREAD
+bool ActionQueue::isActionQueueDone() const
+{
+  boost::lock_guard<boost::mutex> lock(mtx_actionqueue);
+  return (CurrentAction == actionqueue.size());
+}
+
+bool ActionQueue::isTempQueueDone() const
+{
+  boost::lock_guard<boost::mutex> lock(mtx_tempqueue);
+  return tempqueue.empty();
+}
+#endif
 
 void ActionQueue::insertAction(Action *_action, enum Action::QueryOptions state)
@@ -142 +158 @@
   Action *newaction = _action->clone(state);
   newaction->prepare(state);
-  mtx_queue.lock();
-  tempqueue.push_back( newaction );
-  setRunThreadIdle( !((CurrentAction != actionqueue.size()) || !tempqueue.empty()) );
-  mtx_queue.unlock();
+  {
+    boost::lock_guard<boost::mutex> lock(mtx_tempqueue);
+    tempqueue.push_back( newaction );
+  }
+  {
+    bool new_run_thread_isIdle = getrun_thread_isIdle();
+    new_run_thread_isIdle &= isTempQueueDone();
+    setrun_thread_isIdle(new_run_thread_isIdle);
+  }
 #endif
 }
@@ -152 +173 @@
 void ActionQueue::run()
 {
+  {
+    boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
+    run_thread_running = true;
+  }
   bool Interrupted = false;
   do {
@@ -157 +182 @@
     try {
 #if BOOST_VERSION < 105000
-      run_thread.sleep(boost::get_system_time() + boost::posix_time::milliseconds(100));
+      boost::this_thread::sleep(boost::get_system_time() + boost::posix_time::milliseconds(100));
 #else
       boost::this_thread::sleep_for(boost::chrono::milliseconds(100));
@@ -167 +192 @@
 //    LOG(1, "DEBUG: Start of ActionQueue's run() loop.");
     // call all currently present Actions
-    mtx_queue.lock();
     insertTempQueue();
-    bool status = (CurrentAction != actionqueue.size());
-    mtx_queue.unlock();
-    while (status) {
+    while ((!Interrupted) && (!isActionQueueDone())) {
       //      boost::this_thread::disable_interruption di;
+      // access actionqueue, hence using mutex
+      mtx_actionqueue.lock();
       LOG(0, "Calling Action " << actionqueue[CurrentAction]->getName() << " ... ");
       try {
@@ -192 +216 @@
         std::cerr << "Remaining Actions cleared from queue." << std::endl;
       }
+      // remember action we juse executed
+      const Action *lastaction = actionqueue[CurrentAction];
+      // step on to next action and check for end
+      CurrentAction++;
+      // insert new actions (before [CurrentAction]) if they have been spawned
+      // we must have an extra vector for this, as we cannot change actionqueue
+      // while an action instance is "in-use"
+      mtx_actionqueue.unlock();
+
+      insertTempQueue();
+
+      // set last action
       if (lastActionOk) {
         OBSERVE;
         NOTIFY(ActionQueued);
-        _lastchangedaction = actionqueue[CurrentAction];
-        mtx_queue.lock();
-        CurrentAction++;
-        mtx_queue.unlock();
+        _lastchangedaction = lastaction;
       }
-      // access actionqueue, hence using mutex
-      mtx_queue.lock();
-      // insert new actions (before [CurrentAction]) if they have been spawned
-      // we must have an extra vector for this, as we cannot change actionqueue
-      // while an action instance is "in-use"
-      insertTempQueue();
-      status = (CurrentAction != actionqueue.size());
-      mtx_queue.unlock();
-    }
-    setRunThreadIdle( !((CurrentAction != actionqueue.size()) || !tempqueue.empty()) );
+    }
+    {
+      bool new_run_thread_isIdle = isActionQueueDone();
+      new_run_thread_isIdle &= isTempQueueDone();
+      setrun_thread_isIdle(new_run_thread_isIdle);
+    }
     cond_idle.notify_one();
 //    LOG(1, "DEBUG: End of ActionQueue's run() loop.");
   } while (!Interrupted);
+  {
+    boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
+    run_thread_running = false;
+  }
 }
 
 void ActionQueue::insertTempQueue()
 {
+  boost::lock_guard<boost::mutex> lock(mtx_tempqueue);
   if (!tempqueue.empty()) {
+    boost::lock_guard<boost::mutex> lock(mtx_actionqueue);
     ActionQueue_t::iterator InsertionIter = actionqueue.begin();
     std::advance(InsertionIter, CurrentAction);
@@ -225 +260 @@
 }
 
+void ActionQueue::setrun_thread_isIdle(
+    const bool _run_thread_isIdle)
+{
+  boost::unique_lock<boost::mutex> lock(mtx_run_thread_isIdle);
+  run_thread_isIdle = _run_thread_isIdle;
+}
+
+bool ActionQueue::getrun_thread_isIdle() const
+{
+  boost::unique_lock<boost::mutex> lock(mtx_run_thread_isIdle);
+  return run_thread_isIdle;
+}
+
 void ActionQueue::wait()
 {
-  boost::unique_lock<boost::mutex> lock(mtx_idle);
-  while(!run_thread_isIdle)
-  {
-      cond_idle.wait(lock);
-  }
-}
-#endif
-
-#ifdef HAVE_ACTION_THREAD
+  if (run_thread_running) {
+    boost::unique_lock<boost::mutex> lock(mtx_run_thread_isIdle);
+    while(!run_thread_isIdle)
+    {
+        cond_idle.wait(lock);
+    }
+  }
+}
+
 void ActionQueue::stop()
 {
@@ -311 +359 @@
 void ActionQueue::clearQueue(const size_t _fromAction)
 {
-#ifdef HAVE_ACTION_THREAD
-  mtx_queue.lock();
-#endif
-  LOG(1, "Removing all Actions from position " << _fromAction << " onward.");
-  // free all actions still to be called contained in actionqueue
-  ActionQueue_t::iterator inititer = actionqueue.begin();
-  std::advance(inititer, _fromAction);
-  for (ActionQueue_t::iterator iter = inititer; iter != actionqueue.end(); ++iter)
-    delete *iter;
-  actionqueue.erase(inititer, actionqueue.end());
-  LOG(1, "There are " << actionqueue.size() << " remaining Actions.");
-#ifdef HAVE_ACTION_THREAD
-  CurrentAction = actionqueue.size();
-  mtx_queue.unlock();
-#endif
+  // free all actions contained in actionqueue
+  {
+#ifdef HAVE_ACTION_THREAD
+    boost::lock_guard<boost::mutex> lock(mtx_actionqueue);
+#endif
+    LOG(1, "Removing all Actions from position " << _fromAction << " onward.");
+    // free all actions still to be called contained in actionqueue
+    ActionQueue_t::iterator inititer = actionqueue.begin();
+    std::advance(inititer, _fromAction);
+    for (ActionQueue_t::iterator iter = inititer; iter != actionqueue.end(); ++iter)
+      delete *iter;
+    actionqueue.erase(inititer, actionqueue.end());
+    LOG(1, "There are " << actionqueue.size() << " remaining Actions.");
+#ifdef HAVE_ACTION_THREAD
+    CurrentAction = actionqueue.size();
+#endif
+  }
 }
 
@@ -332 +382 @@
 {
   // free all actions contained in tempqueue
-  for (ActionQueue_t::iterator iter = tempqueue.begin();
-      !tempqueue.empty(); iter = tempqueue.begin()) {
-    delete *iter;
-    tempqueue.erase(iter);
-  }
-}
-
-void ActionQueue::setRunThreadIdle(const bool _flag)
-{
-  {
-    boost::unique_lock<boost::mutex> lock(mtx_idle);
-    run_thread_isIdle = _flag;
+  {
+    boost::lock_guard<boost::mutex> lock(mtx_tempqueue);
+    for (ActionQueue_t::iterator iter = tempqueue.begin();
+        !tempqueue.empty(); iter = tempqueue.begin()) {
+      delete *iter;
+      tempqueue.erase(iter);
+    }
   }
 }