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

FairMQ: Integration of Plugins through PluginServices

Browse Source
This commit is contained in:
Dennis Klein 2017-06-26 15:08:26 +02:00 committed by Mohammad Al-Turany
parent 60d929b0bd
commit 9b61b924b2
26 changed files with 244 additions and 701 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
fairmq/CMakeLists.txt
View File

@ -46,11 +46,6 @@ endif()
if(BUILD_TESTING)
add_subdirectory(test)
endif()
if(DDS_FOUND)
add_subdirectory(plugins/config)
add_subdirectory(plugins/control)
endif(DDS_FOUND)
##########################
# libFairMQ header files #
@ -61,9 +56,7 @@ set(FAIRMQ_DEPRECATED_HEADER_FILES
set(FAIRMQ_HEADER_FILES
${FAIRMQ_DEPRECATED_HEADER_FILES}
FairMQChannel.h
FairMQConfigPlugin.h
FairMQConfigurable.h
FairMQControlPlugin.h
FairMQDevice.h
FairMQLogger.h
FairMQMessage.h
@ -90,6 +83,7 @@ set(FAIRMQ_HEADER_FILES
options/FairProgOptionsHelper.h
Plugin.h
PluginManager.h
PluginServices.h
runFairMQDevice.h
shmem/FairMQMessageSHM.h
shmem/FairMQPollerSHM.h
@ -145,6 +139,7 @@ set(FAIRMQ_SOURCE_FILES
options/FairProgOptions.cxx
Plugin.cxx
PluginManager.cxx
PluginServices.cxx
shmem/FairMQMessageSHM.cxx
shmem/FairMQPollerSHM.cxx
shmem/FairMQSocketSHM.cxx
@ -199,9 +194,9 @@ target_include_directories(FairMQ
##################
target_link_libraries(FairMQ
INTERFACE # only consumers link against interface dependencies
dl
PUBLIC # libFairMQ AND consumers of libFairMQ link aginst public dependencies
dl
pthread
Boost::boost
Boost::log

27
fairmq/FairMQConfigPlugin.h
View File

@ -1,27 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQCONFIGPLUGIN_H_
#define FAIRMQCONFIGPLUGIN_H_
class FairMQDevice;
extern "C"
struct FairMQConfigPlugin
{
typedef void (*init_t)(FairMQDevice&);
init_t initConfig;
typedef void (*handleInitialConfig_t)(FairMQDevice&);
handleInitialConfig_t handleInitialConfig;
typedef void (*stop_t)();
stop_t stopConfig;
};
#endif /* FAIRMQCONFIGPLUGIN_H_ */

27
fairmq/FairMQControlPlugin.h
View File

@ -1,27 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQCONTROLPLUGIN_H_
#define FAIRMQCONTROLPLUGIN_H_
class FairMQDevice;
extern "C"
struct FairMQControlPlugin
{
typedef void (*init_t)(FairMQDevice&);
init_t initControl;
typedef void (*handleStateChanges_t)(FairMQDevice&);
handleStateChanges_t handleStateChanges;
typedef void (*stop_t)();
stop_t stopControl;
};
#endif /* FAIRMQCONTROLPLUGIN_H_ */

2
fairmq/FairMQDevice.cxx
View File

@ -933,7 +933,7 @@ void FairMQDevice::CreateOwnConfig()
char arg0[] = "undefined"; // executable name
char arg1[] = "--id";
char* arg2 = const_cast<char*>(id.c_str()); // device ID
char* argv[] = { &arg0[0], &arg1[0], arg2, nullptr };
const char* argv[] = { &arg0[0], &arg1[0], arg2, nullptr };
int argc = static_cast<int>((sizeof(argv) / sizeof(argv[0])) - 1);
fConfig->ParseAll(argc, &argv[0]);

11
fairmq/Plugin.cxx
View File

@ -11,11 +11,12 @@
using namespace std;
fair::mq::Plugin::Plugin(const string name, const Version version, const string maintainer, const string homepage)
: fkName(name)
, fkVersion(version)
, fkMaintainer(maintainer)
, fkHomepage(homepage)
fair::mq::Plugin::Plugin(const string name, const Version version, const string maintainer, const string homepage, PluginServices& pluginServices)
: fkName{name}
, fkVersion{version}
, fkMaintainer{maintainer}
, fkHomepage{homepage}
, fPluginServices{pluginServices}
{
LOG(DEBUG) << "Loaded plugin: " << *this;
}

10
fairmq/Plugin.h
View File

@ -9,7 +9,8 @@
#ifndef FAIR_MQ_PLUGIN_H
#define FAIR_MQ_PLUGIN_H
#include <fairmq/tools/Strings.h>
#include <fairmq/Tools.h>
#include <fairmq/PluginServices.h>
#include <boost/dll/alias.hpp>
#include <boost/optional.hpp>
#include <boost/program_options.hpp>
@ -49,7 +50,7 @@ class Plugin
};
Plugin() = delete;
Plugin(const std::string name, const Version version, const std::string maintainer, const std::string homepage);
Plugin(const std::string name, const Version version, const std::string maintainer, const std::string homepage, PluginServices& pluginServices);
virtual ~Plugin();
auto GetName() const -> const std::string& { return fkName; }
@ -74,6 +75,7 @@ class Plugin
const Version fkVersion;
const std::string fkMaintainer;
const std::string fkHomepage;
PluginServices& fPluginServices;
}; /* class Plugin */
@ -81,9 +83,9 @@ class Plugin
} /* namespace fair */
#define REGISTER_FAIRMQ_PLUGIN(KLASS, NAME, VERSION, MAINTAINER, HOMEPAGE, PROGOPTIONS) \
static auto Make_##NAME##_Plugin() -> std::shared_ptr<KLASS> \
static auto Make_##NAME##_Plugin(fair::mq::PluginServices& pluginServices) -> std::shared_ptr<KLASS> \
{ \
return std::make_shared<KLASS>(std::string{#NAME}, VERSION, std::string{MAINTAINER}, std::string{HOMEPAGE}); \
return std::make_shared<KLASS>(std::string{#NAME}, VERSION, std::string{MAINTAINER}, std::string{HOMEPAGE}, pluginServices); \
} \
BOOST_DLL_ALIAS(Make_##NAME##_Plugin, make_##NAME##_plugin) \
BOOST_DLL_ALIAS(PROGOPTIONS, get_##NAME##_plugin_progoptions)

54
fairmq/PluginManager.cxx
View File

@ -14,6 +14,7 @@
#include <memory>
#include <sstream>
#include <string>
#include <utility>
using namespace std;
using fair::mq::tools::ToString;
@ -27,7 +28,7 @@ using boost::optional;
const std::string fair::mq::PluginManager::fgkLibPrefix = "FairMQPlugin_";
fair::mq::PluginManager::PluginManager()
: fSearchPaths({"."})
: fSearchPaths{{"."}}
{
}
@ -56,7 +57,7 @@ auto fair::mq::PluginManager::PrependSearchPath(const fs::path& path) -> void
fSearchPaths.insert(begin(fSearchPaths), path);
}
auto fair::mq::PluginManager::ProgramOptions() -> const optional<po::options_description>
auto fair::mq::PluginManager::ProgramOptions() -> const po::options_description
{
auto plugin_options = po::options_description{"Plugin Manager"};
plugin_options.add_options()
@ -69,14 +70,14 @@ auto fair::mq::PluginManager::ProgramOptions() -> const optional<po::options_des
"* If you mix the overriding and appending/prepending syntaxes, the overriding paths act as default search path, e.g.\n"
" -S /usr/lib >/lib </home/user/lib /usr/local/lib results in /home/user/lib,/usr/local/lib,/usr/lib/,/lib")
("plugin,P", po::value<vector<string>>(),
"List of plugin names to load in order, e.g. if the file is called 'libFairMQPlugin_example.so', just list 'example' or 'd:example' here. To load a static plugin, list 's:example' here (The plugin must be already linked into the executable).");
"List of plugin names to load in order, e.g. if the file is called 'libFairMQPlugin_example.so', just list 'example' or 'd:example' here. To load a prelinked plugin, list 'p:example' here.");
return plugin_options;
}
auto fair::mq::PluginManager::MakeFromCommandLineOptions(const vector<string> args) -> shared_ptr<PluginManager>
{
// Parse command line options
auto options = ProgramOptions().value();
auto options = ProgramOptions();
auto vm = po::variables_map{};
try
{
@ -87,7 +88,7 @@ auto fair::mq::PluginManager::MakeFromCommandLineOptions(const vector<string> ar
{
throw ProgramOptionsParseError{ToString("Error occured while parsing the 'Plugin Manager' program options: ", e.what())};
}
// Process plugin search paths
auto append = vector<fs::path>{};
auto prepend = vector<fs::path>{};
@ -107,13 +108,7 @@ auto fair::mq::PluginManager::MakeFromCommandLineOptions(const vector<string> ar
mgr->SetSearchPaths(searchPaths);
for(const auto& path : prepend) { mgr->PrependSearchPath(path); }
for(const auto& path : append) { mgr->AppendSearchPath(path); }
if (vm.count("plugin"))
{
for (const auto& plugin : vm["plugin"].as<vector<string>>())
{
mgr->LoadPlugin(plugin);
}
}
if (vm.count("plugin")) { mgr->LoadPlugins(vm["plugin"].as<vector<string>>()); }
// Return the plugin manager and command line options, that have not been recognized.
return mgr;
@ -121,10 +116,10 @@ auto fair::mq::PluginManager::MakeFromCommandLineOptions(const vector<string> ar
auto fair::mq::PluginManager::LoadPlugin(const string& pluginName) -> void
{
if (pluginName.substr(0,2) == "s:")
if (pluginName.substr(0,2) == "p:")
{
// Mechanism A: static
LoadPluginStatic(pluginName.substr(2));
// Mechanism A: prelinked dynamic
LoadPluginPrelinkedDynamic(pluginName.substr(2));
}
else if (pluginName.substr(0,2) == "d:")
{
@ -138,7 +133,7 @@ auto fair::mq::PluginManager::LoadPlugin(const string& pluginName) -> void
}
}
auto fair::mq::PluginManager::LoadPluginStatic(const string& pluginName) -> void
auto fair::mq::PluginManager::LoadPluginPrelinkedDynamic(const string& pluginName) -> void
{
// Load symbol
if (fPluginFactories.find(pluginName) == fPluginFactories.end())
@ -149,11 +144,9 @@ auto fair::mq::PluginManager::LoadPluginStatic(const string& pluginName) -> void
}
catch (boost::system::system_error& e)
{
throw PluginLoadError(ToString("An error occurred while loading static plugin ", pluginName, ": ", e.what()));
throw PluginLoadError(ToString("An error occurred while loading prelinked dynamic plugin ", pluginName, ": ", e.what()));
}
}
InstantiatePlugin(pluginName);
}
auto fair::mq::PluginManager::LoadPluginDynamic(const string& pluginName) -> void
@ -172,6 +165,7 @@ auto fair::mq::PluginManager::LoadPluginDynamic(const string& pluginName) -> voi
dll::load_mode::append_decorations
);
fPluginOrder.push_back(pluginName);
success = true;
break;
}
@ -183,17 +177,29 @@ auto fair::mq::PluginManager::LoadPluginDynamic(const string& pluginName) -> voi
}
}
}
if(!success) throw PluginLoadError(ToString("The plugin ", pluginName, " could not be found in the plugin search paths."));
if(!success) { throw PluginLoadError(ToString("The plugin ", pluginName, " could not be found in the plugin search paths.")); }
}
InstantiatePlugin(pluginName);
}
auto fair::mq::PluginManager::InstantiatePlugin(const string& pluginName) -> void
{
if (fPlugins.find(pluginName) == fPlugins.end())
{
fPlugins[pluginName] = fPluginFactories[pluginName]();
fPluginOrder.push_back(pluginName);
fPlugins[pluginName] = fPluginFactories[pluginName](*fPluginServices);
}
}
auto fair::mq::PluginManager::InstantiatePlugins() -> void
{
for(const auto& pluginName : fPluginOrder)
{
try
{
InstantiatePlugin(pluginName);
}
catch (std::exception& e)
{
throw PluginInstantiationError(ToString("An error occurred while instantiating plugin ", pluginName, ": ", e.what()));
}
}
}

18
fairmq/PluginManager.h
View File

@ -10,7 +10,9 @@
#define FAIR_MQ_PLUGINMANAGER_H
#include <fairmq/Plugin.h>
#include <fairmq/PluginServices.h>
#include <fairmq/Tools.h>
#include <FairMQDevice.h>
#define BOOST_FILESYSTEM_VERSION 3
#define BOOST_FILESYSTEM_NO_DEPRECATED
#include <boost/filesystem.hpp>
@ -38,15 +40,15 @@ namespace mq
* @brief manages and owns plugin instances
*
* The plugin manager is responsible for the whole plugin lifecycle. It
* facilitates three plugin mechanisms:
* A static (built-in) plugins
* facilitates two plugin mechanisms:
* A prelinked dynamic plugins (shared libraries)
* B dynamic plugins (shared libraries)
*/
class PluginManager
{
public:
using PluginFactory = std::shared_ptr<fair::mq::Plugin>();
using PluginFactory = std::shared_ptr<fair::mq::Plugin>(PluginServices&);
using PluginProgOptions = const boost::optional<boost::program_options::options_description>();
PluginManager();
@ -60,8 +62,10 @@ class PluginManager
auto LoadPlugin(const std::string& pluginName) -> void;
auto LoadPlugins(const std::vector<std::string>& pluginNames) -> void { for(const auto& pluginName : pluginNames) { LoadPlugin(pluginName); } }
struct PluginLoadError : std::runtime_error { using std::runtime_error::runtime_error; };
auto InstantiatePlugins() -> void;
struct PluginInstantiationError : std::runtime_error { using std::runtime_error::runtime_error; };
static auto ProgramOptions() -> const boost::optional<boost::program_options::options_description>;
static auto ProgramOptions() -> const boost::program_options::options_description;
static auto MakeFromCommandLineOptions(const std::vector<std::string>) -> std::shared_ptr<PluginManager>;
struct ProgramOptionsParseError : std::runtime_error { using std::runtime_error::runtime_error; };
@ -70,11 +74,14 @@ class PluginManager
auto ForEachPlugin(std::function<void (Plugin&)> func) -> void { for(const auto& p : fPluginOrder) { func(*fPlugins[p]); } }
auto ForEachPluginProgOptions(std::function<void (const boost::program_options::options_description&)> func) const -> void { for(const auto& pair : fPluginProgOptions) { func(pair.second); } }
template<typename... Args>
auto EmplacePluginServices(Args&&... args) -> void { fPluginServices = fair::mq::tools::make_unique<PluginServices>(std::forward<Args>(args)...); };
private:
static auto ValidateSearchPath(const boost::filesystem::path&) -> void;
auto LoadPluginStatic(const std::string& pluginName) -> void;
auto LoadPluginPrelinkedDynamic(const std::string& pluginName) -> void;
auto LoadPluginDynamic(const std::string& pluginName) -> void;
template<typename... Args>
auto LoadSymbols(const std::string& pluginName, Args&&... args) -> void
@ -107,6 +114,7 @@ class PluginManager
std::map<std::string, std::shared_ptr<Plugin>> fPlugins;
std::vector<std::string> fPluginOrder;
std::map<std::string, boost::program_options::options_description> fPluginProgOptions;
std::unique_ptr<PluginServices> fPluginServices;
}; /* class PluginManager */

8
fairmq/plugins/config/FairMQDDSConfigPlugin.h → fairmq/PluginServices.cxx
View File

@ -1,7 +1,11 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/PluginServices.h>
using namespace std;

88
fairmq/PluginServices.h Normal file
View File

@ -0,0 +1,88 @@
/********************************************************************************
* 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_PLUGINSERVICES_H
#define FAIR_MQ_PLUGINSERVICES_H
#include <FairMQDevice.h>
#include <options/FairMQProgOptions.h>
#include <functional>
#include <string>
namespace fair
{
namespace mq
{
/**
* @class PluginServices PluginServices.h <fairmq/PluginServices.h>
* @brief Facilitates communication between devices and plugins
*
* - Configuration interface
* - Control interface
*/
class PluginServices
{
public:
PluginServices() = delete;
PluginServices(FairMQProgOptions& config, FairMQDevice& device)
: fDevice{device}
, fConfig{config}
{}
// Control
//enum class DeviceState
//{
//Error,
//Idle,
//Initializing_device,
//Device_ready,
//Initializing_task,
//Ready,
//Running,
//Paused,
//Resetting_task,
//Resetting_device,
//Exiting
//}
//auto ToDeviceState(std::string state) const -> DeviceState;
//auto ChangeDeviceState(DeviceState next) -> void;
//auto SubscribeToDeviceStateChange(std::function<void(DeviceState [>new<])> callback) -> void;
//auto UnsubscribeFromDeviceChange() -> void;
//// Configuration
//// Writing only works during Initializing_device state
//template<typename T>
//auto SetProperty(const std::string& key, T val) -> void;
//template<typename T>
//auto GetProperty(const std::string& key) const -> T;
//auto GetPropertyAsString(const std::string& key) const -> std::string;
//template<typename T>
//auto SubscribeToPropertyChange(
//const std::string& key,
//std::function<void(const std::string& [>key*/, const T /*newValue<])> callback
//) const -> void;
//auto UnsubscribeFromPropertyChange(const std::string& key) -> void;
private:
FairMQDevice& fDevice;
FairMQProgOptions& fConfig;
}; /* class PluginServices */
} /* namespace mq */
} /* namespace fair */
#endif /* FAIR_MQ_PLUGINSERVICES_H */

