Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 08:41:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

StateMachine: Guard access to subscription containers

Browse Source
This commit is contained in:
Alexey Rybalchenko 2023-11-10 10:51:16 +01:00 committed by Dennis Klein
parent 6122010694
commit fbb6577625
1 changed files with 9 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
fairmq/StateMachine.cxx
View File

@ -177,6 +177,7 @@ struct Machine_ : public state_machine_def<Machine_>
atomic<bool> fLastTransitionResult; atomic<bool> fLastTransitionResult;
mutex fStateMtx; mutex fStateMtx;
mutex fSubscriptionsMtx;
atomic<bool> fNewStatePending; atomic<bool> fNewStatePending;
condition_variable fNewStatePendingCV; 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. // Check if the key has a integer value as prefix, if yes, decode it.
int i = strtol(key.c_str(), nullptr, 10); 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) void StateMachine::UnsubscribeFromStateChange(const string& key)
{ {
auto fsm = static_pointer_cast<FairMQFSM>(fFsm); auto fsm = static_pointer_cast<FairMQFSM>(fFsm);
lock_guard<mutex> lock(fsm->fSubscriptionsMtx);
if (fsm->fStateChangeSignalsMap.count(key)) { if (fsm->fStateChangeSignalsMap.count(key)) {
fsm->fStateChangeSignalsMap.at(key).disconnect(); fsm->fStateChangeSignalsMap.at(key).disconnect();
fsm->fStateChangeSignalsMap.erase(key); fsm->fStateChangeSignalsMap.erase(key);
@ -357,12 +361,15 @@ void StateMachine::StopHandlingStates()
void StateMachine::SubscribeToNewTransition(const string& key, function<void(const Transition)> callback) 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) void StateMachine::UnsubscribeFromNewTransition(const string& key)
{ {
auto fsm = static_pointer_cast<FairMQFSM>(fFsm); auto fsm = static_pointer_cast<FairMQFSM>(fFsm);
lock_guard<mutex> lock(fsm->fSubscriptionsMtx);
if (fsm->fNewTransitionSignalsMap.count(key)) { if (fsm->fNewTransitionSignalsMap.count(key)) {
fsm->fNewTransitionSignalsMap.at(key).disconnect(); fsm->fNewTransitionSignalsMap.at(key).disconnect();
fsm->fNewTransitionSignalsMap.erase(key); fsm->fNewTransitionSignalsMap.erase(key);