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
parent 436f79bee5
commit 81299ce005
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
{ {