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

simplify state change callback handling

Browse Source
This commit is contained in:
Alexey Rybalchenko 2017-08-28 13:35:45 +02:00 committed by Mohammad Al-Turany
parent eeb994a584
commit 4dd08051cc
3 changed files with 29 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
fairmq/FairMQDevice.cxx
View File

@ -261,10 +261,7 @@ void FairMQDevice::InitWrapper()
} }
} }
} }
if (!fStateChangeCallback.empty()) CallStateChangeCallbacks(INITIALIZING_DEVICE);
{
fStateChangeCallback(INITIALIZING_DEVICE);
}
// Bind channels. Here one run is enough, because bind settings should be available locally // Bind channels. Here one run is enough, because bind settings should be available locally
// If necessary this could be handled in the same way as the connecting channels // If necessary this could be handled in the same way as the connecting channels
@ -452,10 +449,7 @@ bool FairMQDevice::BindEndpoint(FairMQSocket& socket, string& endpoint)
void FairMQDevice::InitTaskWrapper() void FairMQDevice::InitTaskWrapper()
{ {
if (!fStateChangeCallback.empty()) CallStateChangeCallbacks(INITIALIZING_TASK);
{
fStateChangeCallback(INITIALIZING_TASK);
}
InitTask(); InitTask();
@ -517,10 +511,7 @@ void FairMQDevice::PrintChannel(const string& name)
void FairMQDevice::RunWrapper() void FairMQDevice::RunWrapper()
{ {
if (!fStateChangeCallback.empty()) CallStateChangeCallbacks(RUNNING);
{
fStateChangeCallback(RUNNING);
}
LOG(INFO) << "DEVICE: Running..."; LOG(INFO) << "DEVICE: Running...";
@ -793,10 +784,7 @@ void FairMQDevice::PostRun()
void FairMQDevice::PauseWrapper() void FairMQDevice::PauseWrapper()
{ {
if (!fStateChangeCallback.empty()) CallStateChangeCallbacks(PAUSED);
{
fStateChangeCallback(PAUSED);
}
Pause(); Pause();
} }
@ -1237,10 +1225,7 @@ void FairMQDevice::Unblock()
void FairMQDevice::ResetTaskWrapper() void FairMQDevice::ResetTaskWrapper()
{ {
if (!fStateChangeCallback.empty()) CallStateChangeCallbacks(RESETTING_TASK);
{
fStateChangeCallback(RESETTING_TASK);
}
ResetTask(); ResetTask();
@ -1253,10 +1238,7 @@ void FairMQDevice::ResetTask()
void FairMQDevice::ResetWrapper() void FairMQDevice::ResetWrapper()
{ {
if (!fStateChangeCallback.empty()) CallStateChangeCallbacks(RESETTING_DEVICE);
{
fStateChangeCallback(RESETTING_DEVICE);
}
Reset(); Reset();

6
fairmq/FairMQStateMachine.cxx
View File

@ -228,11 +228,11 @@ bool FairMQStateMachine::WaitForEndOfStateForMs(std::string event, int durationI
void FairMQStateMachine::SubscribeToStateChange(const std::string& key, std::function<void(const State)> callback) void FairMQStateMachine::SubscribeToStateChange(const std::string& key, std::function<void(const State)> callback)
{ {
fStateChangeCallbacks.insert({key, fStateChangeCallback.connect(callback)}); fStateChangeSignalsMap.insert({key, fStateChangeSignal.connect(callback)});
} }
void FairMQStateMachine::UnsubscribeFromStateChange(const std::string& key) void FairMQStateMachine::UnsubscribeFromStateChange(const std::string& key)
{ {
fStateChangeCallbacks.at(key).disconnect(); fStateChangeSignalsMap.at(key).disconnect();
fStateChangeCallbacks.erase(key); fStateChangeSignalsMap.erase(key);
} }

53
fairmq/FairMQStateMachine.h
View File

@ -85,8 +85,8 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
, fWorkAvailable(false) , fWorkAvailable(false)
, fState() , fState()
, fChangeStateMutex() , fChangeStateMutex()
, fStateChangeCallback() , fStateChangeSignal()
, fStateChangeCallbacks() , fStateChangeSignalsMap()
{} {}
// Destructor // Destructor
@ -134,10 +134,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
{ {
LOG(STATE) << "Entering IDLE state"; LOG(STATE) << "Entering IDLE state";
fsm.fState = IDLE; fsm.fState = IDLE;
if (!fsm.fStateChangeCallbacks.empty()) fsm.CallStateChangeCallbacks(IDLE);
{
fsm.fStateChangeCallback(IDLE);
}
} }
}; };
@ -167,10 +164,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
{ {
LOG(STATE) << "Entering DEVICE READY state"; LOG(STATE) << "Entering DEVICE READY state";
fsm.fState = DEVICE_READY; fsm.fState = DEVICE_READY;
if (!fsm.fStateChangeCallback.empty()) fsm.CallStateChangeCallbacks(DEVICE_READY);
{
fsm.fStateChangeCallback(DEVICE_READY);
}
} }
}; };
@ -194,10 +188,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
{ {
LOG(STATE) << "Entering READY state"; LOG(STATE) << "Entering READY state";
fsm.fState = READY; fsm.fState = READY;
if (!fsm.fStateChangeCallback.empty()) fsm.CallStateChangeCallbacks(READY);
{
fsm.fStateChangeCallback(READY);
}
} }
}; };
@ -266,10 +257,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
{ {
LOG(STATE) << "Entering READY state"; LOG(STATE) << "Entering READY state";
fsm.fState = READY; fsm.fState = READY;
if (!fsm.fStateChangeCallback.empty()) fsm.CallStateChangeCallbacks(READY);
{
fsm.fStateChangeCallback(READY);
}
fsm.Unblock(); fsm.Unblock();
std::unique_lock<std::mutex> lock(fsm.fWorkMutex); std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
@ -287,10 +275,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
{ {
LOG(STATE) << "RUNNING state finished without an external event, entering READY state"; LOG(STATE) << "RUNNING state finished without an external event, entering READY state";
fsm.fState = READY; fsm.fState = READY;
if (!fsm.fStateChangeCallback.empty()) fsm.CallStateChangeCallbacks(READY);
{
fsm.fStateChangeCallback(READY);
}
fsm.Unblock(); fsm.Unblock();
} }
@ -341,10 +326,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
{ {
LOG(STATE) << "Entering EXITING state"; LOG(STATE) << "Entering EXITING state";
fsm.fState = EXITING; fsm.fState = EXITING;
if (!fsm.fStateChangeCallback.empty()) fsm.CallStateChangeCallbacks(EXITING);
{
fsm.fStateChangeCallback(EXITING);
}
// terminate worker thread // terminate worker thread
{ {
@ -370,10 +352,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
{ {
LOG(STATE) << "Entering ERROR state"; LOG(STATE) << "Entering ERROR state";
fsm.fState = ERROR; fsm.fState = ERROR;
if (!fsm.fStateChangeCallback.empty()) fsm.CallStateChangeCallbacks(ERROR);
{
fsm.fStateChangeCallback(ERROR);
}
} }
}; };
@ -488,7 +467,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
EXITING EXITING
}; };
std::string GetStateName(int state) const static std::string GetStateName(const int state)
{ {
switch(state) switch(state)
{ {
@ -526,6 +505,14 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
bool CheckCurrentState(int state) const { return state == fState; } bool CheckCurrentState(int state) const { return state == fState; }
bool CheckCurrentState(std::string state) const { return state == GetCurrentStateName(); } bool CheckCurrentState(std::string state) const { return state == GetCurrentStateName(); }
void CallStateChangeCallbacks(const State state) const
{
if (!fStateChangeSignal.empty())
{
fStateChangeSignal(state);
}
}
// this is to run certain functions in a separate thread // this is to run certain functions in a separate thread
std::thread fWorkerThread; std::thread fWorkerThread;
@ -542,8 +529,8 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
std::atomic<State> fState; std::atomic<State> fState;
std::mutex fChangeStateMutex; std::mutex fChangeStateMutex;
boost::signals2::signal<void(const State)> fStateChangeCallback; boost::signals2::signal<void(const State)> fStateChangeSignal;
std::unordered_map<std::string, boost::signals2::connection> fStateChangeCallbacks; std::unordered_map<std::string, boost::signals2::connection> fStateChangeSignalsMap;
}; };
// reactivate the warning for non-virtual destructor // reactivate the warning for non-virtual destructor