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

Improve compilation speed

Browse Source
This commit is contained in:
Alexey Rybalchenko 2018-05-16 18:06:57 +02:00 committed by Dennis Klein
parent 436f79bee5
commit e4c349888d
24 changed files with 985 additions and 812 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
fairmq/CMakeLists.txt
View File

@ -148,6 +148,9 @@ set(FAIRMQ_SOURCE_FILES
shmem/Manager.cxx shmem/Manager.cxx
shmem/Monitor.cxx shmem/Monitor.cxx
shmem/Region.cxx shmem/Region.cxx
tools/Network.cxx
tools/Process.cxx
tools/Unique.cxx
zeromq/FairMQMessageZMQ.cxx zeromq/FairMQMessageZMQ.cxx
zeromq/FairMQPollerZMQ.cxx zeromq/FairMQPollerZMQ.cxx
zeromq/FairMQUnmanagedRegionZMQ.cxx zeromq/FairMQUnmanagedRegionZMQ.cxx

642
fairmq/FairMQStateMachine.cxx
View File

@ -14,14 +14,531 @@
#include "FairMQStateMachine.h" #include "FairMQStateMachine.h"
FairMQStateMachine::FairMQStateMachine() // Increase maximum number of boost::msm states (default is 10)
// This #define has to be before any msm header includes
#define FUSION_MAX_VECTOR_SIZE 20
#include <boost/mpl/for_each.hpp>
#include <boost/msm/back/state_machine.hpp>
#include <boost/msm/back/tools.hpp>
#include <boost/msm/back/metafunctions.hpp>
#include <boost/msm/front/state_machine_def.hpp>
#include <boost/msm/front/functor_row.hpp>
#include <boost/signals2.hpp> // signal/slot for onStateChange callbacks
#include <atomic>
#include <condition_variable>
#include <thread>
#include <chrono>
#include <unordered_map>
using namespace std;
namespace msmf = boost::msm::front;
namespace fair
{ {
start(); namespace mq
{
namespace fsm
{
// defining events for the boost MSM state machine
struct INIT_DEVICE_E { string name() const { return "INIT_DEVICE"; } };
struct internal_DEVICE_READY_E { string name() const { return "internal_DEVICE_READY"; } };
struct INIT_TASK_E { string name() const { return "INIT_TASK"; } };
struct internal_READY_E { string name() const { return "internal_READY"; } };
struct RUN_E { string name() const { return "RUN"; } };
struct PAUSE_E { string name() const { return "PAUSE"; } };
struct STOP_E { string name() const { return "STOP"; } };
struct RESET_TASK_E { string name() const { return "RESET_TASK"; } };
struct RESET_DEVICE_E { string name() const { return "RESET_DEVICE"; } };
struct internal_IDLE_E { string name() const { return "internal_IDLE"; } };
struct END_E { string name() const { return "END"; } };
struct ERROR_FOUND_E { string name() const { return "ERROR_FOUND"; } };
// deactivate the warning for non-virtual destructor thrown in the boost library
#if defined(__clang__)
_Pragma("clang diagnostic push")
_Pragma("clang diagnostic ignored \"-Wnon-virtual-dtor\"")
#elif defined(__GNUC__) || defined(__GNUG__)
_Pragma("GCC diagnostic push")
_Pragma("GCC diagnostic ignored \"-Wnon-virtual-dtor\"")
#endif
// defining the boost MSM state machine
struct Machine_ : public msmf::state_machine_def<Machine_>
{
public:
Machine_()
: fState()
, fWork()
, fWorkAvailableCondition()
, fWorkDoneCondition()
, fWorkMutex()
, fWorkerTerminated(false)
, fWorkActive(false)
, fWorkAvailable(false)
, fStateChangeSignal()
, fStateChangeSignalsMap()
, fTerminationRequested(false)
, fWorkerThread()
{}
virtual ~Machine_()
{}
template<typename Event, typename FSM>
void on_entry(Event const&, FSM& fsm)
{
LOG(state) << "Starting FairMQ state machine";
fState = FairMQStateMachine::IDLE;
fsm.CallStateChangeCallbacks(FairMQStateMachine::IDLE);
// start a worker thread to execute user states in.
fsm.fWorkerThread = thread(&Machine_::Worker, &fsm);
}
template<typename Event, typename FSM>
void on_exit(Event const&, FSM& /*fsm*/)
{
LOG(state) << "Exiting FairMQ state machine";
}
// list of FSM states
struct OK_FSM : public msmf::state<> {};
struct ERROR_FSM : public msmf::terminate_state<> {};
struct IDLE_FSM : public msmf::state<> {};
struct INITIALIZING_DEVICE_FSM : public msmf::state<> {};
struct DEVICE_READY_FSM : public msmf::state<> {};
struct INITIALIZING_TASK_FSM : public msmf::state<> {};
struct READY_FSM : public msmf::state<> {};
struct RUNNING_FSM : public msmf::state<> {};
struct PAUSED_FSM : public msmf::state<> {};
struct RESETTING_TASK_FSM : public msmf::state<> {};
struct RESETTING_DEVICE_FSM : public msmf::state<> {};
struct EXITING_FSM : public msmf::state<> {};
// initial states
using initial_state = boost::mpl::vector<IDLE_FSM, OK_FSM>;
// actions
struct IdleFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering IDLE state";
fsm.fState = FairMQStateMachine::IDLE;
}
};
struct InitDeviceFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::INITIALIZING_DEVICE;
unique_lock<mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering INITIALIZING DEVICE state";
fsm.fWork = fsm.fInitWrapperHandler;
fsm.fWorkAvailableCondition.notify_one();
}
};
struct DeviceReadyFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering DEVICE READY state";
fsm.fState = FairMQStateMachine::DEVICE_READY;
}
};
struct InitTaskFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::INITIALIZING_TASK;
unique_lock<mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering INITIALIZING TASK state";
fsm.fWork = fsm.fInitTaskWrapperHandler;
fsm.fWorkAvailableCondition.notify_one();
}
};
struct ReadyFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering READY state";
fsm.fState = FairMQStateMachine::READY;
}
};
struct RunFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::RUNNING;
unique_lock<mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering RUNNING state";
fsm.fWork = fsm.fRunWrapperHandler;
fsm.fWorkAvailableCondition.notify_one();
}
};
struct PauseFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::PAUSED;
fsm.fUnblockHandler();
unique_lock<mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering PAUSED state";
fsm.fWork = fsm.fPauseWrapperHandler;
fsm.fWorkAvailableCondition.notify_one();
}
};
struct ResumeFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::RUNNING;
unique_lock<mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering RUNNING state";
fsm.fWork = fsm.fRunWrapperHandler;
fsm.fWorkAvailableCondition.notify_one();
}
};
struct StopFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::READY;
fsm.fUnblockHandler();
unique_lock<mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
LOG(state) << "Entering READY state";
}
};
struct InternalStopFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::READY;
fsm.fUnblockHandler();
LOG(state) << "RUNNING state finished without an external event, entering READY state";
}
};
struct ResetTaskFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::RESETTING_TASK;
unique_lock<mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering RESETTING TASK state";
fsm.fWork = fsm.fResetTaskWrapperHandler;
fsm.fWorkAvailableCondition.notify_one();
}
};
struct ResetDeviceFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = FairMQStateMachine::RESETTING_DEVICE;
unique_lock<mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering RESETTING DEVICE state";
fsm.fWork = fsm.fResetWrapperHandler;
fsm.fWorkAvailableCondition.notify_one();
}
};
struct ExitingFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering EXITING state";
fsm.fState = FairMQStateMachine::EXITING;
fsm.fTerminationRequested = true;
fsm.CallStateChangeCallbacks(FairMQStateMachine::EXITING);
// terminate worker thread
{
lock_guard<mutex> lock(fsm.fWorkMutex);
fsm.fWorkerTerminated = true;
fsm.fWorkAvailableCondition.notify_one();
}
// join the worker thread (executing user states)
if (fsm.fWorkerThread.joinable())
{
fsm.fWorkerThread.join();
}
fsm.fExitHandler();
}
};
struct ErrorFoundFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering ERROR state";
fsm.fState = FairMQStateMachine::Error;
fsm.CallStateChangeCallbacks(FairMQStateMachine::Error);
}
};
// Transition table for Machine_
struct transition_table : boost::mpl::vector<
// Start Event Next Action Guard
msmf::Row<IDLE_FSM, INIT_DEVICE_E, INITIALIZING_DEVICE_FSM, InitDeviceFct, msmf::none>,
msmf::Row<IDLE_FSM, END_E, EXITING_FSM, ExitingFct, msmf::none>,
msmf::Row<INITIALIZING_DEVICE_FSM, internal_DEVICE_READY_E, DEVICE_READY_FSM, DeviceReadyFct, msmf::none>,
msmf::Row<DEVICE_READY_FSM, INIT_TASK_E, INITIALIZING_TASK_FSM, InitTaskFct, msmf::none>,
msmf::Row<DEVICE_READY_FSM, RESET_DEVICE_E, RESETTING_DEVICE_FSM, ResetDeviceFct, msmf::none>,
msmf::Row<INITIALIZING_TASK_FSM, internal_READY_E, READY_FSM, ReadyFct, msmf::none>,
msmf::Row<READY_FSM, RUN_E, RUNNING_FSM, RunFct, msmf::none>,
msmf::Row<READY_FSM, RESET_TASK_E, RESETTING_TASK_FSM, ResetTaskFct, msmf::none>,
msmf::Row<RUNNING_FSM, PAUSE_E, PAUSED_FSM, PauseFct, msmf::none>,
msmf::Row<RUNNING_FSM, STOP_E, READY_FSM, StopFct, msmf::none>,
msmf::Row<RUNNING_FSM, internal_READY_E, READY_FSM, InternalStopFct, msmf::none>,
msmf::Row<PAUSED_FSM, RUN_E, RUNNING_FSM, ResumeFct, msmf::none>,
msmf::Row<RESETTING_TASK_FSM, internal_DEVICE_READY_E, DEVICE_READY_FSM, DeviceReadyFct, msmf::none>,
msmf::Row<RESETTING_DEVICE_FSM, internal_IDLE_E, IDLE_FSM, IdleFct, msmf::none>,
msmf::Row<OK_FSM, ERROR_FOUND_E, ERROR_FSM, ErrorFoundFct, msmf::none>>
{};
// replaces the default no-transition response.
template<typename FSM, typename Event>
void no_transition(Event const& e, FSM&, int state)
{
using recursive_stt = typename boost::msm::back::recursive_get_transition_table<FSM>::type;
using all_states = typename boost::msm::back::generate_state_set<recursive_stt>::type;
string stateName;
boost::mpl::for_each<all_states, boost::msm::wrap<boost::mpl::placeholders::_1>>(boost::msm::back::get_state_name<recursive_stt>(stateName, state));
stateName = stateName.substr(24);
size_t pos = stateName.find("_FSME");
stateName.erase(pos);
if (stateName == "1RUNNING" || stateName == "6DEVICE_READY" || stateName == "0PAUSED" || stateName == "8RESETTING_TASK" || stateName == "0RESETTING_DEVICE")
{
stateName = stateName.substr(1);
}
if (stateName != "OK")
{
LOG(state) << "No transition from state " << stateName << " on event " << e.name();
}
// LOG(state) << "no transition from state " << GetStateName(state) << " (" << state << ") on event " << e.name();
}
static string GetStateName(const int state)
{
switch(state)
{
case FairMQStateMachine::OK:
return "OK";
case FairMQStateMachine::Error:
return "Error";
case FairMQStateMachine::IDLE:
return "IDLE";
case FairMQStateMachine::INITIALIZING_DEVICE:
return "INITIALIZING_DEVICE";
case FairMQStateMachine::DEVICE_READY:
return "DEVICE_READY";
case FairMQStateMachine::INITIALIZING_TASK:
return "INITIALIZING_TASK";
case FairMQStateMachine::READY:
return "READY";
case FairMQStateMachine::RUNNING:
return "RUNNING";
case FairMQStateMachine::PAUSED:
return "PAUSED";
case FairMQStateMachine::RESETTING_TASK:
return "RESETTING_TASK";
case FairMQStateMachine::RESETTING_DEVICE:
return "RESETTING_DEVICE";
case FairMQStateMachine::EXITING:
return "EXITING";
default:
return "requested name for non-existent state...";
}
}
void CallStateChangeCallbacks(const FairMQStateMachine::State state) const
{
if (!fStateChangeSignal.empty())
{
fStateChangeSignal(state);
}
}
function<void(void)> fInitWrapperHandler;
function<void(void)> fInitTaskWrapperHandler;
function<void(void)> fRunWrapperHandler;
function<void(void)> fPauseWrapperHandler;
function<void(void)> fResetWrapperHandler;
function<void(void)> fResetTaskWrapperHandler;
function<void(void)> fExitHandler;
function<void(void)> fUnblockHandler;
// function to execute user states in a worker thread
function<void(void)> fWork;
condition_variable fWorkAvailableCondition;
condition_variable fWorkDoneCondition;
mutex fWorkMutex;
bool fWorkerTerminated;
bool fWorkActive;
bool fWorkAvailable;
boost::signals2::signal<void(const FairMQStateMachine::State)> fStateChangeSignal;
unordered_map<string, boost::signals2::connection> fStateChangeSignalsMap;
atomic<bool> fTerminationRequested;
atomic<FairMQStateMachine::State> fState;
private:
void Worker()
{
while (true)
{
{
unique_lock<mutex> lock(fWorkMutex);
// Wait for work to be done.
while (!fWorkAvailable && !fWorkerTerminated)
{
fWorkAvailableCondition.wait(lock);
}
if (fWorkerTerminated)
{
break;
}
fWorkActive = true;
}
fWork();
{
lock_guard<mutex> lock(fWorkMutex);
fWorkActive = false;
fWorkAvailable = false;
fWorkDoneCondition.notify_one();
}
CallStateChangeCallbacks(fState);
}
}
// run state handlers in a separate thread
thread fWorkerThread;
}; // Machine_
using FairMQFSM = boost::msm::back::state_machine<Machine_>;
// reactivate the warning for non-virtual destructor
#if defined(__clang__)
_Pragma("clang diagnostic pop")
#elif defined(__GNUC__) || defined(__GNUG__)
_Pragma("GCC diagnostic pop")
#endif
} // namespace fsm
} // namespace mq
} // namespace fair
using namespace fair::mq::fsm;
FairMQStateMachine::FairMQStateMachine()
: fFsm(new FairMQFSM)
, fChangeStateMutex()
{
static_pointer_cast<FairMQFSM>(fFsm)->fInitWrapperHandler = bind(&FairMQStateMachine::InitWrapper, this);
static_pointer_cast<FairMQFSM>(fFsm)->fInitTaskWrapperHandler = bind(&FairMQStateMachine::InitTaskWrapper, this);
static_pointer_cast<FairMQFSM>(fFsm)->fRunWrapperHandler = bind(&FairMQStateMachine::RunWrapper, this);
static_pointer_cast<FairMQFSM>(fFsm)->fPauseWrapperHandler = bind(&FairMQStateMachine::PauseWrapper, this);
static_pointer_cast<FairMQFSM>(fFsm)->fResetWrapperHandler = bind(&FairMQStateMachine::ResetWrapper, this);
static_pointer_cast<FairMQFSM>(fFsm)->fResetTaskWrapperHandler = bind(&FairMQStateMachine::ResetTaskWrapper, this);
static_pointer_cast<FairMQFSM>(fFsm)->fExitHandler = bind(&FairMQStateMachine::Exit, this);
static_pointer_cast<FairMQFSM>(fFsm)->fUnblockHandler = bind(&FairMQStateMachine::Unblock, this);
static_pointer_cast<FairMQFSM>(fFsm)->start();
} }
FairMQStateMachine::~FairMQStateMachine() FairMQStateMachine::~FairMQStateMachine()
{ {
stop(); static_pointer_cast<FairMQFSM>(fFsm)->stop();
} }
int FairMQStateMachine::GetInterfaceVersion() const int FairMQStateMachine::GetInterfaceVersion() const
@ -37,85 +554,85 @@ bool FairMQStateMachine::ChangeState(int event)
{ {
case INIT_DEVICE: case INIT_DEVICE:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::INIT_DEVICE()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(INIT_DEVICE_E());
return true; return true;
} }
case internal_DEVICE_READY: case internal_DEVICE_READY:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::internal_DEVICE_READY()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(internal_DEVICE_READY_E());
return true; return true;
} }
case INIT_TASK: case INIT_TASK:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::INIT_TASK()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(INIT_TASK_E());
return true; return true;
} }
case internal_READY: case internal_READY:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::internal_READY()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(internal_READY_E());
return true; return true;
} }
case RUN: case RUN:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::RUN()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(RUN_E());
return true; return true;
} }
case PAUSE: case PAUSE:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::PAUSE()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(PAUSE_E());
return true; return true;
} }
case STOP: case STOP:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::STOP()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(STOP_E());
return true; return true;
} }
case RESET_DEVICE: case RESET_DEVICE:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::RESET_DEVICE()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(RESET_DEVICE_E());
return true; return true;
} }
case RESET_TASK: case RESET_TASK:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::RESET_TASK()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(RESET_TASK_E());
return true; return true;
} }
case internal_IDLE: case internal_IDLE:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::internal_IDLE()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(internal_IDLE_E());
return true; return true;
} }
case END: case END:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::END()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(END_E());
return true; return true;
} }
case ERROR_FOUND: case ERROR_FOUND:
{ {
std::lock_guard<std::mutex> lock(fChangeStateMutex); lock_guard<mutex> lock(fChangeStateMutex);
process_event(fair::mq::fsm::ERROR_FOUND()); static_pointer_cast<FairMQFSM>(fFsm)->process_event(ERROR_FOUND_E());
return true; return true;
} }
default: default:
{ {
LOG(error) << "Requested state transition with an unsupported event: " << event << std::endl LOG(error) << "Requested state transition with an unsupported event: " << event << endl
<< "Supported are: INIT_DEVICE, INIT_TASK, RUN, PAUSE, STOP, RESET_TASK, RESET_DEVICE, END, ERROR_FOUND"; << "Supported are: INIT_DEVICE, INIT_TASK, RUN, PAUSE, STOP, RESET_TASK, RESET_DEVICE, END, ERROR_FOUND";
return false; return false;
} }
} }
} }
catch (std::exception& e) catch (exception& e)
{ {
LOG(error) << "Exception in FairMQStateMachine::ChangeState(): " << e.what(); LOG(error) << "Exception in FairMQStateMachine::ChangeState(): " << e.what();
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
@ -123,7 +640,7 @@ bool FairMQStateMachine::ChangeState(int event)
return false; return false;
} }
bool FairMQStateMachine::ChangeState(const std::string& event) bool FairMQStateMachine::ChangeState(const string& event)
{ {
return ChangeState(GetEventNumber(event)); return ChangeState(GetEventNumber(event));
} }
@ -140,10 +657,10 @@ void FairMQStateMachine::WaitForEndOfState(int event)
case RESET_TASK: case RESET_TASK:
case RESET_DEVICE: case RESET_DEVICE:
{ {
std::unique_lock<std::mutex> lock(fWorkMutex); unique_lock<mutex> lock(static_pointer_cast<FairMQFSM>(fFsm)->fWorkMutex);
while (fWorkActive || fWorkAvailable) while (static_pointer_cast<FairMQFSM>(fFsm)->fWorkActive || static_pointer_cast<FairMQFSM>(fFsm)->fWorkAvailable)
{ {
fWorkDoneCondition.wait_for(lock, std::chrono::seconds(1)); static_pointer_cast<FairMQFSM>(fFsm)->fWorkDoneCondition.wait_for(lock, chrono::seconds(1));
} }
break; break;
@ -153,13 +670,13 @@ void FairMQStateMachine::WaitForEndOfState(int event)
break; break;
} }
} }
catch (std::exception& e) catch (exception& e)
{ {
LOG(error) << "Exception in FairMQStateMachine::WaitForEndOfState(): " << e.what(); LOG(error) << "Exception in FairMQStateMachine::WaitForEndOfState(): " << e.what();
} }
} }
void FairMQStateMachine::WaitForEndOfState(const std::string& event) void FairMQStateMachine::WaitForEndOfState(const string& event)
{ {
return WaitForEndOfState(GetEventNumber(event)); return WaitForEndOfState(GetEventNumber(event));
} }
@ -176,11 +693,11 @@ bool FairMQStateMachine::WaitForEndOfStateForMs(int event, int durationInMs)
case RESET_TASK: case RESET_TASK:
case RESET_DEVICE: case RESET_DEVICE:
{ {
std::unique_lock<std::mutex> lock(fWorkMutex); unique_lock<mutex> lock(static_pointer_cast<FairMQFSM>(fFsm)->fWorkMutex);
while (fWorkActive || fWorkAvailable) while (static_pointer_cast<FairMQFSM>(fFsm)->fWorkActive || static_pointer_cast<FairMQFSM>(fFsm)->fWorkAvailable)
{ {
fWorkDoneCondition.wait_for(lock, std::chrono::milliseconds(durationInMs)); static_pointer_cast<FairMQFSM>(fFsm)->fWorkDoneCondition.wait_for(lock, chrono::milliseconds(durationInMs));
if (fWorkActive) if (static_pointer_cast<FairMQFSM>(fFsm)->fWorkActive)
{ {
return false; return false;
} }
@ -192,32 +709,59 @@ bool FairMQStateMachine::WaitForEndOfStateForMs(int event, int durationInMs)
return false; return false;
} }
} }
catch (std::exception& e) catch (exception& e)
{ {
LOG(error) << "Exception in FairMQStateMachine::WaitForEndOfStateForMs(): " << e.what(); LOG(error) << "Exception in FairMQStateMachine::WaitForEndOfStateForMs(): " << e.what();
} }
return false; return false;
} }
bool FairMQStateMachine::WaitForEndOfStateForMs(const std::string& event, int durationInMs) bool FairMQStateMachine::WaitForEndOfStateForMs(const string& event, int durationInMs)
{ {
return WaitForEndOfStateForMs(GetEventNumber(event), durationInMs); return WaitForEndOfStateForMs(GetEventNumber(event), durationInMs);
} }
void FairMQStateMachine::SubscribeToStateChange(const std::string& key, std::function<void(const State)> callback) void FairMQStateMachine::SubscribeToStateChange(const string& key, function<void(const State)> callback)
{ {
fStateChangeSignalsMap.insert({key, fStateChangeSignal.connect(callback)}); static_pointer_cast<FairMQFSM>(fFsm)->fStateChangeSignalsMap.insert({key, static_pointer_cast<FairMQFSM>(fFsm)->fStateChangeSignal.connect(callback)});
} }
void FairMQStateMachine::UnsubscribeFromStateChange(const std::string& key) void FairMQStateMachine::UnsubscribeFromStateChange(const string& key)
{ {
if (fStateChangeSignalsMap.count(key)) if (static_pointer_cast<FairMQFSM>(fFsm)->fStateChangeSignalsMap.count(key))
{ {
fStateChangeSignalsMap.at(key).disconnect(); static_pointer_cast<FairMQFSM>(fFsm)->fStateChangeSignalsMap.at(key).disconnect();
fStateChangeSignalsMap.erase(key); static_pointer_cast<FairMQFSM>(fFsm)->fStateChangeSignalsMap.erase(key);
} }
} }
int FairMQStateMachine::GetEventNumber(const std::string& event) void FairMQStateMachine::CallStateChangeCallbacks(const State state) const
{
static_pointer_cast<FairMQFSM>(fFsm)->CallStateChangeCallbacks(state);
}
string FairMQStateMachine::GetCurrentStateName() const
{
return static_pointer_cast<FairMQFSM>(fFsm)->GetStateName(static_pointer_cast<FairMQFSM>(fFsm)->fState);
}
int FairMQStateMachine::GetCurrentState() const
{
return static_pointer_cast<FairMQFSM>(fFsm)->fState;
}
bool FairMQStateMachine::CheckCurrentState(int state) const
{
return state == static_pointer_cast<FairMQFSM>(fFsm)->fState;
}
bool FairMQStateMachine::CheckCurrentState(string state) const
{
return state == GetCurrentStateName();
}
bool FairMQStateMachine::Terminated()
{
return static_pointer_cast<FairMQFSM>(fFsm)->fTerminationRequested;
}
int FairMQStateMachine::GetEventNumber(const string& event)
{ {
if (event == "INIT_DEVICE") return INIT_DEVICE; if (event == "INIT_DEVICE") return INIT_DEVICE;
if (event == "INIT_TASK") return INIT_TASK; if (event == "INIT_TASK") return INIT_TASK;
@ -228,7 +772,7 @@ int FairMQStateMachine::GetEventNumber(const std::string& event)
if (event == "RESET_TASK") return RESET_TASK; if (event == "RESET_TASK") return RESET_TASK;
if (event == "END") return END; if (event == "END") return END;
if (event == "ERROR_FOUND") return ERROR_FOUND; if (event == "ERROR_FOUND") return ERROR_FOUND;
LOG(error) << "Requested number for non-existent event... " << event << std::endl LOG(error) << "Requested number for non-existent event... " << event << endl
<< "Supported are: INIT_DEVICE, INIT_TASK, RUN, PAUSE, STOP, RESET_DEVICE, RESET_TASK, END, ERROR_FOUND"; << "Supported are: INIT_DEVICE, INIT_TASK, RUN, PAUSE, STOP, RESET_DEVICE, RESET_TASK, END, ERROR_FOUND";
return -1; return -1;
} }

