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

feat!: Remove deprecated components sdk, sdk_commands, dds_plugin

Browse Source 
BREAKING CHANGE: Components have been moved to ODC project, see
https://github.com/FairRootGroup/FairMQ/discussions/392 for details.
This commit is contained in:
Dennis Klein
2022-03-22 12:58:10 +01:00
parent 226733c653
commit cda7282422
47 changed files with 23 additions and 6582 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
fairmq/CMakeLists.txt
View File

@@ -1,12 +1,12 @@
################################################################################
# Copyright (C) 2012-2021 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
# Copyright (C) 2012-2022 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" #
################################################################################
if(BUILD_FAIRMQ OR BUILD_SDK)
if(BUILD_FAIRMQ)
if(BUILD_TIDY_TOOL)
include(FairMQTidy)
@@ -452,20 +452,9 @@ if(BUILD_FAIRMQ)
endforeach()
endif()
if(BUILD_SDK_COMMANDS)
add_subdirectory(sdk/commands)
endif()
if(BUILD_SDK)
add_subdirectory(sdk)
endif()
####################
# external plugins #
####################
if(BUILD_DDS_PLUGIN)
add_subdirectory(plugins/DDS)
endif()
if(BUILD_PMIX_PLUGIN)
add_subdirectory(plugins/PMIx)
endif()

26
fairmq/SDK.h
View File

@@ -1,26 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_H
#define FAIR_MQ_SDK_H
// IWYU pragma: begin_exports
#include <fairmq/sdk/AsioAsyncOp.h>
#include <fairmq/sdk/AsioBase.h>
#include <fairmq/sdk/DDSAgent.h>
#include <fairmq/sdk/DDSEnvironment.h>
#include <fairmq/sdk/DDSInfo.h>
#include <fairmq/sdk/DDSSession.h>
#include <fairmq/sdk/DDSTask.h>
#include <fairmq/sdk/DDSTopology.h>
#include <fairmq/sdk/Error.h>
#include <fairmq/sdk/Topology.h>
#include <fairmq/sdk/Traits.h>
// IWYU pragma: end_exports
#endif // FAIR_MQ_SDK_H

24
fairmq/plugins/DDS/CMakeLists.txt
View File

