From fbb6577625c0d24e258df886a06a77ee98cb13ae Mon Sep 17 00:00:00 2001
From: Alexey Rybalchenko <alexryba@gmail.com>
Date: Fri, 10 Nov 2023 10:51:16 +0100
Subject: [PATCH] StateMachine: Guard access to subscription containers

---
 fairmq/StateMachine.cxx | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/fairmq/StateMachine.cxx b/fairmq/StateMachine.cxx
index cc2ee6eb..41d6c2d8 100644
--- a/fairmq/StateMachine.cxx
+++ b/fairmq/StateMachine.cxx
@@ -177,6 +177,7 @@ struct Machine_ : public state_machine_def<Machine_>
     atomic<bool> fLastTransitionResult;
 
     mutex fStateMtx;
+    mutex fSubscriptionsMtx;
     atomic<bool> fNewStatePending;
     condition_variable fNewStatePendingCV;
 
@@ -312,12 +313,15 @@ void StateMachine::SubscribeToStateChange(const string& key, function<void(const
 {
     // Check if the key has a integer value as prefix, if yes, decode it.
     int i = strtol(key.c_str(), nullptr, 10);
-    static_pointer_cast<FairMQFSM>(fFsm)->fStateChangeSignalsMap.insert({key, static_pointer_cast<FairMQFSM>(fFsm)->fStateChangeSignal.connect(i, callback)});
+    auto fsm = static_pointer_cast<FairMQFSM>(fFsm);
+    lock_guard<mutex> lock(fsm->fSubscriptionsMtx);
+    fsm->fStateChangeSignalsMap.insert({key, fsm->fStateChangeSignal.connect(i, callback)});
 }
 
 void StateMachine::UnsubscribeFromStateChange(const string& key)
 {
     auto fsm = static_pointer_cast<FairMQFSM>(fFsm);
+    lock_guard<mutex> lock(fsm->fSubscriptionsMtx);
     if (fsm->fStateChangeSignalsMap.count(key)) {
         fsm->fStateChangeSignalsMap.at(key).disconnect();
         fsm->fStateChangeSignalsMap.erase(key);
@@ -357,12 +361,15 @@ void StateMachine::StopHandlingStates()
 
 void StateMachine::SubscribeToNewTransition(const string& key, function<void(const Transition)> callback)
 {
-    static_pointer_cast<FairMQFSM>(fFsm)->fNewTransitionSignalsMap.insert({key, static_pointer_cast<FairMQFSM>(fFsm)->fNewTransitionSignal.connect(callback)});
+    auto fsm = static_pointer_cast<FairMQFSM>(fFsm);
+    lock_guard<mutex> lock(fsm->fSubscriptionsMtx);
+    fsm->fNewTransitionSignalsMap.insert({key, fsm->fNewTransitionSignal.connect(callback)});
 }
 
 void StateMachine::UnsubscribeFromNewTransition(const string& key)
 {
     auto fsm = static_pointer_cast<FairMQFSM>(fFsm);
+    lock_guard<mutex> lock(fsm->fSubscriptionsMtx);
     if (fsm->fNewTransitionSignalsMap.count(key)) {
         fsm->fNewTransitionSignalsMap.at(key).disconnect();
         fsm->fNewTransitionSignalsMap.erase(key);