2
fairmq/options/FairMQProgOptions.cxx
View File

@ -37,7 +37,7 @@ FairMQProgOptions::~FairMQProgOptions()
{
}
void FairMQProgOptions::ParseAll(const int argc, char** argv, bool allowUnregistered)
void FairMQProgOptions::ParseAll(const int argc, const char** argv, bool allowUnregistered)
{
// init description
InitOptionDescription();

3
fairmq/options/FairMQProgOptions.h
View File

@ -17,6 +17,7 @@
#define FAIRMQPROGOPTIONS_H
#include <unordered_map>
#include <functional>
#include <map>
#include <set>
#include <mutex>
@ -38,7 +39,7 @@ class FairMQProgOptions : public FairProgOptions , public FairMQEventManager
// parse command line and txt/INI configuration file.
// default parser for the mq-configuration file (JSON/XML) is called if command line key mq-config is called
virtual void ParseAll(const int argc, char** argv, bool allowUnregistered = false);
virtual void ParseAll(const int argc, const char** argv, bool allowUnregistered = false);
// external parser, store function
template <typename T, typename ...Args>

4
fairmq/options/FairProgOptions.cxx
View File

@ -124,7 +124,7 @@ void FairProgOptions::UseConfigFile(const string& filename)
/// //////////////////////////////////////////////////////////////////////////////////////////////////////
/// Parser
int FairProgOptions::ParseCmdLine(const int argc, char** argv, const po::options_description& desc, po::variables_map& varmap, bool allowUnregistered)
int FairProgOptions::ParseCmdLine(const int argc, const char** argv, const po::options_description& desc, po::variables_map& varmap, bool allowUnregistered)
{
// get options from cmd line and store in variable map
// here we use command_line_parser instead of parse_command_line, to allow unregistered and positional options
@ -151,7 +151,7 @@ int FairProgOptions::ParseCmdLine(const int argc, char** argv, const po::options
return 0;
}
int FairProgOptions::ParseCmdLine(const int argc, char** argv, const po::options_description& desc, bool allowUnregistered)
int FairProgOptions::ParseCmdLine(const int argc, const char** argv, const po::options_description& desc, bool allowUnregistered)
{
return ParseCmdLine(argc, argv, desc, fVarMap, allowUnregistered);
}

6
fairmq/options/FairProgOptions.h
View File

@ -151,8 +151,8 @@ class FairProgOptions
const po::variables_map& GetVarMap() const { return fVarMap; }
// boost prog options parsers
int ParseCmdLine(const int argc, char** argv, const po::options_description& desc, po::variables_map& varmap, bool allowUnregistered = false);
int ParseCmdLine(const int argc, char** argv, const po::options_description& desc, bool allowUnregistered = false);
int ParseCmdLine(const int argc, const char** argv, const po::options_description& desc, po::variables_map& varmap, bool allowUnregistered = false);
int ParseCmdLine(const int argc, const char** argv, const po::options_description& desc, bool allowUnregistered = false);
int ParseCfgFile(const std::string& filename, const po::options_description& desc, po::variables_map& varmap, bool allowUnregistered = false);
int ParseCfgFile(const std::string& filename, const po::options_description& desc, bool allowUnregistered = false);
@ -161,7 +161,7 @@ class FairProgOptions
int ParseEnvironment(const std::function<std::string(std::string)>&);
virtual void ParseAll(const int argc, char** argv, bool allowUnregistered = false) = 0;// TODO change return type to bool and propagate to executable
virtual void ParseAll(const int argc, const char** argv, bool allowUnregistered = false) = 0;// TODO change return type to bool and propagate to executable
virtual int PrintOptions();
virtual int PrintOptionsRaw();

46
fairmq/plugins/config/CMakeLists.txt
View File

@ -1,46 +0,0 @@
################################################################################
# Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
# #
# This software is distributed under the terms of the #
# GNU Lesser General Public Licence version 3 (LGPL) version 3, #
# copied verbatim in the file "LICENSE" #
################################################################################
set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/plugins/config
${CMAKE_SOURCE_DIR}/fairmq/options
)
set(SYSTEM_INCLUDE_DIRECTORIES
${SYSTEM_INCLUDE_DIRECTORIES}
${Boost_INCLUDE_DIR}
${DDS_INCLUDE_DIR}
)
include_directories(${INCLUDE_DIRECTORIES})
include_directories(SYSTEM ${SYSTEM_INCLUDE_DIRECTORIES})
set(LINK_DIRECTORIES
${LINK_DIRECTORIES}
${Boost_LIBRARY_DIRS}
)
link_directories(${LINK_DIRECTORIES})
set(SRCS
"FairMQDDSConfigPlugin.cxx"
)
set(LIBRARY_NAME FairMQDDSConfigPlugin)
set(DEPENDENCIES
${DEPENDENCIES}
FairMQ
pthread
${DDS_INTERCOM_LIBRARY_SHARED}
${DDS_PROTOCOL_LIBRARY_SHARED}
${DDS_USER_DEFAULTS_LIBRARY_SHARED}
)
GENERATE_LIBRARY()

187
fairmq/plugins/config/FairMQDDSConfigPlugin.cxx
View File

@ -1,187 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "dds_intercom.h"
#include "FairMQConfigPlugin.h"
#include "FairMQLogger.h"
#include "FairMQDevice.h"
#include "FairMQChannel.h"
#include "FairMQProgOptions.h"
#include <vector>
#include <map>
#include <string>
#include <exception>
#include <unordered_map>
using namespace std;
using namespace dds::intercom_api;
// container to hold channel config and corresponding dds key values
struct DDSConfig
{
DDSConfig()
: subChannels()
, ddsValues()
{}
// container of sub channels, e.g. 'i' in data[i]
vector<FairMQChannel*> subChannels;
// dds values for the channel
unordered_map<string, string> ddsValues;
};
class FairMQConfigPluginDDS
{
public:
static FairMQConfigPluginDDS* GetInstance()
{
if (fInstance == NULL)
{
fInstance = new FairMQConfigPluginDDS();
}
return fInstance;
}
static void ResetInstance()
{
try
{
delete fInstance;
fInstance = NULL;
}
catch (exception& e)
{
LOG(ERROR) << "Error: " << e.what() << endl;
return;
}
}
void Init(FairMQDevice& device)
{
for (auto& mi : device.fChannels)
{
if ((mi.second).at(0).GetMethod() == "bind")
{
for (auto& vi : mi.second)
{
fBindingChans.push_back(&vi);
}
}
else if ((mi.second).at(0).GetMethod() == "connect")
{
// try some trickery with forwarding emplacing values into map
fConnectingChans.emplace(piecewise_construct, forward_as_tuple(mi.first), forward_as_tuple());
LOG(DEBUG) << "preparing to connect: " << (mi.second).at(0).GetChannelPrefix() << " with " << mi.second.size() << " sockets.";
for (auto& vi : mi.second)
{
fConnectingChans.at(mi.first).subChannels.push_back(&vi);
}
}
else
{
LOG(ERROR) << "Cannot update address configuration. Socket method (bind/connect) not specified.";
return;
}
}
if (fConnectingChans.size() > 0)
{
LOG(DEBUG) << "Subscribing for DDS properties.";
fDDSKeyValue.subscribe([&] (const string& propertyId, const string& key, const string& value)
{
LOG(DEBUG) << "Received update for " << propertyId << ": key=" << key << " value=" << value;
fConnectingChans.at(propertyId).ddsValues.insert(make_pair<string, string>(key.c_str(), value.c_str()));
// update channels and remove them from unfinished container
for (auto mi = fConnectingChans.begin(); mi != fConnectingChans.end(); /* no increment */)
{
if (mi->second.subChannels.size() == mi->second.ddsValues.size())
{
auto it = mi->second.ddsValues.begin();
for (unsigned int i = 0; i < mi->second.subChannels.size(); ++i)
{
mi->second.subChannels.at(i)->UpdateAddress(it->second);
++it;
}
// when multiple subChannels are used, their order on every device should be the same, irregardless of arrival order from DDS.
device.SortChannel(mi->first);
fConnectingChans.erase(mi++);
}
else
{
++mi;
}
}
});
}
}
void Run(FairMQDevice& /*device*/)
{
// start DDS intercom service
fService.start();
// publish bound addresses via DDS at keys corresponding to the channel prefixes, e.g. 'data' in data[i]
for (const auto& i : fBindingChans)
{
LOG(DEBUG) << "Publishing " << i->GetChannelPrefix() << " address to DDS under '" << i->GetChannelPrefix() << "' property name.";
fDDSKeyValue.putValue(i->GetChannelPrefix(), i->GetAddress());
}
}
private:
FairMQConfigPluginDDS()
: fService()
, fDDSKeyValue(fService)
, fBindingChans()
, fConnectingChans()
{
fService.subscribeOnError([](EErrorCode errorCode, const string& msg) {
LOG(ERROR) << "DDS key-value error code: " << errorCode << ", message: " << msg;
});
}
static FairMQConfigPluginDDS* fInstance;
CIntercomService fService;
CKeyValue fDDSKeyValue;
// container for binding channels
vector<FairMQChannel*> fBindingChans;
// container for connecting channels
map<string, DDSConfig> fConnectingChans;
};
FairMQConfigPluginDDS* FairMQConfigPluginDDS::fInstance = NULL;
void initConfig(FairMQDevice& device)
{
FairMQConfigPluginDDS::GetInstance()->Init(device);
}
/// Handles channels addresses of the device with configuration from DDS
/// Addresses of binding channels are published via DDS using channels names as keys
/// Addresses of connecting channels are collected from DDS using channels names as keys
/// \param device Reference to FairMQDevice whose channels to handle
void handleInitialConfig(FairMQDevice& device)
{
FairMQConfigPluginDDS::GetInstance()->Run(device);
}
void stopConfig()
{
LOG(DEBUG) << "[FairMQConfigPluginDDS]: " << "Resetting instance.";
FairMQConfigPluginDDS::ResetInstance();
LOG(DEBUG) << "[FairMQConfigPluginDDS]: " << "Instance has been reset.";
}
FairMQConfigPlugin fairmqConfigPlugin = { initConfig, handleInitialConfig, stopConfig };

65
fairmq/plugins/control/CMakeLists.txt
View File

@ -1,65 +0,0 @@
################################################################################
# Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
# #
# This software is distributed under the terms of the #
# GNU Lesser General Public Licence version 3 (LGPL) version 3, #
# copied verbatim in the file "LICENSE" #
################################################################################
set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
${CMAKE_SOURCE_DIR}/fairmq/plugins/control
)
set(SYSTEM_INCLUDE_DIRECTORIES
${SYSTEM_INCLUDE_DIRECTORIES}
${Boost_INCLUDE_DIR}
${DDS_INCLUDE_DIR}
)
include_directories(${INCLUDE_DIRECTORIES})
include_directories(SYSTEM ${SYSTEM_INCLUDE_DIRECTORIES})
set(LINK_DIRECTORIES
${LINK_DIRECTORIES}
${Boost_LIBRARY_DIRS}
)
link_directories(${LINK_DIRECTORIES})
set(SRCS
"FairMQDDSControlPlugin.cxx"
)
set(LIBRARY_NAME FairMQDDSControlPlugin)
set(DEPENDENCIES
${DEPENDENCIES}
FairMQ
pthread
${DDS_INTERCOM_LIBRARY_SHARED}
${DDS_PROTOCOL_LIBRARY_SHARED}
${DDS_USER_DEFAULTS_LIBRARY_SHARED}
)
GENERATE_LIBRARY()
set(Exe_Names
fairmq-dds-command-ui
)
set(Exe_Source
../../run/runDDSCommandUI.cxx
)
list(LENGTH Exe_Names _length)
math(EXPR _length ${_length}-1)
foreach(_file RANGE 0 ${_length})
list(GET Exe_Names ${_file} _name)
list(GET Exe_Source ${_file} _src)
set(EXE_NAME ${_name})
set(SRCS ${_src})
set(DEPENDENCIES FairMQ pthread ${DDS_INTERCOM_LIBRARY_SHARED} ${DDS_PROTOCOL_LIBRARY_SHARED} ${DDS_USER_DEFAULTS_LIBRARY_SHARED})
GENERATE_EXECUTABLE()
endforeach(_file RANGE 0 ${_length})