@@ -1,24 +0,0 @@
################################################################################
# Copyright (C) 2012-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" #
################################################################################
set(plugin FairMQPlugin_dds)
add_library(${plugin} SHARED ${CMAKE_CURRENT_SOURCE_DIR}/DDS.cxx ${CMAKE_CURRENT_SOURCE_DIR}/DDS.h)
target_compile_features(${plugin} PUBLIC cxx_std_17)
target_link_libraries(${plugin} PUBLIC FairMQ StateMachine DDS::dds_intercom_lib DDS::dds_protocol_lib Boost::boost PRIVATE Commands asio::asio)
target_include_directories(${plugin} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
set_target_properties(${plugin} PROPERTIES CXX_VISIBILITY_PRESET hidden)
set_target_properties(${plugin} PROPERTIES
VERSION ${PROJECT_VERSION}
SOVERSION "${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}"
LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/fairmq
)
install(TARGETS ${plugin}
EXPORT ${PROJECT_EXPORT_SET}
LIBRARY DESTINATION ${PROJECT_INSTALL_LIBDIR}
)

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

@@ -1,457 +0,0 @@
/********************************************************************************
* 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 "DDS.h"
#include <fairmq/tools/Strings.h>
#include <boost/algorithm/string/join.hpp>
#include <boost/algorithm/string/split.hpp>
#include <boost/algorithm/string/classification.hpp>
#include <asio/post.hpp>
#include <cstdlib>
#include <stdexcept>
#include <sstream>
using namespace std;
using fair::mq::tools::ToString;
namespace fair::mq::plugins
{
DDS::DDS(const string& name,
const Plugin::Version version,
const string& maintainer,
const string& homepage,
PluginServices* pluginServices)
: Plugin(name, version, maintainer, homepage, pluginServices)
, fDDSTaskId(dds::env_prop<dds::task_id>())
, fCurrentState(DeviceState::Idle)
, fLastState(DeviceState::Idle)
, fDeviceTerminationRequested(false)
, fLastExternalController(0)
, fExitingAckedByLastExternalController(false)
, fUpdatesAllowed(false)
, fWorkGuard(fWorkerQueue.get_executor())
{
try {
TakeDeviceControl();
string deviceId(GetProperty<string>("id"));
if (deviceId.empty()) {
SetProperty<string>("id", dds::env_prop<dds::task_path>());
}
string sessionId(GetProperty<string>("session"));
if (sessionId == "default") {
SetProperty<string>("session", dds::env_prop<dds::dds_session_id>());
}
auto control = GetProperty<string>("control");
if (control == "static") {
LOG(error) << "DDS Plugin: static mode is not supported";
throw invalid_argument("DDS Plugin: static mode is not supported");
} else if (control == "dynamic" || control == "external" || control == "interactive") {
LOG(debug) << "Running DDS controller: external";
} else {
LOG(error) << "Unrecognized control mode '" << control << "' requested. " << "Ignoring and starting in external control mode.";
}
SubscribeForCustomCommands();
SubscribeForConnectingChannels();
// subscribe to device state changes, pushing new state changes into the event queue
SubscribeToDeviceStateChange([&](DeviceState newState) {
switch (newState) {
case DeviceState::Bound: {
// Receive addresses of connecting channels from DDS
// and propagate addresses of bound channels to DDS.
FillChannelContainers();
// allow updates from key value after channel containers are filled
{
lock_guard<mutex> lk(fUpdateMutex);
fUpdatesAllowed = true;
}
fUpdateCondition.notify_one();
// publish bound addresses via DDS at keys corresponding to the channel
// prefixes, e.g. 'data' in data[i]
PublishBoundChannels();
} break;
case DeviceState::ResettingDevice: {
{
lock_guard<mutex> lk(fUpdateMutex);
fUpdatesAllowed = false;
}
EmptyChannelContainers();
} break;
case DeviceState::Exiting: {
if (!fControllerThread.joinable()) {
fControllerThread = thread(&DDS::WaitForExitingAck, this);
}
fWorkGuard.reset();
fDeviceTerminationRequested = true;
UnsubscribeFromDeviceStateChange();
ReleaseDeviceControl();
} break;
default:
break;
}
using namespace sdk::cmd;
auto now = chrono::steady_clock::now();
string id = GetProperty<string>("id");
fLastState = fCurrentState;
fCurrentState = newState;
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
for (auto it = fStateChangeSubscribers.cbegin(); it != fStateChangeSubscribers.end();) {
// if a subscriber did not send a heartbeat in more than 3 times the promised interval,
// remove it from the subscriber list
if (chrono::duration<double>(now - it->second.first).count() > 3 * it->second.second) {
LOG(warn) << "Controller '" << it->first
<< "' did not send heartbeats since over 3 intervals ("
<< 3 * it->second.second << " ms), removing it.";
fStateChangeSubscribers.erase(it++);
} else {
LOG(debug) << "Publishing state-change: " << fLastState << "->" << fCurrentState << " to " << it->first;
Cmds cmds(make<StateChange>(id, fDDSTaskId, fLastState, fCurrentState));
fDDS.Send(cmds.Serialize(), to_string(it->first));
++it;
}
}
});
StartWorkerThread();
fDDS.Start();
} catch (PluginServices::DeviceControlError& e) {
LOG(debug) << e.what();
} catch (exception& e) {
LOG(error) << "Error in plugin initialization: " << e.what();
}
}
void DDS::EmptyChannelContainers()
{
fBindingChans.clear();
fConnectingChans.clear();
}
auto DDS::StartWorkerThread() -> void
{
fWorkerThread = thread([this]() {
fWorkerQueue.run();
});
}
auto DDS::WaitForExitingAck() -> void
{
unique_lock<mutex> lock(fStateChangeSubscriberMutex);
auto timeout = GetProperty<unsigned int>("wait-for-exiting-ack-timeout");
fExitingAcked.wait_for(lock, chrono::milliseconds(timeout), [this]() {
return fExitingAckedByLastExternalController || fStateChangeSubscribers.empty();
});
}
auto DDS::FillChannelContainers() -> void
{
try {
unordered_map<string, int> channelInfo(GetChannelInfo());
// fill binding and connecting chans
for (const auto& c : channelInfo) {
string methodKey{"chans." + c.first + "." + to_string(c.second - 1) + ".method"};
if (GetProperty<string>(methodKey) == "bind") {
fBindingChans.insert(make_pair(c.first, vector<string>()));
for (int i = 0; i < c.second; ++i) {
fBindingChans.at(c.first).push_back(GetProperty<string>(string{"chans." + c.first + "." + to_string(i) + ".address"}));
}
} else if (GetProperty<string>(methodKey) == "connect") {
fConnectingChans.insert(make_pair(c.first, DDSConfig()));
LOG(debug) << "preparing to connect: " << c.first << " with " << c.second << " sub-channels.";
fConnectingChans.at(c.first).fNumSubChannels = c.second;
} else {
LOG(error) << "Cannot update address configuration. Channel method (bind/connect) not specified.";
return;
}
}
// save properties that will have multiple values arriving (with only some of them to be used)
vector<string> iValues;
if (PropertyExists("dds-i")) {
iValues = GetProperty<vector<string>>("dds-i");
}
vector<string> inValues;
if (PropertyExists("dds-i-n")) {
inValues = GetProperty<vector<string>>("dds-i-n");
}
for (const auto& vi : iValues) {
size_t pos = vi.find(":");
string chanName = vi.substr(0, pos);
// check if provided name is a valid channel name
if (fConnectingChans.find(chanName) == fConnectingChans.end()) {
throw invalid_argument(ToString("channel provided to dds-i is not an actual connecting channel of this device: ", chanName));
}
int i = stoi(vi.substr(pos + 1));
LOG(debug) << "dds-i: adding " << chanName << " -> i of " << i;
fI.insert(make_pair(chanName, i));
}
for (const auto& vi : inValues) {
size_t pos = vi.find(":");
string chanName = vi.substr(0, pos);
// check if provided name is a valid channel name
if (fConnectingChans.find(chanName) == fConnectingChans.end()) {
throw invalid_argument(ToString("channel provided to dds-i-n is not an actual connecting channel of this device: ", chanName));
}
string i_n = vi.substr(pos + 1);
pos = i_n.find("-");
int i = stoi(i_n.substr(0, pos));
int n = stoi(i_n.substr(pos + 1));
LOG(debug) << "dds-i-n: adding " << chanName << " -> i: " << i << " n: " << n;
fIofN.insert(make_pair(chanName, IofN(i, n)));
}
} catch (const exception& e) {
LOG(error) << "Error filling channel containers: " << e.what();
}
}
auto DDS::SubscribeForConnectingChannels() -> void
{
LOG(debug) << "Subscribing for DDS properties.";
fDDS.SubscribeKeyValue([&] (const string& key, const string& value, uint64_t senderTaskID) {
LOG(debug) << "Received property: key=" << key << ", value=" << value << ", senderTaskID=" << senderTaskID;
if (key.compare(0, 8, "fmqchan_") != 0) {
LOG(debug) << "property update is not a channel info update: " << key;
return;
}
string channelName = key.substr(8);
LOG(info) << "Update for channel name: " << channelName;
asio::post(fWorkerQueue, [=]() {
try {
{
unique_lock<mutex> lk(fUpdateMutex);
fUpdateCondition.wait(lk, [&]{ return fUpdatesAllowed; });
}
if (fConnectingChans.find(channelName) == fConnectingChans.end()) {
LOG(error) << "Received an update for a connecting channel, but either no channel with given channel name exists or it has already been configured: '" << channelName << "', ignoring...";
return;
}
string val = value;
// check if it is to handle as one out of multiple values
auto it = fIofN.find(channelName);
if (it != fIofN.end()) {
it->second.fEntries.push_back(value);
if (it->second.fEntries.size() == it->second.fN) {
sort(it->second.fEntries.begin(), it->second.fEntries.end());
val = it->second.fEntries.at(it->second.fI);
} else {
LOG(debug) << "received " << it->second.fEntries.size() << " values for " << channelName << ", expecting total of " << it->second.fN;
return;
}
}
vector<string> connectionStrings;
boost::algorithm::split(connectionStrings, val, boost::algorithm::is_any_of(","));
if (connectionStrings.size() > 1) { // multiple bound channels received
auto it2 = fI.find(channelName);
if (it2 != fI.end()) {
LOG(debug) << "adding connecting channel " << channelName << " : " << connectionStrings.at(it2->second);
fConnectingChans.at(channelName).fDDSValues.insert({senderTaskID, connectionStrings.at(it2->second).c_str()});
} else {
LOG(error) << "multiple bound channels received, but no task index specified, only assigning the first";
fConnectingChans.at(channelName).fDDSValues.insert({senderTaskID, connectionStrings.at(0).c_str()});
}
} else { // only one bound channel received
fConnectingChans.at(channelName).fDDSValues.insert({senderTaskID, val.c_str()});
}
for (const auto& mi : fConnectingChans) {
if (mi.second.fNumSubChannels == mi.second.fDDSValues.size()) {
int i = 0;
for (const auto& e : mi.second.fDDSValues) {
auto result = UpdateProperty<string>(string{"chans." + mi.first + "." + to_string(i) + ".address"}, e.second);
if (!result) {
LOG(error) << "UpdateProperty failed for: " << "chans." << mi.first << "." << to_string(i) << ".address" << " - property does not exist";
}
++i;
}
}
}
} catch (const exception& e) {
LOG(error) << "Error handling DDS property: key=" << key << ", value=" << value << ", senderTaskID=" << senderTaskID << ": " << e.what();
}
});
});
}
auto DDS::PublishBoundChannels() -> void
{
for (const auto& chan : fBindingChans) {
string joined = boost::algorithm::join(chan.second, ",");
LOG(debug) << "Publishing bound addresses (" << chan.second.size() << ") of channel '" << chan.first << "' to DDS under '" << "fmqchan_" + chan.first << "' property name.";
fDDS.PutValue("fmqchan_" + chan.first, joined);
}
}
auto DDS::SubscribeForCustomCommands() -> void
{
LOG(debug) << "Subscribing for DDS custom commands.";
string id = GetProperty<string>("id");
fDDS.SubscribeCustomCmd([id, this](const string& cmdStr, const string& cond, uint64_t senderId) {
// LOG(info) << "Received command: '" << cmdStr << "' from " << senderId;
sdk::cmd::Cmds inCmds;
inCmds.Deserialize(cmdStr);
for (const auto& cmd : inCmds) {
HandleCmd(id, *cmd, cond, senderId);
}
});
}
auto DDS::HandleCmd(const string& id, sdk::cmd::Cmd& cmd, const string& cond, uint64_t senderId) -> void
{
using namespace fair::mq::sdk;
using namespace fair::mq::sdk::cmd;
// LOG(info) << "Received command type: '" << cmd.GetType() << "' from " << senderId;
switch (cmd.GetType()) {
case Type::check_state: {
fDDS.Send(Cmds(make<CurrentState>(id, GetCurrentDeviceState())).Serialize(), to_string(senderId));
} break;
case Type::change_state: {
Transition transition = static_cast<ChangeState&>(cmd).GetTransition();
if (ChangeDeviceState(transition)) {
Cmds outCmds(make<TransitionStatus>(id, fDDSTaskId, Result::Ok, transition, GetCurrentDeviceState()));
fDDS.Send(outCmds.Serialize(), to_string(senderId));
} else {
Cmds outCmds(make<TransitionStatus>(id, fDDSTaskId, Result::Failure, transition, GetCurrentDeviceState()));
fDDS.Send(outCmds.Serialize(), to_string(senderId));
}
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fLastExternalController = senderId;
}
} break;
case Type::dump_config: {
stringstream ss;
for (const auto& pKey : GetPropertyKeys()) {
ss << id << ": " << pKey << " -> " << GetPropertyAsString(pKey) << "\n";
}
Cmds outCmds(make<Config>(id, ss.str()));
fDDS.Send(outCmds.Serialize(), to_string(senderId));
} break;
case Type::state_change_exiting_received: {
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
if (fLastExternalController == senderId) {
fExitingAckedByLastExternalController = true;
}
}
fExitingAcked.notify_one();
} break;
case Type::subscribe_to_state_change: {
auto _cmd = static_cast<cmd::SubscribeToStateChange&>(cmd);
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fStateChangeSubscribers.emplace(senderId, make_pair(chrono::steady_clock::now(), _cmd.GetInterval()));
LOG(debug) << "Publishing state-change: " << fLastState << "->" << fCurrentState << " to " << senderId;
Cmds outCmds(make<StateChangeSubscription>(id, fDDSTaskId, Result::Ok),
make<StateChange>(id, fDDSTaskId, fLastState, fCurrentState));
fDDS.Send(outCmds.Serialize(), to_string(senderId));
} break;
case Type::subscription_heartbeat: {
try {
auto _cmd = static_cast<cmd::SubscriptionHeartbeat&>(cmd);
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fStateChangeSubscribers.at(senderId) = make_pair(chrono::steady_clock::now(), _cmd.GetInterval());
} catch(out_of_range& oor) {
LOG(warn) << "Received subscription heartbeat from an unknown controller with id '" << senderId << "'";
}
} break;
case Type::unsubscribe_from_state_change: {
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fStateChangeSubscribers.erase(senderId);
}
Cmds outCmds(make<StateChangeUnsubscription>(id, fDDSTaskId, Result::Ok));
fDDS.Send(outCmds.Serialize(), to_string(senderId));
} break;
case Type::get_properties: {
auto _cmd = static_cast<cmd::GetProperties&>(cmd);
auto const request_id(_cmd.GetRequestId());
auto result(Result::Ok);
vector<pair<string, string>> props;
try {
for (auto const& prop : GetPropertiesAsString(_cmd.GetQuery())) {
props.push_back({prop.first, prop.second});
}
} catch (exception const& e) {
LOG(warn) << "Getting properties (request id: " << request_id << ") failed: " << e.what();
result = Result::Failure;
}
Cmds const outCmds(make<cmd::Properties>(id, request_id, result, props));
fDDS.Send(outCmds.Serialize(), to_string(senderId));
} break;
case Type::set_properties: {
auto _cmd(static_cast<cmd::SetProperties&>(cmd));
auto const request_id(_cmd.GetRequestId());
auto result(Result::Ok);
try {
fair::mq::Properties props;
for (auto const& prop : _cmd.GetProps()) {
props.insert({prop.first, fair::mq::Property(prop.second)});
}
// TODO Handle builtin keys with different value type than string
SetProperties(props);
} catch (exception const& e) {
LOG(warn) << "Setting properties (request id: " << request_id << ") failed: " << e.what();
result = Result::Failure;
}
Cmds const outCmds(make<PropertiesSet>(id, request_id, result));
fDDS.Send(outCmds.Serialize(), to_string(senderId));
} break;
default:
LOG(warn) << "Unexpected/unknown command received: " << cmd.GetType();
LOG(warn) << "Origin: " << senderId;
LOG(warn) << "Destination: " << cond;
break;
}
}
DDS::~DDS()
{
UnsubscribeFromDeviceStateChange();
ReleaseDeviceControl();
if (fControllerThread.joinable()) {
fControllerThread.join();
}
fWorkGuard.reset();
if (fWorkerThread.joinable()) {
fWorkerThread.join();
}
}
} // namespace fair::mq::plugins

199
fairmq/plugins/DDS/DDS.h
View File

@@ -1,199 +0,0 @@
/********************************************************************************
* Copyright (C) 2017-2021 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" *
********************************************************************************/
#ifndef FAIR_MQ_PLUGINS_DDS
#define FAIR_MQ_PLUGINS_DDS
#include <fairmq/Plugin.h>
#include <fairmq/StateQueue.h>
#include <fairmq/Version.h>
#include <fairmq/sdk/commands/Commands.h>
#define BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <dds/dds.h>
#undef BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <asio/executor.hpp>
#include <asio/executor_work_guard.hpp>
#include <asio/io_context.hpp>
#include <cassert>
#include <chrono>
#include <condition_variable>
#include <mutex>
#include <string>
#include <atomic>
#include <thread>
#include <map>
#include <unordered_map>
#include <utility> // pair
#include <vector>
namespace fair::mq::plugins
{
struct DDSConfig
{
// container of sub channel addresses
unsigned int fNumSubChannels;
// dds values for the channel
std::map<uint64_t, std::string> fDDSValues;
};
struct DDSSubscription
{
DDSSubscription()
: fDDSCustomCmd(fService)
, fDDSKeyValue(fService)
{
LOG(debug) << "$DDS_TASK_PATH: " << dds::env_prop<dds::task_path>();
LOG(debug) << "$DDS_GROUP_NAME: " << dds::env_prop<dds::group_name>();
LOG(debug) << "$DDS_COLLECTION_NAME: " << dds::env_prop<dds::collection_name>();
LOG(debug) << "$DDS_TASK_NAME: " << dds::env_prop<dds::task_name>();
LOG(debug) << "$DDS_TASK_INDEX: " << dds::env_prop<dds::task_index>();
LOG(debug) << "$DDS_COLLECTION_INDEX: " << dds::env_prop<dds::collection_index>();
LOG(debug) << "$DDS_TASK_ID: " << dds::env_prop<dds::task_id>();
LOG(debug) << "$DDS_LOCATION: " << dds::env_prop<dds::dds_location>();
std::string dds_session_id(dds::env_prop<dds::dds_session_id>());
LOG(debug) << "$DDS_SESSION_ID: " << dds_session_id;
// subscribe for DDS service errors.
fService.subscribeOnError([](const dds::intercom_api::EErrorCode errorCode, const std::string& errorMsg) {
LOG(error) << "DDS Error received: error code: " << errorCode << ", error message: " << errorMsg;
});
// fDDSCustomCmd.subscribe([](const std::string& cmd, const std::string& cond, uint64_t senderId) {
// LOG(debug) << "cmd: " << cmd << ", cond: " << cond << ", senderId: " << senderId;
// });
assert(!dds_session_id.empty());
}
auto Start() -> void {
fService.start(dds::env_prop<dds::dds_session_id>());
}
~DDSSubscription() {
fDDSKeyValue.unsubscribe();
fDDSCustomCmd.unsubscribe();
}
template<typename... Args>
auto SubscribeCustomCmd(Args&&... args) -> void
{
fDDSCustomCmd.subscribe(std::forward<Args>(args)...);
}
template<typename... Args>
auto SubscribeKeyValue(Args&&... args) -> void
{
fDDSKeyValue.subscribe(std::forward<Args>(args)...);
}
template<typename... Args>
auto Send(Args&&... args) -> void
{
fDDSCustomCmd.send(std::forward<Args>(args)...);
}
template<typename... Args>
auto PutValue(Args&&... args) -> void
{
fDDSKeyValue.putValue(std::forward<Args>(args)...);
}
private:
dds::intercom_api::CIntercomService fService;
dds::intercom_api::CCustomCmd fDDSCustomCmd;
dds::intercom_api::CKeyValue fDDSKeyValue;
};
struct IofN
{
IofN(int i, int n)
: fI(i)
, fN(n)
{}
unsigned int fI;
unsigned int fN;
std::vector<std::string> fEntries;
};
class DDS : public Plugin
{
public:
DDS(const std::string& name, const Plugin::Version version, const std::string& maintainer, const std::string& homepage, PluginServices* pluginServices);
~DDS();
private:
auto WaitForExitingAck() -> void;
auto StartWorkerThread() -> void;
auto FillChannelContainers() -> void;
auto EmptyChannelContainers() -> void;
auto SubscribeForConnectingChannels() -> void;
auto PublishBoundChannels() -> void;
auto SubscribeForCustomCommands() -> void;
auto HandleCmd(const std::string& id, sdk::cmd::Cmd& cmd, const std::string& cond, uint64_t senderId) -> void;
DDSSubscription fDDS;
size_t fDDSTaskId;
std::unordered_map<std::string, std::vector<std::string>> fBindingChans;
std::unordered_map<std::string, DDSConfig> fConnectingChans;
std::unordered_map<std::string, int> fI;
std::unordered_map<std::string, IofN> fIofN;
std::thread fControllerThread;
DeviceState fCurrentState, fLastState;
std::atomic<bool> fDeviceTerminationRequested;
std::unordered_map<uint64_t, std::pair<std::chrono::steady_clock::time_point, int64_t>> fStateChangeSubscribers;
uint64_t fLastExternalController;
bool fExitingAckedByLastExternalController;
std::condition_variable fExitingAcked;
std::mutex fStateChangeSubscriberMutex;
bool fUpdatesAllowed;
std::mutex fUpdateMutex;
std::condition_variable fUpdateCondition;
std::thread fWorkerThread;
asio::io_context fWorkerQueue;
asio::executor_work_guard<asio::executor> fWorkGuard;
};
Plugin::ProgOptions DDSProgramOptions()
{
boost::program_options::options_description options{"DDS Plugin"};
options.add_options()
("dds-i", boost::program_options::value<std::vector<std::string>>()->multitoken()->composing(), "Task index for chosing connection target (single channel n to m). When all values come via same update.")
("dds-i-n", boost::program_options::value<std::vector<std::string>>()->multitoken()->composing(), "Task index for chosing connection target (one out of n values to take). When values come as independent updates.")
("wait-for-exiting-ack-timeout", boost::program_options::value<unsigned int>()->default_value(1000), "Wait timeout for EXITING state-change acknowledgement by external controller in milliseconds.");
return options;
}
REGISTER_FAIRMQ_PLUGIN(
DDS, // Class name
dds, // Plugin name (string, lower case chars only)
(Plugin::Version{FAIRMQ_VERSION_MAJOR,
FAIRMQ_VERSION_MINOR,
FAIRMQ_VERSION_PATCH}), // Version
"FairRootGroup <fairroot@gsi.de>", // Maintainer
"https://github.com/FairRootGroup/FairMQ", // Homepage
DDSProgramOptions // custom program options for the plugin
)
} // namespace fair::mq::plugins
#endif /* FAIR_MQ_PLUGINS_DDS */

6
fairmq/plugins/PMIx/CMakeLists.txt
View File

@@ -1,5 +1,5 @@
################################################################################
# Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
# Copyright (C) 2019-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
# #
# This software is distributed under the terms of the #
# GNU Lesser General Public Licence (LGPL) version 3, #
@@ -14,7 +14,7 @@ add_library(${plugin} SHARED
${CMAKE_CURRENT_SOURCE_DIR}/PMIx.hpp
)
target_compile_features(${plugin} PUBLIC cxx_std_17)
target_link_libraries(${plugin} PUBLIC FairMQ PMIx::libpmix PRIVATE Commands)
target_link_libraries(${plugin} PUBLIC FairMQ PMIx::libpmix)
target_include_directories(${plugin} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
set_target_properties(${plugin} PROPERTIES
CXX_VISIBILITY_PRESET hidden
@@ -24,7 +24,7 @@ set_target_properties(${plugin} PROPERTIES
set(exe fairmq-pmix-command-ui)
add_executable(${exe} ${CMAKE_CURRENT_SOURCE_DIR}/runPMIxCommandUI.cxx)
target_link_libraries(${exe} FairMQ Commands StateMachine PMIx::libpmix)
target_link_libraries(${exe} FairMQ StateMachine PMIx::libpmix)
target_include_directories(${exe} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
install(TARGETS ${plugin} ${exe}

108
fairmq/plugins/PMIx/PMIxPlugin.cxx
View File

@@ -1,5 +1,5 @@
/********************************************************************************
* Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* Copyright (C) 2019-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
@@ -8,7 +8,6 @@
#include "PMIxPlugin.h"
#include <fairmq/sdk/commands/Commands.h>
#include <fairmq/tools/Strings.h>
#include <sstream>
@@ -16,7 +15,6 @@
#include <cstdint> // UINT32_MAX
using namespace std;
using namespace fair::mq::sdk::cmd;
namespace fair::mq::plugins
{
@@ -31,8 +29,7 @@ PMIxPlugin::PMIxPlugin(const string& name,
, fPid(getpid())
, fPMIxClient(tools::ToString("PMIx client(pid=", fPid, ") "))
, fDeviceId(string(fProcess.nspace) + "_" + to_string(fProcess.rank))
, fCommands(fProcess)
, fLastExternalController(UINT32_MAX)
// , fLastExternalController(UINT32_MAX)
, fExitingAckedByLastExternalController(false)
, fCurrentState(DeviceState::Idle)
, fLastState(DeviceState::Idle)
@@ -42,12 +39,6 @@ PMIxPlugin::PMIxPlugin(const string& name,
SetProperty<string>("id", fDeviceId);
Fence("pmix::init");
SubscribeForCommands();
Fence("subscribed");
// fCommands.Send("test1");
// fCommands.Send("test2", 0);
// fCommands.Send("test3", 0);
// LOG(info) << "PMIX_EXTERNAL_ERR_BASE: " << PMIX_EXTERNAL_ERR_BASE;
@@ -101,11 +92,9 @@ PMIxPlugin::PMIxPlugin(const string& name,
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fLastState = fCurrentState;
fCurrentState = newState;
for (auto subscriberId : fStateChangeSubscribers) {
LOG(debug) << "Publishing state-change: " << fLastState << "->" << newState << " to " << subscriberId;
Cmds cmds(make<StateChange>(fDeviceId, 0, fLastState, fCurrentState));
fCommands.Send(cmds.Serialize(Format::JSON), static_cast<pmix::rank>(subscriberId));
}
// for (auto subscriberId : fStateChangeSubscribers) {
// LOG(debug) << "Publishing state-change: " << fLastState << "->" << newState << " to " << subscriberId;
// }
});
}
@@ -113,7 +102,6 @@ PMIxPlugin::~PMIxPlugin()
{
LOG(debug) << "Destroying PMIxPlugin";
ReleaseDeviceControl();
fCommands.Unsubscribe();
while (pmix::initialized()) {
try {
pmix::finalize();
@@ -124,92 +112,6 @@ PMIxPlugin::~PMIxPlugin()
}
}
auto PMIxPlugin::SubscribeForCommands() -> void
{
fCommands.Subscribe([this](const string& cmdStr, const pmix::proc& sender) {
// LOG(info) << "PMIx Plugin received message: '" << cmdStr << "', from " << sender;
Cmds inCmds;
inCmds.Deserialize(cmdStr, Format::JSON);
for (const auto& cmd : inCmds) {
LOG(info) << "Received command type: '" << cmd->GetType() << "' from " << sender;
switch (cmd->GetType()) {
case Type::check_state:
fCommands.Send(Cmds(make<CurrentState>(fDeviceId, GetCurrentDeviceState()))
.Serialize(Format::JSON),
{sender});
break;
case Type::change_state: {
Transition transition = static_cast<ChangeState&>(*cmd).GetTransition();
if (ChangeDeviceState(transition)) {
fCommands.Send(
Cmds(make<TransitionStatus>(fDeviceId, 0, Result::Ok, transition, GetCurrentDeviceState()))
.Serialize(Format::JSON),
{sender});
} else {
fCommands.Send(
Cmds(make<TransitionStatus>(fDeviceId, 0, Result::Failure, transition, GetCurrentDeviceState()))
.Serialize(Format::JSON),
{sender});
}
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fLastExternalController = sender.rank;
}
}
break;
case Type::subscribe_to_state_change: {
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fStateChangeSubscribers.insert(sender.rank);
}
LOG(debug) << "Publishing state-change: " << fLastState << "->" << fCurrentState
<< " to " << sender;
Cmds outCmds(make<StateChangeSubscription>(fDeviceId, fProcess.rank, Result::Ok),
make<StateChange>(fDeviceId, 0, fLastState, fCurrentState));
fCommands.Send(outCmds.Serialize(Format::JSON), {sender});
}
break;
case Type::unsubscribe_from_state_change: {
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fStateChangeSubscribers.erase(sender.rank);
}
fCommands.Send(Cmds(make<StateChangeUnsubscription>(fDeviceId, fProcess.rank, Result::Ok))
.Serialize(Format::JSON),
{sender});
}
break;
case Type::state_change_exiting_received: {
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
if (fLastExternalController == sender.rank) {
fExitingAckedByLastExternalController = true;
}
}
fExitingAcked.notify_one();
}
break;
case Type::dump_config: {
stringstream ss;
for (const auto& k: GetPropertyKeys()) {
ss << fDeviceId << ": " << k << " -> " << GetPropertyAsString(k) << "\n";
}
fCommands.Send(Cmds(make<Config>(fDeviceId, ss.str())).Serialize(Format::JSON),
{sender});
}
break;
default:
LOG(warn) << "Unexpected/unknown command received: " << cmdStr;
LOG(warn) << "Origin: " << sender;
break;
}
}
});
}
auto PMIxPlugin::Init() -> pmix::proc
{
if (!pmix::initialized()) {

6
fairmq/plugins/PMIx/PMIxPlugin.h
View File

@@ -1,5 +1,5 @@
/********************************************************************************
* Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* Copyright (C) 2019-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
@@ -44,10 +44,9 @@ class PMIxPlugin : public Plugin
pid_t fPid;
std::string fPMIxClient;
std::string fDeviceId;
pmix::Commands fCommands;
std::set<uint32_t> fStateChangeSubscribers;
uint32_t fLastExternalController;
// uint32_t fLastExternalController;
bool fExitingAckedByLastExternalController;
std::condition_variable fExitingAcked;
std::mutex fStateChangeSubscriberMutex;
@@ -61,7 +60,6 @@ class PMIxPlugin : public Plugin
auto Fence(const std::string& label) -> void;
auto Lookup() -> void;
auto SubscribeForCommands() -> void;
auto WaitForExitingAck() -> void;
};

89
fairmq/plugins/PMIx/runPMIxCommandUI.cxx
View File

@@ -1,12 +1,11 @@
/********************************************************************************
* Copyright (C) 2014-2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* Copyright (C) 2014-2022 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 <fairlogger/Logger.h>
@@ -29,7 +28,6 @@
#include <vector>
using namespace std;
using namespace fair::mq::sdk::cmd;
namespace bpo = boost::program_options;
const std::map<fair::mq::Transition, fair::mq::State> expected =
@@ -46,23 +44,6 @@ const std::map<fair::mq::Transition, fair::mq::State> expected =
{ fair::mq::Transition::End, fair::mq::State::Exiting }
};
struct StateSubscription
{
pmix::Commands& fCommands;
explicit StateSubscription(pmix::Commands& commands)
: fCommands(commands)
{
fCommands.Send(Cmds(make<SubscribeToStateChange>(600000)).Serialize(Format::JSON));
}
~StateSubscription()
{
fCommands.Send(Cmds(make<UnsubscribeFromStateChange>()).Serialize(Format::JSON));
this_thread::sleep_for(chrono::milliseconds(100)); // give PMIx a chance to complete request
}
};
struct MiniTopo
{
explicit MiniTopo(unsigned int n)
@@ -141,74 +122,6 @@ int main(int argc, char* argv[])
LOG(warn) << "pmix::fence() [pmix::init] OK";
MiniTopo topo(numDevices);
pmix::Commands commands(process);
commands.Subscribe([&](const string& msg, const pmix::proc& sender) {
// LOG(info) << "Received '" << msg << "' from " << sender;
Cmds cmds;
cmds.Deserialize(msg, Format::JSON);
// 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;
topo.Update(sender.rank, static_cast<StateChange&>(*cmd).GetCurrentState());
if (static_cast<StateChange&>(*cmd).GetCurrentState() == fair::mq::State::Exiting) {
commands.Send(Cmds(make<StateChangeExitingReceived>()).Serialize(Format::JSON), {sender});
}
}
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;
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: " << sender << endl;
break;
}
}
});
pmix::fence({all});
LOG(warn) << "pmix::fence() [subscribed] OK";
StateSubscription stateSubscription(commands);
for (auto transition : { fair::mq::Transition::InitDevice,
fair::mq::Transition::CompleteInit,
fair::mq::Transition::Bind,
fair::mq::Transition::Connect,
fair::mq::Transition::InitTask,
fair::mq::Transition::Run,
fair::mq::Transition::Stop,
fair::mq::Transition::ResetTask,
fair::mq::Transition::ResetDevice,
fair::mq::Transition::End }) {
commands.Send(Cmds(make<ChangeState>(transition)).Serialize(Format::JSON));
topo.WaitFor(expected.at(transition));
}
} catch (exception& e) {
LOG(error) << "Error: " << e.what();
return EXIT_FAILURE;

223
fairmq/sdk/AsioAsyncOp.h
View File

@@ -1,223 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_ASIOASYNCOP_H
#define FAIR_MQ_SDK_ASIOASYNCOP_H
#include <asio/associated_allocator.hpp>
#include <asio/associated_executor.hpp>
#include <asio/executor_work_guard.hpp>
#include <asio/dispatch.hpp>
#include <asio/system_executor.hpp>
#include <chrono>
#include <exception>
#include <fairmq/sdk/Error.h>
#include <fairmq/sdk/Traits.h>
#include <functional>
#include <memory>
#include <system_error>
#include <type_traits>
#include <utility>
#include <fairlogger/Logger.h>
#ifndef FAIR_LOG
#define FAIR_LOG LOG
#endif /* ifndef FAIR_LOG */
namespace fair::mq::sdk
{
template<typename... SignatureArgTypes>
struct AsioAsyncOpImplBase
{
virtual auto Complete(std::error_code, SignatureArgTypes...) -> void = 0;
virtual auto IsCompleted() const -> bool = 0;
};
/**
* @tparam Executor1 Associated I/O executor, see https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.associated_i_o_executor
* @tparam Allocator1 Default allocation strategy, see https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.allocation_of_intermediate_storage
*/
template<typename Executor1, typename Allocator1, typename Handler, typename... SignatureArgTypes>
struct AsioAsyncOpImpl : AsioAsyncOpImplBase<SignatureArgTypes...>
{
/// See https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.allocation_of_intermediate_storage
using Allocator2 = typename asio::associated_allocator<Handler, Allocator1>::type;
/// See https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.associated_completion_handler_executor
using Executor2 = typename asio::associated_executor<Handler, Executor1>::type;
/// Ctor
AsioAsyncOpImpl(const Executor1& ex1, Allocator1 alloc1, Handler&& handler)
: fWork1(ex1)
, fWork2(asio::get_associated_executor(handler, ex1))
, fHandler(std::move(handler))
, fAlloc1(std::move(alloc1))
{}
auto GetAlloc2() const -> Allocator2 { return asio::get_associated_allocator(fHandler, fAlloc1); }
auto GetEx2() const -> Executor2 { return asio::get_associated_executor(fWork2); }
auto Complete(std::error_code ec, SignatureArgTypes... args) -> void override
{
if (IsCompleted()) {
throw RuntimeError("Async operation already completed");
}
asio::dispatch(GetEx2(),
[=, handler = std::move(fHandler)]() mutable {
try {
handler(ec, args...);
} catch (const std::exception& e) {
FAIR_LOG(error) << "Uncaught exception in AsioAsyncOp completion handler: " << e.what();
} catch (...) {
FAIR_LOG(error) << "Unknown uncaught exception in AsioAsyncOp completion handler.";
}
});
fWork1.reset();
fWork2.reset();
}
auto IsCompleted() const -> bool override
{
return !fWork1.owns_work() && !fWork2.owns_work();
}
private:
/// See https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.outstanding_work
asio::executor_work_guard<Executor1> fWork1;
asio::executor_work_guard<Executor2> fWork2;
Handler fHandler;
Allocator1 fAlloc1;
};
/**
* @class AsioAsyncOp AsioAsyncOp.h <fairmq/sdk/AsioAsyncOp.h>
* @tparam Executor Associated I/O executor, see https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.associated_i_o_executor
* @tparam Allocator Default allocation strategy, see https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.allocation_of_intermediate_storage
* @tparam CompletionSignature
* @brief Interface for Asio-compliant asynchronous operation, see https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html
*
* @par Thread Safety
* @e Distinct @e objects: Safe.@n
* @e Shared @e objects: Unsafe.
*
* primary template
*/
template<typename Executor, typename Allocator, typename CompletionSignature>
struct AsioAsyncOp
{
};
/**
* @tparam Executor See primary template
* @tparam Allocator See primary template
* @tparam SignatureReturnType Return type of CompletionSignature, see primary template
* @tparam SignatureFirstArgType Type of first argument of CompletionSignature, see primary template
* @tparam SignatureArgTypes Types of the rest of arguments of CompletionSignature
*
* partial specialization to deconstruct CompletionSignature
*/
template<typename Executor,
typename Allocator,
typename SignatureReturnType,
typename SignatureFirstArgType,
typename... SignatureArgTypes>
struct AsioAsyncOp<Executor,
Allocator,
SignatureReturnType(SignatureFirstArgType, SignatureArgTypes...)>
{
static_assert(std::is_void<SignatureReturnType>::value,
"return value of CompletionSignature must be void");
static_assert(std::is_same<SignatureFirstArgType, std::error_code>::value,
"first argument of CompletionSignature must be std::error_code");
using Duration = std::chrono::milliseconds;
private:
using Impl = AsioAsyncOpImplBase<SignatureArgTypes...>;
using ImplPtr = std::unique_ptr<Impl, std::function<void(Impl*)>>;
ImplPtr fImpl;
public:
/// Default Ctor
AsioAsyncOp()
: fImpl(nullptr)
{}
/// Ctor with handler
template<typename Handler>
AsioAsyncOp(Executor ex1, Allocator alloc1, Handler&& handler)
: AsioAsyncOp()
{
// Async operation type to be allocated and constructed
using Op = AsioAsyncOpImpl<Executor, Allocator, Handler, SignatureArgTypes...>;
// Create allocator for concrete op type
// Allocator2, see https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.allocation_of_intermediate_storage
using OpAllocator =
typename std::allocator_traits<typename Op::Allocator2>::template rebind_alloc<Op>;
OpAllocator opAlloc;
// Allocate memory
auto mem(std::allocator_traits<OpAllocator>::allocate(opAlloc, 1));
// Construct object
auto ptr(new (mem) Op(std::move(ex1),
std::move(alloc1),
std::forward<Handler>(handler)));
// Assign ownership to this object
fImpl = ImplPtr(ptr, [opAlloc](Impl* p) mutable {
std::allocator_traits<OpAllocator>::deallocate(opAlloc, static_cast<Op*>(p), 1);
});
}
/// Ctor with handler #2
template<typename Handler>
AsioAsyncOp(Executor ex1, Handler&& handler)
: AsioAsyncOp(std::move(ex1), Allocator(), std::forward<Handler>(handler))
{}
/// Ctor with handler #3
template<typename Handler>
explicit AsioAsyncOp(Handler&& handler)
: AsioAsyncOp(asio::system_executor(), std::forward<Handler>(handler))
{}
auto IsCompleted() -> bool { return (fImpl == nullptr) || fImpl->IsCompleted(); }
auto Complete(std::error_code ec, SignatureArgTypes... args) -> void
{
if(IsCompleted()) {
throw RuntimeError("Async operation already completed");
}
fImpl->Complete(ec, args...);
fImpl.reset(nullptr);
}
auto Complete(SignatureArgTypes... args) -> void
{
Complete(std::error_code(), args...);
}
auto Cancel(SignatureArgTypes... args) -> void
{
Complete(MakeErrorCode(ErrorCode::OperationCanceled), args...);
}
auto Timeout(SignatureArgTypes... args) -> void
{
Complete(MakeErrorCode(ErrorCode::OperationTimeout), args...);
}
};
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_ASIOASYNCOP_H */

73
fairmq/sdk/AsioBase.h
View File

@@ -1,73 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_ASIOBASE_H
#define FAIR_MQ_SDK_ASIOBASE_H
#include <asio/any_io_executor.hpp>
#include <fairmq/sdk/Traits.h>
#include <memory>
#include <utility>
namespace fair::mq::sdk
{
using DefaultExecutor = asio::any_io_executor;
using DefaultAllocator = std::allocator<int>;
/**
* @class AsioBase AsioBase.h <fairmq/sdk/AsioBase.h>
* @tparam Executor Associated I/O executor
* @tparam Allocator Associated default allocator
* @brief Base for creating Asio-enabled I/O objects
*
* @par Thread Safety
* @e Distinct @e objects: Safe.@n
* @e Shared @e objects: Unsafe.
*/
template<typename Executor, typename Allocator>
class AsioBase
{
public:
/// Member type of associated I/O executor
using ExecutorType = Executor;
/// Get associated I/O executor
auto GetExecutor() const noexcept -> ExecutorType { return fExecutor; }
/// Member type of associated default allocator
using AllocatorType = Allocator;
/// Get associated default allocator
auto GetAllocator() const noexcept -> AllocatorType { return fAllocator; }
/// NO default ctor
AsioBase() = delete;
/// Construct with associated I/O executor
explicit AsioBase(Executor ex, Allocator alloc)
: fExecutor(std::move(ex))
, fAllocator(std::move(alloc))
{}
/// NOT copyable
AsioBase(const AsioBase&) = delete;
AsioBase& operator=(const AsioBase&) = delete;
/// movable
AsioBase(AsioBase&&) noexcept = default;
AsioBase& operator=(AsioBase&&) noexcept = default;
~AsioBase() = default;
private:
ExecutorType fExecutor;
AllocatorType fAllocator;
};
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_ASIOBASE_H */

117
fairmq/sdk/CMakeLists.txt
View File

@@ -1,117 +0,0 @@
################################################################################
# Copyright (C) 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" #
################################################################################
if(BUILD_TIDY_TOOL)
include(FairMQTidy)
endif()
#################
# libFairMQ_SDK #
#################
configure_file(DDSInfo.h.in ${CMAKE_CURRENT_BINARY_DIR}/DDSInfo.h @ONLY)
set(target SDK)
set(SDK_PUBLIC_HEADER_FILES
../SDK.h
AsioAsyncOp.h
AsioBase.h
DDSAgent.h
DDSCollection.h
DDSEnvironment.h
DDSSession.h
DDSTask.h
DDSTopology.h
Error.h
Topology.h
Traits.h
)
set(SDK_PRIVATE_HEADER_FILES
${CMAKE_CURRENT_BINARY_DIR}/DDSInfo.h
)
set(SDK_SOURCE_FILES
DDSEnvironment.cxx
DDSSession.cxx
DDSTopology.cxx
Topology.cxx
)
add_library(${target}
${SDK_SOURCE_FILES}
${SDK_PUBLIC_HEADER_FILES} # for IDE integration
${SDK_PRIVATE_HEADER_FILES} # for IDE integration
)
target_compile_features(${target} PUBLIC cxx_std_17)
set_target_properties(${target} PROPERTIES LABELS coverage)
target_include_directories(${target}
PUBLIC
$<BUILD_INTERFACE:${CMAKE_SOURCE_DIR}>
$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}>
$<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
)
target_link_libraries(${target}
PUBLIC
asio::asio
Boost::boost
Boost::filesystem
FairLogger::FairLogger
Threads::Threads
Tools
StateMachine
Commands
PRIVATE
DDS::dds_intercom_lib
DDS::dds_tools_lib
DDS::dds_topology_lib
)
set_target_properties(${target} PROPERTIES
VERSION ${PROJECT_VERSION}
SOVERSION "${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}"
OUTPUT_NAME FairMQ_${target}
)
if(BUILD_TIDY_TOOL AND RUN_FAIRMQ_TIDY)
fairmq_target_tidy(TARGET ${target})
endif()
###############
# executables #
###############
add_executable(fairmq-dds-command-ui ${CMAKE_CURRENT_SOURCE_DIR}/runDDSCommandUI.cxx)
target_link_libraries(fairmq-dds-command-ui
FairMQ
Commands
SDK
StateMachine
)
install(
TARGETS
SDK
fairmq-dds-command-ui
EXPORT ${PROJECT_EXPORT_SET}
RUNTIME DESTINATION ${PROJECT_INSTALL_BINDIR}
LIBRARY DESTINATION ${PROJECT_INSTALL_LIBDIR}
ARCHIVE DESTINATION ${PROJECT_INSTALL_LIBDIR}
)
# preserve relative path and prepend fairmq
foreach(HEADER IN LISTS SDK_PUBLIC_HEADER_FILES)
get_filename_component(_path ${HEADER} DIRECTORY)
file(TO_CMAKE_PATH ${PROJECT_INSTALL_INCDIR}/sdk/${_path} _destination)
install(FILES ${HEADER}
DESTINATION ${_destination}
)
endforeach()
install(FILES ${CMAKE_CURRENT_BINARY_DIR}/DDSInfo.h
DESTINATION ${PROJECT_INSTALL_INCDIR}/sdk
)

78
fairmq/sdk/DDSAgent.h
View File

@@ -1,78 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_DDSSAGENT_H
#define FAIR_MQ_SDK_DDSSAGENT_H
#include <fairmq/sdk/DDSSession.h>
#include <ostream>
#include <string>
#include <chrono>
#include <cstdint>
namespace fair::mq::sdk
{
/**
* @class DDSAgent <fairmq/sdk/DDSAgent.h>
* @brief Represents a DDS agent
*/
class DDSAgent
{
public:
using Id = uint64_t;
using Pid = uint32_t;
explicit DDSAgent(DDSSession session,
Id id,
Pid pid,
std::string path,
std::string host,
std::chrono::milliseconds startupTime,
std::string username)
: fSession(std::move(session))
, fId(id)
, fPid(pid)
, fDDSPath(std::move(path))
, fHost(std::move(host))
, fStartupTime(startupTime)
, fUsername(std::move(username))
{}
DDSSession GetSession() const { return fSession; }
Id GetId() const { return fId; }
Pid GetPid() const { return fPid; }
std::string GetHost() const { return fHost; }
std::string GetDDSPath() const { return fDDSPath; }
std::chrono::milliseconds GetStartupTime() const { return fStartupTime; }
std::string GetUsername() const { return fUsername; }
friend auto operator<<(std::ostream& os, const DDSAgent& agent) -> std::ostream&
{
return os << "DDSAgent id: " << agent.fId
<< ", pid: " << agent.fPid
<< ", path: " << agent.fDDSPath
<< ", host: " << agent.fHost
<< ", startupTime: " << agent.fStartupTime.count()
<< ", username: " << agent.fUsername;
}
private:
DDSSession fSession;
Id fId;
Pid fPid;
std::string fDDSPath;
std::string fHost;
std::chrono::milliseconds fStartupTime;
std::string fUsername;
};
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSSAGENT_H */

46
fairmq/sdk/DDSCollection.h
View File

@@ -1,46 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_DDSCOLLECTION_H
#define FAIR_MQ_SDK_DDSCOLLECTION_H
// #include <fairmq/sdk/DDSAgent.h>
#include <ostream>
#include <cstdint>
namespace fair::mq::sdk
{
/**
* @class DDSCollection <fairmq/sdk/DDSCollection.h>
* @brief Represents a DDS collection
*/
class DDSCollection
{
public:
using Id = std::uint64_t;
explicit DDSCollection(Id id)
: fId(id)
{}
Id GetId() const { return fId; }
friend auto operator<<(std::ostream& os, const DDSCollection& collection) -> std::ostream&
{
return os << "DDSCollection id: " << collection.fId;
}
private:
Id fId;
};
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSCOLLECTION_H */

86
fairmq/sdk/DDSEnvironment.cxx
View File

@@ -1,86 +0,0 @@
/********************************************************************************
* Copyright (C) 2019-2021 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 "DDSEnvironment.h"
#include <cstdlib>
#define BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <dds/dds.h>
#undef BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <fairlogger/Logger.h>
#include <fairmq/tools/InstanceLimit.h>
#include <fairmq/sdk/DDSInfo.h>
#include <sstream>
#include <utility>
namespace fair::mq::sdk
{
struct DDSEnvironment::Impl
{
explicit Impl(Path configHome)
: fLocation(DDSInstallPrefix)
, fConfigHome(std::move(configHome))
{
SetupPath();
SetupConfigHome();
}
auto SetupConfigHome() -> void
{
if (fConfigHome.empty()) {
fConfigHome = GetEnv("HOME");
} else {
setenv("HOME", fConfigHome.c_str(), 1);
}
}
auto SetupPath() -> void
{
std::string path(GetEnv("PATH"));
Path ddsExecDir = (fLocation == DDSInstallPrefix) ? DDSExecutableDir : fLocation / Path("bin");
path = ddsExecDir.string() + std::string(":") + path;
setenv("PATH", path.c_str(), 1);
}
auto GetEnv(const std::string& key) const -> std::string
{
auto value = std::getenv(key.c_str());
if (value) {
return {value};
}
return {};
}
struct Tag {};
friend auto operator<<(std::ostream& os, Tag) -> std::ostream& { return os << "DDSEnvironment"; }
tools::InstanceLimiter<Tag, 1> fCount;
Path fLocation;
Path fConfigHome;
};
DDSEnvironment::DDSEnvironment()
: DDSEnvironment(Path())
{}
DDSEnvironment::DDSEnvironment(Path configHome)
: fImpl(std::make_shared<Impl>(std::move(configHome)))
{}
auto DDSEnvironment::GetLocation() const -> Path { return fImpl->fLocation; }
auto DDSEnvironment::GetConfigHome() const -> Path { return fImpl->fConfigHome; }
auto operator<<(std::ostream& os, DDSEnvironment env) -> std::ostream&
{
return os << "$DDS_LOCATION: " << env.GetLocation() << ", "
<< "$DDS_CONFIG_HOME: " << env.GetConfigHome() / DDSEnvironment::Path(".DDS");
}
} // namespace fair::mq::sdk

44
fairmq/sdk/DDSEnvironment.h
View File

@@ -1,44 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_DDSENVIRONMENT_H
#define FAIR_MQ_SDK_DDSENVIRONMENT_H
#include <boost/filesystem.hpp>
#include <memory>
#include <ostream>
namespace fair::mq::sdk
{
/**
* @class DDSEnvironment DDSSession.h <fairmq/sdk/DDSSession.h>
* @brief Sets up the DDS environment (object helper)
*/
class DDSEnvironment
{
public:
using Path = boost::filesystem::path;
DDSEnvironment();
explicit DDSEnvironment(Path);
auto GetLocation() const -> Path;
auto GetConfigHome() const -> Path;
friend auto operator<<(std::ostream& os, DDSEnvironment env) -> std::ostream&;
private:
struct Impl;
std::shared_ptr<Impl> fImpl;
};
using DDSEnv = DDSEnvironment;
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSENVIRONMENT_H */

29
fairmq/sdk/DDSInfo.h.in
View File

@@ -1,29 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_DDSINFO_H
#define FAIR_MQ_SDK_DDSINFO_H
#include <string>
namespace dds::tools_api { class CSession; }
namespace dds::topology_api { class CTopology; }
namespace fair::mq::sdk
{
const std::string DDSVersion("@DDS_VERSION@");
const std::string DDSInstallPrefix("@DDS_INSTALL_PREFIX@");
const std::string DDSExecutableDir("@DDS_BINDIR@");
const std::string DDSIncludeDir("@DDS_INCDIR@");
const std::string DDSLibraryDir("@DDS_LIBDIR@");
const std::string DDSPluginDir("@DDS_PLUGINDIR@");
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSINFO_H */

389
fairmq/sdk/DDSSession.cxx
View File

@@ -1,389 +0,0 @@
/********************************************************************************
* Copyright (C) 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 "DDSSession.h"
#include <boost/process.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <cassert>
#include <cstdlib>
#define BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <dds/dds.h>
#undef BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <fairlogger/Logger.h>
#include <fairmq/sdk/DDSAgent.h>
#include <fairmq/sdk/DDSEnvironment.h>
#include <fairmq/sdk/DDSTopology.h>
#include <fairmq/tools/Semaphore.h>
#include <fairmq/tools/Strings.h>
#include <mutex>
#include <sstream>
#include <thread>
#include <unordered_map>
#include <utility>
#include <vector>
namespace fair::mq::sdk
{
auto operator<<(std::ostream& os, DDSRMSPlugin plugin) -> std::ostream&
{
switch (plugin) {
case DDSRMSPlugin::ssh:
return os << "ssh";
case DDSRMSPlugin::localhost:
return os << "localhost";
default:
__builtin_unreachable();
}
}
auto operator>>(std::istream& is, DDSRMSPlugin& plugin) -> std::istream&
{
std::string value;
if (is >> value) {
if (value == "ssh") {
plugin = DDSRMSPlugin::ssh;
} else if (value == "localhost") {
plugin = DDSRMSPlugin::localhost;
} else {
throw std::runtime_error("Unknown or unsupported DDSRMSPlugin");
}
}
return is;
}
struct DDSSession::Impl
{
explicit Impl(DDSEnvironment env)
: fEnv(std::move(env))
, fRMSPlugin(DDSRMSPlugin::localhost)
, fSession(std::make_shared<dds::tools_api::CSession>())
, fDDSCustomCmd(fDDSService)
, fId(to_string(fSession->create()))
, fStopOnDestruction(false)
{
fDDSService.subscribeOnError([](const dds::intercom_api::EErrorCode errorCode, const std::string& msg) {
std::cerr << "DDS error, error code: " << errorCode << ", error message: " << msg << std::endl;
});
}
explicit Impl(Id existing, DDSEnvironment env)
: fEnv(std::move(env))
, fRMSPlugin(DDSRMSPlugin::localhost)
, fSession(std::make_shared<dds::tools_api::CSession>())
, fDDSCustomCmd(fDDSService)
, fId(std::move(existing))
, fStopOnDestruction(false)
{
fSession->attach(fId);
fDDSService.subscribeOnError([](const dds::intercom_api::EErrorCode errorCode, const std::string& msg) {
std::cerr << "DDS error, error code: " << errorCode << ", error message: " << msg << std::endl;
});
}
explicit Impl(std::shared_ptr<dds::tools_api::CSession> nativeSession, DDSEnv env)
: fEnv(std::move(env))
, fRMSPlugin(DDSRMSPlugin::localhost)
, fSession(std::move(nativeSession))
, fDDSCustomCmd(fDDSService)
, fId(to_string(fSession->getSessionID()))
, fStopOnDestruction(false)
{
// Sanity check
if (!fSession->IsRunning()) {
throw std::runtime_error("Given CSession must be running");
}
}
~Impl()
{
if (fStopOnDestruction) {
fSession->shutdown();
}
}
Impl() = delete;
Impl(const Impl&) = delete;
Impl& operator=(const Impl&) = delete;
Impl(Impl&&) = delete;
Impl& operator=(Impl&&) = delete;
DDSEnvironment fEnv;
DDSRMSPlugin fRMSPlugin;
Path fRMSConfig;
std::shared_ptr<dds::tools_api::CSession> fSession;
dds::intercom_api::CIntercomService fDDSService;
dds::intercom_api::CCustomCmd fDDSCustomCmd;
Id fId;
bool fStopOnDestruction;
};
DDSSession::DDSSession(DDSEnvironment env)
: fImpl(std::make_shared<Impl>(std::move(env)))
{}
DDSSession::DDSSession(Id existing, DDSEnvironment env)
: fImpl(std::make_shared<Impl>(std::move(existing), std::move(env)))
{}
DDSSession::DDSSession(std::shared_ptr<dds::tools_api::CSession> nativeSession, DDSEnv env)
: fImpl(std::make_shared<Impl>(std::move(nativeSession), std::move(env)))
{}
auto DDSSession::GetEnv() const -> DDSEnvironment { return fImpl->fEnv; }
auto DDSSession::IsRunning() const -> bool { return fImpl->fSession->IsRunning(); }
auto DDSSession::GetId() const -> Id { return fImpl->fId; }
auto DDSSession::Stop() -> void { return fImpl->fSession->shutdown(); }
auto DDSSession::GetRMSPlugin() const -> DDSRMSPlugin { return fImpl->fRMSPlugin; }
auto DDSSession::SetRMSPlugin(DDSRMSPlugin plugin) -> void { fImpl->fRMSPlugin = plugin; }
auto DDSSession::GetRMSConfig() const -> Path { return fImpl->fRMSConfig; }
auto DDSSession::SetRMSConfig(Path configFile) const -> void
{
fImpl->fRMSConfig = std::move(configFile);
}
auto DDSSession::IsStoppedOnDestruction() const -> bool { return fImpl->fStopOnDestruction; }
auto DDSSession::StopOnDestruction(bool stop) -> void { fImpl->fStopOnDestruction = stop; }
auto DDSSession::SubmitAgents(Quantity agents) -> void
{
// Requesting to submit 0 agents is not meaningful
assert(agents > 0);
using namespace dds::tools_api;
SSubmitRequestData submitInfo;
submitInfo.m_rms = tools::ToString(GetRMSPlugin());
submitInfo.m_instances = 1;
submitInfo.m_slots = agents; // TODO new api: get slots from agents
submitInfo.m_config = GetRMSConfig().string();
tools::SharedSemaphore blocker;
auto submitRequest = SSubmitRequest::makeRequest(submitInfo);
submitRequest->setMessageCallback([](const SMessageResponseData& message){
LOG(debug) << message.m_msg;
});
submitRequest->setDoneCallback([agents, blocker]() mutable {
LOG(debug) << agents << " Agents submitted";
blocker.Signal();
});
fImpl->fSession->sendRequest<SSubmitRequest>(submitRequest);
blocker.Wait();
WaitForIdleAgents(agents);
}
auto DDSSession::RequestAgentCount() -> AgentCount
{
using namespace dds::tools_api;
SAgentCountRequest::response_t res;
fImpl->fSession->syncSendRequest<SAgentCountRequest>(SAgentCountRequest::request_t(), res);
AgentCount count;
count.active = res.m_activeSlotsCount;
count.idle = res.m_idleSlotsCount;
count.executing = res.m_executingSlotsCount;
return count;
}
auto DDSSession::RequestAgentInfo() -> std::vector<DDSAgent>
{
using namespace dds::tools_api;
SAgentInfoRequest::responseVector_t res;
fImpl->fSession->syncSendRequest<SAgentInfoRequest>(SAgentInfoRequest::request_t(), res);
std::vector<DDSAgent> agentInfo;
agentInfo.reserve(res.size());
for (const auto& a : res) {
agentInfo.emplace_back(
*this,
a.m_agentID,
a.m_agentPid,
a.m_DDSPath,
a.m_host,
a.m_startUpTime,
a.m_username
// a.m_nSlots
);
}
return agentInfo;
}
auto DDSSession::RequestTaskInfo() -> std::vector<DDSTask>
{
using namespace dds::tools_api;
SAgentInfoRequest::responseVector_t res;
fImpl->fSession->syncSendRequest<SAgentInfoRequest>(SAgentInfoRequest::request_t(), res);
std::vector<DDSTask> taskInfo;
taskInfo.reserve(res.size());
for (auto& a : res) {
//taskInfo.emplace_back(a.m_taskID, 0);
(void)a;
taskInfo.emplace_back(0, 0);
}
return taskInfo;
}
auto DDSSession::RequestCommanderInfo() -> CommanderInfo
{
using namespace dds::tools_api;
SCommanderInfoRequestData commanderInfo;
tools::SharedSemaphore blocker;
std::string error;
auto commanderInfoRequest = SCommanderInfoRequest::makeRequest(commanderInfo);
CommanderInfo info;
commanderInfoRequest->setResponseCallback([&info](const SCommanderInfoResponseData& _response) {
info.pid = _response.m_pid;
info.activeTopologyName = _response.m_activeTopologyName;
});
commanderInfoRequest->setMessageCallback([&](const SMessageResponseData& _message) {
if (_message.m_severity == dds::intercom_api::EMsgSeverity::error) {
error = _message.m_msg;
blocker.Signal();
} else {
LOG(debug) << _message.m_msg;
}
});
commanderInfoRequest->setDoneCallback([blocker]() mutable { blocker.Signal(); });
fImpl->fSession->sendRequest<SCommanderInfoRequest>(commanderInfoRequest);
blocker.Wait();
if (!error.empty()) {
throw std::runtime_error(error);
}
return info;
}
auto DDSSession::WaitForExecutingAgents(Quantity minCount) -> void
{
auto count(RequestAgentCount());
int interval(8);
while (count.executing < minCount) {
std::this_thread::sleep_for(std::chrono::milliseconds(interval));
interval = std::min(256, interval * 2);
count = RequestAgentCount();
}
}
auto DDSSession::WaitForIdleAgents(Quantity minCount) -> void
{
auto count(RequestAgentCount());
int interval(8);
while (count.idle < minCount) {
std::this_thread::sleep_for(std::chrono::milliseconds(interval));
interval = std::min(256, interval * 2);
count = RequestAgentCount();
}
}
auto DDSSession::ActivateTopology(DDSTopology topo) -> void
{
using namespace dds::tools_api;
STopologyRequestData topologyInfo;
topologyInfo.m_updateType = STopologyRequestData::EUpdateType::ACTIVATE;
topologyInfo.m_topologyFile = topo.GetTopoFile().string();
tools::SharedSemaphore blocker;
auto topologyRequest = STopologyRequest::makeRequest(topologyInfo);
topologyRequest->setMessageCallback([](const SMessageResponseData& _message) {
LOG(debug) << _message.m_msg;
});
topologyRequest->setDoneCallback([blocker]() mutable { blocker.Signal(); });
fImpl->fSession->sendRequest<STopologyRequest>(topologyRequest);
blocker.Wait();
WaitForExecutingAgents(topo.GetNumRequiredAgents());
}
auto DDSSession::ActivateTopology(const Path& topoFile) -> void
{
ActivateTopology(DDSTopology(topoFile, GetEnv()));
}
void DDSSession::StartDDSService() { fImpl->fDDSService.start(fImpl->fId); }
void DDSSession::SubscribeToCommands(std::function<void(const std::string& msg, const std::string& condition, uint64_t senderId)> cb)
{
fImpl->fDDSCustomCmd.subscribe(cb);
// fImpl->fDDSCustomCmd.subscribeOnReply([](const std::string& reply) {
// LOG(debug) << reply;
// });
}
void DDSSession::UnsubscribeFromCommands()
{
fImpl->fDDSCustomCmd.unsubscribe();
}
void DDSSession::SendCommand(const std::string& cmd, const std::string& path /* = "" */) { fImpl->fDDSCustomCmd.send(cmd, path); }
void DDSSession::SendCommand(const std::string& cmd, DDSChannel::Id recipient)
{
fImpl->fDDSCustomCmd.send(cmd, std::to_string(recipient));
}
auto operator<<(std::ostream& os, const DDSSession& session) -> std::ostream&
{
return os << "$DDS_SESSION_ID: " << session.GetId();
}
auto getMostRecentRunningDDSSession(DDSEnv env) -> DDSSession
{
boost::process::ipstream pipeStream;
boost::process::child c("dds-session list all", boost::process::std_out > pipeStream);
std::string lastLine;
std::string currentLine;
while (pipeStream && std::getline(pipeStream, currentLine) && !currentLine.empty()) {
lastLine = currentLine;
}
c.wait();
std::string sessionId;
if (!lastLine.empty()) {
std::vector<std::string> words;
std::istringstream iss(lastLine);
for (std::string s; iss >> s;) {
if (s != "*") {
words.push_back(s);
}
}
if (words.back() == "RUNNING") {
sessionId = words.front();
}
}
if (sessionId.empty()) {
throw std::runtime_error("could not find most recent DDS session");
}
return DDSSession(DDSSession::Id(sessionId), std::move(env));
}
} // namespace fair::mq::sdk

118
fairmq/sdk/DDSSession.h
View File

@@ -1,118 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_DDSSESSION_H
#define FAIR_MQ_SDK_DDSSESSION_H
#include <fairmq/sdk/DDSEnvironment.h>
#include <fairmq/sdk/DDSInfo.h>
#include <fairmq/sdk/DDSTask.h>
#include <boost/filesystem.hpp>
#include <cstdint>
#include <istream>
#include <memory>
#include <ostream>
#include <stdexcept>
#include <string>
#include <functional>
#include <vector>
namespace fair::mq::sdk
{
/**
* @enum DDSRMSPlugin DDSSession.h <fairmq/sdk/DDSSession.h>
* @brief Supported DDS resource management system plugins
*/
enum class DDSRMSPlugin
{
localhost,
ssh
};
auto operator<<(std::ostream& os, DDSRMSPlugin plugin) -> std::ostream&;
auto operator>>(std::istream& is, DDSRMSPlugin& plugin) -> std::istream&;
class DDSTopology;
class DDSAgent;
class DDSChannel
{
public:
using Id = std::uint64_t;
};
/**
* @class DDSSession DDSSession.h <fairmq/sdk/DDSSession.h>
* @brief Represents a DDS session
*/
class DDSSession
{
public:
using Id = std::string;
using Quantity = std::uint32_t;
using Path = boost::filesystem::path;
explicit DDSSession(DDSEnvironment env = DDSEnvironment());
explicit DDSSession(Id existing, DDSEnvironment env = DDSEnvironment());
/// @brief Construct with already existing native DDS API objects
/// @param nativeSession Existing and initialized CSession (either via create() or attach())
/// @param env Optional DDSEnv
explicit DDSSession(std::shared_ptr<dds::tools_api::CSession> nativeSession, DDSEnv env = {});
auto GetEnv() const -> DDSEnvironment;
auto GetId() const -> Id;
auto GetRMSPlugin() const -> DDSRMSPlugin;
auto SetRMSPlugin(DDSRMSPlugin) -> void;
auto GetRMSConfig() const -> Path;
auto SetRMSConfig(Path) const -> void;
auto IsStoppedOnDestruction() const -> bool;
auto StopOnDestruction(bool stop = true) -> void;
auto IsRunning() const -> bool;
auto SubmitAgents(Quantity agents) -> void;
struct AgentCount {
Quantity idle = 0;
Quantity active = 0;
Quantity executing = 0;
};
auto RequestAgentCount() -> AgentCount;
auto RequestAgentInfo() -> std::vector<DDSAgent>;
auto RequestTaskInfo() -> std::vector<DDSTask>;
struct CommanderInfo {
int pid = -1;
std::string activeTopologyName;
};
auto RequestCommanderInfo() -> CommanderInfo;
auto WaitForIdleAgents(Quantity) -> void;
auto WaitForOnlyIdleAgents() -> void;
auto WaitForExecutingAgents(Quantity) -> void;
auto ActivateTopology(const Path& topoFile) -> void;
auto ActivateTopology(DDSTopology) -> void;
auto Stop() -> void;
void StartDDSService();
void SubscribeToCommands(std::function<void(const std::string& msg, const std::string& condition, uint64_t senderId)>);
void UnsubscribeFromCommands();
void SendCommand(const std::string&, const std::string& = "");
void SendCommand(const std::string&, DDSChannel::Id);
auto GetTaskId(DDSChannel::Id) const -> DDSTask::Id;
friend auto operator<<(std::ostream& os, const DDSSession& session) -> std::ostream&;
private:
struct Impl;
std::shared_ptr<Impl> fImpl;
};
auto getMostRecentRunningDDSSession(DDSEnv env = {}) -> DDSSession;
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSSESSION_H */

49
fairmq/sdk/DDSTask.h
View File

@@ -1,49 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_DDSTASK_H
#define FAIR_MQ_SDK_DDSTASK_H
#include <fairmq/sdk/DDSCollection.h>
#include <ostream>
#include <cstdint>
namespace fair::mq::sdk
{
/**
* @class DDSTask <fairmq/sdk/DDSTask.h>
* @brief Represents a DDS task
*/
class DDSTask
{
public:
using Id = std::uint64_t;
explicit DDSTask(Id id, Id collectionId)
: fId(id)
, fCollectionId(collectionId)
{}
Id GetId() const { return fId; }
DDSCollection::Id GetCollectionId() const { return fCollectionId; }
friend auto operator<<(std::ostream& os, const DDSTask& task) -> std::ostream&
{
return os << "DDSTask id: " << task.fId << ", collection id: " << task.fCollectionId;
}
private:
Id fId;
DDSCollection::Id fCollectionId;
};
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSTASK_H */

116
fairmq/sdk/DDSTopology.cxx
View File

@@ -1,116 +0,0 @@
/********************************************************************************
* Copyright (C) 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 "DDSTopology.h"
#include <boost/range/iterator_range.hpp>
#define BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <dds/dds.h>
#undef BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <fairlogger/Logger.h>
#include <fairmq/sdk/DDSEnvironment.h>
#include <memory>
#include <sstream>
#include <stdexcept>
#include <utility>
namespace fair::mq::sdk
{
struct DDSTopology::Impl
{
explicit Impl(Path topoFile, DDSEnvironment env)
: fEnv(std::move(env))
, fTopoFile(std::move(topoFile))
, fTopo(fTopoFile.string())
{}
explicit Impl(dds::topology_api::CTopology nativeTopology, DDSEnvironment env)
: fEnv(std::move(env))
, fTopo(std::move(nativeTopology))
{}
DDSEnvironment fEnv;
Path fTopoFile;
dds::topology_api::CTopology fTopo;
};
DDSTopology::DDSTopology(Path topoFile, DDSEnvironment env)
: fImpl(std::make_shared<Impl>(std::move(topoFile), std::move(env)))
{}
DDSTopology::DDSTopology(dds::topology_api::CTopology nativeTopology, DDSEnv env)
: fImpl(std::make_shared<Impl>(std::move(nativeTopology), std::move(env)))
{}
auto DDSTopology::GetEnv() const -> DDSEnvironment { return fImpl->fEnv; }
auto DDSTopology::GetTopoFile() const -> Path
{
auto file(fImpl->fTopoFile);
if (file.string().empty()) {
throw std::runtime_error("DDS topology xml spec file unknown");
}
return file;
}
auto DDSTopology::GetNumRequiredAgents() const -> std::size_t
{
return fImpl->fTopo.getRequiredNofAgents();
}
auto DDSTopology::GetTasks(const std::string& path /* = "" */) const -> std::vector<DDSTask>
{
std::vector<DDSTask> list;
dds::topology_api::STopoRuntimeTask::FilterIteratorPair_t itPair;
if (path.empty()) {
itPair = fImpl->fTopo.getRuntimeTaskIterator(nullptr); // passing nullptr will get all tasks
} else {
itPair = fImpl->fTopo.getRuntimeTaskIteratorMatchingPath(path);
}
auto tasks = boost::make_iterator_range(itPair.first, itPair.second);
for (const auto& task : tasks) {
// LOG(debug) << "Found task with id: " << task.first << ", "
// << "Path: " << task.second.m_taskPath << ", "
// << "Collection id: " << task.second.m_taskCollectionId << ", "
// << "Name: " << task.second.m_task->getName() << "_" << task.second.m_taskIndex;
list.emplace_back(task.first, task.second.m_taskCollectionId);
}
return list;
}
auto DDSTopology::GetCollections() const -> std::vector<DDSCollection>
{
std::vector<DDSCollection> list;
auto itPair = fImpl->fTopo.getRuntimeCollectionIterator(nullptr); // passing nullptr will get all collections
auto collections = boost::make_iterator_range(itPair.first, itPair.second);
for (const auto& c : collections) {
LOG(debug) << "Found collection with id: " << c.first << ", "
<< "Index: " << c.second.m_collectionIndex << ", "
<< "Path: " << c.second.m_collectionPath;
list.emplace_back(c.first);
}
return list;
}
auto DDSTopology::GetName() const -> std::string { return fImpl->fTopo.getName(); }
auto operator<<(std::ostream& os, const DDSTopology& t) -> std::ostream&
try {
return os << "DDS topology: " << t.GetName() << " (loaded from " << t.GetTopoFile() << ")";
} catch (std::runtime_error&) {
return os << "DDS topology: " << t.GetName();
}
} // namespace fair::mq::sdk

75
fairmq/sdk/DDSTopology.h
View File

@@ -1,75 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_DDSTOPOLOGY_H
#define FAIR_MQ_SDK_DDSTOPOLOGY_H
#include <boost/filesystem.hpp>
#include <fairmq/sdk/DDSCollection.h>
#include <fairmq/sdk/DDSEnvironment.h>
#include <fairmq/sdk/DDSInfo.h>
#include <fairmq/sdk/DDSTask.h>
#include <memory>
#include <string>
#include <vector>
namespace fair::mq::sdk
{
/**
* @class DDSTopology DDSTopology.h <fairmq/sdk/DDSTopology.h>
* @brief Represents a DDS topology
*/
class DDSTopology
{
public:
using Path = boost::filesystem::path;
DDSTopology() = delete;
/// @brief Construct from file
/// @param topoFile DDS topology xml file
/// @param env DDS environment
explicit DDSTopology(Path topoFile, DDSEnvironment env = DDSEnvironment());
/// @brief Construct with already existing native DDS API objects
/// @param nativeTopology Existing and initialized CTopology
/// @param env Optional DDSEnv
explicit DDSTopology(dds::topology_api::CTopology nativeTopology, DDSEnv env = {});
/// @brief Get associated DDS environment
auto GetEnv() const -> DDSEnvironment;
/// @brief Get path to DDS topology xml, if it is known
/// @throw std::runtime_error
auto GetTopoFile() const -> Path;
/// @brief Get number of required agents for this topology
auto GetNumRequiredAgents() const -> std::size_t;
/// @brief Get list of tasks in this topology, optionally matching provided path
auto GetTasks(const std::string& = "") const -> std::vector<DDSTask>;
/// @brief Get list of tasks in this topology
auto GetCollections() const -> std::vector<DDSCollection>;
/// @brief Get the name of the topology
auto GetName() const -> std::string;
friend auto operator<<(std::ostream&, const DDSTopology&) -> std::ostream&;
private:
struct Impl;
std::shared_ptr<Impl> fImpl;
};
using DDSTopo = DDSTopology;
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSTOPOLOGY_H */

20
fairmq/sdk/Error.h
View File

@@ -1,20 +0,0 @@
/********************************************************************************
* Copyright (C) 2019-2021 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_ERROR_H
#define FAIR_MQ_SDK_ERROR_H
#include <fairmq/Error.h>
namespace fair::mq::sdk {
using RuntimeError = fair::mq::RuntimeError;
} /* namespace fair::mq::sdk */
#endif /* FAIR_MQ_SDK_ERROR_H */

31
fairmq/sdk/Topology.cxx
View File

@@ -1,31 +0,0 @@
/********************************************************************************
* Copyright (C) 2019-2021 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 "Topology.h"
#define BOOST_BIND_GLOBAL_PLACEHOLDERS
#include <dds/dds.h>
#undef BOOST_BIND_GLOBAL_PLACEHOLDERS
namespace fair::mq::sdk
{
/// @brief Helper to (Re)Construct a FairMQ topology based on already existing native DDS API objects
/// @param nativeSession Existing and initialized CSession (either via create() or attach())
/// @param nativeTopo Existing CTopology that is activated on the given nativeSession
/// @param env Optional DDSEnv (needed primarily for unit testing)
/// @param blockUntilConnected if true, ctor will wait for all tasks to confirm subscriptions
auto MakeTopology(dds::topology_api::CTopology nativeTopo,
std::shared_ptr<dds::tools_api::CSession> nativeSession,
DDSEnv env,
bool blockUntilConnected) -> Topology
{
return {DDSTopo(std::move(nativeTopo), env), DDSSession(std::move(nativeSession), env), blockUntilConnected};
}
} // namespace fair::mq::sdk

1367
fairmq/sdk/Topology.h
View File

File diff suppressed because it is too large Load Diff

48
fairmq/sdk/Traits.h
View File

@@ -1,48 +0,0 @@
/********************************************************************************
* Copyright (C) 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_TRAITS_H
#define FAIR_MQ_SDK_TRAITS_H
#include <asio/associated_allocator.hpp>
#include <asio/associated_executor.hpp>
#include <type_traits>
namespace asio::detail {
/// Specialize to match our coding conventions
template<typename T, typename Executor>
struct associated_executor_impl<T,
Executor,
std::enable_if_t<is_executor<typename T::ExecutorType>::value>>
{
using type = typename T::ExecutorType;
static auto get(const T& obj, const Executor& /*ex = Executor()*/) noexcept -> type
{
return obj.GetExecutor();
}
};
/// Specialize to match our coding conventions
template<typename T, typename Allocator>
struct associated_allocator_impl<T,
Allocator,
std::enable_if_t<T::AllocatorType>>
{
using type = typename T::AllocatorType;
static auto get(const T& obj, const Allocator& /*alloc = Allocator()*/) noexcept -> type
{
return obj.GetAllocator();
}
};
} /* namespace asio::detail */
#endif /* FAIR_MQ_SDK_TRAITS_H */

60
fairmq/sdk/commands/CMakeLists.txt
View File

@@ -1,60 +0,0 @@
################################################################################
# Copyright (C) 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" #
################################################################################
set(target Commands)
add_custom_command(
OUTPUT ${CMAKE_CURRENT_BINARY_DIR}/CommandsFormat.h
COMMAND $<TARGET_FILE:flatbuffers::flatc> -c -o ${CMAKE_CURRENT_BINARY_DIR} CommandsFormat.fbs
COMMAND ${CMAKE_COMMAND} -E rename ${CMAKE_CURRENT_BINARY_DIR}/CommandsFormat_generated.h ${CMAKE_CURRENT_BINARY_DIR}/CommandsFormat.h
WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
)
# JSON serialization needs to see the .fbs file at run time, save it as constexpr string instead of locating/opening it every time
file(STRINGS CommandsFormat.fbs tmp)
list(JOIN tmp "\n" commands_format_def_fbs)
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/CommandsFormatDef.h.in ${CMAKE_CURRENT_BINARY_DIR}/CommandsFormatDef.h)
add_library(${target} Commands.cxx Commands.h ${CMAKE_CURRENT_BINARY_DIR}/CommandsFormat.h ${CMAKE_CURRENT_BINARY_DIR}/CommandsFormatDef.h)
add_library(FairMQ::${target} ALIAS ${target})
set(_flatbuffers flatbuffers::flatbuffers_shared)
if(NOT TARGET flatbuffers::flatbuffers_shared AND TARGET flatbuffers::flatbuffers)
set(_flatbuffers flatbuffers::flatbuffers)
endif()
target_link_libraries(${target}
PUBLIC
StateMachine
Tools
PRIVATE
${_flatbuffers}
)
target_compile_features(${target} PUBLIC cxx_std_17)
set_target_properties(${target} PROPERTIES
VERSION ${PROJECT_VERSION}
SOVERSION "${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}"
OUTPUT_NAME FairMQ${target}
)
target_include_directories(${target}
PUBLIC
$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}>
$<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
)
install(
TARGETS ${target}
EXPORT ${PROJECT_EXPORT_SET}
DESTINATION ${PROJECT_INSTALL_LIBDIR}
)
install(FILES Commands.h
DESTINATION ${PROJECT_INSTALL_INCDIR}/sdk/commands
)

479
fairmq/sdk/commands/Commands.cxx
View File

@@ -1,479 +0,0 @@
/********************************************************************************
* Copyright (C) 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 "Commands.h"
#include <fairmq/sdk/commands/CommandsFormatDef.h>
#include <fairmq/sdk/commands/CommandsFormat.h>
#include <flatbuffers/idl.h>
#include <array>
using namespace std;
namespace fair::mq::sdk::cmd
{
array<Result, 2> fbResultToResult =
{
{
Result::Ok,
Result::Failure
}
};
array<FBResult, 2> resultToFBResult =
{
{
FBResult::FBResult_Ok,
FBResult::FBResult_Failure
}
};
array<string, 2> resultNames =
{
{
"Ok",
"Failure"
}
};
array<string, 17> typeNames =
{
{
"CheckState",
"ChangeState",
"DumpConfig",
"SubscribeToStateChange",
"UnsubscribeFromStateChange",
"StateChangeExitingReceived",
"GetProperties",
"SetProperties",
"SubscriptionHeartbeat",
"CurrentState",
"TransitionStatus",
"Config",
"StateChangeSubscription",
"StateChangeUnsubscription",
"StateChange",
"Properties",
"PropertiesSet"
}
};
array<fair::mq::State, 16> fbStateToMQState =
{
{
fair::mq::State::Undefined,
fair::mq::State::Ok,
fair::mq::State::Error,
fair::mq::State::Idle,
fair::mq::State::InitializingDevice,
fair::mq::State::Initialized,
fair::mq::State::Binding,
fair::mq::State::Bound,
fair::mq::State::Connecting,
fair::mq::State::DeviceReady,
fair::mq::State::InitializingTask,
fair::mq::State::Ready,
fair::mq::State::Running,
fair::mq::State::ResettingTask,
fair::mq::State::ResettingDevice,
fair::mq::State::Exiting
}
};
array<sdk::cmd::FBState, 16> mqStateToFBState =
{
{
sdk::cmd::FBState_Undefined,
sdk::cmd::FBState_Ok,
sdk::cmd::FBState_Error,
sdk::cmd::FBState_Idle,
sdk::cmd::FBState_InitializingDevice,
sdk::cmd::FBState_Initialized,
sdk::cmd::FBState_Binding,
sdk::cmd::FBState_Bound,
sdk::cmd::FBState_Connecting,
sdk::cmd::FBState_DeviceReady,
sdk::cmd::FBState_InitializingTask,
sdk::cmd::FBState_Ready,
sdk::cmd::FBState_Running,
sdk::cmd::FBState_ResettingTask,
sdk::cmd::FBState_ResettingDevice,
sdk::cmd::FBState_Exiting
}
};
array<fair::mq::Transition, 12> fbTransitionToMQTransition =
{
{
fair::mq::Transition::Auto,
fair::mq::Transition::InitDevice,
fair::mq::Transition::CompleteInit,
fair::mq::Transition::Bind,
fair::mq::Transition::Connect,
fair::mq::Transition::InitTask,
fair::mq::Transition::Run,
fair::mq::Transition::Stop,
fair::mq::Transition::ResetTask,
fair::mq::Transition::ResetDevice,
fair::mq::Transition::End,
fair::mq::Transition::ErrorFound
}
};
array<sdk::cmd::FBTransition, 12> mqTransitionToFBTransition =
{
{
sdk::cmd::FBTransition_Auto,
sdk::cmd::FBTransition_InitDevice,
sdk::cmd::FBTransition_CompleteInit,
sdk::cmd::FBTransition_Bind,
sdk::cmd::FBTransition_Connect,
sdk::cmd::FBTransition_InitTask,
sdk::cmd::FBTransition_Run,
sdk::cmd::FBTransition_Stop,
sdk::cmd::FBTransition_ResetTask,
sdk::cmd::FBTransition_ResetDevice,
sdk::cmd::FBTransition_End,
sdk::cmd::FBTransition_ErrorFound
}
};
array<FBCmd, 17> typeToFBCmd =
{
{
FBCmd::FBCmd_check_state,
FBCmd::FBCmd_change_state,
FBCmd::FBCmd_dump_config,
FBCmd::FBCmd_subscribe_to_state_change,
FBCmd::FBCmd_unsubscribe_from_state_change,
FBCmd::FBCmd_state_change_exiting_received,
FBCmd::FBCmd_get_properties,
FBCmd::FBCmd_set_properties,
FBCmd::FBCmd_subscription_heartbeat,
FBCmd::FBCmd_current_state,
FBCmd::FBCmd_transition_status,
FBCmd::FBCmd_config,
FBCmd::FBCmd_state_change_subscription,
FBCmd::FBCmd_state_change_unsubscription,
FBCmd::FBCmd_state_change,
FBCmd::FBCmd_properties,
FBCmd::FBCmd_properties_set
}
};
array<Type, 17> fbCmdToType =
{
{
Type::check_state,
Type::change_state,
Type::dump_config,
Type::subscribe_to_state_change,
Type::unsubscribe_from_state_change,
Type::state_change_exiting_received,
Type::get_properties,
Type::set_properties,
Type::subscription_heartbeat,
Type::current_state,
Type::transition_status,
Type::config,
Type::state_change_subscription,
Type::state_change_unsubscription,
Type::state_change,
Type::properties,
Type::properties_set
}
};
fair::mq::State GetMQState(const FBState state) { return fbStateToMQState.at(state); }
FBState GetFBState(const fair::mq::State state) { return mqStateToFBState.at(static_cast<int>(state)); }
fair::mq::Transition GetMQTransition(const FBTransition transition) { return fbTransitionToMQTransition.at(transition); }
FBTransition GetFBTransition(const fair::mq::Transition transition) { return mqTransitionToFBTransition.at(static_cast<int>(transition)); }
Result GetResult(const FBResult result) { return fbResultToResult.at(result); }
FBResult GetFBResult(const Result result) { return resultToFBResult.at(static_cast<int>(result)); }
string GetResultName(const Result result) { return resultNames.at(static_cast<int>(result)); }
string GetTypeName(const Type type) { return typeNames.at(static_cast<int>(type)); }
FBCmd GetFBCmd(const Type type) { return typeToFBCmd.at(static_cast<int>(type)); }
string Cmds::Serialize(const Format type) const
{
flatbuffers::FlatBufferBuilder fbb;
vector<flatbuffers::Offset<FBCommand>> commandOffsets;
for (auto& cmd : fCmds) {
flatbuffers::Offset<FBCommand> cmdOffset;
unique_ptr<FBCommandBuilder> cmdBuilder; // delay the creation of the builder, because child strings need to be constructed first (which are conditional)
switch (cmd->GetType()) {
case Type::check_state: {
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
}
break;
case Type::change_state: {
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_transition(GetFBTransition(static_cast<ChangeState&>(*cmd).GetTransition()));
}
break;
case Type::dump_config: {
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
}
break;
break;
case Type::subscribe_to_state_change: {
auto _cmd = static_cast<SubscribeToStateChange&>(*cmd);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_interval(_cmd.GetInterval());
}
break;
case Type::unsubscribe_from_state_change: {
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
}
break;
case Type::state_change_exiting_received: {
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
}
break;
case Type::get_properties: {
auto _cmd = static_cast<GetProperties&>(*cmd);
auto query = fbb.CreateString(_cmd.GetQuery());
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_request_id(_cmd.GetRequestId());
cmdBuilder->add_property_query(query);
}
break;
case Type::set_properties: {
auto _cmd = static_cast<SetProperties&>(*cmd);
std::vector<flatbuffers::Offset<FBProperty>> propsVector;
for (auto const& e : _cmd.GetProps()) {
auto const key(fbb.CreateString(e.first));
auto const val(fbb.CreateString(e.second));
propsVector.push_back(CreateFBProperty(fbb, key, val));
}
auto props = fbb.CreateVector(propsVector);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_request_id(_cmd.GetRequestId());
cmdBuilder->add_properties(props);
}
break;
case Type::subscription_heartbeat: {
auto _cmd = static_cast<SubscriptionHeartbeat&>(*cmd);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_interval(_cmd.GetInterval());
}
break;
case Type::current_state: {
auto _cmd = static_cast<CurrentState&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_current_state(GetFBState(_cmd.GetCurrentState()));
}
break;
case Type::transition_status: {
auto _cmd = static_cast<TransitionStatus&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
cmdBuilder->add_transition(GetFBTransition(_cmd.GetTransition()));
cmdBuilder->add_current_state(GetFBState(_cmd.GetCurrentState()));
}
break;
case Type::config: {
auto _cmd = static_cast<Config&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
auto config = fbb.CreateString(_cmd.GetConfig());
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_config_string(config);
}
break;
case Type::state_change_subscription: {
auto _cmd = static_cast<StateChangeSubscription&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
}
break;
case Type::state_change_unsubscription: {
auto _cmd = static_cast<StateChangeUnsubscription&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
}
break;
case Type::state_change: {
auto _cmd = static_cast<StateChange&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_last_state(GetFBState(_cmd.GetLastState()));
cmdBuilder->add_current_state(GetFBState(_cmd.GetCurrentState()));
}
break;
case Type::properties: {
auto _cmd = static_cast<Properties&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
std::vector<flatbuffers::Offset<FBProperty>> propsVector;
for (const auto& e : _cmd.GetProps()) {
auto key = fbb.CreateString(e.first);
auto val = fbb.CreateString(e.second);
auto prop = CreateFBProperty(fbb, key, val);
propsVector.push_back(prop);
}
auto props = fbb.CreateVector(propsVector);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_request_id(_cmd.GetRequestId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
cmdBuilder->add_properties(props);
}
break;
case Type::properties_set: {
auto _cmd = static_cast<PropertiesSet&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_request_id(_cmd.GetRequestId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
}
break;
default:
throw CommandFormatError("unrecognized command type given to fair::mq::cmd::Cmds::Serialize()");
break;
}
cmdBuilder->add_command_id(typeToFBCmd.at(static_cast<int>(cmd->GetType())));
cmdOffset = cmdBuilder->Finish();
commandOffsets.push_back(cmdOffset);
}
auto commands = fbb.CreateVector(commandOffsets);
auto cmds = CreateFBCommands(fbb, commands);
fbb.Finish(cmds);
if (type == Format::Binary) {
return string(reinterpret_cast<char*>(fbb.GetBufferPointer()), fbb.GetSize());
} else { // Type == Format::JSON
flatbuffers::Parser parser;
if (!parser.Parse(commandsFormatDefFbs)) {
throw CommandFormatError("Serialize couldn't parse commands format");
}
std::string json;
if (!flatbuffers::GenerateText(parser, fbb.GetBufferPointer(), &json)) {
throw CommandFormatError("Serialize couldn't serialize parsed data to JSON!");
}
return json;
}
}
void Cmds::Deserialize(const string& str, const Format type)
{
fCmds.clear();
const flatbuffers::Vector<flatbuffers::Offset<FBCommand>>* cmds = nullptr;
flatbuffers::Parser parser;
if (type == Format::Binary) {
cmds = cmd::GetFBCommands(const_cast<char*>(str.c_str()))->commands();
} else { // Type == Format::JSON
if (!parser.Parse(commandsFormatDefFbs)) {
throw CommandFormatError("Deserialize couldn't parse commands format");
}
if (!parser.Parse(str.c_str())) {
throw CommandFormatError("Deserialize couldn't parse incoming JSON string");
}
cmds = cmd::GetFBCommands(parser.builder_.GetBufferPointer())->commands();
}
for (unsigned int i = 0; i < cmds->size(); ++i) {
const fair::mq::sdk::cmd::FBCommand& cmdPtr = *(cmds->Get(i));
switch (cmdPtr.command_id()) {
case FBCmd_check_state:
fCmds.emplace_back(make<CheckState>());
break;
case FBCmd_change_state:
fCmds.emplace_back(make<ChangeState>(GetMQTransition(cmdPtr.transition())));
break;
case FBCmd_dump_config:
fCmds.emplace_back(make<DumpConfig>());
break;
case FBCmd_subscribe_to_state_change:
fCmds.emplace_back(make<SubscribeToStateChange>(cmdPtr.interval()));
break;
case FBCmd_unsubscribe_from_state_change:
fCmds.emplace_back(make<UnsubscribeFromStateChange>());
break;
case FBCmd_state_change_exiting_received:
fCmds.emplace_back(make<StateChangeExitingReceived>());
break;
case FBCmd_get_properties:
fCmds.emplace_back(make<GetProperties>(cmdPtr.request_id(), cmdPtr.property_query()->str()));
break;
case FBCmd_set_properties: {
std::vector<std::pair<std::string, std::string>> properties;
auto props = cmdPtr.properties();
for (unsigned int j = 0; j < props->size(); ++j) {
properties.emplace_back(props->Get(j)->key()->str(), props->Get(j)->value()->str());
}
fCmds.emplace_back(make<SetProperties>(cmdPtr.request_id(), properties));
} break;
case FBCmd_subscription_heartbeat:
fCmds.emplace_back(make<SubscriptionHeartbeat>(cmdPtr.interval()));
break;
case FBCmd_current_state:
fCmds.emplace_back(make<CurrentState>(cmdPtr.device_id()->str(), GetMQState(cmdPtr.current_state())));
break;
case FBCmd_transition_status:
fCmds.emplace_back(make<TransitionStatus>(cmdPtr.device_id()->str(), cmdPtr.task_id(), GetResult(cmdPtr.result()), GetMQTransition(cmdPtr.transition()), GetMQState(cmdPtr.current_state())));
break;
case FBCmd_config:
fCmds.emplace_back(make<Config>(cmdPtr.device_id()->str(), cmdPtr.config_string()->str()));
break;
case FBCmd_state_change_subscription:
fCmds.emplace_back(make<StateChangeSubscription>(cmdPtr.device_id()->str(), cmdPtr.task_id(), GetResult(cmdPtr.result())));
break;
case FBCmd_state_change_unsubscription:
fCmds.emplace_back(make<StateChangeUnsubscription>(cmdPtr.device_id()->str(), cmdPtr.task_id(), GetResult(cmdPtr.result())));
break;
case FBCmd_state_change:
fCmds.emplace_back(make<StateChange>(cmdPtr.device_id()->str(), cmdPtr.task_id(), GetMQState(cmdPtr.last_state()), GetMQState(cmdPtr.current_state())));
break;
case FBCmd_properties: {
std::vector<std::pair<std::string, std::string>> properties;
auto props = cmdPtr.properties();
for (unsigned int j = 0; j < props->size(); ++j) {
properties.emplace_back(props->Get(j)->key()->str(), props->Get(j)->value()->str());
}
fCmds.emplace_back(make<Properties>(cmdPtr.device_id()->str(), cmdPtr.request_id(), GetResult(cmdPtr.result()), properties));
} break;
case FBCmd_properties_set:
fCmds.emplace_back(make<PropertiesSet>(cmdPtr.device_id()->str(), cmdPtr.request_id(), GetResult(cmdPtr.result())));
break;
default:
throw CommandFormatError("unrecognized command type given to fair::mq::cmd::Cmds::Deserialize()");
break;
}
}
}
} // namespace fair::mq::sdk::cmd

414
fairmq/sdk/commands/Commands.h
View File

@@ -1,414 +0,0 @@
/********************************************************************************
* 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" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_COMMANDFACTORY
#define FAIR_MQ_SDK_COMMANDFACTORY
#include <fairmq/States.h>
#include <vector>
#include <string>
#include <memory>
#include <type_traits>
#include <stdexcept>
#include <utility> // move
namespace fair::mq::sdk::cmd
{
enum class Format : int {
Binary,
JSON
};
enum class Result : int {
Ok,
Failure
};
enum class Type : int
{
check_state, // args: { }
change_state, // args: { transition }
dump_config, // args: { }
subscribe_to_state_change, // args: { }
unsubscribe_from_state_change, // args: { }
state_change_exiting_received, // args: { }
get_properties, // args: { request_id, property_query }
set_properties, // args: { request_id, properties }
subscription_heartbeat, // args: { interval }
current_state, // args: { device_id, current_state }
transition_status, // args: { device_id, task_id, Result, transition, current_state }
config, // args: { device_id, config_string }
state_change_subscription, // args: { device_id, task_id, Result }
state_change_unsubscription, // args: { device_id, task_id, Result }
state_change, // args: { device_id, task_id, last_state, current_state }
properties, // args: { device_id, request_id, Result, properties }
properties_set // args: { device_id, request_id, Result }
};
struct Cmd
{
explicit Cmd(const Type type) : fType(type) {}
virtual ~Cmd() = default;
Type GetType() const { return fType; }
private:
Type fType;
};
struct CheckState : Cmd
{
explicit CheckState() : Cmd(Type::check_state) {}
};
struct ChangeState : Cmd
{
explicit ChangeState(Transition transition)
: Cmd(Type::change_state)
, fTransition(transition)
{}
Transition GetTransition() const { return fTransition; }
void SetTransition(Transition transition) { fTransition = transition; }
private:
Transition fTransition;
};
struct DumpConfig : Cmd
{
explicit DumpConfig() : Cmd(Type::dump_config) {}
};
struct SubscribeToStateChange : Cmd
{
explicit SubscribeToStateChange(int64_t interval)
: Cmd(Type::subscribe_to_state_change)
, fInterval(interval)
{}
int64_t GetInterval() const { return fInterval; }
void SetInterval(int64_t interval) { fInterval = interval; }
private:
int64_t fInterval;
};
struct UnsubscribeFromStateChange : Cmd
{
explicit UnsubscribeFromStateChange() : Cmd(Type::unsubscribe_from_state_change) {}
};
struct StateChangeExitingReceived : Cmd
{
explicit StateChangeExitingReceived() : Cmd(Type::state_change_exiting_received) {}
};
struct GetProperties : Cmd
{
GetProperties(std::size_t request_id, std::string query)
: Cmd(Type::get_properties)
, fRequestId(request_id)
, fQuery(std::move(query))
{}
auto GetRequestId() const -> std::size_t { return fRequestId; }
auto SetRequestId(std::size_t requestId) -> void { fRequestId = requestId; }
auto GetQuery() const -> std::string { return fQuery; }
auto SetQuery(std::string query) -> void { fQuery = std::move(query); }
private:
std::size_t fRequestId;
std::string fQuery;
};
struct SetProperties : Cmd
{
SetProperties(std::size_t request_id, std::vector<std::pair<std::string, std::string>> properties)
: Cmd(Type::set_properties)
, fRequestId(request_id)
, fProperties(std::move(properties))
{}
auto GetRequestId() const -> std::size_t { return fRequestId; }
auto SetRequestId(std::size_t requestId) -> void { fRequestId = requestId; }
auto GetProps() const -> std::vector<std::pair<std::string, std::string>> { return fProperties; }
auto SetProps(std::vector<std::pair<std::string, std::string>> properties) -> void { fProperties = std::move(properties); }
private:
std::size_t fRequestId;
std::vector<std::pair<std::string, std::string>> fProperties;
};
struct SubscriptionHeartbeat : Cmd
{
explicit SubscriptionHeartbeat(int64_t interval)
: Cmd(Type::subscription_heartbeat)
, fInterval(interval)
{}
int64_t GetInterval() const { return fInterval; }
void SetInterval(int64_t interval) { fInterval = interval; }
private:
int64_t fInterval;
};
struct CurrentState : Cmd
{
explicit CurrentState(std::string id, State currentState)
: Cmd(Type::current_state)
, fDeviceId(std::move(id))
, fCurrentState(currentState)
{}
std::string GetDeviceId() const { return fDeviceId; }
void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; }
fair::mq::State GetCurrentState() const { return fCurrentState; }
void SetCurrentState(fair::mq::State state) { fCurrentState = state; }
private:
std::string fDeviceId;
fair::mq::State fCurrentState;
};
struct TransitionStatus : Cmd
{
explicit TransitionStatus(std::string deviceId, const uint64_t taskId, const Result result, const Transition transition, State currentState)
: Cmd(Type::transition_status)
, fDeviceId(std::move(deviceId))
, fTaskId(taskId)
, fResult(result)
, fTransition(transition)
, fCurrentState(currentState)
{}
std::string GetDeviceId() const { return fDeviceId; }
void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; }
uint64_t GetTaskId() const { return fTaskId; }
void SetTaskId(const uint64_t taskId) { fTaskId = taskId; }
Result GetResult() const { return fResult; }
void SetResult(const Result result) { fResult = result; }
Transition GetTransition() const { return fTransition; }
void SetTransition(const Transition transition) { fTransition = transition; }
fair::mq::State GetCurrentState() const { return fCurrentState; }
void SetCurrentState(fair::mq::State state) { fCurrentState = state; }
private:
std::string fDeviceId;
uint64_t fTaskId;
Result fResult;
Transition fTransition;
fair::mq::State fCurrentState;
};
struct Config : Cmd
{
explicit Config(std::string id, std::string config)
: Cmd(Type::config)
, fDeviceId(std::move(id))
, fConfig(std::move(config))
{}
std::string GetDeviceId() const { return fDeviceId; }
void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; }
std::string GetConfig() const { return fConfig; }
void SetConfig(const std::string& config) { fConfig = config; }
private:
std::string fDeviceId;
std::string fConfig;
};
struct StateChangeSubscription : Cmd
{
explicit StateChangeSubscription(std::string id, const uint64_t taskId, const Result result)
: Cmd(Type::state_change_subscription)
, fDeviceId(std::move(id))
, fTaskId(taskId)
, fResult(result)
{}
std::string GetDeviceId() const { return fDeviceId; }
void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; }
uint64_t GetTaskId() const { return fTaskId; }
void SetTaskId(const uint64_t taskId) { fTaskId = taskId; }
Result GetResult() const { return fResult; }
void SetResult(const Result result) { fResult = result; }
private:
std::string fDeviceId;
uint64_t fTaskId;
Result fResult;
};
struct StateChangeUnsubscription : Cmd
{
explicit StateChangeUnsubscription(std::string id, const uint64_t taskId, const Result result)
: Cmd(Type::state_change_unsubscription)
, fDeviceId(std::move(id))
, fTaskId(taskId)
, fResult(result)
{}
std::string GetDeviceId() const { return fDeviceId; }
void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; }
uint64_t GetTaskId() const { return fTaskId; }
void SetTaskId(const uint64_t taskId) { fTaskId = taskId; }
Result GetResult() const { return fResult; }
void SetResult(const Result result) { fResult = result; }
private:
std::string fDeviceId;
uint64_t fTaskId;
Result fResult;
};
struct StateChange : Cmd
{
explicit StateChange(std::string deviceId, const uint64_t taskId, const State lastState, const State currentState)
: Cmd(Type::state_change)
, fDeviceId(std::move(deviceId))
, fTaskId(taskId)
, fLastState(lastState)
, fCurrentState(currentState)
{}
std::string GetDeviceId() const { return fDeviceId; }
void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; }
uint64_t GetTaskId() const { return fTaskId; }
void SetTaskId(const uint64_t taskId) { fTaskId = taskId; }
fair::mq::State GetLastState() const { return fLastState; }
void SetLastState(const fair::mq::State state) { fLastState = state; }
fair::mq::State GetCurrentState() const { return fCurrentState; }
void SetCurrentState(const fair::mq::State state) { fCurrentState = state; }
private:
std::string fDeviceId;
uint64_t fTaskId;
fair::mq::State fLastState;
fair::mq::State fCurrentState;
};
struct Properties : Cmd
{
Properties(std::string deviceId, std::size_t requestId, const Result result, std::vector<std::pair<std::string, std::string>> properties)
: Cmd(Type::properties)
, fDeviceId(std::move(deviceId))
, fRequestId(requestId)
, fResult(result)
, fProperties(std::move(properties))
{}
auto GetDeviceId() const -> std::string { return fDeviceId; }
auto SetDeviceId(std::string deviceId) -> void { fDeviceId = std::move(deviceId); }
auto GetRequestId() const -> std::size_t { return fRequestId; }
auto SetRequestId(std::size_t requestId) -> void { fRequestId = requestId; }
auto GetResult() const -> Result { return fResult; }
auto SetResult(Result result) -> void { fResult = result; }
auto GetProps() const -> std::vector<std::pair<std::string, std::string>> { return fProperties; }
auto SetProps(std::vector<std::pair<std::string, std::string>> properties) -> void { fProperties = std::move(properties); }
private:
std::string fDeviceId;
std::size_t fRequestId;
Result fResult;
std::vector<std::pair<std::string, std::string>> fProperties;
};
struct PropertiesSet : Cmd {
PropertiesSet(std::string deviceId, std::size_t requestId, Result result)
: Cmd(Type::properties_set)
, fDeviceId(std::move(deviceId))
, fRequestId(requestId)
, fResult(result)
{}
auto GetDeviceId() const -> std::string { return fDeviceId; }
auto SetDeviceId(std::string deviceId) -> void { fDeviceId = std::move(deviceId); }
auto GetRequestId() const -> std::size_t { return fRequestId; }
auto SetRequestId(std::size_t requestId) -> void { fRequestId = requestId; }
auto GetResult() const -> Result { return fResult; }
auto SetResult(Result result) -> void { fResult = result; }
private:
std::string fDeviceId;
std::size_t fRequestId;
Result fResult;
};
template<typename C, typename... Args>
std::unique_ptr<Cmd> make(Args&&... args)
{
return std::make_unique<C>(std::forward<Args>(args)...);
}
struct Cmds
{
using container = std::vector<std::unique_ptr<Cmd>>;
struct CommandFormatError : std::runtime_error { using std::runtime_error::runtime_error; };
explicit Cmds() = default;
template<typename... Rest>
explicit Cmds(std::unique_ptr<Cmd>&& first, Rest&&... rest)
{
Unpack(std::forward<std::unique_ptr<Cmd>&&>(first), std::forward<Rest>(rest)...);
}
void Add(std::unique_ptr<Cmd>&& cmd) { fCmds.emplace_back(std::move(cmd)); }
template<typename C, typename... Args>
void Add(Args&&... args)
{
static_assert(std::is_base_of<Cmd, C>::value, "Only types derived from fair::mq::cmd::Cmd are allowed");
Add(make<C>(std::forward<Args>(args)...));
}
Cmd& At(size_t i) { return *(fCmds.at(i)); }
size_t Size() const { return fCmds.size(); }
void Reset() { fCmds.clear(); }
std::string Serialize(const Format type = Format::Binary) const;
void Deserialize(const std::string&, const Format type = Format::Binary);
private:
container fCmds;
void Unpack() {}
template <class... Rest>
void Unpack(std::unique_ptr<Cmd>&& first, Rest&&... rest)
{
fCmds.emplace_back(std::move(first));
Unpack(std::forward<Rest>(rest)...);
}
public:
using iterator = container::iterator;
using const_iterator = container::const_iterator;
auto begin() -> decltype(fCmds.begin()) { return fCmds.begin(); }
auto end() -> decltype(fCmds.end()) { return fCmds.end(); }
auto cbegin() -> decltype(fCmds.cbegin()) { return fCmds.cbegin(); }
auto cend() -> decltype(fCmds.cend()) { return fCmds.cend(); }
};
std::string GetResultName(const Result result);
std::string GetTypeName(const Type type);
inline std::ostream& operator<<(std::ostream& os, const Result& result) { return os << GetResultName(result); }
inline std::ostream& operator<<(std::ostream& os, const Type& type) { return os << GetTypeName(type); }
} // namespace fair::mq::sdk::cmd
#endif /* FAIR_MQ_SDK_COMMANDFACTORY */

89
fairmq/sdk/commands/CommandsFormat.fbs
View File

@@ -1,89 +0,0 @@
namespace fair.mq.sdk.cmd;
enum FBResult:byte {
Ok,
Failure
}
enum FBState:byte {
Undefined,
Ok,
Error,
Idle,
InitializingDevice,
Initialized,
Binding,
Bound,
Connecting,
DeviceReady,
InitializingTask,
Ready,
Running,
ResettingTask,
ResettingDevice,
Exiting
}
enum FBTransition:byte {
Auto,
InitDevice,
CompleteInit,
Bind,
Connect,
InitTask,
Run,
Stop,
ResetTask,
ResetDevice,
End,
ErrorFound
}
table FBProperty {
key:string;
value:string;
}
enum FBCmd:byte {
check_state, // args: { }
change_state, // args: { transition }
dump_config, // args: { }
subscribe_to_state_change, // args: { interval }
unsubscribe_from_state_change, // args: { }
state_change_exiting_received, // args: { }
get_properties, // args: { request_id, property_query }
set_properties, // args: { request_id, properties }
subscription_heartbeat, // args: { interval }
current_state, // args: { device_id, current_state }
transition_status, // args: { device_id, task_id, Result, transition, current_state }
config, // args: { device_id, config_string }
state_change_subscription, // args: { device_id, task_id, Result }
state_change_unsubscription, // args: { device_id, task_id, Result }
state_change, // args: { device_id, task_id, last_state, current_state }
properties, // args: { device_id, request_id, Result, properties }
properties_set // args: { device_id, request_id, Result }
}
table FBCommand {
command_id:FBCmd;
device_id:string;
task_id:uint64;
request_id:uint64;
interval:int64;
state:FBState;
transition:FBTransition;
result:FBResult;
config_string:string;
last_state:FBState;
current_state:FBState;
debug:string;
properties:[FBProperty];
property_query:string;
}
table FBCommands {
commands:[FBCommand];
}
root_type FBCommands;

18
fairmq/sdk/commands/CommandsFormatDef.h.in
View File

@@ -1,18 +0,0 @@
/********************************************************************************
* Copyright (C) 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 <string>
namespace fair::mq::sdk
{
namespace cmd {
constexpr auto commandsFormatDefFbs = R"(@commands_format_def_fbs@)";
} // namespace cmd
} // namespace fair::mq::sdk

256
fairmq/sdk/runDDSCommandUI.cxx
View File

@@ -1,256 +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 <fairmq/tools/Strings.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, [p] set property" << endl;
cout << "To quit press Ctrl+C" << endl;
}
void handleCommand(const string& command, const string& path, unsigned int timeout, Topology& topo, const string& pKey, const string& pVal)
{
std::pair<std::error_code, fair::mq::sdk::TopologyState> changeStateResult;
if (command == "c") {
cout << "> checking state of the devices" << endl;
auto const result = topo.GetCurrentState();
bool error = false;
for (const auto& d : result) {
cout << d.taskId << " : " << d.state << endl;
if (d.state == sdk::DeviceState::Error) {
error = true;
}
}
if (error) {
throw runtime_error("Some of the devices are in the Error state");
}
return;
} else if (command == "o") {
cout << "> dumping config of " << (path.empty() ? "all" : path) << 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));
if (result.first != std::error_code()) {
cout << "ERROR: GetProperties failed for '" << path << "': " << result.first.message() << endl;
throw runtime_error(tools::ToString("GetProperties failed for '", path, "': ", result.first.message()));
}
for (const auto& d : result.second.devices) {
for (auto const& p : d.second.props) {
cout << d.first << ": " << p.first << " : " << p.second << endl;
}
}
return;
} else if (command == "p") {
if (pKey.empty() || pVal.empty()) {
cout << "cannot send property with empty key and/or value! given key: '" << pKey << "', value: '" << pVal << "'." << endl;
throw runtime_error(tools::ToString("cannot send property with empty key and/or value! given key: '", pKey, "', value: '", pVal, "'."));
}
const DeviceProperties props{{pKey, pVal}};
cout << "> setting properties --> " << (path.empty() ? "all" : path) << endl;
auto const result = topo.SetProperties(props, path);
if (result.first != std::error_code()) {
cout << "ERROR: SetProperties failed for '" << path << "': " << result.first.message() << endl;
throw runtime_error(tools::ToString("SetProperties failed for '", path, "': ", result.first.message()));
}
// give dds time to complete request
this_thread::sleep_for(chrono::milliseconds(100));
return;
} else if (command == "i") {
cout << "> initiating InitDevice transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::InitDevice, path, std::chrono::milliseconds(timeout));
} else if (command == "k") {
cout << "> initiating CompleteInit transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::CompleteInit, path, std::chrono::milliseconds(timeout));
} else if (command == "b") {
cout << "> initiating Bind transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::Bind, path, std::chrono::milliseconds(timeout));
} else if (command == "x") {
cout << "> initiating Connect transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::Connect, path, std::chrono::milliseconds(timeout));
} else if (command == "j") {
cout << "> initiating InitTask transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::InitTask, path, std::chrono::milliseconds(timeout));
} else if (command == "r") {
cout << "> initiating Run transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::Run, path, std::chrono::milliseconds(timeout));
} else if (command == "s") {
cout << "> initiating Stop transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::Stop, path, std::chrono::milliseconds(timeout));
} else if (command == "t") {
cout << "> initiating ResetTask transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::ResetTask, path, std::chrono::milliseconds(timeout));
} else if (command == "d") {
cout << "> initiating ResetDevice transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::ResetDevice, path, std::chrono::milliseconds(timeout));
} else if (command == "q") {
cout << "> initiating End transition --> " << (path.empty() ? "all" : path) << endl;
changeStateResult = topo.ChangeState(TopologyTransition::End, path, std::chrono::milliseconds(timeout));
} else if (command == "h") {
cout << "> help" << endl;
printControlsHelp();
return;
} else {
cout << "\033[01;32mInvalid input: [" << command << "]\033[0m" << endl;
printControlsHelp();
throw runtime_error(tools::ToString("\033[01;32mInvalid input: [", command, "]\033[0m"));
}
if (changeStateResult.first != std::error_code()) {
cout << "ERROR: ChangeState failed for '" << path << "': " << changeStateResult.first.message() << endl;
throw runtime_error(tools::ToString("ERROR: ChangeState failed for '", path, "': ", changeStateResult.first.message()));
}
}
void sendCommand(const string& commandIn, const string& path, unsigned int timeout, Topology& topo, const string& pKey, const string& pVal)
{
if (!commandIn.empty()) {
handleCommand(commandIn, path, timeout, topo, pKey, pVal);
return;
}
char c = 0;
string command;
TerminalConfig tconfig;
printControlsHelp();
cin >> c;
command = c;
while (true) {
try {
handleCommand(command, path, timeout, topo, pKey, pVal);
} catch(exception& e) {
cout << "Error: " << e.what() << endl;
}
cin >> c;
command = c;
}
}
int main(int argc, char* argv[])
try {
string sessionID;
string topoFile;
string command;
string path;
string targetState;
string pKey;
string pVal;
unsigned int timeout = 0;
fair::Logger::SetConsoleSeverity("debug");
fair::Logger::SetConsoleColor(true);
bpo::options_description options("Common options");
auto envSessionId = getenv("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)");
} else {
options.add_options()("session,s", bpo::value<string>(&sessionID)->required(), "DDS Session ID (overrides any value in env var $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>(&path)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)")
("property-key", bpo::value<string>(&pKey)->default_value(""), "property key to be used with 'p' command")
("property-value", bpo::value<string>(&pVal)->default_value(""), "property value to be used with 'p' command")
("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(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, [p] set property" << endl;
return EXIT_SUCCESS;
}
bpo::notify(vm);
DDSEnvironment env;
DDSSession session(sessionID, env);
DDSTopology ddsTopo(DDSTopology::Path(topoFile), env);
Topology topo(ddsTopo, session, true);
if (!targetState.empty()) {
if (!command.empty()) {
sendCommand(command, path, timeout, topo, pKey, pVal);
}
size_t pos = targetState.find("->");
cout << "> waiting for " << (path.empty() ? "all" : path) << " to reach " << targetState << endl;
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, pKey, pVal);
}
return EXIT_SUCCESS;
} catch (exception& e) {
cerr << "Error: " << e.what() << endl;
return EXIT_FAILURE;
}