589
fairmq/FairMQStateMachine.h
View File

@ -17,555 +17,14 @@
#define FAIRMQ_INTERFACE_VERSION 3 #define FAIRMQ_INTERFACE_VERSION 3
#include <string>
#include <atomic>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <chrono>
#include <functional>
#include <unordered_map>
// Increase maximum number of boost::msm states (default is 10)
// This #define has to be before any msm header includes
#define FUSION_MAX_VECTOR_SIZE 20
#include <boost/mpl/for_each.hpp>
#include <boost/msm/back/state_machine.hpp>
#include <boost/msm/back/tools.hpp>
#include <boost/msm/back/metafunctions.hpp>
#include <boost/msm/front/state_machine_def.hpp>
#include <boost/msm/front/functor_row.hpp>
#include <boost/signals2.hpp> // signal/slot for onStateChange callbacks
#include "FairMQLogger.h" #include "FairMQLogger.h"
namespace msmf = boost::msm::front; #include <string>
#include <memory>
#include <functional>
#include <mutex>
namespace fair class FairMQStateMachine
{
namespace mq
{
namespace fsm
{
// defining events for the boost MSM state machine
struct INIT_DEVICE { std::string name() const { return "INIT_DEVICE"; } };
struct internal_DEVICE_READY { std::string name() const { return "internal_DEVICE_READY"; } };
struct INIT_TASK { std::string name() const { return "INIT_TASK"; } };
struct internal_READY { std::string name() const { return "internal_READY"; } };
struct RUN { std::string name() const { return "RUN"; } };
struct PAUSE { std::string name() const { return "PAUSE"; } };
struct STOP { std::string name() const { return "STOP"; } };
struct RESET_TASK { std::string name() const { return "RESET_TASK"; } };
struct RESET_DEVICE { std::string name() const { return "RESET_DEVICE"; } };
struct internal_IDLE { std::string name() const { return "internal_IDLE"; } };
struct END { std::string name() const { return "END"; } };
struct ERROR_FOUND { std::string name() const { return "ERROR_FOUND"; } };
// deactivate the warning for non-virtual destructor thrown in the boost library
#if defined(__clang__)
_Pragma("clang diagnostic push")
_Pragma("clang diagnostic ignored \"-Wnon-virtual-dtor\"")
#elif defined(__GNUC__) || defined(__GNUG__)
_Pragma("GCC diagnostic push")
_Pragma("GCC diagnostic ignored \"-Wnon-virtual-dtor\"")
#endif
// defining the boost MSM state machine
struct FairMQFSM : public msmf::state_machine_def<FairMQFSM>
{
public:
FairMQFSM()
: fState()
, fChangeStateMutex()
, fWork()
, fWorkAvailableCondition()
, fWorkDoneCondition()
, fWorkMutex()
, fWorkerTerminated(false)
, fWorkActive(false)
, fWorkAvailable(false)
, fStateChangeSignal()
, fStateChangeSignalsMap()
, fTerminationRequested(false)
, fWorkerThread()
{}
virtual ~FairMQFSM()
{}
template<typename Event, typename FSM>
void on_entry(Event const&, FSM& fsm)
{
LOG(state) << "Starting FairMQ state machine";
fState = IDLE;
fsm.CallStateChangeCallbacks(IDLE);
// start a worker thread to execute user states in.
fsm.fWorkerThread = std::thread(&FairMQFSM::Worker, &fsm);
}
template<typename Event, typename FSM>
void on_exit(Event const&, FSM& /*fsm*/)
{
LOG(state) << "Exiting FairMQ state machine";
}
// list of FSM states
struct OK_FSM : public msmf::state<> {};
struct ERROR_FSM : public msmf::terminate_state<> {};
struct IDLE_FSM : public msmf::state<> {};
struct INITIALIZING_DEVICE_FSM : public msmf::state<> {};
struct DEVICE_READY_FSM : public msmf::state<> {};
struct INITIALIZING_TASK_FSM : public msmf::state<> {};
struct READY_FSM : public msmf::state<> {};
struct RUNNING_FSM : public msmf::state<> {};
struct PAUSED_FSM : public msmf::state<> {};
struct RESETTING_TASK_FSM : public msmf::state<> {};
struct RESETTING_DEVICE_FSM : public msmf::state<> {};
struct EXITING_FSM : public msmf::state<> {};
// initial states
using initial_state = boost::mpl::vector<IDLE_FSM, OK_FSM>;
// actions
struct IdleFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering IDLE state";
fsm.fState = IDLE;
}
};
struct InitDeviceFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = INITIALIZING_DEVICE;
std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering INITIALIZING DEVICE state";
fsm.fWork = std::bind(&FairMQFSM::InitWrapper, &fsm);
fsm.fWorkAvailableCondition.notify_one();
}
};
struct DeviceReadyFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering DEVICE READY state";
fsm.fState = DEVICE_READY;
}
};
struct InitTaskFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = INITIALIZING_TASK;
std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering INITIALIZING TASK state";
fsm.fWork = std::bind(&FairMQFSM::InitTaskWrapper, &fsm);
fsm.fWorkAvailableCondition.notify_one();
}
};
struct ReadyFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering READY state";
fsm.fState = READY;
}
};
struct RunFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = RUNNING;
std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering RUNNING state";
fsm.fWork = std::bind(&FairMQFSM::RunWrapper, &fsm);
fsm.fWorkAvailableCondition.notify_one();
}
};
struct PauseFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = PAUSED;
fsm.Unblock();
std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering PAUSED state";
fsm.fWork = std::bind(&FairMQFSM::PauseWrapper, &fsm);
fsm.fWorkAvailableCondition.notify_one();
}
};
struct ResumeFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = RUNNING;
std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering RUNNING state";
fsm.fWork = std::bind(&FairMQFSM::RunWrapper, &fsm);
fsm.fWorkAvailableCondition.notify_one();
}
};
struct StopFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = READY;
fsm.Unblock();
std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
LOG(state) << "Entering READY state";
}
};
struct InternalStopFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = READY;
fsm.Unblock();
LOG(state) << "RUNNING state finished without an external event, entering READY state";
}
};
struct ResetTaskFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = RESETTING_TASK;
std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering RESETTING TASK state";
fsm.fWork = std::bind(&FairMQFSM::ResetTaskWrapper, &fsm);
fsm.fWorkAvailableCondition.notify_one();
}
};
struct ResetDeviceFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
fsm.fState = RESETTING_DEVICE;
std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
while (fsm.fWorkActive)
{
fsm.fWorkDoneCondition.wait(lock);
}
fsm.fWorkAvailable = true;
LOG(state) << "Entering RESETTING DEVICE state";
fsm.fWork = std::bind(&FairMQFSM::ResetWrapper, &fsm);
fsm.fWorkAvailableCondition.notify_one();
}
};
struct ExitingFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering EXITING state";
fsm.fState = EXITING;
fsm.fTerminationRequested = true;
fsm.CallStateChangeCallbacks(EXITING);
// terminate worker thread
{
std::lock_guard<std::mutex> lock(fsm.fWorkMutex);
fsm.fWorkerTerminated = true;
fsm.fWorkAvailableCondition.notify_one();
}
// join the worker thread (executing user states)
if (fsm.fWorkerThread.joinable())
{
fsm.fWorkerThread.join();
}
fsm.Exit();
}
};
struct ErrorFoundFct
{
template<typename EVT, typename FSM, typename SourceState, typename TargetState>
void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
{
LOG(state) << "Entering ERROR state";
fsm.fState = Error;
fsm.CallStateChangeCallbacks(Error);
}
};
// Transition table for FairMQFSM
struct transition_table : boost::mpl::vector<
// Start Event Next Action Guard
msmf::Row<IDLE_FSM, INIT_DEVICE, INITIALIZING_DEVICE_FSM, InitDeviceFct, msmf::none>,
msmf::Row<IDLE_FSM, END, EXITING_FSM, ExitingFct, msmf::none>,
msmf::Row<INITIALIZING_DEVICE_FSM, internal_DEVICE_READY, DEVICE_READY_FSM, DeviceReadyFct, msmf::none>,
msmf::Row<DEVICE_READY_FSM, INIT_TASK, INITIALIZING_TASK_FSM, InitTaskFct, msmf::none>,
msmf::Row<DEVICE_READY_FSM, RESET_DEVICE, RESETTING_DEVICE_FSM, ResetDeviceFct, msmf::none>,
msmf::Row<INITIALIZING_TASK_FSM, internal_READY, READY_FSM, ReadyFct, msmf::none>,
msmf::Row<READY_FSM, RUN, RUNNING_FSM, RunFct, msmf::none>,
msmf::Row<READY_FSM, RESET_TASK, RESETTING_TASK_FSM, ResetTaskFct, msmf::none>,
msmf::Row<RUNNING_FSM, PAUSE, PAUSED_FSM, PauseFct, msmf::none>,
msmf::Row<RUNNING_FSM, STOP, READY_FSM, StopFct, msmf::none>,
msmf::Row<RUNNING_FSM, internal_READY, READY_FSM, InternalStopFct, msmf::none>,
msmf::Row<PAUSED_FSM, RUN, RUNNING_FSM, ResumeFct, msmf::none>,
msmf::Row<RESETTING_TASK_FSM, internal_DEVICE_READY, DEVICE_READY_FSM, DeviceReadyFct, msmf::none>,
msmf::Row<RESETTING_DEVICE_FSM, internal_IDLE, IDLE_FSM, IdleFct, msmf::none>,
msmf::Row<OK_FSM, ERROR_FOUND, ERROR_FSM, ErrorFoundFct, msmf::none>>
{};
// replaces the default no-transition response.
template<typename FSM, typename Event>
void no_transition(Event const& e, FSM&, int state)
{
using recursive_stt = typename boost::msm::back::recursive_get_transition_table<FSM>::type;
using all_states = typename boost::msm::back::generate_state_set<recursive_stt>::type;
std::string stateName;
boost::mpl::for_each<all_states, boost::msm::wrap<boost::mpl::placeholders::_1>>(boost::msm::back::get_state_name<recursive_stt>(stateName, state));
stateName = stateName.substr(24);
std::size_t pos = stateName.find("_FSME");
stateName.erase(pos);
if (stateName == "1RUNNING" || stateName == "6DEVICE_READY" || stateName == "0PAUSED" || stateName == "8RESETTING_TASK" || stateName == "0RESETTING_DEVICE")
{
stateName = stateName.substr(1);
}
if (stateName != "OK")
{
LOG(state) << "No transition from state " << stateName << " on event " << e.name();
}
// LOG(state) << "no transition from state " << GetStateName(state) << " (" << state << ") on event " << e.name();
}
// backward compatibility to FairMQStateMachine
enum State
{
OK,
Error,
IDLE,
INITIALIZING_DEVICE,
DEVICE_READY,
INITIALIZING_TASK,
READY,
RUNNING,
PAUSED,
RESETTING_TASK,
RESETTING_DEVICE,
EXITING
};
static std::string GetStateName(const int state)
{
switch(state)
{
case OK:
return "OK";
case Error:
return "Error";
case IDLE:
return "IDLE";
case INITIALIZING_DEVICE:
return "INITIALIZING_DEVICE";
case DEVICE_READY:
return "DEVICE_READY";
case INITIALIZING_TASK:
return "INITIALIZING_TASK";
case READY:
return "READY";
case RUNNING:
return "RUNNING";
case PAUSED:
return "PAUSED";
case RESETTING_TASK:
return "RESETTING_TASK";
case RESETTING_DEVICE:
return "RESETTING_DEVICE";
case EXITING:
return "EXITING";
default:
return "requested name for non-existent state...";
}
}
std::string GetCurrentStateName() const
{
return GetStateName(fState);
}
int GetCurrentState() const
{
return fState;
}
bool CheckCurrentState(int state) const
{
return state == fState;
}
bool CheckCurrentState(std::string state) const
{
return state == GetCurrentStateName();
}
// actions to be overwritten by derived classes
virtual void InitWrapper() {}
virtual void InitTaskWrapper() {}
virtual void RunWrapper() {}
virtual void PauseWrapper() {}
virtual void ResetWrapper() {}
virtual void ResetTaskWrapper() {}
virtual void Exit() {}
virtual void Unblock() {}
bool Terminated()
{
return fTerminationRequested;
}
protected:
std::atomic<State> fState;
std::mutex fChangeStateMutex;
// function to execute user states in a worker thread
std::function<void(void)> fWork;
std::condition_variable fWorkAvailableCondition;
std::condition_variable fWorkDoneCondition;
std::mutex fWorkMutex;
bool fWorkerTerminated;
bool fWorkActive;
bool fWorkAvailable;
boost::signals2::signal<void(const State)> fStateChangeSignal;
std::unordered_map<std::string, boost::signals2::connection> fStateChangeSignalsMap;
std::atomic<bool> fTerminationRequested;
void CallStateChangeCallbacks(const State state) const
{
if (!fStateChangeSignal.empty())
{
fStateChangeSignal(state);
}
}
private:
void Worker()
{
while (true)
{
{
std::unique_lock<std::mutex> lock(fWorkMutex);
// Wait for work to be done.
while (!fWorkAvailable && !fWorkerTerminated)
{
fWorkAvailableCondition.wait(lock);
}
if (fWorkerTerminated)
{
break;
}
fWorkActive = true;
}
fWork();
{
std::lock_guard<std::mutex> lock(fWorkMutex);
fWorkActive = false;
fWorkAvailable = false;
fWorkDoneCondition.notify_one();
}
CallStateChangeCallbacks(fState);
}
}
// run state handlers in a separate thread
std::thread fWorkerThread;
};
// reactivate the warning for non-virtual destructor
#if defined(__clang__)
_Pragma("clang diagnostic pop")
#elif defined(__GNUC__) || defined(__GNUG__)
_Pragma("GCC diagnostic pop")
#endif
} // namespace fsm
} // namespace mq
} // namespace fair
class FairMQStateMachine : public boost::msm::back::state_machine<fair::mq::fsm::FairMQFSM>
{ {
public: public:
enum Event enum Event
@ -584,6 +43,22 @@ class FairMQStateMachine : public boost::msm::back::state_machine<fair::mq::fsm:
ERROR_FOUND ERROR_FOUND
}; };
enum State
{
OK,
Error,
IDLE,
INITIALIZING_DEVICE,
DEVICE_READY,
INITIALIZING_TASK,
READY,
RUNNING,
PAUSED,
RESETTING_TASK,
RESETTING_DEVICE,
EXITING
};
FairMQStateMachine(); FairMQStateMachine();
virtual ~FairMQStateMachine(); virtual ~FairMQStateMachine();
@ -601,8 +76,30 @@ class FairMQStateMachine : public boost::msm::back::state_machine<fair::mq::fsm:
void SubscribeToStateChange(const std::string& key, std::function<void(const State)> callback); void SubscribeToStateChange(const std::string& key, std::function<void(const State)> callback);
void UnsubscribeFromStateChange(const std::string& key); void UnsubscribeFromStateChange(const std::string& key);
void CallStateChangeCallbacks(const State state) const;
std::string GetCurrentStateName() const;
int GetCurrentState() const;
bool CheckCurrentState(int state) const;
bool CheckCurrentState(std::string state) const;
bool Terminated();
// actions to be overwritten by derived classes
virtual void InitWrapper() {}
virtual void InitTaskWrapper() {}
virtual void RunWrapper() {}
virtual void PauseWrapper() {}
virtual void ResetWrapper() {}
virtual void ResetTaskWrapper() {}
virtual void Exit() {}
virtual void Unblock() {}
private: private:
int GetEventNumber(const std::string& event); int GetEventNumber(const std::string& event);
std::mutex fChangeStateMutex;
std::shared_ptr<void> fFsm;
}; };
#endif /* FAIRMQSTATEMACHINE_H_ */ #endif /* FAIRMQSTATEMACHINE_H_ */