164
fairmq/plugins/control/FairMQDDSControlPlugin.cxx
View File

@ -1,164 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "dds_intercom.h"
#include "FairMQControlPlugin.h"
#include "FairMQLogger.h"
#include "FairMQDevice.h"
#include <string>
#include <exception>
#include <condition_variable>
#include <mutex>
#include <chrono>
#include <set>
using namespace std;
using namespace dds::intercom_api;
class FairMQControlPluginDDS
{
public:
static FairMQControlPluginDDS* GetInstance()
{
if (fInstance == NULL)
{
fInstance = new FairMQControlPluginDDS();
}
return fInstance;
}
static void ResetInstance()
{
try
{
delete fInstance;
fInstance = NULL;
}
catch (exception& e)
{
LOG(ERROR) << "Error: " << e.what() << endl;
return;
}
}
void Init(FairMQDevice& device)
{
string id = device.GetId();
string pid(to_string(getpid()));
try
{
fDDSCustomCmd.subscribe([id, pid, &device, this](const string& cmd, const string& cond, uint64_t senderId)
{
LOG(INFO) << "Received command: " << cmd;
if (cmd == "check-state")
{
fDDSCustomCmd.send(id + ": " + device.GetCurrentStateName() + " (pid: " + pid + ")", to_string(senderId));
}
else if (fEvents.find(cmd) != fEvents.end())
{
fDDSCustomCmd.send(id + ": attempting to " + cmd, to_string(senderId));
device.ChangeState(cmd);
}
else if (cmd == "END")
{
fDDSCustomCmd.send(id + ": attempting to " + cmd, to_string(senderId));
device.ChangeState(cmd);
fDDSCustomCmd.send(id + ": " + device.GetCurrentStateName(), to_string(senderId));
if (device.GetCurrentStateName() == "EXITING")
{
unique_lock<mutex> lock(fMtx);
fStopCondition.notify_one();
}
}
else
{
LOG(WARN) << "Unknown command: " << cmd;
LOG(WARN) << "Origin: " << senderId;
LOG(WARN) << "Destination: " << cond;
}
});
}
catch (exception& e)
{
LOG(ERROR) << "Error: " << e.what() << endl;
return;
}
}
void Run(FairMQDevice& device)
{
try
{
fService.start();
LOG(INFO) << "Listening for commands from DDS...";
unique_lock<mutex> lock(fMtx);
while (!device.Terminated())
{
fStopCondition.wait_for(lock, chrono::seconds(1));
}
LOG(DEBUG) << "Stopping DDS control plugin";
}
catch (exception& e)
{
LOG(ERROR) << "Error: " << e.what() << endl;
return;
}
}
private:
FairMQControlPluginDDS()
: fService()
, fDDSCustomCmd(fService)
, fMtx()
, fStopCondition()
, fEvents({ "INIT_DEVICE", "INIT_TASK", "PAUSE", "RUN", "STOP", "RESET_TASK", "RESET_DEVICE" })
{
fService.subscribeOnError([](const EErrorCode errorCode, const string& errorMsg) {
LOG(ERROR) << "Error received: error code: " << errorCode << ", error message: " << errorMsg << endl;
});
}
static FairMQControlPluginDDS* fInstance;
CIntercomService fService;
CCustomCmd fDDSCustomCmd;
mutex fMtx;
condition_variable fStopCondition;
const set<string> fEvents;
};
FairMQControlPluginDDS* FairMQControlPluginDDS::fInstance = NULL;
void initControl(FairMQDevice& device)
{
FairMQControlPluginDDS::GetInstance()->Init(device);
}
/// Controls device state via DDS custom commands interface
/// \param device Reference to FairMQDevice whose state to control
void handleStateChanges(FairMQDevice& device)
{
FairMQControlPluginDDS::GetInstance()->Run(device);
}
void stopControl()
{
LOG(DEBUG) << "[FairMQControlPluginDDS]: " << "Resetting instance.";
FairMQControlPluginDDS::ResetInstance();
LOG(DEBUG) << "[FairMQControlPluginDDS]: " << "Instance has been reset.";
}
FairMQControlPlugin fairmqControlPlugin = { initControl, handleStateChanges, stopControl };

