Source · fairmq/
StateMachine.cxxsource
Described in Device states, Threads in a device, Control a device from code, States and transitions, Your first device, Logging.
1/********************************************************************************2 * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#include <fairmq/StateMachine.h>10#include <fairmq/tools/Exceptions.h>11 12#include <fairlogger/Logger.h>13 14// Increase maximum number of boost::msm states (default is 10)15// This #define has to be before any msm header includes16#define FUSION_MAX_VECTOR_SIZE 2017 18#include <boost/mpl/for_each.hpp>19#include <boost/msm/back/state_machine.hpp>20#include <boost/msm/back/tools.hpp>21#include <boost/msm/back/metafunctions.hpp>22#include <boost/msm/front/state_machine_def.hpp>23#include <boost/msm/front/functor_row.hpp>24#include <boost/core/demangle.hpp>25#include <boost/signals2.hpp> // signal/slot for onStateChange callbacks26 27#include <atomic>28#include <condition_variable>29#include <chrono>30#include <unordered_map>31#include <mutex>32 33using namespace std;34using namespace boost::msm;35using namespace boost::msm::front;36using namespace boost::msm::back;37namespace bmpl = boost::mpl;38 39namespace fair::mq40{41namespace fsm42{43 44// list of FSM states45struct OK_S : public state<> { static string Name() { return "OK"; } static State Type() { return State::Ok; } };46 47struct IDLE_S : public state<> { static string Name() { return "IDLE"; } static State Type() { return State::Idle; } };48struct INITIALIZING_DEVICE_S : public state<> { static string Name() { return "INITIALIZING_DEVICE"; } static State Type() { return State::InitializingDevice; } };49struct INITIALIZED_S : public state<> { static string Name() { return "INITIALIZED"; } static State Type() { return State::Initialized; } };50struct BINDING_S : public state<> { static string Name() { return "BINDING"; } static State Type() { return State::Binding; } };51struct BOUND_S : public state<> { static string Name() { return "BOUND"; } static State Type() { return State::Bound; } };52struct CONNECTING_S : public state<> { static string Name() { return "CONNECTING"; } static State Type() { return State::Connecting; } };53struct DEVICE_READY_S : public state<> { static string Name() { return "DEVICE_READY"; } static State Type() { return State::DeviceReady; } };54struct INITIALIZING_TASK_S : public state<> { static string Name() { return "INITIALIZING_TASK"; } static State Type() { return State::InitializingTask; } };55struct READY_S : public state<> { static string Name() { return "READY"; } static State Type() { return State::Ready; } };56struct RUNNING_S : public state<> { static string Name() { return "RUNNING"; } static State Type() { return State::Running; } };57struct RESETTING_TASK_S : public state<> { static string Name() { return "RESETTING_TASK"; } static State Type() { return State::ResettingTask; } };58struct RESETTING_DEVICE_S : public state<> { static string Name() { return "RESETTING_DEVICE"; } static State Type() { return State::ResettingDevice; } };59struct EXITING_S : public state<> { static string Name() { return "EXITING"; } static State Type() { return State::Exiting; } };60 61struct ERROR_S : public terminate_state<> { static string Name() { return "ERROR"; } static State Type() { return State::Error; } };62 63// list of FSM transitions (events)64struct AUTO_E { static string Name() { return "AUTO"; } static Transition Type() { return Transition::Auto; } };65struct INIT_DEVICE_E { static string Name() { return "INIT_DEVICE"; } static Transition Type() { return Transition::InitDevice; } };66struct COMPLETE_INIT_E { static string Name() { return "COMPLETE_INIT"; } static Transition Type() { return Transition::CompleteInit; } };67struct BIND_E { static string Name() { return "BIND"; } static Transition Type() { return Transition::Bind; } };68struct CONNECT_E { static string Name() { return "CONNECT"; } static Transition Type() { return Transition::Connect; } };69struct INIT_TASK_E { static string Name() { return "INIT_TASK"; } static Transition Type() { return Transition::InitTask; } };70struct RUN_E { static string Name() { return "RUN"; } static Transition Type() { return Transition::Run; } };71struct STOP_E { static string Name() { return "STOP"; } static Transition Type() { return Transition::Stop; } };72struct RESET_TASK_E { static string Name() { return "RESET_TASK"; } static Transition Type() { return Transition::ResetTask; } };73struct RESET_DEVICE_E { static string Name() { return "RESET_DEVICE"; } static Transition Type() { return Transition::ResetDevice; } };74struct END_E { static string Name() { return "END"; } static Transition Type() { return Transition::End; } };75struct ERROR_FOUND_E { static string Name() { return "ERROR_FOUND"; } static Transition Type() { return Transition::ErrorFound; } };76 77// defining the boost MSM state machine78struct Machine_ : public state_machine_def<Machine_>79{80 public:81 Machine_()82 : fState(State::Ok)83 , fNewState(State::Ok)84 , fLastTransitionResult(true)85 , fNewStatePending(false)86 {}87 88 // initial states89 using initial_state = bmpl::vector<IDLE_S, OK_S>;90 91 template<typename Transition, typename FSM>92 void on_entry(Transition const&, FSM& /* fsm */)93 {94 LOG(state) << "Starting FairMQ state machine --> IDLE";95 fState = State::Idle;96 }97 98 template<typename Transition, typename FSM>99 void on_exit(Transition const&, FSM& /*fsm*/)100 {101 LOG(state) << "Exiting FairMQ state machine";102 }103 104 struct DefaultFct105 {106 template<typename EVT, typename FSM, typename SourceState, typename TargetState>107 void operator()(EVT const& e, FSM& fsm, SourceState& /* ss */, TargetState& ts)108 {109 fsm.fNewState = ts.Type();110 fsm.fLastTransitionResult = true;111 fsm.CallNewTransitionCallbacks(e.Type());112 fsm.fNewStatePending = true;113 fsm.fNewStatePendingCV.notify_all();114 }115 };116 117 struct transition_table : bmpl::vector<118 // Start Transition Next Action Guard119 Row<IDLE_S, END_E, EXITING_S, DefaultFct, none>,120 Row<IDLE_S, INIT_DEVICE_E, INITIALIZING_DEVICE_S, DefaultFct, none>,121 122 Row<INITIALIZING_DEVICE_S, COMPLETE_INIT_E, INITIALIZED_S, DefaultFct, none>,123 Row<INITIALIZED_S, BIND_E, BINDING_S, DefaultFct, none>,124 Row<INITIALIZED_S, RESET_DEVICE_E, RESETTING_DEVICE_S, DefaultFct, none>,125 126 Row<BINDING_S, AUTO_E, BOUND_S, DefaultFct, none>,127 Row<BOUND_S, CONNECT_E, CONNECTING_S, DefaultFct, none>,128 Row<BOUND_S, RESET_DEVICE_E, RESETTING_DEVICE_S, DefaultFct, none>,129 130 Row<CONNECTING_S, AUTO_E, DEVICE_READY_S, DefaultFct, none>,131 Row<DEVICE_READY_S, INIT_TASK_E, INITIALIZING_TASK_S, DefaultFct, none>,132 Row<DEVICE_READY_S, RESET_DEVICE_E, RESETTING_DEVICE_S, DefaultFct, none>,133 134 Row<INITIALIZING_TASK_S, AUTO_E, READY_S, DefaultFct, none>,135 136 Row<READY_S, RUN_E, RUNNING_S, DefaultFct, none>,137 Row<READY_S, RESET_TASK_E, RESETTING_TASK_S, DefaultFct, none>,138 139 Row<RUNNING_S, STOP_E, READY_S, DefaultFct, none>,140 141 Row<RESETTING_TASK_S, AUTO_E, DEVICE_READY_S, DefaultFct, none>,142 Row<RESETTING_DEVICE_S, AUTO_E, IDLE_S, DefaultFct, none>,143 144 Row<OK_S, ERROR_FOUND_E, ERROR_S, DefaultFct, none>> {};145 146 void CallStateChangeCallbacks(const State state) const147 {148 if (!fStateChangeSignal.empty()) {149 fStateChangeSignal(state);150 }151 }152 153 void CallStateHandler(const State state) const154 {155 if (!fStateHandleSignal.empty()) {156 fStateHandleSignal(state);157 }158 }159 160 void CallStatePrep(const State state) const161 {162 if (!fStatePrepSignal.empty()) {163 fStatePrepSignal(state);164 }165 }166 167 void CallNewTransitionCallbacks(const Transition transition) const168 {169 if (!fNewTransitionSignal.empty()) {170 fNewTransitionSignal(transition);171 }172 }173 174 175 atomic<State> fState;176 atomic<State> fNewState;177 atomic<bool> fLastTransitionResult;178 179 mutex fStateMtx;180 mutex fSubscriptionsMtx;181 atomic<bool> fNewStatePending;182 condition_variable fNewStatePendingCV;183 184 boost::signals2::signal<void(const State)> fStateChangeSignal;185 boost::signals2::signal<void(const State)> fStateHandleSignal;186 boost::signals2::signal<void(const State)> fStatePrepSignal;187 boost::signals2::signal<void(const Transition)> fNewTransitionSignal;188 unordered_map<string, boost::signals2::connection> fStateChangeSignalsMap;189 unordered_map<string, boost::signals2::connection> fNewTransitionSignalsMap;190 191 void ProcessWork()192 {193 bool stop = false;194 195 while (!stop) {196 {197 unique_lock<mutex> lock(fStateMtx);198 199 fNewStatePendingCV.wait(lock, [this]{ return fNewStatePending.load(); });200 201 LOG(state) << fState << " ---> " << fNewState;202 fState = static_cast<State>(fNewState);203 fNewStatePending = false;204 205 if (fState == State::Exiting || fState == State::Error) {206 stop = true;207 }208 }209 210 CallStatePrep(fState);211 CallStateChangeCallbacks(fState);212 CallStateHandler(fState);213 }214 215 if (fState == State::Error) {216 LOG(trace) << "Device transitioned to error state";217 throw StateMachine::ErrorStateException("Device transitioned to error state");218 }219 }220 221 // replaces the default no-transition response.222 template<typename FSM, typename Transition>223 void no_transition(Transition const& t, FSM& fsm, int state)224 {225 using RecursiveStt = typename recursive_get_transition_table<FSM>::type;226 using AllStates = typename generate_state_set<RecursiveStt>::type;227 228 string stateName;229 230 bmpl::for_each<AllStates, wrap<bmpl::placeholders::_1>>(get_state_name<RecursiveStt>(stateName, state));231 232 stateName = boost::core::demangle(stateName.c_str());233 size_t pos = stateName.rfind(':');234 stateName = stateName.substr(pos + 1);235 size_t pos2 = stateName.rfind('_');236 stateName = stateName.substr(0, pos2);237 238 if (stateName != "OK") {239 LOG(state) << "No transition from state " << stateName << " on transition " << t.Name();240 }241 fsm.fLastTransitionResult = false;242 }243}; // Machine_244 245using FairMQFSM = state_machine<Machine_>;246 247} // namespace fsm248} // namespace fair::mq249 250using namespace fair::mq::fsm;251using namespace fair::mq;252 253StateMachine::StateMachine() : fFsm(new FairMQFSM) {}254void StateMachine::Start() { static_pointer_cast<FairMQFSM>(fFsm)->start(); }255StateMachine::~StateMachine() { static_pointer_cast<FairMQFSM>(fFsm)->stop(); }256 257bool StateMachine::ChangeState(Transition transition)258try {259 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);260 lock_guard<mutex> lock(fsm->fStateMtx);261 if (!static_cast<bool>(fsm->fNewStatePending) || transition == Transition::ErrorFound) {262 switch (transition) {263 case Transition::Auto:264 fsm->process_event(AUTO_E());265 return fsm->fLastTransitionResult;266 case Transition::InitDevice:267 fsm->process_event(INIT_DEVICE_E());268 return fsm->fLastTransitionResult;269 case Transition::CompleteInit:270 fsm->process_event(COMPLETE_INIT_E());271 return fsm->fLastTransitionResult;272 case Transition::Bind:273 fsm->process_event(BIND_E());274 return fsm->fLastTransitionResult;275 case Transition::Connect:276 fsm->process_event(CONNECT_E());277 return fsm->fLastTransitionResult;278 case Transition::InitTask:279 fsm->process_event(INIT_TASK_E());280 return fsm->fLastTransitionResult;281 case Transition::Run:282 fsm->process_event(RUN_E());283 return fsm->fLastTransitionResult;284 case Transition::Stop:285 fsm->process_event(STOP_E());286 return fsm->fLastTransitionResult;287 case Transition::ResetDevice:288 fsm->process_event(RESET_DEVICE_E());289 return fsm->fLastTransitionResult;290 case Transition::ResetTask:291 fsm->process_event(RESET_TASK_E());292 return fsm->fLastTransitionResult;293 case Transition::End:294 fsm->process_event(END_E());295 return fsm->fLastTransitionResult;296 case Transition::ErrorFound:297 fsm->process_event(ERROR_FOUND_E());298 return fsm->fLastTransitionResult;299 default:300 LOG(error) << "Requested unsupported state transition: " << transition << endl;301 return false;302 }303 } else {304 LOG(state) << "Transition " << GetTransitionName(transition) << " incoming, but another state transition is already ongoing.";305 return false;306 }307} catch (exception& e) {308 LOG(error) << "Exception in StateMachine::ChangeState(): " << e.what();309 return false;310}311 312void StateMachine::SubscribeToStateChange(const string& key, function<void(const State)> callback)313{314 // Check if the key has a integer value as prefix, if yes, decode it.315 int i = strtol(key.c_str(), nullptr, 10);316 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);317 lock_guard<mutex> lock(fsm->fSubscriptionsMtx);318 fsm->fStateChangeSignalsMap.insert({key, fsm->fStateChangeSignal.connect(i, callback)});319}320 321void StateMachine::UnsubscribeFromStateChange(const string& key)322{323 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);324 lock_guard<mutex> lock(fsm->fSubscriptionsMtx);325 if (fsm->fStateChangeSignalsMap.count(key)) {326 fsm->fStateChangeSignalsMap.at(key).disconnect();327 fsm->fStateChangeSignalsMap.erase(key);328 }329}330 331void StateMachine::PrepareState(std::function<void(const State)> callback)332{333 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);334 if (fsm->fStatePrepSignal.empty()) {335 fsm->fStatePrepSignal.connect(callback);336 } else {337 LOG(error) << "state preparation handler is already set";338 }339}340 341void StateMachine::HandleStates(function<void(const State)> callback)342{343 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);344 if (fsm->fStateHandleSignal.empty()) {345 fsm->fStateHandleSignal.connect(callback);346 } else {347 LOG(error) << "state handler is already set";348 }349}350 351void StateMachine::StopHandlingStates()352{353 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);354 if (!fsm->fStatePrepSignal.empty()) {355 fsm->fStatePrepSignal.disconnect_all_slots();356 }357 if (!fsm->fStateHandleSignal.empty()) {358 fsm->fStateHandleSignal.disconnect_all_slots();359 }360}361 362void StateMachine::SubscribeToNewTransition(const string& key, function<void(const Transition)> callback)363{364 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);365 lock_guard<mutex> lock(fsm->fSubscriptionsMtx);366 fsm->fNewTransitionSignalsMap.insert({key, fsm->fNewTransitionSignal.connect(callback)});367}368 369void StateMachine::UnsubscribeFromNewTransition(const string& key)370{371 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);372 lock_guard<mutex> lock(fsm->fSubscriptionsMtx);373 if (fsm->fNewTransitionSignalsMap.count(key)) {374 fsm->fNewTransitionSignalsMap.at(key).disconnect();375 fsm->fNewTransitionSignalsMap.erase(key);376 }377}378 379State StateMachine::GetCurrentState() const { return static_pointer_cast<FairMQFSM>(fFsm)->fState; }380string StateMachine::GetCurrentStateName() const { return GetStateName(static_pointer_cast<FairMQFSM>(fFsm)->fState); }381 382bool StateMachine::NewStatePending() const { return static_cast<bool>(static_pointer_cast<FairMQFSM>(fFsm)->fNewStatePending); }383void StateMachine::WaitForPendingState() const384{385 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);386 unique_lock<mutex> lock(fsm->fStateMtx);387 fsm->fNewStatePendingCV.wait(lock, [&]{ return static_cast<bool>(fsm->fNewStatePending); });388}389bool StateMachine::WaitForPendingStateFor(int durationInMs) const390{391 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);392 unique_lock<mutex> lock(fsm->fStateMtx);393 return fsm->fNewStatePendingCV.wait_for(lock, std::chrono::milliseconds(durationInMs), [&]{ return static_cast<bool>(fsm->fNewStatePending); });394}395 396void StateMachine::ProcessWork()397{398 auto fsm = static_pointer_cast<FairMQFSM>(fFsm);399 400 fair::mq::tools::CallOnDestruction cod([&](){401 LOG(debug) << "Exception caught in ProcessWork(), going to Error state";402 {403 lock_guard<mutex> lock(fsm->fStateMtx);404 fsm->fState = State::Error;405 fsm->CallStateChangeCallbacks(State::Error);406 }407 ChangeState(Transition::ErrorFound);408 });409 410 fsm->CallStateChangeCallbacks(State::Idle);411 fsm->ProcessWork();412 413 cod.disable();414}