1
fairmq/options/FairMQParser.cxx
View File

@ -15,6 +15,7 @@
#include "FairMQParser.h" #include "FairMQParser.h"
#include "FairMQLogger.h" #include "FairMQLogger.h"
#include <boost/property_tree/json_parser.hpp> #include <boost/property_tree/json_parser.hpp>
#include <boost/property_tree/ptree.hpp>
using namespace std; using namespace std;

2
fairmq/options/FairMQParser.h
View File

@ -13,7 +13,7 @@
#include <map> #include <map>
#include <unordered_map> #include <unordered_map>
#include <boost/property_tree/ptree.hpp> #include <boost/property_tree/ptree_fwd.hpp>
#include "FairMQChannel.h" #include "FairMQChannel.h"

1
fairmq/plugins/DDS/DDS.cxx
View File

@ -12,6 +12,7 @@
#include <boost/algorithm/string/join.hpp> #include <boost/algorithm/string/join.hpp>
#include <boost/algorithm/string/split.hpp> #include <boost/algorithm/string/split.hpp>
#include <boost/algorithm/string/classification.hpp>
#include <termios.h> // for the interactive mode #include <termios.h> // for the interactive mode
#include <poll.h> // for the interactive mode #include <poll.h> // for the interactive mode

192
fairmq/tools/Network.cxx Normal file
View File