7
fairmq/plugins/control/FairMQDDSControlPlugin.h
View File

@ -1,7 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/

29
fairmq/runFairMQDevice.h
View File

@ -1,18 +1,18 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <FairMQLogger.h>
#include <options/FairMQProgOptions.h>
#include <FairMQDevice.h>
#include <tools/runSimpleMQStateMachine.h>
#include <fairmq/PluginManager.h>
#include <boost/program_options.hpp>
#include "FairMQLogger.h"
#include "options/FairMQProgOptions.h"
#include "FairMQDevice.h"
#include "tools/runSimpleMQStateMachine.h"
template <typename R>
class GenericFairMQDevice : public FairMQDevice
{
@ -40,7 +40,7 @@ FairMQDevicePtr getDevice(const FairMQProgOptions& config);
// to be implemented by the user to add custom command line options (or just with empty body)
void addCustomOptions(boost::program_options::options_description&);
int main(int argc, char** argv)
int main(int argc, const char** argv)
{
try
{
@ -49,7 +49,14 @@ int main(int argc, char** argv)
FairMQProgOptions config;
config.AddToCmdLineOptions(customOptions);
config.ParseAll(argc, argv);
auto pluginManager = fair::mq::PluginManager::MakeFromCommandLineOptions(fair::mq::tools::ToStrVector(argc, argv));
config.AddToCmdLineOptions(pluginManager->ProgramOptions());
pluginManager->ForEachPluginProgOptions([&config](const boost::program_options::options_description& options){
config.AddToCmdLineOptions(options);
});
config.ParseAll(argc, argv, true);
std::unique_ptr<FairMQDevice> device(getDevice(config));
if (!device)
@ -57,6 +64,10 @@ int main(int argc, char** argv)
LOG(ERROR) << "getDevice(): no valid device provided. Exiting.";
return 1;
}
pluginManager->EmplacePluginServices(config, *device);
pluginManager->InstantiatePlugins();
int result = runStateMachine(*device, config);
if (result > 0)

4
fairmq/test/helper/plugins/dummy.h.in
View File

@ -26,8 +26,8 @@ class DummyPlugin : public fair::mq::Plugin
{
public:
DummyPlugin(const std::string name, const Version version, const std::string maintainer, const std::string homepage)
: Plugin(name, version, maintainer, homepage)
DummyPlugin(const std::string name, const Version version, const std::string maintainer, const std::string homepage, PluginServices& pluginServices)
: Plugin(name, version, maintainer, homepage, pluginServices)
{
}

4
fairmq/test/helper/plugins/dummy2.h.in
View File

@ -22,8 +22,8 @@ class Dummy2Plugin : public fair::mq::Plugin
{
public:
Dummy2Plugin(const std::string name, const Version version, const std::string maintainer, const std::string homepage)
: Plugin(name, version, maintainer, homepage)
Dummy2Plugin(const std::string name, const Version version, const std::string maintainer, const std::string homepage, PluginServices& pluginServices)
: Plugin(name, version, maintainer, homepage, pluginServices)
{
}
}; /* class Dummy2Plugin */

35
fairmq/test/plugins/_plugin.cxx
View File

@ -8,30 +8,55 @@
#include <gtest/gtest.h>
#include <fairmq/Plugin.h>
#include <fairmq/PluginServices.h>
#include <FairMQDevice.h>
#include <options/FairMQProgOptions.h>
#include <sstream>
#include <string>
#include <vector>
namespace
{
using namespace std;
using fair::mq::Plugin;
using namespace fair::mq;
auto control(FairMQDevice& device) -> void
{
device.SetTransport("zeromq");
for (const auto event : {
FairMQDevice::INIT_DEVICE,
FairMQDevice::RESET_DEVICE,
FairMQDevice::END,
}) {
device.ChangeState(event);
if (event != FairMQDevice::END) device.WaitForEndOfState(event);
}
}
TEST(Plugin, Operators)
{
auto p1 = Plugin{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git"};
auto p2 = Plugin{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git"};
auto p3 = Plugin{"file", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/file.git"};
auto config = FairMQProgOptions{};
FairMQDevice device{};
auto services = PluginServices{config, device};
auto p1 = Plugin{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", services};
auto p2 = Plugin{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", services};
auto p3 = Plugin{"file", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/file.git", services};
EXPECT_EQ(p1, p2);
EXPECT_NE(p1, p3);
control(device);
}
TEST(Plugin, OstreamOperators)
{
auto p1 = Plugin{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git"};
auto config = FairMQProgOptions{};
FairMQDevice device{};
auto services = PluginServices{config, device};
auto p1 = Plugin{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", services};
stringstream ss;
ss << p1;
EXPECT_EQ(ss.str(), string{"'dds', version '1.0.0', maintainer 'Foo Bar <foo.bar@test.net>', homepage 'https://git.test.net/dds.git'"});
control(device);
}
TEST(PluginVersion, Operators)

26
fairmq/test/plugins/_plugin_manager.cxx
View File

@ -8,6 +8,9 @@
#include <gtest/gtest.h>
#include <fairmq/PluginManager.h>
#include <fairmq/PluginServices.h>
#include <FairMQDevice.h>
#include <options/FairMQProgOptions.h>
#include <FairMQLogger.h>
#include <fstream>
#include <vector>
@ -20,16 +23,35 @@ using namespace boost::filesystem;
using namespace boost::program_options;
using namespace std;
auto control(FairMQDevice& device) -> void
{
device.SetTransport("zeromq");
for (const auto event : {
FairMQDevice::INIT_DEVICE,
FairMQDevice::RESET_DEVICE,
FairMQDevice::END,
}) {
device.ChangeState(event);
if (event != FairMQDevice::END) device.WaitForEndOfState(event);
}
}
TEST(PluginManager, LoadPlugin)
{
auto config = FairMQProgOptions{};
FairMQDevice device{};
auto mgr = PluginManager{};
mgr.EmplacePluginServices(config, device);
mgr.PrependSearchPath("./lib");
ASSERT_NO_THROW(mgr.LoadPlugin("test_dummy"));
ASSERT_NO_THROW(mgr.LoadPlugin("test_dummy2"));
ASSERT_NO_THROW(mgr.InstantiatePlugins());
// check order
auto expected = vector<string>{"test_dummy", "test_dummy2"};
const auto expected = vector<string>{"test_dummy", "test_dummy2"};
auto actual = vector<string>{};
mgr.ForEachPlugin([&](Plugin& plugin){ actual.push_back(plugin.GetName()); });
ASSERT_TRUE(actual == expected);
@ -38,6 +60,8 @@ TEST(PluginManager, LoadPlugin)
auto count = 0;
mgr.ForEachPluginProgOptions([&count](const options_description& d){ ++count; });
ASSERT_EQ(count, 1);
control(device);
}
TEST(PluginManager, Factory)

6
fairmq/test/plugins/_plugin_manager_static.cxx
View File

@ -8,6 +8,8 @@
#include <gtest/gtest.h>
#include <fairmq/PluginManager.h>
#include <FairMQDevice.h>
#include <options/FairMQProgOptions.h>
#include <FairMQLogger.h>
#include <vector>
@ -21,8 +23,8 @@ TEST(PluginManager, LoadPluginStatic)
{
auto mgr = PluginManager{};
ASSERT_NO_THROW(mgr.LoadPlugin("s:test_dummy"));
ASSERT_NO_THROW(mgr.LoadPlugin("s:test_dummy2"));
ASSERT_NO_THROW(mgr.LoadPlugin("p:test_dummy"));
ASSERT_NO_THROW(mgr.LoadPlugin("p:test_dummy2"));
}
} /* namespace */

101
fairmq/tools/runSimpleMQStateMachine.h
View File

@ -9,8 +9,6 @@
#define RUNSIMPLEMQSTATEMACHINE_H
#include "FairMQLogger.h"
#include "FairMQConfigPlugin.h"
#include "FairMQControlPlugin.h"
#include "options/FairMQParser.h"
#include "options/FairMQProgOptions.h"
@ -47,12 +45,6 @@ inline int runStateMachine(TMQDevice& device, FairMQProgOptions& cfg)
std::string config = cfg.GetValue<std::string>("config");
std::string control = cfg.GetValue<std::string>("control");
// plugin objects
void* ldConfigHandle = nullptr;
void* ldControlHandle = nullptr;
FairMQConfigPlugin* fairmqConfigPlugin = nullptr;
FairMQControlPlugin* fairmqControlPlugin = nullptr;
std::clock_t cStart = std::clock();
auto tStart = std::chrono::high_resolution_clock::now();
@ -60,75 +52,6 @@ inline int runStateMachine(TMQDevice& device, FairMQProgOptions& cfg)
// Wait for the binding channels to bind
device.WaitForInitialValidation();
if (config != "static")
{
LOG(DEBUG) << "Opening config plugin: " << config;
ldConfigHandle = dlopen(config.c_str(), RTLD_LAZY);
if (!ldConfigHandle)
{
LOG(ERROR) << "Cannot open library: " << dlerror();
return 1;
}
// load the fairmqConfigPlugin
dlerror();
fairmqConfigPlugin = static_cast<FairMQConfigPlugin*>(dlsym(ldConfigHandle, "fairmqConfigPlugin"));
const char* dlsymError = dlerror();
if (dlsymError)
{
LOG(ERROR) << "Cannot load fairmqConfigPlugin() from: " << dlsymError;
fairmqConfigPlugin = nullptr;
dlclose(ldConfigHandle);
return 1;
}
fairmqConfigPlugin->initConfig(device);
}
if (control != "interactive" && control != "static")
{
LOG(DEBUG) << "Opening control plugin: " << control;
ldControlHandle = dlopen(control.c_str(), RTLD_LAZY);
if (!ldControlHandle)
{
LOG(ERROR) << "Cannot open library: " << dlerror();
if (ldConfigHandle)
{
dlclose(ldConfigHandle);
}
return 1;
}
// load the fairmqControlPlugin
dlerror();
fairmqControlPlugin = static_cast<FairMQControlPlugin*>(dlsym(ldControlHandle, "fairmqControlPlugin"));
const char* dlsymError = dlerror();
if (dlsymError)
{
LOG(ERROR) << "Cannot load fairmqControlPlugin(): " << dlsymError;
fairmqControlPlugin = nullptr;
dlclose(ldControlHandle);
// also close the config plugin before quiting with error.
if (ldConfigHandle)
{
dlclose(ldConfigHandle);
}
return 1;
}
fairmqControlPlugin->initControl(device);
}
if (config != "static")
{
if (fairmqConfigPlugin)
{
fairmqConfigPlugin->handleInitialConfig(device);
}
}
device.WaitForEndOfState(TMQDevice::INIT_DEVICE);
std::clock_t cEnd = std::clock();
@ -163,30 +86,6 @@ inline int runStateMachine(TMQDevice& device, FairMQProgOptions& cfg)
}
else
{
if (fairmqControlPlugin)
{
fairmqControlPlugin->handleStateChanges(device);
}
}
if (config != "static")
{
if (fairmqConfigPlugin)
{
LOG(DEBUG) << "Closing FairMQConfigPlugin...";
fairmqConfigPlugin->stopConfig();
dlclose(ldConfigHandle);
}
}
if (control != "interactive" && control != "static")
{
if (fairmqControlPlugin)
{
LOG(DEBUG) << "Closing FairMQControlPlugin...";
fairmqControlPlugin->stopControl();
dlclose(ldControlHandle);
}
}
return 0;