diff --git a/fairmq/FairMQDevice.cxx b/fairmq/FairMQDevice.cxx index fe738540..d7519855 100644 --- a/fairmq/FairMQDevice.cxx +++ b/fairmq/FairMQDevice.cxx @@ -228,6 +228,11 @@ void FairMQDevice::InitWrapper() } } + if (!fStateChangeCallback.empty()) + { + fStateChangeCallback(INITIALIZING_DEVICE); + } + // 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 AttachChannels(uninitializedBindingChannels); @@ -414,6 +419,11 @@ bool FairMQDevice::BindEndpoint(FairMQSocket& socket, string& endpoint) void FairMQDevice::InitTaskWrapper() { + if (!fStateChangeCallback.empty()) + { + fStateChangeCallback(INITIALIZING_TASK); + } + InitTask(); ChangeState(internal_READY); @@ -472,30 +482,13 @@ void FairMQDevice::PrintChannel(const string& name) } } -void FairMQDevice::OnData(const string& channelName, InputMsgCallback callback) -{ - fDataCallbacks = true; - fMsgInputs.insert(make_pair(channelName, callback)); - - if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end()) - { - fInputChannelKeys.push_back(channelName); - } -} - -void FairMQDevice::OnData(const string& channelName, InputMultipartCallback callback) -{ - fDataCallbacks = true; - fMultipartInputs.insert(make_pair(channelName, callback)); - - if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end()) - { - fInputChannelKeys.push_back(channelName); - } -} - void FairMQDevice::RunWrapper() { + if (!fStateChangeCallback.empty()) + { + fStateChangeCallback(RUNNING); + } + LOG(INFO) << "DEVICE: Running..."; // start the rate logger thread @@ -765,6 +758,16 @@ void FairMQDevice::PostRun() { } +void FairMQDevice::PauseWrapper() +{ + if (!fStateChangeCallback.empty()) + { + fStateChangeCallback(PAUSED); + } + + Pause(); +} + void FairMQDevice::Pause() { while (CheckCurrentState(PAUSED)) @@ -1195,6 +1198,11 @@ void FairMQDevice::Unblock() void FairMQDevice::ResetTaskWrapper() { + if (!fStateChangeCallback.empty()) + { + fStateChangeCallback(RESETTING_TASK); + } + ResetTask(); ChangeState(internal_DEVICE_READY); @@ -1206,6 +1214,11 @@ void FairMQDevice::ResetTask() void FairMQDevice::ResetWrapper() { + if (!fStateChangeCallback.empty()) + { + fStateChangeCallback(RESETTING_DEVICE); + } + Reset(); ChangeState(internal_IDLE); diff --git a/fairmq/FairMQDevice.h b/fairmq/FairMQDevice.h index d8a3d8e2..4ee3fa92 100644 --- a/fairmq/FairMQDevice.h +++ b/fairmq/FairMQDevice.h @@ -356,7 +356,17 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable } } - void OnData(const std::string& channelName, InputMsgCallback); + void OnData(const std::string& channelName, InputMsgCallback callback) + { + fDataCallbacks = true; + fMsgInputs.insert(make_pair(channelName, callback)); + + if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end()) + { + fInputChannelKeys.push_back(channelName); + } + } + template void OnData(const std::string& channelName, bool (T::* memberFunction)(FairMQParts& parts, int index)) @@ -373,7 +383,16 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable } } - void OnData(const std::string& channelName, InputMultipartCallback); + void OnData(const std::string& channelName, InputMultipartCallback callback) + { + fDataCallbacks = true; + fMultipartInputs.insert(make_pair(channelName, callback)); + + if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end()) + { + fInputChannelKeys.push_back(channelName); + } + } bool Terminated(); @@ -487,6 +506,8 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable void InitTaskWrapper(); /// Handles the Run() method void RunWrapper(); + /// Handles the Pause() method + void PauseWrapper(); /// Handles the ResetTask() method void ResetTaskWrapper(); /// Handles the Reset() method diff --git a/fairmq/FairMQStateMachine.cxx b/fairmq/FairMQStateMachine.cxx index 18c0531b..a4c9c656 100644 --- a/fairmq/FairMQStateMachine.cxx +++ b/fairmq/FairMQStateMachine.cxx @@ -225,3 +225,9 @@ bool FairMQStateMachine::WaitForEndOfStateForMs(std::string event, int durationI { return WaitForEndOfStateForMs(GetEventNumber(event), durationInMs); } + +void FairMQStateMachine::OnStateChange(std::function callback) +{ + fStateChangeCallback.connect(callback); +} + diff --git a/fairmq/FairMQStateMachine.h b/fairmq/FairMQStateMachine.h index 60b4acfd..2905e457 100644 --- a/fairmq/FairMQStateMachine.h +++ b/fairmq/FairMQStateMachine.h @@ -35,6 +35,8 @@ #include #include +#include // signal/slot for onStateChange callbacks + #include "FairMQLogger.h" namespace msm = boost::msm; @@ -82,6 +84,7 @@ struct FairMQFSM_ : public msmf::state_machine_def , fWorkAvailable(false) , fState() , fChangeStateMutex() + , fStateChangeCallback() {} // Destructor @@ -129,6 +132,10 @@ struct FairMQFSM_ : public msmf::state_machine_def { LOG(STATE) << "Entering IDLE state"; fsm.fState = IDLE; + if (!fsm.fStateChangeCallback.empty()) + { + fsm.fStateChangeCallback(IDLE); + } } }; @@ -158,6 +165,10 @@ struct FairMQFSM_ : public msmf::state_machine_def { LOG(STATE) << "Entering DEVICE READY state"; fsm.fState = DEVICE_READY; + if (!fsm.fStateChangeCallback.empty()) + { + fsm.fStateChangeCallback(DEVICE_READY); + } } }; @@ -181,6 +192,10 @@ struct FairMQFSM_ : public msmf::state_machine_def { LOG(STATE) << "Entering READY state"; fsm.fState = READY; + if (!fsm.fStateChangeCallback.empty()) + { + fsm.fStateChangeCallback(READY); + } } }; @@ -218,7 +233,7 @@ struct FairMQFSM_ : public msmf::state_machine_def fsm.fWorkDoneCondition.wait(lock); } fsm.fWorkAvailable = true; - fsm.fWork = std::bind(&FairMQFSM_::Pause, &fsm); + fsm.fWork = std::bind(&FairMQFSM_::PauseWrapper, &fsm); fsm.fWorkAvailableCondition.notify_one(); } }; @@ -249,6 +264,10 @@ struct FairMQFSM_ : public msmf::state_machine_def { LOG(STATE) << "Entering READY state"; fsm.fState = READY; + if (!fsm.fStateChangeCallback.empty()) + { + fsm.fStateChangeCallback(READY); + } fsm.Unblock(); std::unique_lock lock(fsm.fWorkMutex); @@ -266,6 +285,10 @@ struct FairMQFSM_ : public msmf::state_machine_def { LOG(STATE) << "RUNNING state finished without an external event, entering READY state"; fsm.fState = READY; + if (!fsm.fStateChangeCallback.empty()) + { + fsm.fStateChangeCallback(READY); + } } }; @@ -314,6 +337,10 @@ struct FairMQFSM_ : public msmf::state_machine_def { LOG(STATE) << "Entering EXITING state"; fsm.fState = EXITING; + if (!fsm.fStateChangeCallback.empty()) + { + fsm.fStateChangeCallback(EXITING); + } // terminate worker thread { @@ -338,8 +365,11 @@ struct FairMQFSM_ : public msmf::state_machine_def void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&) { LOG(STATE) << "Entering ERROR state"; - fsm.fState = ERROR; + if (!fsm.fStateChangeCallback.empty()) + { + fsm.fStateChangeCallback(ERROR); + } } }; @@ -350,6 +380,7 @@ struct FairMQFSM_ : public msmf::state_machine_def virtual void InitTask() {} virtual void RunWrapper() {} virtual void Run() {} + virtual void PauseWrapper() {} virtual void Pause() {} virtual void ResetWrapper() {} virtual void Reset() {} @@ -535,6 +566,8 @@ struct FairMQFSM_ : public msmf::state_machine_def protected: std::atomic fState; std::mutex fChangeStateMutex; + + boost::signals2::signal fStateChangeCallback; }; // reactivate the warning for non-virtual destructor @@ -582,6 +615,8 @@ class FairMQStateMachine : public FairMQFSM::FairMQFSM bool WaitForEndOfStateForMs(int state, int durationInMs); bool WaitForEndOfStateForMs(std::string state, int durationInMs); + + void OnStateChange(std::function callback); }; #endif /* FAIRMQSTATEMACHINE_H_ */ diff --git a/fairmq/options/FairMQProgOptions.cxx b/fairmq/options/FairMQProgOptions.cxx index ea723c0f..277ccd0a 100644 --- a/fairmq/options/FairMQProgOptions.cxx +++ b/fairmq/options/FairMQProgOptions.cxx @@ -351,8 +351,8 @@ void FairMQProgOptions::InitOptionDescription() else { fMQOptionsInCmd.add_options() - ("id", po::value(), "Device ID (required argument)") - ("io-threads", po::value()->default_value(1), "Number of I/O threads") + ("id", po::value(), "Device ID (required argument).") + ("io-threads", po::value()->default_value(1), "Number of I/O threads.") ("transport", po::value()->default_value("zeromq"), "Transport ('zeromq'/'nanomsg').") ("config", po::value()->default_value("static"), "Config source ('static'/).") ("control", po::value()->default_value("interactive"), "States control ('interactive'/'static'/).")