@ -0,0 +1,192 @@
/********************************************************************************
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/tools/Network.h>
#ifndef _GNU_SOURCE
#define _GNU_SOURCE // To get defns of NI_MAXSERV and NI_MAXHOST
#endif
#include "FairMQLogger.h"
#include <sys/socket.h>
#include <sys/types.h>
#include <netdb.h>
#include <ifaddrs.h>
#include <stdio.h>
#include <boost/algorithm/string.hpp> // trim
#include <boost/asio.hpp>
#include <map>
#include <string>
#include <iostream>
#include <array>
#include <exception>
#include <algorithm>
using namespace std;
namespace fair
{
namespace mq
{
namespace tools
{
// returns a map with network interface names as keys and their IP addresses as values
int getHostIPs(map<string, string>& addressMap)
{
struct ifaddrs *ifaddr, *ifa;
int s;
char host[NI_MAXHOST];
if (getifaddrs(&ifaddr) == -1)
{
perror("getifaddrs");
return -1;
}
for (ifa = ifaddr; ifa != NULL; ifa = ifa->ifa_next)
{
if (ifa->ifa_addr == NULL)
{
continue;
}
if (ifa->ifa_addr->sa_family == AF_INET)
{
s = getnameinfo(ifa->ifa_addr, sizeof(struct sockaddr_in), host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST);
if (s != 0)
{
cout << "getnameinfo() failed: " << gai_strerror(s) << endl;
return -1;
}
addressMap.insert(pair<string, string>(ifa->ifa_name, host));
}
}
freeifaddrs(ifaddr);
return 0;
}
// get IP address of a given interface name
string getInterfaceIP(const string& interface)
{
map<string, string> IPs;
getHostIPs(IPs);
if (IPs.count(interface))
{
return IPs[interface];
}
else
{
LOG(error) << "Could not find provided network interface: \"" << interface << "\"!, exiting.";
return "";
}
}
// get name of the default route interface
string getDefaultRouteNetworkInterface()
{
array<char, 128> buffer;
string interfaceName;
#ifdef __APPLE__ // MacOS
unique_ptr<FILE, decltype(pclose) *> file(popen("route -n get default | grep interface | cut -d \":\" -f 2", "r"), pclose);
#else // Linux
unique_ptr<FILE, decltype(pclose) *> file(popen("ip route | grep default | cut -d \" \" -f 5 | head -n 1", "r"), pclose);
#endif
if (!file)
{
LOG(error) << "Could not detect default route network interface name - popen() failed!";
return "";
}
while (!feof(file.get()))
{
if (fgets(buffer.data(), 128, file.get()) != NULL)
{
interfaceName += buffer.data();
}
}
boost::algorithm::trim(interfaceName);
if (interfaceName == "")
{
LOG(error) << "Could not detect default route network interface name";
}
else
{
LOG(debug) << "Detected network interface name for the default route: " << interfaceName;
}
return interfaceName;
}
string getIpFromHostname(const string& hostname)
{
try {
namespace bai = boost::asio::ip;
boost::asio::io_service ios;
bai::tcp::resolver resolver(ios);
bai::tcp::resolver::query query(hostname, "");
bai::tcp::resolver::iterator end;
auto it = find_if(static_cast<bai::basic_resolver_iterator<bai::tcp>>(resolver.resolve(query)), end, [](const bai::tcp::endpoint& ep) {
return ep.address().is_v4();
});
if (it != end) {
stringstream ss;
ss << static_cast<bai::tcp::endpoint>(*it).address();
return ss.str();
}
LOG(warn) << "could not find ipv4 address for hostname '" << hostname << "'";
return "";
} catch (exception& e) {
LOG(error) << "could not resolve hostname '" << hostname << "', reason: " << e.what();
return "";
}
}
string getIpFromHostname(const string& hostname, boost::asio::io_service& ios)
{
try {
namespace bai = boost::asio::ip;
bai::tcp::resolver resolver(ios);
bai::tcp::resolver::query query(hostname, "");
bai::tcp::resolver::iterator end;
auto it = find_if(static_cast<bai::basic_resolver_iterator<bai::tcp>>(resolver.resolve(query)), end, [](const bai::tcp::endpoint& ep) {
return ep.address().is_v4();
});
if (it != end) {
stringstream ss;
ss << static_cast<bai::tcp::endpoint>(*it).address();
return ss.str();
}
LOG(warn) << "could not find ipv4 address for hostname '" << hostname << "'";
return "";
} catch (exception& e) {
LOG(error) << "could not resolve hostname '" << hostname << "', reason: " << e.what();
return "";
}
}
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */

