Changeset 0427d1 for src/Actions/ActionQueue.cpp

Timestamp:

Feb 27, 2024, 10:18:17 PM (20 months ago)

Author:

Frederik Heber <frederik.heber@…>

Branches:

Candidate_v1.7.0, stable

Children:

399c69

Parents:

d203d1e

git-author:

Frederik Heber <frederik.heber@…> (02/25/24 23:14:04)

git-committer:

Frederik Heber <frederik.heber@…> (02/27/24 22:18:17)

Message:

FIX: ActionQueue no locked on CurrentAction access.

• actionqueue[CurrentAction] without mutex locking.
• getCurrentAction() without mutex locking.
• extracted stepOnToNextAction().
• using recursive_mutex instead of normal to allow nested locks.

File:

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

src/Actions/ActionQueue.cpp

@@ -43 +43 @@
 
 #include <boost/date_time/posix_time/posix_time.hpp>
+#include <boost/thread/recursive_mutex.hpp>
 #include <boost/version.hpp>
 #include <iterator>
@@ -147 +148 @@
   Action *newaction = _action->clone(state);
   newaction->prepare(state);
-  mtx_queue.lock();
-  tempqueue.push_back( newaction );
-  const bool tempqueue_notempty = !tempqueue.empty();
-  mtx_queue.unlock();
+  bool tempqueue_notempty;
+  {
+    boost::recursive_mutex::scoped_lock lock(mtx_queue);
+    tempqueue.push_back( newaction );
+    tempqueue_notempty = !tempqueue.empty();
+  }
   setRunThreadIdle( !((!isIdle()) || tempqueue_notempty) );
 #endif
@@ -158 +161 @@
 {
 #ifdef HAVE_ACTION_THREAD
-  boost::unique_lock<boost::mutex> lock(mtx_queue);
+  boost::recursive_mutex::scoped_lock lock(mtx_queue);
 #endif
   bool status = (CurrentAction == actionqueue.size());
@@ -188 +191 @@
 const Action& ActionQueue::getCurrentAction() const
 {
+#ifdef HAVE_ACTION_THREAD
+  boost::recursive_mutex::scoped_lock lock(mtx_queue);
+#endif
   return *const_cast<const Action *>(actionqueue[CurrentAction]);
 }
@@ -209 +215 @@
 //    LOG(1, "DEBUG: Start of ActionQueue's run() loop.");
     // call all currently present Actions
-    mtx_queue.lock();
     insertTempQueue();
-    mtx_queue.unlock();
     bool status = !isIdle();
     while (status) {
       //      boost::this_thread::disable_interruption di;
-      LOG(0, "Calling Action " << actionqueue[CurrentAction]->getName() << " ... ");
       try {
-        if (!isDryRun(actionqueue[CurrentAction]))
-          actionqueue[CurrentAction]->call();
-        pushStatus("SUCCESS: Action "+actionqueue[CurrentAction]->getName()+" successful.");
+        // pick next action to run
+        Action * action_to_run;
+        {
+          boost::recursive_mutex::scoped_lock lock(mtx_queue);
+          action_to_run = actionqueue[CurrentAction];
+        }
+        LOG(0, "Calling Action " << action_to_run->getName() << " ... ");
+        // run action
+        if (!isDryRun(action_to_run))
+          action_to_run->call();
+        pushStatus("SUCCESS: Action "+action_to_run->getName()+" successful.");
         lastActionOk = true;
       } catch(ActionFailureException &e) {
@@ -239 +250 @@
         OBSERVE;
         NOTIFY(ActionQueued);
-        _lastchangedaction = actionqueue[CurrentAction];
-        mtx_queue.lock();
-        CurrentAction++;
-        mtx_queue.unlock();
+        stepOnToNextAction();
       }
-      // 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();
-      mtx_queue.unlock();
       status = !isIdle();
     }
-    mtx_queue.lock();
-    const bool tempqueue_notempty = !tempqueue.empty();
-    mtx_queue.unlock();
+    bool tempqueue_notempty;
+    {
+      boost::recursive_mutex::scoped_lock lock(mtx_queue);
+      tempqueue_notempty = !tempqueue.empty();
+    }
     setRunThreadIdle( !((!isIdle()) || tempqueue_notempty) );
     cond_idle.notify_one();
@@ -262 +269 @@
 }
 
+void ActionQueue::stepOnToNextAction()
+{
+   boost::recursive_mutex::scoped_lock lock(mtx_queue);
+   _lastchangedaction = actionqueue[CurrentAction];
+   CurrentAction++;
+}
+
 void ActionQueue::insertTempQueue()
 {
   if (!tempqueue.empty()) {
-    ActionQueue_t::iterator InsertionIter = actionqueue.begin();
-    std::advance(InsertionIter, CurrentAction);
-    actionqueue.insert( InsertionIter, tempqueue.begin(), tempqueue.end() );
+    // access actionqueue, hence using mutex
+    boost::recursive_mutex::scoped_lock lock(mtx_queue);
+    actionqueue.insert( actionqueue.end(), tempqueue.begin(), tempqueue.end() );
     tempqueue.clear();
   }
@@ -359 +373 @@
 {
 #ifdef HAVE_ACTION_THREAD
-  mtx_queue.lock();
+  boost::recursive_mutex::scoped_lock lock(mtx_queue);
 #endif
   LOG(1, "Removing all Actions from position " << _fromAction << " onward.");
@@ -371 +385 @@
 #ifdef HAVE_ACTION_THREAD
   CurrentAction = actionqueue.size();
-  mtx_queue.unlock();
 #endif
 }