mirror of
https://github.com/FairRootGroup/FairMQ.git
synced 2025-10-13 08:41:16 +00:00
Use SDK in dds-command-ui
This commit is contained in:
Alexey Rybalchenko
Dennis Klein
parent
e853d121bf
commit
e221242f9a
|
|
@ -25,10 +25,10 @@ plugin=${1:-localhost}
|
|||
|
||||
exec 5>&1
|
||||
output=$(dds-session start | tee >(cat - >&5))
|
||||
export DDS_SESSION_ID=$(echo ${output} | grep "DDS session ID: " | cut -d' ' -f4)
|
||||
echo "SESSION ID: ${DDS_SESSION_ID}"
|
||||
export FAIRMQ_DDS_SESSION_ID=$(echo ${output} | grep "DDS session ID: " | cut -d' ' -f4)
|
||||
echo "SESSION ID: ${FAIRMQ_DDS_SESSION_ID}"
|
||||
|
||||
trap "cleanup ${DDS_SESSION_ID}" EXIT
|
||||
trap "cleanup ${FAIRMQ_DDS_SESSION_ID}" EXIT
|
||||
|
||||
requiredNofSlots=12
|
||||
if [[ "$plugin" == "ssh" ]]; then
|
||||
|
|
@ -39,12 +39,12 @@ fi
|
|||
echo "...waiting for ${requiredNofSlots} idle slots..."
|
||||
dds-info --idle-count --wait ${requiredNofSlots}
|
||||
|
||||
topologyFile=@DATA_DIR@/ex-dds-topology.xml
|
||||
echo "TOPOLOGY FILE: ${topologyFile}"
|
||||
echo "TOPOLOGY NAME: $(dds-topology --disable-validation --topology-name ${topologyFile})"
|
||||
export FAIRMQ_DDS_TOPO_FILE=@DATA_DIR@/ex-dds-topology.xml
|
||||
echo "TOPOLOGY FILE: ${FAIRMQ_DDS_TOPO_FILE}"
|
||||
echo "TOPOLOGY NAME: $(dds-topology --disable-validation --topology-name ${FAIRMQ_DDS_TOPO_FILE})"
|
||||
|
||||
dds-info --active-topology
|
||||
dds-topology --activate ${topologyFile}
|
||||
dds-topology --activate ${FAIRMQ_DDS_TOPO_FILE}
|
||||
dds-info --active-topology
|
||||
echo "...waiting for ${requiredNofSlots} executing slots..."
|
||||
dds-info --executing-count --wait ${requiredNofSlots}
|
||||
|
|
@ -52,21 +52,20 @@ dds-info --executing-count --wait ${requiredNofSlots}
|
|||
echo "------------------------"
|
||||
echo "...waiting for Topology to finish..."
|
||||
# TODO Retrieve number of devices from DDS topology API instead of having the user pass it explicitely
|
||||
fairmq-dds-command-ui -w "IDLE" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c i -w "INITIALIZING DEVICE" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c k -w "INITIALIZED" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c b -w "BOUND" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c x -w "DEVICE READY" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c j -w "READY" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -w "IDLE"
|
||||
fairmq-dds-command-ui -c i
|
||||
fairmq-dds-command-ui -c k
|
||||
fairmq-dds-command-ui -c b
|
||||
fairmq-dds-command-ui -c x
|
||||
fairmq-dds-command-ui -c j
|
||||
fairmq-dds-command-ui -c r
|
||||
sampler_and_sink="main/(Sampler|Sink)"
|
||||
# processors="main/ProcessorGroup/Processor"
|
||||
fairmq-dds-command-ui -p $sampler_and_sink -w "RUNNING->READY" -n 2
|
||||
sampler_and_sink="main/(Sampler|Sink).*"
|
||||
fairmq-dds-command-ui -w "RUNNING->READY" -p $sampler_and_sink
|
||||
echo "...$sampler_and_sink are READY, sending shutdown..."
|
||||
fairmq-dds-command-ui -c s -w "RUNNING->READY" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c t -w "DEVICE READY" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c d -w "IDLE" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c q -w "EXITING" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c s
|
||||
fairmq-dds-command-ui -c t
|
||||
fairmq-dds-command-ui -c d
|
||||
fairmq-dds-command-ui -c q
|
||||
echo "...waiting for ${requiredNofSlots} idle slots..."
|
||||
dds-info --idle-count --wait ${requiredNofSlots}
|
||||
echo "------------------------"
|
||||
|
|
|
|||
|
|
@ -45,21 +45,20 @@ dds-info --executing-count --wait ${requiredNofSlots}
|
|||
echo "------------------------"
|
||||
echo "...waiting for Topology to finish..."
|
||||
# TODO Retrieve number of devices from DDS topology API instead of having the user pass it explicitely
|
||||
fairmq-dds-command-ui -w "IDLE" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c i -w "INITIALIZING DEVICE" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c k -w "INITIALIZED" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c b -w "BOUND" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c x -w "DEVICE READY" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c j -w "READY" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -w "IDLE"
|
||||
fairmq-dds-command-ui -c i
|
||||
fairmq-dds-command-ui -c k
|
||||
fairmq-dds-command-ui -c b
|
||||
fairmq-dds-command-ui -c x
|
||||
fairmq-dds-command-ui -c j
|
||||
fairmq-dds-command-ui -c r
|
||||
sampler_and_sink="main/(Sampler|Sink)"
|
||||
# processors="main/ProcessorGroup/Processor"
|
||||
fairmq-dds-command-ui -p $sampler_and_sink -w "RUNNING->READY" -n 2
|
||||
sampler_and_sink="main/(Sampler|Sink).*"
|
||||
fairmq-dds-command-ui -w "RUNNING->READY" -p $sampler_and_sink
|
||||
echo "...$sampler_and_sink are READY, sending shutdown..."
|
||||
fairmq-dds-command-ui -c s -w "RUNNING->READY" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c t -w "DEVICE READY" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c d -w "IDLE" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c q -w "EXITING" -n ${requiredNofSlots}
|
||||
fairmq-dds-command-ui -c s
|
||||
fairmq-dds-command-ui -c t
|
||||
fairmq-dds-command-ui -c d
|
||||
fairmq-dds-command-ui -c q
|
||||
echo "...waiting for ${requiredNofSlots} idle slots..."
|
||||
dds-info --idle-count --wait ${requiredNofSlots}
|
||||
echo "------------------------"
|
||||
|
|
|
|||
|
|
@ -19,15 +19,10 @@ set_target_properties(${plugin} PROPERTIES
|
|||
|
||||
set(exe1 fairmq-dds-command-ui)
|
||||
add_executable(${exe1} ${CMAKE_CURRENT_SOURCE_DIR}/runDDSCommandUI.cxx)
|
||||
target_link_libraries(${exe1} FairMQ Commands StateMachine DDS::dds_intercom_lib DDS::dds_protocol_lib)
|
||||
target_link_libraries(${exe1} FairMQ Commands SDK StateMachine DDS::dds_intercom_lib DDS::dds_protocol_lib)
|
||||
target_include_directories(${exe1} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
|
||||
|
||||
set(exe2 fairmq-dds-command-ui-new)
|
||||
add_executable(${exe2} ${CMAKE_CURRENT_SOURCE_DIR}/runDDSCommandUInew.cxx)
|
||||
target_link_libraries(${exe2} FairMQ Commands SDK StateMachine DDS::dds_intercom_lib DDS::dds_protocol_lib)
|
||||
target_include_directories(${exe2} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
|
||||
|
||||
install(TARGETS ${plugin} ${exe1} ${exe2}
|
||||
install(TARGETS ${plugin} ${exe1}
|
||||
EXPORT ${PROJECT_EXPORT_SET}
|
||||
LIBRARY DESTINATION ${PROJECT_INSTALL_LIBDIR}
|
||||
RUNTIME DESTINATION ${PROJECT_INSTALL_BINDIR}
|
||||
|
|
|
|||
|
|
@ -8,23 +8,23 @@
|
|||
|
||||
#include <fairmq/sdk/commands/Commands.h>
|
||||
#include <fairmq/States.h>
|
||||
|
||||
#include <dds/dds.h>
|
||||
#include <fairmq/SDK.h>
|
||||
|
||||
#include <boost/program_options.hpp>
|
||||
|
||||
#include <termios.h> // raw mode console input
|
||||
#include <unistd.h>
|
||||
#include <condition_variable>
|
||||
#include <cstdlib>
|
||||
#include <iostream>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <termios.h> // raw mode console input
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
#include <unistd.h>
|
||||
|
||||
using namespace std;
|
||||
using namespace dds::intercom_api;
|
||||
using namespace fair::mq;
|
||||
using namespace fair::mq::sdk;
|
||||
using namespace fair::mq::sdk::cmd;
|
||||
namespace bpo = boost::program_options;
|
||||
|
||||
|
|
@ -49,24 +49,6 @@ struct TerminalConfig
|
|||
}
|
||||
};
|
||||
|
||||
struct StateSubscription
|
||||
{
|
||||
const string& fTopologyPath;
|
||||
CCustomCmd& fDdsCustomCmd;
|
||||
|
||||
explicit StateSubscription(const string& topologyPath, CCustomCmd& ddsCustomCmd)
|
||||
: fTopologyPath(topologyPath)
|
||||
, fDdsCustomCmd(ddsCustomCmd)
|
||||
{
|
||||
fDdsCustomCmd.send(Cmds(make<SubscribeToStateChange>()).Serialize(), fTopologyPath);
|
||||
}
|
||||
|
||||
~StateSubscription() {
|
||||
fDdsCustomCmd.send(Cmds(make<UnsubscribeFromStateChange>()).Serialize(), fTopologyPath);
|
||||
this_thread::sleep_for(chrono::milliseconds(100)); // give dds a chance to complete request
|
||||
}
|
||||
};
|
||||
|
||||
void printControlsHelp()
|
||||
{
|
||||
cout << "Use keys to control the devices:" << endl;
|
||||
|
|
@ -74,170 +56,117 @@ void printControlsHelp()
|
|||
cout << "To quit press Ctrl+C" << endl;
|
||||
}
|
||||
|
||||
void sendCommand(const string& commandIn, const string& topologyPath, CCustomCmd& ddsCustomCmd)
|
||||
void handleCommand(const string& command, const string& path, unsigned int timeout, Topology& topo)
|
||||
{
|
||||
char c;
|
||||
string command(commandIn);
|
||||
TerminalConfig tconfig;
|
||||
|
||||
if (command == "") {
|
||||
printControlsHelp();
|
||||
cin >> c;
|
||||
command = c;
|
||||
}
|
||||
|
||||
while (true) {
|
||||
if (command == "c") {
|
||||
cout << "> checking state of the devices" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<CheckState>()).Serialize(), topologyPath);
|
||||
auto const result = topo.GetCurrentState();
|
||||
for (const auto& d : result) {
|
||||
cout << d.taskId << " : " << d.state << endl;
|
||||
}
|
||||
} else if (command == "o") {
|
||||
cout << "> dumping config of the devices" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<DumpConfig>()).Serialize(), topologyPath);
|
||||
// TODO: extend this regex to return all properties, once command size limitation is removed.
|
||||
auto const result = topo.GetProperties("^(session|id)$", path, std::chrono::milliseconds(timeout));
|
||||
for (const auto& d : result.second.devices) {
|
||||
for (auto const& p : d.second.props) {
|
||||
cout << d.first << ": " << p.first << " : " << p.second << endl;
|
||||
}
|
||||
}
|
||||
} else if (command == "i") {
|
||||
cout << "> init devices" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::InitDevice)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::InitDevice, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "k") {
|
||||
cout << "> complete init" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::CompleteInit)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::CompleteInit, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "b") {
|
||||
cout << "> bind devices" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::Bind)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::Bind, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "x") {
|
||||
cout << "> connect devices" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::Connect)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::Connect, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "j") {
|
||||
cout << "> init tasks" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::InitTask)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::InitTask, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "r") {
|
||||
cout << "> run tasks" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::Run)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::Run, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "s") {
|
||||
cout << "> stop devices" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::Stop)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::Stop, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "t") {
|
||||
cout << "> reset tasks" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::ResetTask)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::ResetTask, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "d") {
|
||||
cout << "> reset devices" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::ResetDevice)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::ResetDevice, std::chrono::milliseconds(timeout));
|
||||
} else if (command == "h") {
|
||||
cout << "> help" << endl;
|
||||
printControlsHelp();
|
||||
} else if (command == "q") {
|
||||
cout << "> end" << endl;
|
||||
ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::End)).Serialize(), topologyPath);
|
||||
topo.ChangeState(TopologyTransition::End, std::chrono::milliseconds(timeout));
|
||||
} else {
|
||||
cout << "\033[01;32mInvalid input: [" << c << "]\033[0m" << endl;
|
||||
cout << "\033[01;32mInvalid input: [" << command << "]\033[0m" << endl;
|
||||
printControlsHelp();
|
||||
}
|
||||
|
||||
if (commandIn != "") {
|
||||
this_thread::sleep_for(chrono::milliseconds(100)); // give dds a chance to complete request
|
||||
break;
|
||||
} else {
|
||||
cin >> c;
|
||||
command = c;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct WaitMode
|
||||
void sendCommand(const string& commandIn, const string& path, unsigned int timeout, Topology& topo)
|
||||
{
|
||||
explicit WaitMode(const string& targetState)
|
||||
: fTransitionedCount(0)
|
||||
{
|
||||
if (targetState != "") {
|
||||
size_t n = targetState.find("->");
|
||||
if (n == string::npos) {
|
||||
fTargetStatePair.first = fair::mq::State::Ok;
|
||||
fTargetStatePair.second = fair::mq::GetState(targetState);
|
||||
} else {
|
||||
fTargetStatePair.first = fair::mq::GetState(targetState.substr(0, n));
|
||||
fTargetStatePair.second = fair::mq::GetState(targetState.substr(n + 2));
|
||||
}
|
||||
}
|
||||
if (commandIn != "") {
|
||||
handleCommand(commandIn, path, timeout, topo);
|
||||
return;
|
||||
}
|
||||
|
||||
void Run(const chrono::milliseconds& timeout, const string& topologyPath, CCustomCmd& ddsCustomCmd, unsigned int numDevices, const string& command = "")
|
||||
{
|
||||
if (command != "") {
|
||||
sendCommand(command, topologyPath, ddsCustomCmd);
|
||||
}
|
||||
char c;
|
||||
string command;
|
||||
TerminalConfig tconfig;
|
||||
|
||||
// TODO once DDS provides an API to retrieve actual number of tasks, use it here
|
||||
auto condition = [&] {
|
||||
bool res = fTransitionedCount == numDevices;
|
||||
if (fTargetStatePair.first == fair::mq::State::Ok) {
|
||||
cout << "Waiting for " << numDevices << " devices to reach " << fTargetStatePair.second << ", condition check: " << res << endl;
|
||||
} else {
|
||||
cout << "Waiting for " << numDevices << " devices to reach " << fTargetStatePair.first << "->" << fTargetStatePair.second << ", condition check: " << res << endl;
|
||||
}
|
||||
return res;
|
||||
};
|
||||
printControlsHelp();
|
||||
cin >> c;
|
||||
command = c;
|
||||
|
||||
unique_lock<mutex> lock(fMtx);
|
||||
|
||||
if (timeout > chrono::milliseconds(0)) {
|
||||
if (!fCV.wait_for(lock, timeout, condition)) {
|
||||
throw runtime_error("timeout");
|
||||
while (true) {
|
||||
handleCommand(command, path, timeout, topo);
|
||||
cin >> c;
|
||||
command = c;
|
||||
}
|
||||
} else {
|
||||
fCV.wait(lock, condition);
|
||||
}
|
||||
|
||||
// cout << "WaitMode.Run() finished" << endl;
|
||||
}
|
||||
|
||||
void CountStates(fair::mq::State lastState, fair::mq::State currentState)
|
||||
{
|
||||
{
|
||||
unique_lock<mutex> lock(fMtx);
|
||||
if (fTargetStatePair.first == fair::mq::State::Ok) {
|
||||
if (fTargetStatePair.second == currentState) {
|
||||
fTransitionedCount++;
|
||||
// cout << "fTransitionedCount = " << fTransitionedCount << " for single value" << endl;
|
||||
}
|
||||
} else {
|
||||
if (fTargetStatePair.first == lastState && fTargetStatePair.second == currentState) {
|
||||
fTransitionedCount++;
|
||||
// cout << "fTransitionedCount = " << fTransitionedCount << " for double value" << endl;
|
||||
}
|
||||
}
|
||||
}
|
||||
fCV.notify_one();
|
||||
}
|
||||
|
||||
mutex fMtx;
|
||||
condition_variable fCV;
|
||||
pair<fair::mq::State, fair::mq::State> fTargetStatePair;
|
||||
unsigned int fTransitionedCount;
|
||||
};
|
||||
}
|
||||
|
||||
int main(int argc, char* argv[])
|
||||
{
|
||||
try {
|
||||
try {
|
||||
string sessionID;
|
||||
string topoFile;
|
||||
|
||||
string command;
|
||||
string topologyPath;
|
||||
string path;
|
||||
string targetState;
|
||||
unsigned int timeout;
|
||||
unsigned int numDevices(0);
|
||||
|
||||
fair::Logger::SetConsoleSeverity("debug");
|
||||
|
||||
bpo::options_description options("Common options");
|
||||
|
||||
auto envSessionId = getenv("DDS_SESSION_ID");
|
||||
auto envSessionId = getenv("FAIRMQ_DDS_SESSION_ID");
|
||||
if (envSessionId) {
|
||||
options.add_options()("session,s", bpo::value<string>(&sessionID)->default_value(envSessionId), "DDS Session ID (overrides any value in env var $DDS_SESSION_ID)");
|
||||
options.add_options()("session,s", bpo::value<string>(&sessionID)->default_value(envSessionId), "DDS Session ID (overrides any value in env var $FAIRMQ_DDS_SESSION_ID)");
|
||||
} else {
|
||||
options.add_options()("session,s", bpo::value<string>(&sessionID)->required(), "DDS Session ID (overrides any value in env var $DDS_SESSION_ID)");
|
||||
options.add_options()("session,s", bpo::value<string>(&sessionID)->required(), "DDS Session ID (overrides any value in env var $FAIRMQ_DDS_SESSION_ID)");
|
||||
}
|
||||
|
||||
auto envTopoFile = getenv("FAIRMQ_DDS_TOPO_FILE");
|
||||
if (envTopoFile) {
|
||||
options.add_options()("topology-file,f", bpo::value<string>(&topoFile)->default_value(envTopoFile), "DDS topology file path");
|
||||
} else {
|
||||
options.add_options()("topology-file,f", bpo::value<string>(&topoFile)->required(), "DDS topology file path");
|
||||
}
|
||||
|
||||
options.add_options()
|
||||
("command,c", bpo::value<string> (&command)->default_value(""), "Command character")
|
||||
("path,p", bpo::value<string> (&topologyPath)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)")
|
||||
("wait-for-state,w", bpo::value<string> (&targetState)->default_value(""), "Wait until targeted FairMQ devices reach the given state")
|
||||
("timeout,t", bpo::value<unsigned int> (&timeout)->default_value(0), "Timeout in milliseconds when waiting for a device state (0 - wait infinitely)")
|
||||
("number-devices,n", bpo::value<unsigned int> (&numDevices)->default_value(0), "Number of devices (will be removed in the future)")
|
||||
("command,c", bpo::value<string>(&command)->default_value(""), "Command character")
|
||||
("path,p", bpo::value<string>(&path)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)")
|
||||
("wait-for-state,w", bpo::value<string>(&targetState)->default_value(""), "Wait until targeted FairMQ devices reach the given state")
|
||||
("timeout,t", bpo::value<unsigned int>(&timeout)->default_value(0), "Timeout in milliseconds when waiting for a device state (0 - wait infinitely)")
|
||||
("help,h", "Produce help message");
|
||||
|
||||
bpo::variables_map vm;
|
||||
|
|
@ -251,77 +180,30 @@ int main(int argc, char* argv[])
|
|||
|
||||
bpo::notify(vm);
|
||||
|
||||
WaitMode waitMode(targetState);
|
||||
DDSEnvironment env;
|
||||
DDSSession session(sessionID, env);
|
||||
DDSTopology ddsTopo(DDSTopology::Path(topoFile), env);
|
||||
|
||||
CIntercomService service;
|
||||
CCustomCmd ddsCustomCmd(service);
|
||||
Topology topo(ddsTopo, session);
|
||||
|
||||
service.subscribeOnError([](const EErrorCode errorCode, const string& errorMsg) {
|
||||
cerr << "DDS error received: error code: " << errorCode << ", error message: " << errorMsg << endl;
|
||||
});
|
||||
|
||||
// subscribe to receive messages from DDS
|
||||
ddsCustomCmd.subscribe([&](const string& msg, const string& /*condition*/, uint64_t senderId) {
|
||||
Cmds cmds;
|
||||
cmds.Deserialize(msg);
|
||||
// cout << "Received " << cmds.Size() << " command(s) with total size of " << msg.length() << " bytes: " << endl;
|
||||
for (const auto& cmd : cmds) {
|
||||
// cout << " > " << cmd->GetType() << endl;
|
||||
switch (cmd->GetType()) {
|
||||
case Type::state_change: {
|
||||
cout << "Received state_change from " << static_cast<StateChange&>(*cmd).GetDeviceId() << ": " << static_cast<StateChange&>(*cmd).GetLastState() << "->" << static_cast<StateChange&>(*cmd).GetCurrentState() << endl;
|
||||
if (static_cast<StateChange&>(*cmd).GetCurrentState() == fair::mq::State::Exiting) {
|
||||
ddsCustomCmd.send(Cmds(make<StateChangeExitingReceived>()).Serialize(), to_string(senderId));
|
||||
if (targetState != "") {
|
||||
if (command != "") {
|
||||
sendCommand(command, path, timeout, topo);
|
||||
}
|
||||
waitMode.CountStates(static_cast<StateChange&>(*cmd).GetLastState(), static_cast<StateChange&>(*cmd).GetCurrentState());
|
||||
}
|
||||
break;
|
||||
case Type::state_change_subscription:
|
||||
if (static_cast<StateChangeSubscription&>(*cmd).GetResult() != Result::Ok) {
|
||||
cout << "State change subscription failed for " << static_cast<StateChangeSubscription&>(*cmd).GetDeviceId() << endl;
|
||||
}
|
||||
break;
|
||||
case Type::state_change_unsubscription:
|
||||
if (static_cast<StateChangeUnsubscription&>(*cmd).GetResult() != Result::Ok) {
|
||||
cout << "State change unsubscription failed for " << static_cast<StateChangeUnsubscription&>(*cmd).GetDeviceId() << endl;
|
||||
}
|
||||
break;
|
||||
case Type::transition_status: {
|
||||
if (static_cast<TransitionStatus&>(*cmd).GetResult() == Result::Ok) {
|
||||
cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " started to transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
|
||||
size_t pos = targetState.find("->");
|
||||
if (pos == string::npos) {
|
||||
/* auto ec = */topo.WaitForState(GetState(targetState), path, std::chrono::milliseconds(timeout));
|
||||
// cout << "WaitForState(" << targetState << ") result: " << ec.message() << endl;
|
||||
} else {
|
||||
cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " cannot transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
|
||||
/* auto ec = */topo.WaitForState(GetState(targetState.substr(0, pos)), GetState(targetState.substr(pos + 2)), path, std::chrono::milliseconds(timeout));
|
||||
// cout << "WaitForState(" << targetState << ") result: " << ec.message() << endl;
|
||||
}
|
||||
}
|
||||
break;
|
||||
case Type::current_state:
|
||||
cout << "Device " << static_cast<CurrentState&>(*cmd).GetDeviceId() << " is in " << static_cast<CurrentState&>(*cmd).GetCurrentState() << " state" << endl;
|
||||
break;
|
||||
case Type::config:
|
||||
cout << "Received config for device " << static_cast<Config&>(*cmd).GetDeviceId() << ":\n" << static_cast<Config&>(*cmd).GetConfig() << endl;
|
||||
break;
|
||||
default:
|
||||
cout << "Unexpected/unknown command received: " << cmd->GetType() << endl;
|
||||
cout << "Origin: " << senderId << endl;
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
service.start(sessionID);
|
||||
|
||||
StateSubscription stateSubscription(topologyPath, ddsCustomCmd);
|
||||
|
||||
if (targetState == "") {
|
||||
sendCommand(command, topologyPath, ddsCustomCmd);
|
||||
} else {
|
||||
waitMode.Run(chrono::milliseconds(timeout), topologyPath, ddsCustomCmd, numDevices, command);
|
||||
sendCommand(command, path, timeout, topo);
|
||||
}
|
||||
|
||||
ddsCustomCmd.unsubscribe();
|
||||
} catch (exception& e) {
|
||||
return EXIT_SUCCESS;
|
||||
} catch (exception& e) {
|
||||
cerr << "Error: " << e.what() << endl;
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,199 +0,0 @@
|
|||
/********************************************************************************
|
||||
* Copyright (C) 2014-2019 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/sdk/commands/Commands.h>
|
||||
#include <fairmq/States.h>
|
||||
#include <fairmq/SDK.h>
|
||||
|
||||
#include <boost/program_options.hpp>
|
||||
|
||||
#include <termios.h> // raw mode console input
|
||||
#include <unistd.h>
|
||||
#include <condition_variable>
|
||||
#include <cstdlib>
|
||||
#include <iostream>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
|
||||
using namespace std;
|
||||
using namespace fair::mq;
|
||||
using namespace fair::mq::sdk;
|
||||
using namespace fair::mq::sdk::cmd;
|
||||
namespace bpo = boost::program_options;
|
||||
|
||||
struct TerminalConfig
|
||||
{
|
||||
explicit TerminalConfig()
|
||||
{
|
||||
termios t;
|
||||
tcgetattr(STDIN_FILENO, &t); // get the current terminal I/O structure
|
||||
t.c_lflag &= ~ICANON; // disable canonical input
|
||||
// t.c_lflag &= ~ECHO; // do not echo input chars
|
||||
tcsetattr(STDIN_FILENO, TCSANOW, &t); // apply the new settings
|
||||
}
|
||||
|
||||
~TerminalConfig()
|
||||
{
|
||||
termios t;
|
||||
tcgetattr(STDIN_FILENO, &t); // get the current terminal I/O structure
|
||||
t.c_lflag |= ICANON; // re-enable canonical input
|
||||
// t.c_lflag |= ECHO; // echo input chars
|
||||
tcsetattr(STDIN_FILENO, TCSANOW, &t); // apply the new settings
|
||||
}
|
||||
};
|
||||
|
||||
void printControlsHelp()
|
||||
{
|
||||
cout << "Use keys to control the devices:" << endl;
|
||||
cout << "[c] check states, [o] dump config, [h] help, [r] run, [s] stop, [t] reset task, [d] reset device, [q] end, [j] init task, [i] init device, [k] complete init, [b] bind, [x] connect" << endl;
|
||||
cout << "To quit press Ctrl+C" << endl;
|
||||
}
|
||||
|
||||
void sendCommand(const string& commandIn, const string& topologyPath, Topology& topo)
|
||||
{
|
||||
char c;
|
||||
string command(commandIn);
|
||||
TerminalConfig tconfig;
|
||||
|
||||
if (command == "") {
|
||||
printControlsHelp();
|
||||
cin >> c;
|
||||
command = c;
|
||||
}
|
||||
|
||||
while (true) {
|
||||
if (command == "c") {
|
||||
cout << "> checking state of the devices" << endl;
|
||||
topo.GetCurrentState();
|
||||
// TODO: extend me
|
||||
} else if (command == "o") {
|
||||
cout << "> dumping config of the devices" << endl;
|
||||
auto const result = topo.GetProperties("^(session|id)$", topologyPath);
|
||||
// TODO: extend me
|
||||
} else if (command == "i") {
|
||||
cout << "> init devices" << endl;
|
||||
topo.ChangeState(TopologyTransition::InitDevice);
|
||||
} else if (command == "k") {
|
||||
cout << "> complete init" << endl;
|
||||
topo.ChangeState(TopologyTransition::CompleteInit);
|
||||
} else if (command == "b") {
|
||||
cout << "> bind devices" << endl;
|
||||
topo.ChangeState(TopologyTransition::Bind);
|
||||
} else if (command == "x") {
|
||||
cout << "> connect devices" << endl;
|
||||
topo.ChangeState(TopologyTransition::Connect);
|
||||
} else if (command == "j") {
|
||||
cout << "> init tasks" << endl;
|
||||
topo.ChangeState(TopologyTransition::InitTask);
|
||||
} else if (command == "r") {
|
||||
cout << "> run tasks" << endl;
|
||||
topo.ChangeState(TopologyTransition::Run);
|
||||
} else if (command == "s") {
|
||||
cout << "> stop devices" << endl;
|
||||
topo.ChangeState(TopologyTransition::Stop);
|
||||
} else if (command == "t") {
|
||||
cout << "> reset tasks" << endl;
|
||||
topo.ChangeState(TopologyTransition::ResetTask);
|
||||
} else if (command == "d") {
|
||||
cout << "> reset devices" << endl;
|
||||
topo.ChangeState(TopologyTransition::ResetDevice);
|
||||
} else if (command == "h") {
|
||||
cout << "> help" << endl;
|
||||
printControlsHelp();
|
||||
} else if (command == "q") {
|
||||
cout << "> end" << endl;
|
||||
topo.ChangeState(TopologyTransition::End);
|
||||
// TODO: extend me..?
|
||||
} else {
|
||||
cout << "\033[01;32mInvalid input: [" << c << "]\033[0m" << endl;
|
||||
printControlsHelp();
|
||||
}
|
||||
|
||||
if (commandIn != "") {
|
||||
this_thread::sleep_for(chrono::milliseconds(100)); // give dds a chance to complete request
|
||||
break;
|
||||
} else {
|
||||
cin >> c;
|
||||
command = c;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
int main(int argc, char* argv[])
|
||||
try {
|
||||
string topoFile;
|
||||
string sessionID;
|
||||
string command;
|
||||
string topologyPath;
|
||||
string targetState;
|
||||
unsigned int timeout;
|
||||
|
||||
bpo::options_description opts("Common options");
|
||||
|
||||
auto envSessionId = getenv("FAIRMQ_DDS_SESSION_ID");
|
||||
if (envSessionId) {
|
||||
opts.add_options()("session,s", bpo::value<string>(&sessionID)->default_value(envSessionId), "DDS Session ID (overrides any value in env var $FAIRMQ_DDS_SESSION_ID)");
|
||||
} else {
|
||||
opts.add_options()("session,s", bpo::value<string>(&sessionID)->required(), "DDS Session ID (overrides any value in env var $FAIRMQ_DDS_SESSION_ID)");
|
||||
}
|
||||
|
||||
auto envTopoFile = getenv("FAIRMQ_DDS_TOPO_FILE");
|
||||
if (envTopoFile) {
|
||||
opts.add_options()("topology-file,f", bpo::value<string>(&topoFile)->default_value(envTopoFile), "DDS topology file path");
|
||||
} else {
|
||||
opts.add_options()("topology-file,f", bpo::value<string>(&topoFile)->required(), "DDS topology file path");
|
||||
}
|
||||
|
||||
opts.add_options()
|
||||
("command,c", bpo::value<string>(&command)->default_value(""), "Command character")
|
||||
("path,p", bpo::value<string>(&topologyPath)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)")
|
||||
("wait-for-state,w", bpo::value<string>(&targetState)->default_value(""), "Wait until targeted FairMQ devices reach the given state")
|
||||
("timeout,t", bpo::value<unsigned int>(&timeout)->default_value(0), "Timeout in milliseconds when waiting for a device state (0 - wait infinitely)")
|
||||
("help,h", "Produce help message");
|
||||
|
||||
bpo::variables_map vm;
|
||||
bpo::store(bpo::command_line_parser(argc, argv).options(opts).run(), vm);
|
||||
|
||||
if (vm.count("help")) {
|
||||
cout << "FairMQ DDS Command UI" << endl << opts << endl;
|
||||
cout << "Commands: [c] check state, [o] dump config, [h] help, [r] run, [s] stop, [t] reset task, [d] reset device, [q] end, [j] init task, [i] init device, [k] complete init, [b] bind, [x] connect" << endl;
|
||||
return EXIT_SUCCESS;
|
||||
}
|
||||
|
||||
bpo::notify(vm);
|
||||
|
||||
DDSEnvironment env;
|
||||
DDSSession session(sessionID, env);
|
||||
DDSTopology ddsTopo(DDSTopology::Path(topoFile), env);
|
||||
|
||||
int n = ddsTopo.GetNumRequiredAgents();
|
||||
cout << "Number of required agents/slots: " << n << endl;
|
||||
cout << "creating Topology" << endl;
|
||||
|
||||
Topology topo(ddsTopo, session);
|
||||
for (auto transition : { TopologyTransition::InitDevice,
|
||||
TopologyTransition::CompleteInit,
|
||||
TopologyTransition::Bind,
|
||||
TopologyTransition::Connect,
|
||||
TopologyTransition::InitTask,
|
||||
TopologyTransition::Run,
|
||||
TopologyTransition::Stop,
|
||||
TopologyTransition::ResetTask,
|
||||
TopologyTransition::ResetDevice,
|
||||
TopologyTransition::End }) {
|
||||
topo.ChangeState(transition);
|
||||
}
|
||||
|
||||
cout << "Finishing..." << endl;
|
||||
return EXIT_SUCCESS;
|
||||
} catch (exception& e) {
|
||||
cerr << "Error: " << e.what() << endl;
|
||||
return EXIT_FAILURE;
|
||||
}
|
||||
|
|
@ -194,9 +194,7 @@ class BasicTopology : public AsioBase<Executor, Allocator>
|
|||
case Type::state_change: {
|
||||
auto _cmd = static_cast<StateChange&>(*cmd);
|
||||
if (_cmd.GetCurrentState() == DeviceState::Exiting) {
|
||||
Cmds outCmds;
|
||||
outCmds.Add<StateChangeExitingReceived>();
|
||||
fDDSSession.SendCommand(outCmds.Serialize(), senderId);
|
||||
fDDSSession.SendCommand(Cmds(make<StateChangeExitingReceived>()).Serialize(), senderId);
|
||||
}
|
||||
HandleCmd(_cmd);
|
||||
} break;
|
||||
|
|
@ -213,7 +211,7 @@ class BasicTopology : public AsioBase<Executor, Allocator>
|
|||
case Type::transition_status: {
|
||||
auto _cmd = static_cast<TransitionStatus&>(*cmd);
|
||||
if (_cmd.GetResult() != Result::Ok) {
|
||||
LOG(error) << "Transition failed for " << _cmd.GetDeviceId();
|
||||
LOG(error) << _cmd.GetTransition() << " transition failed for " << _cmd.GetDeviceId();
|
||||
std::lock_guard<std::mutex> lk(fMtx);
|
||||
if (!fChangeStateOp.IsCompleted() && fStateData.at(fStateIndex.at(_cmd.GetTaskId())).state != fChangeStateTarget) {
|
||||
fChangeStateOpTimer.cancel();
|
||||
|
|
@ -568,7 +566,6 @@ class BasicTopology : public AsioBase<Executor, Allocator>
|
|||
return asio::async_initiate<CompletionToken, WaitForStateCompletionSignature>([&](auto handler) {
|
||||
typename GetPropertiesOp::Id const id(tools::UuidHash());
|
||||
|
||||
// TODO Implement garbage collection of completed ops
|
||||
std::lock_guard<std::mutex> lk(fMtx);
|
||||
|
||||
for (auto it = begin(fWaitForStateOps); it != end(fWaitForStateOps);) {
|
||||
|
|
@ -579,6 +576,8 @@ class BasicTopology : public AsioBase<Executor, Allocator>
|
|||
}
|
||||
}
|
||||
|
||||
LOG(info) << fDDSTopo.GetTasks(path).size();
|
||||
|
||||
auto p = fWaitForStateOps.emplace(
|
||||
std::piecewise_construct,
|
||||
std::forward_as_tuple(id),
|
||||
|
|
|
|||
|
|
@ -134,6 +134,9 @@ TEST_F(Topology, ChangeState)
|
|||
EXPECT_EQ(result.first, std::error_code());
|
||||
EXPECT_NO_THROW(sdk::AggregateState(result.second));
|
||||
EXPECT_EQ(sdk::StateEqualsTo(result.second, sdk::DeviceState::InitializingDevice), true);
|
||||
auto const currentState = topo.GetCurrentState();
|
||||
EXPECT_NO_THROW(sdk::AggregateState(currentState));
|
||||
EXPECT_EQ(sdk::StateEqualsTo(currentState, sdk::DeviceState::InitializingDevice), true);
|
||||
}
|
||||
|
||||
TEST_F(Topology, AsyncChangeStateConcurrent)
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user