176
fairmq/tools/Network.h
View File

@ -9,27 +9,20 @@
#ifndef FAIR_MQ_TOOLS_NETWORK_H #ifndef FAIR_MQ_TOOLS_NETWORK_H
#define FAIR_MQ_TOOLS_NETWORK_H #define FAIR_MQ_TOOLS_NETWORK_H
#ifndef _GNU_SOURCE
#define _GNU_SOURCE // To get defns of NI_MAXSERV and NI_MAXHOST
#endif
#include "FairMQLogger.h"
#include <sys/socket.h>
#include <sys/types.h>
#include <netdb.h>
#include <ifaddrs.h>
#include <stdio.h>
#include <boost/algorithm/string.hpp> // trim
#include <boost/asio.hpp>
#include <map> #include <map>
#include <string> #include <string>
#include <iostream>
#include <array> // forward declarations
#include <exception> namespace boost
#include <algorithm> {
namespace asio
{
class io_context;
typedef class io_context io_service;
} // namespace asio
} // namespace boost
namespace fair namespace fair
{ {
@ -39,152 +32,17 @@ namespace tools
{ {
// returns a map with network interface names as keys and their IP addresses as values // returns a map with network interface names as keys and their IP addresses as values
inline int getHostIPs(std::map<std::string, std::string>& addressMap) int getHostIPs(std::map<std::string, std::string>& addressMap);
{
struct ifaddrs *ifaddr, *ifa;
int s;
char host[NI_MAXHOST];
if (getifaddrs(&ifaddr) == -1)
{
perror("getifaddrs");
return -1;
}
for (ifa = ifaddr; ifa != NULL; ifa = ifa->ifa_next)
{
if (ifa->ifa_addr == NULL)
{
continue;
}
if (ifa->ifa_addr->sa_family == AF_INET)
{
s = getnameinfo(ifa->ifa_addr, sizeof(struct sockaddr_in), host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST);
if (s != 0)
{
std::cout << "getnameinfo() failed: " << gai_strerror(s) << std::endl;
return -1;
}
addressMap.insert(std::pair<std::string, std::string>(ifa->ifa_name, host));
}
}
freeifaddrs(ifaddr);
return 0;
}
// get IP address of a given interface name // get IP address of a given interface name
inline std::string getInterfaceIP(std::string interface) std::string getInterfaceIP(const std::string& interface);
{
std::map<std::string, std::string> IPs;
getHostIPs(IPs);
if (IPs.count(interface))
{
return IPs[interface];
}
else
{
LOG(error) << "Could not find provided network interface: \"" << interface << "\"!, exiting.";
return "";
}
}
// get name of the default route interface // get name of the default route interface
inline std::string getDefaultRouteNetworkInterface() std::string getDefaultRouteNetworkInterface();
{
std::array<char, 128> buffer;
std::string interfaceName;
#ifdef __APPLE__ // MacOS std::string getIpFromHostname(const std::string& hostname);
std::unique_ptr<FILE, decltype(pclose) *> file(popen("route -n get default | grep interface | cut -d \":\" -f 2", "r"), pclose);
#else // Linux
std::unique_ptr<FILE, decltype(pclose) *> file(popen("ip route | grep default | cut -d \" \" -f 5 | head -n 1", "r"), pclose);
#endif
if (!file) std::string getIpFromHostname(const std::string& hostname, boost::asio::io_service& ios);
{
LOG(error) << "Could not detect default route network interface name - popen() failed!";
return "";
}
while (!feof(file.get()))
{
if (fgets(buffer.data(), 128, file.get()) != NULL)
{
interfaceName += buffer.data();
}
}
boost::algorithm::trim(interfaceName);
if (interfaceName == "")
{
LOG(error) << "Could not detect default route network interface name";
}
else
{
LOG(debug) << "Detected network interface name for the default route: " << interfaceName;
}
return interfaceName;
}
inline std::string getIpFromHostname(const std::string& hostname)
{
try {
namespace bai = boost::asio::ip;
boost::asio::io_service ios;
bai::tcp::resolver resolver(ios);
bai::tcp::resolver::query query(hostname, "");
bai::tcp::resolver::iterator end;
auto it = std::find_if(static_cast<bai::basic_resolver_iterator<bai::tcp>>(resolver.resolve(query)), end, [](const bai::tcp::endpoint& ep) {
return ep.address().is_v4();
});
if (it != end) {
std::stringstream ss;
ss << static_cast<bai::tcp::endpoint>(*it).address();
return ss.str();
}
LOG(warn) << "could not find ipv4 address for hostname '" << hostname << "'";
return "";
} catch (std::exception& e) {
LOG(error) << "could not resolve hostname '" << hostname << "', reason: " << e.what();
return "";
}
}
inline std::string getIpFromHostname(const std::string& hostname, boost::asio::io_service& ios)
{
try {
namespace bai = boost::asio::ip;
bai::tcp::resolver resolver(ios);
bai::tcp::resolver::query query(hostname, "");
bai::tcp::resolver::iterator end;
auto it = std::find_if(static_cast<bai::basic_resolver_iterator<bai::tcp>>(resolver.resolve(query)), end, [](const bai::tcp::endpoint& ep) {
return ep.address().is_v4();
});
if (it != end) {
std::stringstream ss;
ss << static_cast<bai::tcp::endpoint>(*it).address();
return ss.str();
}
LOG(warn) << "could not find ipv4 address for hostname '" << hostname << "'";
return "";
} catch (std::exception& e) {
LOG(error) << "could not resolve hostname '" << hostname << "', reason: " << e.what();
return "";
}
}
} /* namespace tools */ } /* namespace tools */
} /* namespace mq */ } /* namespace mq */

