Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-14 00:56:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use new commands format in dds command UI

Browse Source
This commit is contained in:
Alexey Rybalchenko 2019-09-27 15:46:43 +02:00 committed by Dennis Klein
parent a53e95b5f6
commit 1c1509af3e
1 changed files with 106 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
fairmq/plugins/DDS/runDDSCommandUI.cxx
View File

@ -6,27 +6,33 @@
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include <algorithm> #include <fairmq/sdk/commands/Commands.h>
#include <atomic> #include <fairmq/States.h>
#include <DDS/dds_intercom.h>
#include <boost/algorithm/string/classification.hpp> #include <boost/algorithm/string/classification.hpp>
#include <boost/algorithm/string/predicate.hpp> #include <boost/algorithm/string/predicate.hpp>
#include <boost/algorithm/string/split.hpp> #include <boost/algorithm/string/split.hpp>
#include <boost/algorithm/string/trim.hpp> #include <boost/algorithm/string/trim.hpp>
#include <boost/program_options.hpp> #include <boost/program_options.hpp>
#include <algorithm>
#include <atomic>
#include <condition_variable> #include <condition_variable>
#include <cstdlib> #include <cstdlib>
#include <DDS/dds_intercom.h>
#include <exception> #include <exception>
#include <iostream> #include <iostream>
#include <unordered_map>
#include <mutex> #include <mutex>
#include <string>
#include <termios.h> // raw mode console input #include <termios.h> // raw mode console input
#include <thread> #include <thread>
#include <string> #include <utility>
#include <unistd.h> #include <unistd.h>
using namespace std; using namespace std;
using namespace dds::intercom_api; using namespace dds::intercom_api;
using namespace fair::mq::sdk::cmd;
namespace bpo = boost::program_options; namespace bpo = boost::program_options;
struct TerminalConfig struct TerminalConfig
@ -50,7 +56,8 @@ struct TerminalConfig
} }
}; };
struct StateSubscription { struct StateSubscription
{
const string& fTopologyPath; const string& fTopologyPath;
CCustomCmd& fDdsCustomCmd; CCustomCmd& fDdsCustomCmd;
@ -58,11 +65,11 @@ struct StateSubscription {
: fTopologyPath(topologyPath) : fTopologyPath(topologyPath)
, fDdsCustomCmd(ddsCustomCmd) , fDdsCustomCmd(ddsCustomCmd)
{ {
fDdsCustomCmd.send("subscribe-to-state-changes", fTopologyPath); fDdsCustomCmd.send(Cmds(make<SubscribeToStateChange>()).Serialize(), fTopologyPath);
} }
~StateSubscription() { ~StateSubscription() {
fDdsCustomCmd.send("unsubscribe-from-state-changes", fTopologyPath); fDdsCustomCmd.send(Cmds(make<UnsubscribeFromStateChange>()).Serialize(), fTopologyPath);
this_thread::sleep_for(chrono::milliseconds(100)); // give dds a chance to complete request this_thread::sleep_for(chrono::milliseconds(100)); // give dds a chance to complete request
} }
}; };
@ -74,7 +81,7 @@ void printControlsHelp()
cout << "To quit press Ctrl+C" << endl; cout << "To quit press Ctrl+C" << endl;
} }
void commandMode(const string& commandIn, const string& topologyPath, CCustomCmd& ddsCustomCmd) { void sendCommand(const string& commandIn, const string& topologyPath, CCustomCmd& ddsCustomCmd) {
char c; char c;
string command(commandIn); string command(commandIn);
TerminalConfig tconfig; TerminalConfig tconfig;
@ -88,52 +95,43 @@ void commandMode(const string& commandIn, const string& topologyPath, CCustomCmd
while (true) { while (true) {
if (command == "c") { if (command == "c") {
cout << "> checking state of the devices" << endl; cout << "> checking state of the devices" << endl;
ddsCustomCmd.send("check-state", topologyPath); ddsCustomCmd.send(Cmds(make<CheckState>()).Serialize(), topologyPath);
} else if (command == "o") { } else if (command == "o") {
cout << "> dumping config of the devices" << endl; cout << "> dumping config of the devices" << endl;
ddsCustomCmd.send("dump-config", topologyPath); ddsCustomCmd.send(Cmds(make<DumpConfig>()).Serialize(), topologyPath);
} else if (command == "i") { } else if (command == "i") {
cout << "> init devices" << endl; cout << "> init devices" << endl;
ddsCustomCmd.send("INIT DEVICE", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::InitDevice)).Serialize(), topologyPath);
} else if (command == "k") { } else if (command == "k") {
cout << "> complete init" << endl; cout << "> complete init" << endl;
ddsCustomCmd.send("COMPLETE INIT", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::CompleteInit)).Serialize(), topologyPath);
} else if (command == "b") { } else if (command == "b") {
cout << "> bind devices" << endl; cout << "> bind devices" << endl;
ddsCustomCmd.send("BIND", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::Bind)).Serialize(), topologyPath);
} else if (command == "x") { } else if (command == "x") {
cout << "> connect devices" << endl; cout << "> connect devices" << endl;
ddsCustomCmd.send("CONNECT", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::Connect)).Serialize(), topologyPath);
} else if (command == "j") { } else if (command == "j") {
cout << "> init tasks" << endl; cout << "> init tasks" << endl;
ddsCustomCmd.send("INIT TASK", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::InitTask)).Serialize(), topologyPath);
} else if (command == "p") {
cout << "> pause devices" << endl;
ddsCustomCmd.send("PAUSE", topologyPath);
} else if (command == "r") { } else if (command == "r") {
cout << "> run tasks" << endl; cout << "> run tasks" << endl;
ddsCustomCmd.send("RUN", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::Run)).Serialize(), topologyPath);
} else if (command == "s") { } else if (command == "s") {
cout << "> stop devices" << endl; cout << "> stop devices" << endl;
ddsCustomCmd.send("STOP", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::Stop)).Serialize(), topologyPath);
} else if (command == "t") { } else if (command == "t") {
cout << "> reset tasks" << endl; cout << "> reset tasks" << endl;
ddsCustomCmd.send("RESET TASK", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::ResetTask)).Serialize(), topologyPath);
} else if (command == "d") { } else if (command == "d") {
cout << "> reset devices" << endl; cout << "> reset devices" << endl;
ddsCustomCmd.send("RESET DEVICE", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::ResetDevice)).Serialize(), topologyPath);
} else if (command == "h") { } else if (command == "h") {
cout << "> help" << endl; cout << "> help" << endl;
printControlsHelp(); printControlsHelp();
} else if (command == "q") { } else if (command == "q") {
cout << "> end" << endl; cout << "> end" << endl;
ddsCustomCmd.send("END", topologyPath); ddsCustomCmd.send(Cmds(make<ChangeState>(fair::mq::Transition::End)).Serialize(), topologyPath);
} else if (command == "q!") {
cout << "> shutdown" << endl;
ddsCustomCmd.send("SHUTDOWN", topologyPath);
} else if (command == "r!") {
cout << "> startup" << endl;
ddsCustomCmd.send("STARTUP", topologyPath);
} else { } else {
cout << "\033[01;32mInvalid input: [" << c << "]\033[0m" << endl; cout << "\033[01;32mInvalid input: [" << c << "]\033[0m" << endl;
printControlsHelp(); printControlsHelp();
@ -152,33 +150,36 @@ void commandMode(const string& commandIn, const string& topologyPath, CCustomCmd
struct WaitMode struct WaitMode
{ {
explicit WaitMode(const string& targetState) explicit WaitMode(const string& targetState)
: fTargetState(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, void Run(const chrono::milliseconds& timeout, const string& topologyPath, CCustomCmd& ddsCustomCmd, unsigned int numDevices, const string& command = "")
const string& topologyPath,
CCustomCmd& ddsCustomCmd,
unsigned int numberDevices,
const string& command = "")
{ {
StateSubscription stateSubscription(topologyPath, ddsCustomCmd); StateSubscription stateSubscription(topologyPath, ddsCustomCmd);
if (command != "") { if (command != "") {
commandMode(command, topologyPath, ddsCustomCmd); sendCommand(command, topologyPath, ddsCustomCmd);
} }
// TODO once DDS provides an API to retrieve actual number of tasks, use it here // TODO once DDS provides an API to retrieve actual number of tasks, use it here
auto condition = [&] { auto condition = [&] {
bool res(!fTargetStates.empty() bool res = fTransitionedCount == numDevices;
&& all_of(fTargetStates.cbegin(), if (fTargetStatePair.first == fair::mq::State::Ok) {
fTargetStates.cend(), cout << "Waiting for " << numDevices << " devices to reach " << fTargetStatePair.second << ", condition check: " << res << endl;
[&](unordered_map<uint64_t, string>::value_type i) { } else {
return boost::algorithm::ends_with(i.second, fTargetState); cout << "Waiting for " << numDevices << " devices to reach " << fTargetStatePair.first << "->" << fTargetStatePair.second << ", condition check: " << res << endl;
}));
if (numberDevices > 0) {
res = res && (fTargetStates.size() == numberDevices);
} }
cout << "waiting for " << numberDevices << " devices to reach " << fTargetState << ", condition check: " << res << endl;
return res; return res;
}; };
@ -191,21 +192,33 @@ struct WaitMode
} else { } else {
fCV.wait(lock, condition); fCV.wait(lock, condition);
} }
// cout << "WaitMode.Run() finished" << endl;
} }
void AddNewStateEntry(uint64_t senderId, const string& state) void CountStates(fair::mq::State lastState, fair::mq::State currentState)
{ {
{ {
unique_lock<mutex> lock(fMtx); unique_lock<mutex> lock(fMtx);
fTargetStates[senderId] = state; 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(); fCV.notify_one();
} }
mutex fMtx; mutex fMtx;
condition_variable fCV; condition_variable fCV;
unordered_map<uint64_t, string> fTargetStates; pair<fair::mq::State, fair::mq::State> fTargetStatePair;
string fTargetState; unsigned int fTransitionedCount;
}; };
int main(int argc, char* argv[]) int main(int argc, char* argv[])
@ -216,7 +229,7 @@ int main(int argc, char* argv[])
string topologyPath; string topologyPath;
string targetState; string targetState;
unsigned int timeout; unsigned int timeout;
unsigned int numberDevices(0); unsigned int numDevices(0);
bpo::options_description options("Common options"); bpo::options_description options("Common options");
@ -232,7 +245,7 @@ int main(int argc, char* argv[])
("path,p", bpo::value<string> (&topologyPath)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)") ("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") ("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)") ("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> (&numberDevices)->default_value(0), "Number of devices (will be removed in the future)") ("number-devices,n", bpo::value<unsigned int> (&numDevices)->default_value(0), "Number of devices (will be removed in the future)")
("help,h", "Produce help message"); ("help,h", "Produce help message");
bpo::variables_map vm; bpo::variables_map vm;
@ -257,40 +270,58 @@ int main(int argc, char* argv[])
// subscribe to receive messages from DDS // subscribe to receive messages from DDS
ddsCustomCmd.subscribe([&](const string& msg, const string& /*condition*/, uint64_t senderId) { ddsCustomCmd.subscribe([&](const string& msg, const string& /*condition*/, uint64_t senderId) {
// cerr << "Received: " << msg << endl; Cmds cmds;
vector<string> parts; cmds.Deserialize(msg);
boost::algorithm::split(parts, msg, boost::algorithm::is_any_of(":,")); // cout << "Received " << cmds.Size() << " command(s) with total size of " << msg.length() << " bytes: " << endl;
if (parts[0] == "state-change") { for (const auto& cmd : cmds) {
// cerr << "Received: " << msg << endl; // cout << " > " << cmd->GetType() << endl;
boost::trim(parts[2]); switch (cmd->GetType()) {
waitMode.AddNewStateEntry(senderId, parts[3]); case Type::state_change: {
if(parts[3] == "IDLE->EXITING") { cout << "Received state_change from " << static_cast<StateChange&>(*cmd).GetDeviceId() << ": " << static_cast<StateChange&>(*cmd).GetLastState() << "->" << static_cast<StateChange&>(*cmd).GetCurrentState() << endl;
ddsCustomCmd.send("state-change-exiting-received", std::to_string(senderId)); if (static_cast<StateChange&>(*cmd).GetCurrentState() == fair::mq::State::Exiting) {
ddsCustomCmd.send(Cmds(make<StateChangeExitingReceived>()).Serialize(), to_string(senderId));
}
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;
// } else {
// cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " cannot transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
// }
}
break;
default:
cout << "Unexpected/unknown command received: " << cmd->GetType() << endl;
cout << "Origin: " << senderId << endl;
break;
} }
} else if (parts[0] == "state-changes-subscription") {
if (parts[2] != "OK") {
cerr << "state-changes-subscription failed with return code: " << parts[2];
}
} else if (parts[0] == "state-changes-unsubscription") {
if (parts[2] != "OK") {
cerr << "state-changes-unsubscription failed with return code: " << parts[2];
}
} else {
cout << "Received: " << msg << endl;
} }
}); });
service.start(sessionID); service.start(sessionID);
if (targetState == "") { if (targetState == "") {
commandMode(command, topologyPath, ddsCustomCmd); sendCommand(command, topologyPath, ddsCustomCmd);
} else { } else {
waitMode.Run(chrono::milliseconds(timeout), topologyPath, ddsCustomCmd, numberDevices, command); waitMode.Run(chrono::milliseconds(timeout), topologyPath, ddsCustomCmd, numDevices, command);
} }
ddsCustomCmd.unsubscribe();
} catch (exception& e) { } catch (exception& e) {
cerr << "Error: " << e.what() << endl; cerr << "Error: " << e.what() << endl;
return EXIT_FAILURE; return EXIT_FAILURE;
} }
return EXIT_SUCCESS; return EXIT_SUCCESS;
} }