State machine update

- Fix regression from last commit (preventing static run from proper shutdown).
 - Guard state changes (for the internal transitions) (msm::process_event is not thread safe).
 - Remove unused transition from RUNNING to EXITING.

fairmq/FairMQStateMachine.cxx

@@ -55,47 +55,85 @@ bool FairMQStateMachine::ChangeState(int event)
         switch (event)
         {
             case INIT_DEVICE:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::INIT_DEVICE());
                 return true;
+            }
             case internal_DEVICE_READY:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::internal_DEVICE_READY());
                 return true;
+            }
             case INIT_TASK:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::INIT_TASK());
                 return true;
+            }
             case internal_READY:
+            {
+                // std::lock_guard<std::mutex> lock(fChangeStateMutex); // InitTask is synchronous, until ROOT workaround is no longer needed.
                 process_event(FairMQFSM::internal_READY());
                 return true;
+            }
             case RUN:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::RUN());
                 return true;
+            }
             case PAUSE:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::PAUSE());
                 return true;
+            }
             case STOP:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::STOP());
                 return true;
+            }
             case RESET_DEVICE:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::RESET_DEVICE());
                 return true;
+            }
             case RESET_TASK:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::RESET_TASK());
                 return true;
+            }
             case internal_IDLE:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::internal_IDLE());
                 return true;
+            }
             case END:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::END());
                 return true;
+            }
             case ERROR_FOUND:
+            {
+                std::lock_guard<std::mutex> lock(fChangeStateMutex);
                 process_event(FairMQFSM::ERROR_FOUND());
                 return true;
+            }
             default:
+            {
                 LOG(ERROR) << "Requested state transition with an unsupported event: " << event << std::endl
                            << "Supported are: INIT_DEVICE, INIT_TASK, RUN, PAUSE, STOP, RESET_TASK, RESET_DEVICE, END, ERROR_FOUND";
                 return false;
             }
         }
+    }
     catch (std::exception& e)
     {
         LOG(ERROR) << "Exception in FairMQStateMachine::ChangeState(): " << e.what();
@@ -123,7 +161,7 @@ void FairMQStateMachine::WaitForEndOfState(int event)
                 std::unique_lock<std::mutex> lock(fWorkMutex);
                 while (fWorkActive || fWorkAvailable)
                 {
-                    fWorkDoneCondition.wait(lock);
+                    fWorkDoneCondition.wait_for(lock, std::chrono::seconds(1));
                 }
 
                 break;

fairmq/FairMQStateMachine.h

@@ -82,6 +82,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         , fWorkActive(false)
         , fWorkAvailable(false)
         , fState()
+        , fChangeStateMutex()
         {}
 
     // Destructor
@@ -100,9 +101,6 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
     template <class Event, class FSM>
     void on_exit(Event const&, FSM& fsm)
     {
-        // join the worker thread (executing user states)
-        fsm.fWorkerThread.join();
-
         LOG(STATE) << "Exiting FairMQ state machine";
     }
 
@@ -320,31 +318,17 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
             fsm.fState = EXITING;
 
             // terminate worker thread
+            {
                 std::lock_guard<std::mutex> lock(fsm.fWorkMutex);
                 fsm.fWorkerTerminated = true;
                 fsm.fWorkAvailableCondition.notify_one();
-
-            fsm.fTerminateStateThread = std::thread(&FairMQFSM_::Terminate, &fsm);
-            fsm.Shutdown();
-            fsm.fTerminateStateThread.join();
             }
-    };
 
-    struct ExitingRunFct
+            // join the worker thread (executing user states)
+            if (fsm.fWorkerThread.joinable())
             {
-        template <class EVT, class FSM, class SourceState, class TargetState>
+                fsm.fWorkerThread.join();
-        void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
+            }
-        {
-            LOG(STATE) << "Entering EXITING state";
-
-            fsm.fState = EXITING;
-
-            fsm.Unblock(); // Unblock potential blocking transfer calls
-
-            // terminate worker thread
-            std::lock_guard<std::mutex> lock(fsm.fWorkMutex);
-            fsm.fWorkerTerminated = true;
-            fsm.fWorkAvailableCondition.notify_one();
 
             fsm.fTerminateStateThread = std::thread(&FairMQFSM_::Terminate, &fsm);
             fsm.Shutdown();
@@ -423,7 +407,6 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         msmf::Row<RUNNING_FSM,             PAUSE,                 PAUSED_FSM,              PauseFct,        msmf::none>,
         msmf::Row<RUNNING_FSM,             STOP,                  READY_FSM,               StopFct,         msmf::none>,
         msmf::Row<RUNNING_FSM,             internal_READY,        READY_FSM,               InternalStopFct, msmf::none>,
-        msmf::Row<RUNNING_FSM,             END,                   EXITING_FSM,             ExitingRunFct,   msmf::none>,
         msmf::Row<PAUSED_FSM,              RUN,                   RUNNING_FSM,             ResumeFct,       msmf::none>,
         msmf::Row<RESETTING_TASK_FSM,      internal_DEVICE_READY, DEVICE_READY_FSM,        DeviceReadyFct,  msmf::none>,
         msmf::Row<RESETTING_DEVICE_FSM,    internal_IDLE,         IDLE_FSM,                IdleFct,         msmf::none>,
@@ -556,6 +539,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
     bool fWorkAvailable;
 
   protected:
+    std::mutex fChangeStateMutex;
     std::atomic<State> fState;
 };