72
fairmq/tools/Process.cxx Normal file
View File

@ -0,0 +1,72 @@
/********************************************************************************
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/tools/Process.h>
#include <boost/process.hpp>
#include <iostream>
using namespace std;
namespace fair
{
namespace mq
{
namespace tools
{
/**
* Execute given command in forked process and capture stdout output
* and exit code.
*
* @param[in] cmd Command to execute
* @param[in] log_prefix How to prefix each captured output line with
* @return Captured stdout output and exit code
*/
execute_result execute(string cmd, string prefix)
{
execute_result result;
stringstream out;
// print full line thread-safe
stringstream printCmd;
printCmd << prefix << cmd << "\n";
cout << printCmd.str() << flush;
out << prefix << cmd << endl;
// Execute command and capture stdout, add prefix line by line
boost::process::ipstream stdout;
boost::process::child c(cmd, boost::process::std_out > stdout);
string line;
while (getline(stdout, line))
{
// print full line thread-safe
stringstream printLine;
printLine << prefix << line << "\n";
cout << printLine.str() << flush;
out << prefix << line << "\n";
}
c.wait();
// Capture exit code
result.exit_code = c.exit_code();
out << prefix << " Exit code: " << result.exit_code << endl;
result.console_out = out.str();
// Return result
return result;
}
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */

