diff --git a/examples/dds/fairmq-start-ex-dds.sh.in b/examples/dds/fairmq-start-ex-dds.sh.in index d0398142..56a340a4 100755 --- a/examples/dds/fairmq-start-ex-dds.sh.in +++ b/examples/dds/fairmq-start-ex-dds.sh.in @@ -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 "------------------------" diff --git a/examples/qc/fairmq-start-ex-qc.sh.in b/examples/qc/fairmq-start-ex-qc.sh.in index 5b9bf346..9fd55615 100755 --- a/examples/qc/fairmq-start-ex-qc.sh.in +++ b/examples/qc/fairmq-start-ex-qc.sh.in @@ -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 "------------------------" diff --git a/fairmq/plugins/DDS/CMakeLists.txt b/fairmq/plugins/DDS/CMakeLists.txt index 52555131..731a4724 100644 --- a/fairmq/plugins/DDS/CMakeLists.txt +++ b/fairmq/plugins/DDS/CMakeLists.txt @@ -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} diff --git a/fairmq/plugins/DDS/runDDSCommandUI.cxx b/fairmq/plugins/DDS/runDDSCommandUI.cxx index 5c7389ee..5cd5f4a3 100644 --- a/fairmq/plugins/DDS/runDDSCommandUI.cxx +++ b/fairmq/plugins/DDS/runDDSCommandUI.cxx @@ -8,23 +8,23 @@ #include #include - -#include +#include #include +#include // raw mode console input +#include #include #include #include #include #include -#include // raw mode console input #include #include -#include 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()).Serialize(), fTopologyPath); - } - - ~StateSubscription() { - fDdsCustomCmd.send(Cmds(make()).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,254 +56,154 @@ 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) { + if (command == "c") { + cout << "> checking state of the devices" << endl; + 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; + // 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; + topo.ChangeState(TopologyTransition::InitDevice, std::chrono::milliseconds(timeout)); + } else if (command == "k") { + cout << "> complete init" << endl; + topo.ChangeState(TopologyTransition::CompleteInit, std::chrono::milliseconds(timeout)); + } else if (command == "b") { + cout << "> bind devices" << endl; + topo.ChangeState(TopologyTransition::Bind, std::chrono::milliseconds(timeout)); + } else if (command == "x") { + cout << "> connect devices" << endl; + topo.ChangeState(TopologyTransition::Connect, std::chrono::milliseconds(timeout)); + } else if (command == "j") { + cout << "> init tasks" << endl; + topo.ChangeState(TopologyTransition::InitTask, std::chrono::milliseconds(timeout)); + } else if (command == "r") { + cout << "> run tasks" << endl; + topo.ChangeState(TopologyTransition::Run, std::chrono::milliseconds(timeout)); + } else if (command == "s") { + cout << "> stop devices" << endl; + topo.ChangeState(TopologyTransition::Stop, std::chrono::milliseconds(timeout)); + } else if (command == "t") { + cout << "> reset tasks" << endl; + topo.ChangeState(TopologyTransition::ResetTask, std::chrono::milliseconds(timeout)); + } else if (command == "d") { + cout << "> reset devices" << endl; + topo.ChangeState(TopologyTransition::ResetDevice, std::chrono::milliseconds(timeout)); + } else if (command == "h") { + cout << "> help" << endl; + printControlsHelp(); + } else if (command == "q") { + cout << "> end" << endl; + topo.ChangeState(TopologyTransition::End, std::chrono::milliseconds(timeout)); + } else { + cout << "\033[01;32mInvalid input: [" << command << "]\033[0m" << endl; + printControlsHelp(); + } +} + +void sendCommand(const string& commandIn, const string& path, unsigned int timeout, Topology& topo) +{ + if (commandIn != "") { + handleCommand(commandIn, path, timeout, topo); + return; + } + char c; - string command(commandIn); + string command; TerminalConfig tconfig; - if (command == "") { - printControlsHelp(); + printControlsHelp(); + cin >> c; + command = c; + + while (true) { + handleCommand(command, path, timeout, topo); cin >> c; command = c; } - - while (true) { - if (command == "c") { - cout << "> checking state of the devices" << endl; - ddsCustomCmd.send(Cmds(make()).Serialize(), topologyPath); - } else if (command == "o") { - cout << "> dumping config of the devices" << endl; - ddsCustomCmd.send(Cmds(make()).Serialize(), topologyPath); - } else if (command == "i") { - cout << "> init devices" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::InitDevice)).Serialize(), topologyPath); - } else if (command == "k") { - cout << "> complete init" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::CompleteInit)).Serialize(), topologyPath); - } else if (command == "b") { - cout << "> bind devices" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::Bind)).Serialize(), topologyPath); - } else if (command == "x") { - cout << "> connect devices" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::Connect)).Serialize(), topologyPath); - } else if (command == "j") { - cout << "> init tasks" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::InitTask)).Serialize(), topologyPath); - } else if (command == "r") { - cout << "> run tasks" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::Run)).Serialize(), topologyPath); - } else if (command == "s") { - cout << "> stop devices" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::Stop)).Serialize(), topologyPath); - } else if (command == "t") { - cout << "> reset tasks" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::ResetTask)).Serialize(), topologyPath); - } else if (command == "d") { - cout << "> reset devices" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::ResetDevice)).Serialize(), topologyPath); - } else if (command == "h") { - cout << "> help" << endl; - printControlsHelp(); - } else if (command == "q") { - cout << "> end" << endl; - ddsCustomCmd.send(Cmds(make(fair::mq::Transition::End)).Serialize(), topologyPath); - } 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; - } - } } -struct WaitMode -{ - 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)); - } - } - } - - void Run(const chrono::milliseconds& timeout, const string& topologyPath, CCustomCmd& ddsCustomCmd, unsigned int numDevices, const string& command = "") - { - if (command != "") { - sendCommand(command, topologyPath, ddsCustomCmd); - } - - // 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; - }; - - unique_lock lock(fMtx); - - if (timeout > chrono::milliseconds(0)) { - if (!fCV.wait_for(lock, timeout, condition)) { - throw runtime_error("timeout"); - } - } else { - fCV.wait(lock, condition); - } - - // cout << "WaitMode.Run() finished" << endl; - } - - void CountStates(fair::mq::State lastState, fair::mq::State currentState) - { - { - unique_lock 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 fTargetStatePair; - unsigned int fTransitionedCount; -}; - int main(int argc, char* argv[]) -{ - try { - string sessionID; - string command; - string topologyPath; - string targetState; - unsigned int timeout; - unsigned int numDevices(0); +try { + string sessionID; + string topoFile; - bpo::options_description options("Common options"); + string command; + string path; + string targetState; + unsigned int timeout; - auto envSessionId = getenv("DDS_SESSION_ID"); - if (envSessionId) { - options.add_options()("session,s", bpo::value(&sessionID)->default_value(envSessionId), "DDS Session ID (overrides any value in env var $DDS_SESSION_ID)"); - } else { - options.add_options()("session,s", bpo::value(&sessionID)->required(), "DDS Session ID (overrides any value in env var $DDS_SESSION_ID)"); - } + fair::Logger::SetConsoleSeverity("debug"); - options.add_options() - ("command,c", bpo::value (&command)->default_value(""), "Command character") - ("path,p", bpo::value (&topologyPath)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)") - ("wait-for-state,w", bpo::value (&targetState)->default_value(""), "Wait until targeted FairMQ devices reach the given state") - ("timeout,t", bpo::value (&timeout)->default_value(0), "Timeout in milliseconds when waiting for a device state (0 - wait infinitely)") - ("number-devices,n", bpo::value (&numDevices)->default_value(0), "Number of devices (will be removed in the future)") - ("help,h", "Produce help message"); + bpo::options_description options("Common options"); - bpo::variables_map vm; - bpo::store(bpo::command_line_parser(argc, argv).options(options).run(), vm); - - if (vm.count("help")) { - cout << "FairMQ DDS Command UI" << endl << options << 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); - - WaitMode waitMode(targetState); - - CIntercomService service; - CCustomCmd ddsCustomCmd(service); - - 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(*cmd).GetDeviceId() << ": " << static_cast(*cmd).GetLastState() << "->" << static_cast(*cmd).GetCurrentState() << endl; - if (static_cast(*cmd).GetCurrentState() == fair::mq::State::Exiting) { - ddsCustomCmd.send(Cmds(make()).Serialize(), to_string(senderId)); - } - waitMode.CountStates(static_cast(*cmd).GetLastState(), static_cast(*cmd).GetCurrentState()); - } - break; - case Type::state_change_subscription: - if (static_cast(*cmd).GetResult() != Result::Ok) { - cout << "State change subscription failed for " << static_cast(*cmd).GetDeviceId() << endl; - } - break; - case Type::state_change_unsubscription: - if (static_cast(*cmd).GetResult() != Result::Ok) { - cout << "State change unsubscription failed for " << static_cast(*cmd).GetDeviceId() << endl; - } - break; - case Type::transition_status: { - if (static_cast(*cmd).GetResult() == Result::Ok) { - cout << "Device " << static_cast(*cmd).GetDeviceId() << " started to transition with " << static_cast(*cmd).GetTransition() << endl; - } else { - cout << "Device " << static_cast(*cmd).GetDeviceId() << " cannot transition with " << static_cast(*cmd).GetTransition() << endl; - } - } - break; - case Type::current_state: - cout << "Device " << static_cast(*cmd).GetDeviceId() << " is in " << static_cast(*cmd).GetCurrentState() << " state" << endl; - break; - case Type::config: - cout << "Received config for device " << static_cast(*cmd).GetDeviceId() << ":\n" << static_cast(*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); - } - - ddsCustomCmd.unsubscribe(); - } catch (exception& e) { - cerr << "Error: " << e.what() << endl; - return EXIT_FAILURE; + auto envSessionId = getenv("FAIRMQ_DDS_SESSION_ID"); + if (envSessionId) { + options.add_options()("session,s", bpo::value(&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(&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(&topoFile)->default_value(envTopoFile), "DDS topology file path"); + } else { + options.add_options()("topology-file,f", bpo::value(&topoFile)->required(), "DDS topology file path"); + } + + options.add_options() + ("command,c", bpo::value(&command)->default_value(""), "Command character") + ("path,p", bpo::value(&path)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)") + ("wait-for-state,w", bpo::value(&targetState)->default_value(""), "Wait until targeted FairMQ devices reach the given state") + ("timeout,t", bpo::value(&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(options).run(), vm); + + if (vm.count("help")) { + cout << "FairMQ DDS Command UI" << endl << options << 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); + + Topology topo(ddsTopo, session); + + if (targetState != "") { + if (command != "") { + sendCommand(command, path, timeout, topo); + } + 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 { + /* 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; + } + } else { + sendCommand(command, path, timeout, topo); + } + return EXIT_SUCCESS; +} catch (exception& e) { + cerr << "Error: " << e.what() << endl; + return EXIT_FAILURE; } diff --git a/fairmq/plugins/DDS/runDDSCommandUInew.cxx b/fairmq/plugins/DDS/runDDSCommandUInew.cxx deleted file mode 100644 index 066125b2..00000000 --- a/fairmq/plugins/DDS/runDDSCommandUInew.cxx +++ /dev/null @@ -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 -#include -#include - -#include - -#include // raw mode console input -#include -#include -#include -#include -#include -#include -#include -#include - -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(&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(&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(&topoFile)->default_value(envTopoFile), "DDS topology file path"); - } else { - opts.add_options()("topology-file,f", bpo::value(&topoFile)->required(), "DDS topology file path"); - } - - opts.add_options() - ("command,c", bpo::value(&command)->default_value(""), "Command character") - ("path,p", bpo::value(&topologyPath)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)") - ("wait-for-state,w", bpo::value(&targetState)->default_value(""), "Wait until targeted FairMQ devices reach the given state") - ("timeout,t", bpo::value(&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; -} diff --git a/fairmq/sdk/Topology.h b/fairmq/sdk/Topology.h index 49af7fce..e8d06f64 100644 --- a/fairmq/sdk/Topology.h +++ b/fairmq/sdk/Topology.h @@ -194,9 +194,7 @@ class BasicTopology : public AsioBase case Type::state_change: { auto _cmd = static_cast(*cmd); if (_cmd.GetCurrentState() == DeviceState::Exiting) { - Cmds outCmds; - outCmds.Add(); - fDDSSession.SendCommand(outCmds.Serialize(), senderId); + fDDSSession.SendCommand(Cmds(make()).Serialize(), senderId); } HandleCmd(_cmd); } break; @@ -213,7 +211,7 @@ class BasicTopology : public AsioBase case Type::transition_status: { auto _cmd = static_cast(*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 lk(fMtx); if (!fChangeStateOp.IsCompleted() && fStateData.at(fStateIndex.at(_cmd.GetTaskId())).state != fChangeStateTarget) { fChangeStateOpTimer.cancel(); @@ -568,7 +566,6 @@ class BasicTopology : public AsioBase return asio::async_initiate([&](auto handler) { typename GetPropertiesOp::Id const id(tools::UuidHash()); - // TODO Implement garbage collection of completed ops std::lock_guard lk(fMtx); for (auto it = begin(fWaitForStateOps); it != end(fWaitForStateOps);) { @@ -579,6 +576,8 @@ class BasicTopology : public AsioBase } } + LOG(info) << fDDSTopo.GetTasks(path).size(); + auto p = fWaitForStateOps.emplace( std::piecewise_construct, std::forward_as_tuple(id), diff --git a/test/sdk/_topology.cxx b/test/sdk/_topology.cxx index c1146374..05ecc6a0 100644 --- a/test/sdk/_topology.cxx +++ b/test/sdk/_topology.cxx @@ -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)