40
fairmq/tools/Process.h
View File

@ -9,8 +9,6 @@
#ifndef FAIR_MQ_TOOLS_PROCESS_H #ifndef FAIR_MQ_TOOLS_PROCESS_H
#define FAIR_MQ_TOOLS_PROCESS_H #define FAIR_MQ_TOOLS_PROCESS_H
#include <boost/process.hpp>
#include <string> #include <string>
namespace fair namespace fair
@ -37,43 +35,7 @@ struct execute_result
* @param[in] log_prefix How to prefix each captured output line with * @param[in] log_prefix How to prefix each captured output line with
* @return Captured stdout output and exit code * @return Captured stdout output and exit code
*/ */
inline execute_result execute(std::string cmd, std::string prefix = "") execute_result execute(std::string cmd, std::string prefix = "");
{
execute_result result;
std::stringstream out;
// print full line thread-safe
std::stringstream printCmd;
printCmd << prefix << cmd << "\n";
std::cout << printCmd.str() << std::flush;
out << prefix << cmd << std::endl;
// Execute command and capture stdout, add prefix line by line
boost::process::ipstream stdout;
boost::process::child c(cmd, boost::process::std_out > stdout);
std::string line;
while (getline(stdout, line))
{
// print full line thread-safe
std::stringstream printLine;
printLine << prefix << line << "\n";
std::cout << printLine.str() << std::flush;
out << prefix << line << "\n";
}
c.wait();
// Capture exit code
result.exit_code = c.exit_code();
out << prefix << " Exit code: " << result.exit_code << std::endl;
result.console_out = out.str();
// Return result
return result;
}
} /* namespace tools */ } /* namespace tools */
} /* namespace mq */ } /* namespace mq */

44
fairmq/tools/Unique.cxx Normal file
View File

@ -0,0 +1,44 @@
/********************************************************************************
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/tools/Unique.h>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <boost/functional/hash.hpp>
using namespace std;
namespace fair
{
namespace mq
{
namespace tools
{
// generates UUID string
string Uuid()
{
boost::uuids::random_generator gen;
boost::uuids::uuid u = gen();
return boost::uuids::to_string(u);
}
// generates UUID and returns its hash
size_t UuidHash()
{
boost::uuids::random_generator gen;
boost::hash<boost::uuids::uuid> uuid_hasher;
boost::uuids::uuid u = gen();
return uuid_hasher(u);
}
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */

20
fairmq/tools/Unique.h
View File

@ -9,11 +9,6 @@
#ifndef FAIR_MQ_TOOLS_UNIQUE_H #ifndef FAIR_MQ_TOOLS_UNIQUE_H
#define FAIR_MQ_TOOLS_UNIQUE_H #define FAIR_MQ_TOOLS_UNIQUE_H
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <boost/functional/hash.hpp>
#include <string> #include <string>
namespace fair namespace fair
@ -24,21 +19,10 @@ namespace tools
{ {
// generates UUID string // generates UUID string
inline std::string Uuid() std::string Uuid();
{
boost::uuids::random_generator gen;
boost::uuids::uuid u = gen();
return boost::uuids::to_string(u);
}
// generates UUID and returns its hash // generates UUID and returns its hash
inline std::size_t UuidHash() std::size_t UuidHash();
{
boost::uuids::random_generator gen;
boost::hash<boost::uuids::uuid> uuid_hasher;
boost::uuids::uuid u = gen();
return uuid_hasher(u);
}
} /* namespace tools */ } /* namespace tools */
} /* namespace mq */ } /* namespace mq */

2
fairmq/zeromq/FairMQMessageZMQ.cxx
View File

@ -18,6 +18,8 @@
#include <fairmq/Tools.h> #include <fairmq/Tools.h>
#include "FairMQUnmanagedRegionZMQ.h" #include "FairMQUnmanagedRegionZMQ.h"
#include <cstring>
using namespace std; using namespace std;
fair::mq::Transport FairMQMessageZMQ::fTransportType = fair::mq::Transport::ZMQ; fair::mq::Transport FairMQMessageZMQ::fTransportType = fair::mq::Transport::ZMQ;

2
fairmq/zeromq/FairMQSocketZMQ.cxx
View File

@ -12,6 +12,8 @@
#include <zmq.h> #include <zmq.h>
#include <cassert>
using namespace std; using namespace std;
atomic<bool> FairMQSocketZMQ::fInterrupted(false); atomic<bool> FairMQSocketZMQ::fInterrupted(false);

1
test/device/TestReceiver.h
View File

@ -13,6 +13,7 @@
#include <FairMQLogger.h> #include <FairMQLogger.h>
#include <string> #include <string>
#include <thread>
namespace fair namespace fair
{ {

1
test/device/TestSender.h
View File

@ -13,6 +13,7 @@
#include <FairMQLogger.h> #include <FairMQLogger.h>
#include <string> #include <string>
#include <thread>
namespace fair namespace fair
{ {

2
test/helper/devices/TestPairLeft.cxx
View File

@ -8,6 +8,8 @@
#include <FairMQDevice.h> #include <FairMQDevice.h>
#include <thread>
namespace fair namespace fair
{ {
namespace mq namespace mq

1
test/helper/devices/TestPairRight.cxx
View File

@ -8,6 +8,7 @@
#include <FairMQDevice.h> #include <FairMQDevice.h>
#include <string> #include <string>
#include <thread>
namespace fair namespace fair
{ {

1
test/helper/devices/TestPollIn.cxx
View File

@ -9,6 +9,7 @@
#include <FairMQDevice.h> #include <FairMQDevice.h>
#include <FairMQLogger.h> #include <FairMQLogger.h>
#include <options/FairMQProgOptions.h> #include <options/FairMQProgOptions.h>
#include <thread>
namespace fair namespace fair
{ {

1
test/helper/devices/TestPollOut.cxx
View File

@ -7,6 +7,7 @@
********************************************************************************/ ********************************************************************************/
#include <FairMQDevice.h> #include <FairMQDevice.h>
#include <thread>
namespace fair namespace fair
{ {

1
test/helper/devices/TestPull.cxx
View File

@ -8,6 +8,7 @@
#include <FairMQDevice.h> #include <FairMQDevice.h>
#include <FairMQLogger.h> #include <FairMQLogger.h>
#include <thread>
namespace fair namespace fair
{ {

1
test/helper/devices/TestPush.cxx
View File

@ -7,6 +7,7 @@
********************************************************************************/ ********************************************************************************/
#include <FairMQDevice.h> #include <FairMQDevice.h>
#include <thread>
namespace fair namespace fair
{ {

1
test/helper/devices/TestRep.cxx
View File

@ -8,6 +8,7 @@
#include <FairMQDevice.h> #include <FairMQDevice.h>
#include <FairMQLogger.h> #include <FairMQLogger.h>
#include <thread>
namespace fair namespace fair
{ {

1
test/helper/devices/TestReq.cxx
View File

@ -8,6 +8,7 @@
#include <FairMQDevice.h> #include <FairMQDevice.h>
#include <FairMQLogger.h> #include <FairMQLogger.h>
#include <thread>
namespace fair namespace fair
{ {