Add config plugin class.

This commit is contained in:
Alexey Rybalchenko
2019-05-29 17:08:20 +02:00
committed by Dennis Klein
parent bf8ec968e7
commit cba6d19781
109 changed files with 1393 additions and 1433 deletions

View File

@@ -25,6 +25,8 @@ DeviceRunner::DeviceRunner(int argc, char* const argv[], bool printLogo)
auto DeviceRunner::Run() -> int
{
fPluginManager.LoadPlugin("s:config");
////// CALL HOOK ///////
fEvents.Emit<hooks::LoadPlugins>(*this);
////////////////////////
@@ -46,19 +48,50 @@ auto DeviceRunner::Run() -> int
fEvents.Emit<hooks::ModifyRawCmdLineArgs>(*this);
////////////////////////
if (fConfig.ParseAll(fRawCmdLineArgs, true)) {
fConfig.ParseAll(fRawCmdLineArgs, true);
if (fConfig.Count("help")) {
fConfig.PrintHelp();
return 0;
}
if (fPrintLogo) {
LOG(info) << std::endl
<< " ______ _ _______ _________ " << std::endl
<< " / ____/___ _(_)_______ |/ /_ __ \\ version " << FAIRMQ_GIT_VERSION << std::endl
<< " / /_ / __ `/ / ___/__ /|_/ /_ / / / build " << FAIRMQ_BUILD_TYPE << std::endl
<< " / __/ / /_/ / / / _ / / / / /_/ / " << FAIRMQ_REPO_URL << std::endl
<< " /_/ \\__,_/_/_/ /_/ /_/ \\___\\_\\ " << FAIRMQ_LICENSE << " © " << FAIRMQ_COPYRIGHT << std::endl;
if (fConfig.Count("print-options")) {
fConfig.PrintOptionsRaw();
return 0;
}
if (fConfig.Count("print-channels") || fConfig.Count("version")) {
fair::Logger::SetConsoleSeverity("nolog");
} else {
std::string severity = fConfig.GetProperty<std::string>("severity");
std::string logFile = fConfig.GetProperty<std::string>("log-to-file");
bool color = fConfig.GetProperty<bool>("color");
std::string verbosity = fConfig.GetProperty<std::string>("verbosity");
fair::Logger::SetVerbosity(verbosity);
if (logFile != "") {
fair::Logger::InitFileSink(severity, logFile);
fair::Logger::SetConsoleSeverity("nolog");
} else {
fair::Logger::SetConsoleColor(color);
fair::Logger::SetConsoleSeverity(severity);
}
if (fPrintLogo) {
LOG(info) << std::endl
<< " ______ _ _______ _________ " << std::endl
<< " / ____/___ _(_)_______ |/ /_ __ \\ version " << FAIRMQ_GIT_VERSION << std::endl
<< " / /_ / __ `/ / ___/__ /|_/ /_ / / / build " << FAIRMQ_BUILD_TYPE << std::endl
<< " / __/ / /_/ / / / _ / / / / /_/ / " << FAIRMQ_REPO_URL << std::endl
<< " /_/ \\__,_/_/_/ /_/ /_/ \\___\\_\\ " << FAIRMQ_LICENSE << " © " << FAIRMQ_COPYRIGHT << std::endl;
}
fConfig.PrintOptions();
}
fConfig.Notify();
////// CALL HOOK ///////
fEvents.Emit<hooks::InstantiateDevice>(*this);
////////////////////////
@@ -80,6 +113,7 @@ auto DeviceRunner::Run() -> int
// Handle --version
if (fConfig.Count("version")) {
std::cout << "FairMQ version: " << FAIRMQ_GIT_VERSION << std::endl;
std::cout << "User device version: " << fDevice->GetVersion() << std::endl;
fDevice->ChangeState(fair::mq::Transition::End);
return 0;

View File

@@ -11,9 +11,9 @@
#include <fairmq/EventManager.h>
#include <fairmq/PluginManager.h>
#include <fairmq/ProgOptions.h>
#include <FairMQDevice.h>
#include <FairMQLogger.h>
#include <options/FairMQProgOptions.h>
#include <functional>
#include <memory>
@@ -68,7 +68,7 @@ class DeviceRunner
}
std::vector<std::string> fRawCmdLineArgs;
FairMQProgOptions fConfig;
fair::mq::ProgOptions fConfig;
std::unique_ptr<FairMQDevice> fDevice;
PluginManager fPluginManager;
const bool fPrintLogo;

View File

@@ -14,7 +14,7 @@
#include <fairmq/Transports.h>
#include <FairMQLogger.h>
#include <FairMQParts.h>
#include <options/Properties.h>
#include <fairmq/Properties.h>
#include <FairMQMessage.h>
#include <boost/any.hpp>

View File

@@ -58,32 +58,28 @@ static map<int, Transition> backwardsCompatibilityChangeStateHelper =
FairMQDevice::FairMQDevice()
: FairMQDevice(nullptr, {0, 0, 0})
{
}
{}
FairMQDevice::FairMQDevice(FairMQProgOptions& config)
FairMQDevice::FairMQDevice(ProgOptions& config)
: FairMQDevice(&config, {0, 0, 0})
{
}
{}
FairMQDevice::FairMQDevice(const tools::Version version)
: FairMQDevice(nullptr, version)
{
}
{}
FairMQDevice::FairMQDevice(FairMQProgOptions& config, const tools::Version version)
FairMQDevice::FairMQDevice(ProgOptions& config, const tools::Version version)
: FairMQDevice(&config, version)
{
}
{}
FairMQDevice::FairMQDevice(FairMQProgOptions* config, const tools::Version version)
FairMQDevice::FairMQDevice(ProgOptions* config, const tools::Version version)
: fTransportFactory(nullptr)
, fTransports()
, fChannels()
, fInternalConfig(config ? nullptr : tools::make_unique<FairMQProgOptions>())
, fInternalConfig(config ? nullptr : tools::make_unique<ProgOptions>())
, fConfig(config ? config : fInternalConfig.get())
, fId()
, fDefaultTransportType(fair::mq::Transport::ZMQ)
, fId(DefaultId)
, fDefaultTransportType(DefaultTransportType)
, fStateMachine()
, fUninitializedBindingChannels()
, fUninitializedConnectingChannels()
@@ -96,8 +92,8 @@ FairMQDevice::FairMQDevice(FairMQProgOptions* config, const tools::Version versi
, fMultitransportMutex()
, fMultitransportProceed(false)
, fVersion(version)
, fRate(0.)
, fMaxRunRuntimeInS(0)
, fRate(DefaultRate)
, fMaxRunRuntimeInS(DefaultMaxRunTime)
, fRawCmdLineArgs()
{
SubscribeToNewTransition("device", [&](Transition transition) {
@@ -192,18 +188,18 @@ void FairMQDevice::InitWrapper()
{
fStateMachine.WaitForPendingState();
fId = fConfig->GetValue<string>("id");
fId = fConfig->GetProperty<string>("id", DefaultId);
Init();
fRate = fConfig->GetValue<float>("rate");
fMaxRunRuntimeInS = fConfig->GetValue<uint64_t>("max-run-time");
fRate = fConfig->GetProperty<float>("rate", DefaultRate);
fMaxRunRuntimeInS = fConfig->GetProperty<uint64_t>("max-run-time", DefaultMaxRunTime);
try {
fDefaultTransportType = fair::mq::TransportTypes.at(fConfig->GetValue<string>("transport"));
fDefaultTransportType = fair::mq::TransportTypes.at(fConfig->GetProperty<string>("transport", DefaultTransportName));
} catch (const exception& e) {
LOG(error) << "exception: " << e.what();
LOG(error) << "invalid transport type provided: " << fConfig->GetValue<string>("transport");
LOG(error) << "invalid transport type provided: " << fConfig->GetProperty<string>("transport", "not provided");
throw;
}
@@ -217,7 +213,7 @@ void FairMQDevice::InitWrapper()
LOG(debug) << "Setting '" << fair::mq::TransportNames.at(fDefaultTransportType) << "' as default transport for the device";
fTransportFactory = AddTransport(fDefaultTransportType);
string networkInterface = fConfig->GetValue<string>("network-interface");
string networkInterface = fConfig->GetProperty<string>("network-interface", DefaultNetworkInterface);
// Fill the uninitialized channel containers
for (auto& mi : fChannels) {
@@ -274,7 +270,7 @@ void FairMQDevice::BindWrapper()
void FairMQDevice::ConnectWrapper()
{
int initializationTimeoutInS = fConfig->GetValue<int>("initialization-timeout");
int initializationTimeoutInS = fConfig->GetProperty<int>("init-timeout", DefaultInitTimeout);
// go over the list of channels until all are initialized (and removed from the uninitialized list)
int numAttempts = 1;
@@ -288,7 +284,7 @@ void FairMQDevice::ConnectWrapper()
for (auto& chan : fUninitializedConnectingChannels) {
string key{"chans." + chan->GetPrefix() + "." + chan->GetIndex() + ".address"};
string newAddress = fConfig->GetValue<string>(key);
string newAddress = fConfig->GetProperty<string>(key);
if (newAddress != chan->GetAddress()) {
chan->UpdateAddress(newAddress);
}
@@ -399,7 +395,7 @@ bool FairMQDevice::AttachChannel(FairMQChannel& chan)
chan.UpdateAddress(newAddress);
// update address in the config, it could have been modified during binding
fConfig->SetValue({"chans." + chan.GetPrefix() + "." + chan.GetIndex() + ".address"}, newAddress);
fConfig->SetProperty({"chans." + chan.GetPrefix() + "." + chan.GetIndex() + ".address"}, newAddress);
}
return true;
@@ -710,7 +706,7 @@ shared_ptr<FairMQTransportFactory> FairMQDevice::AddTransport(fair::mq::Transpor
}
}
void FairMQDevice::SetConfig(FairMQProgOptions& config)
void FairMQDevice::SetConfig(ProgOptions& config)
{
fInternalConfig.reset();
fConfig = &config;

View File

@@ -19,7 +19,7 @@
#include <FairMQParts.h>
#include <FairMQUnmanagedRegion.h>
#include <FairMQLogger.h>
#include <options/FairMQProgOptions.h>
#include <fairmq/ProgOptions.h>
#include <vector>
#include <memory> // unique_ptr
@@ -83,17 +83,17 @@ class FairMQDevice
/// Default constructor
FairMQDevice();
/// Constructor with external FairMQProgOptions
FairMQDevice(FairMQProgOptions& config);
/// Constructor with external fair::mq::ProgOptions
FairMQDevice(fair::mq::ProgOptions& config);
/// Constructor that sets the version
FairMQDevice(const fair::mq::tools::Version version);
/// Constructor that sets the version and external FairMQProgOptions
FairMQDevice(FairMQProgOptions& config, const fair::mq::tools::Version version);
/// Constructor that sets the version and external fair::mq::ProgOptions
FairMQDevice(fair::mq::ProgOptions& config, const fair::mq::tools::Version version);
private:
FairMQDevice(FairMQProgOptions* config, const fair::mq::tools::Version version);
FairMQDevice(fair::mq::ProgOptions* config, const fair::mq::tools::Version version);
public:
/// Copy constructor (disabled)
@@ -296,9 +296,9 @@ class FairMQDevice
std::shared_ptr<FairMQTransportFactory> AddTransport(const fair::mq::Transport transport);
/// Assigns config to the device
void SetConfig(FairMQProgOptions& config);
void SetConfig(fair::mq::ProgOptions& config);
/// Get pointer to the config
FairMQProgOptions* GetConfig() const
fair::mq::ProgOptions* GetConfig() const
{
return fConfig;
}
@@ -399,23 +399,23 @@ class FairMQDevice
const fair::mq::tools::Version GetVersion() const { return fVersion; }
void SetNumIoThreads(int numIoThreads) { fConfig->SetValue<int>("io-threads", numIoThreads);}
int GetNumIoThreads() const { return fConfig->GetValue<int>("io-threads"); }
void SetNumIoThreads(int numIoThreads) { fConfig->SetProperty("io-threads", numIoThreads);}
int GetNumIoThreads() const { return fConfig->GetProperty<int>("io-threads", DefaultIOThreads); }
void SetNetworkInterface(const std::string& networkInterface) { fConfig->SetValue<std::string>("network-interface", networkInterface); }
std::string GetNetworkInterface() const { return fConfig->GetValue<std::string>("network-interface"); }
void SetNetworkInterface(const std::string& networkInterface) { fConfig->SetProperty("network-interface", networkInterface); }
std::string GetNetworkInterface() const { return fConfig->GetProperty<std::string>("network-interface", DefaultNetworkInterface); }
void SetDefaultTransport(const std::string& name) { fConfig->SetValue<std::string>("transport", name); }
std::string GetDefaultTransport() const { return fConfig->GetValue<std::string>("transport"); }
void SetDefaultTransport(const std::string& name) { fConfig->SetProperty("transport", name); }
std::string GetDefaultTransport() const { return fConfig->GetProperty<std::string>("transport", DefaultTransportName); }
void SetInitializationTimeoutInS(int initializationTimeoutInS) { fConfig->SetValue<int>("initialization-timeout", initializationTimeoutInS); }
int GetInitializationTimeoutInS() const { return fConfig->GetValue<int>("initialization-timeout"); }
void SetInitTimeoutInS(int initTimeoutInS) { fConfig->SetProperty("init-timeout", initTimeoutInS); }
int GetInitTimeoutInS() const { return fConfig->GetProperty<int>("init-timeout", DefaultInitTimeout); }
/// Sets the default transport for the device
/// @param transport Transport string ("zeromq"/"nanomsg"/"shmem")
void SetTransport(const std::string& transport) { fConfig->SetValue<std::string>("transport", transport); }
void SetTransport(const std::string& transport) { fConfig->SetProperty("transport", transport); }
/// Gets the default transport name
std::string GetTransportName() const { return fConfig->GetValue<std::string>("transport"); }
std::string GetTransportName() const { return fConfig->GetProperty<std::string>("transport", DefaultTransportName); }
void SetRawCmdLineArgs(const std::vector<std::string>& args) { fRawCmdLineArgs = args; }
std::vector<std::string> GetRawCmdLineArgs() const { return fRawCmdLineArgs; }
@@ -440,8 +440,8 @@ class FairMQDevice
public:
std::unordered_map<std::string, std::vector<FairMQChannel>> fChannels; ///< Device channels
std::unique_ptr<FairMQProgOptions> fInternalConfig; ///< Internal program options configuration
FairMQProgOptions* fConfig; ///< Pointer to config (internal or external)
std::unique_ptr<fair::mq::ProgOptions> fInternalConfig; ///< Internal program options configuration
fair::mq::ProgOptions* fConfig; ///< Pointer to config (internal or external)
void AddChannel(const std::string& name, FairMQChannel&& channel)
{
@@ -514,6 +514,16 @@ class FairMQDevice
struct DeviceStateError : std::runtime_error { using std::runtime_error::runtime_error; };
static constexpr const char* DefaultId = "";
static constexpr int DefaultIOThreads = 1;
static constexpr const char* DefaultTransportName = "zeromq";
static constexpr fair::mq::Transport DefaultTransportType = fair::mq::Transport::ZMQ;
static constexpr const char* DefaultNetworkInterface = "default";
static constexpr int DefaultInitTimeout = 120;
static constexpr uint64_t DefaultMaxRunTime = 0;
static constexpr float DefaultRate = 0.;
static constexpr const char* DefaultSession = "default";
private:
fair::mq::Transport fDefaultTransportType; ///< Default transport for the device
fair::mq::StateMachine fStateMachine;

View File

@@ -27,7 +27,7 @@ FairMQTransportFactory::FairMQTransportFactory(const std::string& id)
{
}
auto FairMQTransportFactory::CreateTransportFactory(const std::string& type, const std::string& id, const FairMQProgOptions* config) -> std::shared_ptr<FairMQTransportFactory>
auto FairMQTransportFactory::CreateTransportFactory(const std::string& type, const std::string& id, const fair::mq::ProgOptions* config) -> std::shared_ptr<FairMQTransportFactory>
{
using namespace std;

View File

@@ -12,6 +12,7 @@
#include <FairMQLogger.h>
#include <FairMQMessage.h>
#include <FairMQPoller.h>
#include <fairmq/ProgOptionsFwd.h>
#include <FairMQSocket.h>
#include <FairMQUnmanagedRegion.h>
#include <fairmq/MemoryResources.h>
@@ -23,7 +24,6 @@
#include <unordered_map>
class FairMQChannel;
class FairMQProgOptions;
class FairMQTransportFactory
{
@@ -83,7 +83,7 @@ class FairMQTransportFactory
virtual ~FairMQTransportFactory() {};
static auto CreateTransportFactory(const std::string& type, const std::string& id = "", const FairMQProgOptions* config = nullptr) -> std::shared_ptr<FairMQTransportFactory>;
static auto CreateTransportFactory(const std::string& type, const std::string& id = "", const fair::mq::ProgOptions* config = nullptr) -> std::shared_ptr<FairMQTransportFactory>;
static void FairMQNoCleanup(void* /*data*/, void* /*obj*/)
{

View File

@@ -6,13 +6,14 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/*
* File: FairMQParser.cxx
* File: JSONParser.cxx
* Author: winckler
*
* Created on May 14, 2015, 5:01 PM
*/
#include "FairMQParser.h"
#include <fairmq/PropertyOutput.h>
#include "JSONParser.h"
#include "FairMQLogger.h"
#include <fairmq/Tools.h>
@@ -20,7 +21,10 @@
#include <boost/property_tree/ptree.hpp>
#include <boost/any.hpp>
#include <ios>
using namespace std;
using namespace fair::mq;
using namespace fair::mq::tools;
using namespace boost::property_tree;
@@ -28,27 +32,27 @@ namespace fair
{
namespace mq
{
namespace parser
{
fair::mq::Properties ptreeToProperties(const ptree& pt, const string& id)
fair::mq::Properties PtreeParser(const ptree& pt, const string& id)
{
if (id == "") {
throw ParserError("no device ID provided. Provide with `--id` cmd option");
}
return Helper::DeviceParser(pt.get_child("fairMQOptions"), id);
// json_parser::write_json(cout, pt);
return helper::DeviceParser(pt.get_child("fairMQOptions"), id);
}
fair::mq::Properties JSON::UserParser(const string& filename, const string& deviceId)
fair::mq::Properties JSONParser(const string& filename, const string& deviceId)
{
ptree input;
ptree pt;
LOG(debug) << "Parsing JSON from " << filename << " ...";
read_json(filename, input);
return ptreeToProperties(input, deviceId);
read_json(filename, pt);
return PtreeParser(pt, deviceId);
}
namespace Helper
namespace helper
{
fair::mq::Properties DeviceParser(const ptree& fairMQOptions, const string& deviceId)
@@ -58,16 +62,8 @@ fair::mq::Properties DeviceParser(const ptree& fairMQOptions, const string& devi
for (const auto& node : fairMQOptions) {
if (node.first == "devices") {
for (const auto& device : node.second) {
string deviceIdKey;
// check if key is provided, otherwise use id
string key = device.second.get<string>("key", "");
if (key != "") {
deviceIdKey = key;
} else {
deviceIdKey = device.second.get<string>("id");
}
string deviceIdKey = device.second.get<string>("key", device.second.get<string>("id", ""));
// if not correct device id, do not fill MQMap
if (deviceId != deviceIdKey) {
@@ -75,7 +71,6 @@ fair::mq::Properties DeviceParser(const ptree& fairMQOptions, const string& devi
}
LOG(trace) << "Found following channels for device ID '" << deviceId << "' :";
ChannelParser(device.second, properties);
}
}
@@ -111,20 +106,9 @@ void ChannelParser(const ptree& tree, fair::mq::Properties& properties)
LOG(trace) << name << ":";
LOG(trace) << "\tnumSockets of " << numSockets << " specified, applying common settings to each:";
// TODO: make a loop out of this
LOG(trace) << "\ttype = " << boost::any_cast<string>(commonProperties.at("type"));
LOG(trace) << "\tmethod = " << boost::any_cast<string>(commonProperties.at("method"));
LOG(trace) << "\taddress = " << boost::any_cast<string>(commonProperties.at("address"));
LOG(trace) << "\ttransport = " << boost::any_cast<string>(commonProperties.at("transport"));
LOG(trace) << "\tsndBufSize = " << boost::any_cast<int>(commonProperties.at("sndBufSize"));
LOG(trace) << "\trcvBufSize = " << boost::any_cast<int>(commonProperties.at("rcvBufSize"));
LOG(trace) << "\tsndKernelSize = " << boost::any_cast<int>(commonProperties.at("sndKernelSize"));
LOG(trace) << "\trcvKernelSize = " << boost::any_cast<int>(commonProperties.at("rcvKernelSize"));
LOG(trace) << "\tlinger = " << boost::any_cast<int>(commonProperties.at("linger"));
LOG(trace) << "\trateLogging = " << boost::any_cast<int>(commonProperties.at("rateLogging"));
LOG(trace) << "\tportRangeMin = " << boost::any_cast<int>(commonProperties.at("portRangeMin"));
LOG(trace) << "\tportRangeMax = " << boost::any_cast<int>(commonProperties.at("portRangeMax"));
LOG(trace) << "\tautoBind = " << boost::any_cast<bool>(commonProperties.at("autoBind"));
for (auto& p : commonProperties) {
LOG(trace) << "\t" << setw(13) << left << p.first << " : " << p.second;
}
for (int i = 0; i < numSockets; ++i) {
for (const auto& p : commonProperties) {
@@ -166,20 +150,9 @@ void SubChannelParser(const ptree& channelTree, fair::mq::Properties& properties
newProperties["autoBind"] = sn.second.get<bool>("autoBind", boost::any_cast<bool>(commonProperties.at("autoBind")));
LOG(trace) << "" << channelName << "[" << i << "]:";
// TODO: make a loop out of this
LOG(trace) << "\ttype = " << boost::any_cast<string>(newProperties.at("type"));
LOG(trace) << "\tmethod = " << boost::any_cast<string>(newProperties.at("method"));
LOG(trace) << "\taddress = " << boost::any_cast<string>(newProperties.at("address"));
LOG(trace) << "\ttransport = " << boost::any_cast<string>(newProperties.at("transport"));
LOG(trace) << "\tsndBufSize = " << boost::any_cast<int>(newProperties.at("sndBufSize"));
LOG(trace) << "\trcvBufSize = " << boost::any_cast<int>(newProperties.at("rcvBufSize"));
LOG(trace) << "\tsndKernelSize = " << boost::any_cast<int>(newProperties.at("sndKernelSize"));
LOG(trace) << "\trcvKernelSize = " << boost::any_cast<int>(newProperties.at("rcvKernelSize"));
LOG(trace) << "\tlinger = " << boost::any_cast<int>(newProperties.at("linger"));
LOG(trace) << "\trateLogging = " << boost::any_cast<int>(newProperties.at("rateLogging"));
LOG(trace) << "\tportRangeMin = " << boost::any_cast<int>(newProperties.at("portRangeMin"));
LOG(trace) << "\tportRangeMax = " << boost::any_cast<int>(newProperties.at("portRangeMax"));
LOG(trace) << "\tautoBind = " << boost::any_cast<bool>(newProperties.at("autoBind"));
for (auto& p : newProperties) {
LOG(trace) << "\t" << setw(13) << left << p.first << " : " << p.second;
}
for (const auto& p : newProperties) {
properties.emplace(ToString("chans.", channelName, ".", i, ".", p.first), p.second);
@@ -187,31 +160,21 @@ void SubChannelParser(const ptree& channelTree, fair::mq::Properties& properties
++i;
}
}
} // end socket loop
}
if (i > 0) {
LOG(trace) << "Found " << i << " socket(s) in channel.";
} else {
// if no sockets are specified, apply common channel properties
LOG(trace) << "" << channelName << ":";
LOG(trace) << "\tNo sockets specified,";
LOG(trace) << "\tapplying common settings to the channel:";
fair::mq::Properties newProperties(commonProperties);
// TODO: make a loop out of this
LOG(trace) << "\ttype = " << boost::any_cast<string>(newProperties.at("type"));
LOG(trace) << "\tmethod = " << boost::any_cast<string>(newProperties.at("method"));
LOG(trace) << "\taddress = " << boost::any_cast<string>(newProperties.at("address"));
LOG(trace) << "\ttransport = " << boost::any_cast<string>(newProperties.at("transport"));
LOG(trace) << "\tsndBufSize = " << boost::any_cast<int>(newProperties.at("sndBufSize"));
LOG(trace) << "\trcvBufSize = " << boost::any_cast<int>(newProperties.at("rcvBufSize"));
LOG(trace) << "\tsndKernelSize = " << boost::any_cast<int>(newProperties.at("sndKernelSize"));
LOG(trace) << "\trcvKernelSize = " << boost::any_cast<int>(newProperties.at("rcvKernelSize"));
LOG(trace) << "\tlinger = " << boost::any_cast<int>(newProperties.at("linger"));
LOG(trace) << "\trateLogging = " << boost::any_cast<int>(newProperties.at("rateLogging"));
LOG(trace) << "\tportRangeMin = " << boost::any_cast<int>(newProperties.at("portRangeMin"));
LOG(trace) << "\tportRangeMax = " << boost::any_cast<int>(newProperties.at("portRangeMax"));
LOG(trace) << "\tautoBind = " << boost::any_cast<bool>(newProperties.at("autoBind"));
for (auto& p : newProperties) {
LOG(trace) << "\t" << setw(13) << left << p.first << " : " << p.second;
}
for (const auto& p : newProperties) {
properties.emplace(ToString("chans.", channelName, ".0.", p.first), p.second);
@@ -219,8 +182,6 @@ void SubChannelParser(const ptree& channelTree, fair::mq::Properties& properties
}
}
} // Helper namespace
} // namespace parser
} // helper namespace
} // namespace mq
} // namespace fair

View File

@@ -12,8 +12,8 @@
* Created on May 14, 2015, 5:01 PM
*/
#ifndef FAIRMQPARSER_H
#define FAIRMQPARSER_H
#ifndef FAIR_MQ_JSONPARSER_H
#define FAIR_MQ_JSONPARSER_H
#include <string>
#include <vector>
@@ -23,35 +23,29 @@
#include <boost/property_tree/ptree_fwd.hpp>
#include "FairMQChannel.h"
#include "Properties.h"
#include <fairmq/Properties.h>
namespace fair
{
namespace mq
{
namespace parser
{
struct ParserError : std::runtime_error { using std::runtime_error::runtime_error; };
fair::mq::Properties ptreeToProperties(const boost::property_tree::ptree& pt, const std::string& deviceId);
fair::mq::Properties PtreeParser(const boost::property_tree::ptree& pt, const std::string& deviceId);
struct JSON
{
fair::mq::Properties UserParser(const std::string& filename, const std::string& deviceId);
};
fair::mq::Properties JSONParser(const std::string& filename, const std::string& deviceId);
namespace Helper
namespace helper
{
fair::mq::Properties DeviceParser(const boost::property_tree::ptree& tree, const std::string& deviceId);
void ChannelParser(const boost::property_tree::ptree& tree, fair::mq::Properties& properties);
void SubChannelParser(const boost::property_tree::ptree& tree, fair::mq::Properties& properties, const std::string& channelName, const fair::mq::Properties& commonProperties);
} // Helper namespace
} // helper namespace
} // namespace parser
} // namespace mq
} // namespace fair
#endif /* FAIRMQPARSER_H */
#endif /* FAIR_MQ_JSONPARSER_H */

View File

@@ -85,13 +85,25 @@ class Plugin
// device config API
// see <fairmq/PluginServices.h> for docs
auto PropertyExists(const std::string& key) -> int { return fPluginServices->PropertyExists(key); }
template<typename T>
auto SetProperty(const std::string& key, T val) -> void { fPluginServices->SetProperty(key, val); }
T GetProperty(const std::string& key) const { return fPluginServices->GetProperty<T>(key); }
template<typename T>
auto GetProperty(const std::string& key) const -> T { return fPluginServices->GetProperty<T>(key); }
auto GetPropertyAsString(const std::string& key) const -> std::string { return fPluginServices->GetPropertyAsString(key); }
T GetProperty(const std::string& key, const T& ifNotFound) const { return fPluginServices->GetProperty(key, ifNotFound); }
std::string GetPropertyAsString(const std::string& key) const { return fPluginServices->GetPropertyAsString(key); }
std::string GetPropertyAsString(const std::string& key, const std::string& ifNotFound) const { return fPluginServices->GetPropertyAsString(key, ifNotFound); }
fair::mq::Properties GetProperties(const std::string& q) const { return fPluginServices->GetProperties(q); }
fair::mq::Properties GetPropertiesStartingWith(const std::string& q) const { return fPluginServices->GetPropertiesStartingWith(q); };
std::map<std::string, std::string> GetPropertiesAsString(const std::string& q) const { return fPluginServices->GetPropertiesAsString(q); }
std::map<std::string, std::string> GetPropertiesAsStringStartingWith(const std::string& q) const { return fPluginServices->GetPropertiesAsStringStartingWith(q); };
auto GetChannelInfo() const -> std::unordered_map<std::string, int> { return fPluginServices->GetChannelInfo(); }
auto GetPropertyKeys() const -> std::vector<std::string> { return fPluginServices->GetPropertyKeys(); }
template<typename T>
auto SetProperty(const std::string& key, T val) -> void { fPluginServices->SetProperty(key, val); }
void SetProperties(const fair::mq::Properties& props) { fPluginServices->SetProperties(props); }
template<typename T>
auto SubscribeToPropertyChange(std::function<void(const std::string& key, T newValue)> callback) -> void { fPluginServices->SubscribeToPropertyChange<T>(fkName, callback); }
template<typename T>

View File

@@ -49,13 +49,11 @@ fair::mq::PluginManager::PluginManager(const vector<string> args)
// Parse command line options
auto options = ProgramOptions();
auto vm = po::variables_map{};
try
{
try {
auto parsed = po::command_line_parser(args).options(options).allow_unregistered().run();
po::store(parsed, vm);
po::notify(vm);
} catch (const po::error& e)
{
} catch (const po::error& e) {
throw ProgramOptionsParseError{ToString("Error occured while parsing the 'Plugin Manager' program options: ", e.what())};
}
@@ -63,10 +61,8 @@ fair::mq::PluginManager::PluginManager(const vector<string> args)
auto append = vector<fs::path>{};
auto prepend = vector<fs::path>{};
auto searchPaths = vector<fs::path>{};
if (vm.count("plugin-search-path"))
{
for (const auto& path : vm["plugin-search-path"].as<vector<string>>())
{
if (vm.count("plugin-search-path")) {
for (const auto& path : vm["plugin-search-path"].as<vector<string>>()) {
if (path.substr(0, 1) == "<") { prepend.emplace_back(path.substr(1)); }
else if (path.substr(0, 1) == ">") { append.emplace_back(path.substr(1)); }
else { searchPaths.emplace_back(path); }
@@ -126,23 +122,16 @@ auto fair::mq::PluginManager::ProgramOptions() -> po::options_description
auto fair::mq::PluginManager::LoadPlugin(const string& pluginName) -> void
{
if (pluginName.substr(0,2) == "p:")
{
if (pluginName.substr(0,2) == "p:") {
// Mechanism A: prelinked dynamic
LoadPluginPrelinkedDynamic(pluginName.substr(2));
}
else if (pluginName.substr(0,2) == "d:")
{
} else if (pluginName.substr(0,2) == "d:") {
// Mechanism B: dynamic
LoadPluginDynamic(pluginName.substr(2));
}
else if (pluginName.substr(0,2) == "s:")
{
} else if (pluginName.substr(0,2) == "s:") {
// Mechanism C: static (builtin)
LoadPluginStatic(pluginName.substr(2));
}
else
{
} else {
// Mechanism B: dynamic (default)
LoadPluginDynamic(pluginName);
}
@@ -151,15 +140,11 @@ auto fair::mq::PluginManager::LoadPlugin(const string& pluginName) -> void
auto fair::mq::PluginManager::LoadPluginPrelinkedDynamic(const string& pluginName) -> void
{
// Load symbol
if (fPluginFactories.find(pluginName) == fPluginFactories.end())
{
try
{
if (fPluginFactories.find(pluginName) == fPluginFactories.end()) {
try {
LoadSymbols(pluginName, dll::program_location());
fPluginOrder.push_back(pluginName);
}
catch (boost::system::system_error& e)
{
} catch (boost::system::system_error& e) {
throw PluginLoadError(ToString("An error occurred while loading prelinked dynamic plugin ", pluginName, ": ", e.what()));
}
}
@@ -168,13 +153,11 @@ auto fair::mq::PluginManager::LoadPluginPrelinkedDynamic(const string& pluginNam
auto fair::mq::PluginManager::LoadPluginDynamic(const string& pluginName) -> void
{
// Search plugin and load, if found
if (fPluginFactories.find(pluginName) == fPluginFactories.end())
{
if (fPluginFactories.find(pluginName) == fPluginFactories.end()) {
auto success = false;
for (const auto& searchPath : SearchPaths()) {
try {
LoadSymbols(pluginName,
searchPath / ToString(LibPrefix(), pluginName),
LoadSymbols(pluginName, searchPath / ToString(LibPrefix(), pluginName),
dll::load_mode::append_decorations | dll::load_mode::rtld_global);
fPluginOrder.push_back(pluginName);
success = true;
@@ -213,15 +196,11 @@ auto fair::mq::PluginManager::LoadPluginDynamic(const string& pluginName) -> voi
auto fair::mq::PluginManager::LoadPluginStatic(const string& pluginName) -> void
{
// Load symbol
if (fPluginFactories.find(pluginName) == fPluginFactories.end())
{
try
{
if (fPluginFactories.find(pluginName) == fPluginFactories.end()) {
try {
LoadSymbols(pluginName, dll::program_location());
fPluginOrder.push_back(pluginName);
}
catch (boost::system::system_error& e)
{
} catch (boost::system::system_error& e) {
throw PluginLoadError(ToString("An error occurred while loading static plugin ", pluginName, ": ", e.what()));
}
}
@@ -229,22 +208,17 @@ auto fair::mq::PluginManager::LoadPluginStatic(const string& pluginName) -> void
auto fair::mq::PluginManager::InstantiatePlugin(const string& pluginName) -> void
{
if (fPlugins.find(pluginName) == fPlugins.end())
{
if (fPlugins.find(pluginName) == fPlugins.end()) {
fPlugins[pluginName] = fPluginFactories[pluginName](*fPluginServices);
}
}
auto fair::mq::PluginManager::InstantiatePlugins() -> void
{
for(const auto& pluginName : fPluginOrder)
{
try
{
for(const auto& pluginName : fPluginOrder) {
try {
InstantiatePlugin(pluginName);
}
catch (std::exception& e)
{
} catch (std::exception& e) {
throw PluginInstantiationError(ToString("An error occurred while instantiating plugin ", pluginName, ": ", e.what()));
}
}

View File

@@ -129,16 +129,11 @@ auto PluginServices::TakeDeviceControl(const std::string& controller) -> void
{
lock_guard<mutex> lock{fDeviceControllerMutex};
if (!fDeviceController)
{
if (!fDeviceController) {
fDeviceController = controller;
}
else if (fDeviceController == controller)
{
} else if (fDeviceController == controller) {
// nothing to do
}
else
{
} else {
throw DeviceControlError{tools::ToString(
"Plugin '", controller, "' is not allowed to take over control. ",
"Currently, plugin '", *fDeviceController, "' has taken control."
@@ -158,12 +153,9 @@ auto PluginServices::ReleaseDeviceControl(const std::string& controller) -> void
{
lock_guard<mutex> lock{fDeviceControllerMutex};
if (fDeviceController == controller)
{
if (fDeviceController == controller) {
fDeviceController = boost::none;
}
else
{
} else {
throw DeviceControlError{tools::ToString("Plugin '", controller, "' cannot release control because it has not taken over control.")};
}
}
@@ -182,8 +174,7 @@ auto PluginServices::WaitForReleaseDeviceControl() -> void
{
unique_lock<mutex> lock{fDeviceControllerMutex};
while (fDeviceController)
{
while (fDeviceController) {
fReleaseDeviceControlCondition.wait(lock);
}
}

View File

@@ -11,7 +11,8 @@
#include <fairmq/Tools.h>
#include <FairMQDevice.h>
#include <options/FairMQProgOptions.h>
#include <fairmq/ProgOptions.h>
#include <fairmq/Properties.h>
#include <boost/optional.hpp>
#include <boost/optional/optional_io.hpp>
@@ -20,6 +21,7 @@
#include <string>
#include <unordered_map>
#include <mutex>
#include <map>
#include <condition_variable>
#include <stdexcept>
@@ -39,7 +41,7 @@ class PluginServices
{
public:
PluginServices() = delete;
PluginServices(FairMQProgOptions& config, FairMQDevice& device)
PluginServices(ProgOptions& config, FairMQDevice& device)
: fConfig(config)
, fDevice(device)
, fDeviceController()
@@ -196,43 +198,76 @@ class PluginServices
|| (currentState == DeviceState::Binding)
|| (currentState == DeviceState::Bound)
|| (currentState == DeviceState::Connecting)
|| (currentState == DeviceState::Ready)
|| (currentState == DeviceState::Idle && key == "channel-config")) {
fConfig.SetValue(key, val);
|| (currentState == DeviceState::Ready)) {
fConfig.SetProperty(key, val);
} else {
throw InvalidStateError{
tools::ToString("PluginServices::SetProperty is not supported in device state ", currentState, ". ",
"Supported state is ", DeviceState::InitializingDevice, ".")};
}
}
void SetProperties(const fair::mq::Properties& props)
{
fConfig.SetProperties(props);
}
struct InvalidStateError : std::runtime_error { using std::runtime_error::runtime_error; };
/// @brief Read config property
/// @param key
/// @return config property value
/// @return config property
///
/// TODO Currently, if a non-existing key is requested and a default constructed object is returned.
/// This behaviour will be changed in the future to throw an exception in that case to provide a proper sentinel.
template<typename T>
auto GetProperty(const std::string& key) const -> T {
auto GetProperty(const std::string& key) const -> T
{
if (PropertyExists(key)) {
return fConfig.GetValue<T>(key);
return fConfig.GetProperty<T>(key);
}
throw PropertyNotFoundError(fair::mq::tools::ToString("Config has no key: ", key));
}
template<typename T>
T GetProperty(const std::string& key, const T& ifNotFound) const
{
return fConfig.GetProperty(key, ifNotFound);
}
/// @brief Read config property as string
/// @param key
/// @return config property value converted to string
/// @return config property converted to string
///
/// If a type is not supported, the user can provide support by overloading the ostream operator for this type
auto GetPropertyAsString(const std::string& key) const -> std::string {
auto GetPropertyAsString(const std::string& key) const -> std::string
{
if (PropertyExists(key)) {
return fConfig.GetStringValue(key);
return fConfig.GetPropertyAsString(key);
}
throw PropertyNotFoundError(fair::mq::tools::ToString("Config has no key: ", key));
}
auto GetPropertyAsString(const std::string& key, const std::string& ifNotFound) const -> std::string
{
return fConfig.GetPropertyAsString(key, ifNotFound);
}
fair::mq::Properties GetProperties(const std::string& q) const
{
return fConfig.GetProperties(q);
}
fair::mq::Properties GetPropertiesStartingWith(const std::string& q) const
{
return fConfig.GetPropertiesStartingWith(q);
}
std::map<std::string, std::string> GetPropertiesAsString(const std::string& q) const
{
return fConfig.GetPropertiesAsString(q);
}
std::map<std::string, std::string> GetPropertiesAsStringStartingWith(const std::string& q) const
{
return fConfig.GetPropertiesAsStringStartingWith(q);
}
auto GetChannelInfo() const -> std::unordered_map<std::string, int> { return fConfig.GetChannelInfo(); }
/// @brief Discover the list of property keys
@@ -282,7 +317,7 @@ class PluginServices
static const std::unordered_map<DeviceStateTransition, fair::mq::Transition, tools::HashEnum<DeviceStateTransition>> fkDeviceStateTransitionMap;
private:
FairMQProgOptions& fConfig;
fair::mq::ProgOptions& fConfig;
FairMQDevice& fDevice;
boost::optional<std::string> fDeviceController;
mutable std::mutex fDeviceControllerMutex;

404
fairmq/ProgOptions.cxx Normal file
View File

@@ -0,0 +1,404 @@
/********************************************************************************
* Copyright (C) 2014-2018 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" *
********************************************************************************/
/*
* File: ProgOptions.cxx
* Author: winckler
*
* Created on March 11, 2015, 10:20 PM
*/
#include "FairMQLogger.h"
#include <fairmq/ProgOptions.h>
#include "tools/Unique.h"
#include <boost/any.hpp>
#include <boost/regex.hpp>
#include <algorithm>
#include <iomanip>
#include <iostream>
#include <exception>
#include <utility> // pair
using namespace std;
using namespace fair::mq;
namespace po = boost::program_options;
struct ValInfo
{
string value;
string type;
string origin;
};
namespace fair
{
namespace mq
{
ValInfo ConvertVarValToValInfo(const po::variable_value& v)
{
string origin;
if (v.defaulted()) {
origin = "[default]";
} else if (v.empty()) {
origin = "[empty]";
} else {
origin = "[provided]";
}
try {
pair<string, string> info = PropertyHelper::GetPropertyInfo(v.value());
return {info.first, info.second, origin};
} catch (out_of_range& oor) {
return {string("[unidentified_type]"), string("[unidentified_type]"), origin};
}
};
string ConvertVarValToString(const po::variable_value& v)
{
return ConvertVarValToValInfo(v).value;
}
ProgOptions::ProgOptions()
: fVarMap()
, fAllOptions("FairMQ Command Line Options")
, fUnregisteredOptions()
, fEvents()
, fMtx()
{
fAllOptions.add_options()
("help,h", "Print help")
("version,v", "Print version")
("severity", po::value<string>()->default_value("debug"), "Log severity level: trace, debug, info, state, warn, error, fatal, nolog")
("verbosity", po::value<string>()->default_value("medium"), "Log verbosity level: veryhigh, high, medium, low")
("color", po::value<bool >()->default_value(true), "Log color (true/false)")
("log-to-file", po::value<string>()->default_value(""), "Log output to a file.")
("print-options", po::value<bool >()->implicit_value(true), "Print options in machine-readable format (<option>:<computed-value>:<type>:<description>)");
ParseDefaults();
}
void ProgOptions::ParseDefaults()
{
vector<string> emptyArgs = {"dummy"};
vector<const char*> argv(emptyArgs.size());
transform(emptyArgs.begin(), emptyArgs.end(), argv.begin(), [](const string& str) {
return str.c_str();
});
po::store(po::parse_command_line(argv.size(), const_cast<char**>(argv.data()), fAllOptions), fVarMap);
}
void ProgOptions::ParseAll(const vector<string>& cmdArgs, bool allowUnregistered)
{
vector<const char*> argv(cmdArgs.size());
transform(cmdArgs.begin(), cmdArgs.end(), argv.begin(), [](const string& str) {
return str.c_str();
});
ParseAll(argv.size(), const_cast<char**>(argv.data()), allowUnregistered);
}
void ProgOptions::ParseAll(const int argc, char const* const* argv, bool allowUnregistered)
{
// clear the container because it was filled with default values and subsequent calls to store() do not overwrite the existing values
fVarMap.clear();
if (allowUnregistered) {
po::command_line_parser parser(argc, argv);
parser.options(fAllOptions).allow_unregistered();
po::parsed_options parsed = parser.run();
fUnregisteredOptions = po::collect_unrecognized(parsed.options, po::include_positional);
po::store(parsed, fVarMap);
} else {
po::store(po::parse_command_line(argc, argv, fAllOptions), fVarMap);
}
}
void ProgOptions::Notify()
{
po::notify(fVarMap);
}
void ProgOptions::AddToCmdLineOptions(const po::options_description optDesc, bool /* visible */)
{
fAllOptions.add(optDesc);
}
po::options_description& ProgOptions::GetCmdLineOptions()
{
return fAllOptions;
}
int ProgOptions::Count(const string& key) const
{
lock_guard<mutex> lock(fMtx);
return fVarMap.count(key);
}
unordered_map<string, int> ProgOptions::GetChannelInfo() const
{
lock_guard<mutex> lock(fMtx);
return GetChannelInfoImpl();
}
unordered_map<string, int> ProgOptions::GetChannelInfoImpl() const
{
unordered_map<string, int> info;
boost::regex re("chans\\..*\\.type");
for (const auto& m : fVarMap) {
if (boost::regex_match(m.first, re)) {
string chan = m.first.substr(6);
string::size_type n = chan.find(".");
string chanName = chan.substr(0, n);
if (info.find(chanName) == info.end()) {
info.emplace(chanName, 1);
} else {
info[chanName] = info[chanName] + 1;
}
}
}
return info;
}
vector<string> ProgOptions::GetPropertyKeys() const
{
lock_guard<mutex> lock(fMtx);
vector<string> keys;
for (const auto& it : fVarMap) {
keys.push_back(it.first.c_str());
}
return keys;
}
string ProgOptions::GetPropertyAsString(const string& key) const
{
lock_guard<mutex> lock(fMtx);
if (fVarMap.count(key)) {
return ConvertVarValToString(fVarMap.at(key));
} else {
LOG(warn) << "Config has no key: " << key << ". Returning default constructed object.";
return string();
}
}
string ProgOptions::GetPropertyAsString(const string& key, const string& ifNotFound) const
{
lock_guard<mutex> lock(fMtx);
if (fVarMap.count(key)) {
return ConvertVarValToString(fVarMap.at(key));
}
return ifNotFound;
}
Properties ProgOptions::GetProperties(const string& q) const
{
boost::regex re(q);
Properties properties;
lock_guard<mutex> lock(fMtx);
for (const auto& m : fVarMap) {
if (boost::regex_match(m.first, re)) {
properties.emplace(m.first, m.second.value());
}
}
if (properties.size() == 0) {
LOG(warn) << "could not find anything with \"" << q << "\"";
}
return properties;
}
Properties ProgOptions::GetPropertiesStartingWith(const string& q) const
{
Properties properties;
lock_guard<mutex> lock(fMtx);
for (const auto& m : fVarMap) {
if (m.first.compare(0, q.length(), q) == 0) {
properties.emplace(m.first, m.second.value());
}
}
return properties;
}
map<string, string> ProgOptions::GetPropertiesAsString(const string& q) const
{
boost::regex re(q);
map<string, string> properties;
lock_guard<mutex> lock(fMtx);
for (const auto& m : fVarMap) {
if (boost::regex_match(m.first, re)) {
properties.emplace(m.first, PropertyHelper::ConvertPropertyToString(m.second.value()));
}
}
if (properties.size() == 0) {
LOG(warn) << "could not find anything with \"" << q << "\"";
}
return properties;
}
map<string, string> ProgOptions::GetPropertiesAsStringStartingWith(const string& q) const
{
map<string, string> properties;
lock_guard<mutex> lock(fMtx);
for (const auto& m : fVarMap) {
if (m.first.compare(0, q.length(), q) == 0) {
properties.emplace(m.first, PropertyHelper::ConvertPropertyToString(m.second.value()));
}
}
return properties;
}
void ProgOptions::SetProperties(const Properties& input)
{
unique_lock<mutex> lock(fMtx);
map<string, boost::program_options::variable_value>& vm = fVarMap;
for (const auto& m : input) {
vm[m.first].value() = m.second;
}
lock.unlock();
for (const auto& m : input) {
PropertyHelper::fEventEmitters.at(m.second.type())(fEvents, m.first, m.second);
fEvents.Emit<PropertyChangeAsString, string>(m.first, PropertyHelper::ConvertPropertyToString(m.second));
}
}
void ProgOptions::DeleteProperty(const string& key)
{
lock_guard<mutex> lock(fMtx);
map<string, boost::program_options::variable_value>& vm = fVarMap;
vm.erase(key);
}
void ProgOptions::AddChannel(const string& name, const FairMQChannel& channel)
{
lock_guard<mutex> lock(fMtx);
unordered_map<string, int> existingChannels = GetChannelInfoImpl();
int index = 0;
if (existingChannels.count(name) > 0) {
index = existingChannels.at(name);
}
string prefix = fair::mq::tools::ToString("chans.", name, ".", index, ".");
SetVarMapValue<string>(string(prefix + "type"), channel.GetType());
SetVarMapValue<string>(string(prefix + "method"), channel.GetMethod());
SetVarMapValue<string>(string(prefix + "address"), channel.GetAddress());
SetVarMapValue<string>(string(prefix + "transport"), channel.GetTransportName());
SetVarMapValue<int>(string(prefix + "sndBufSize"), channel.GetSndBufSize());
SetVarMapValue<int>(string(prefix + "rcvBufSize"), channel.GetRcvBufSize());
SetVarMapValue<int>(string(prefix + "sndKernelSize"), channel.GetSndKernelSize());
SetVarMapValue<int>(string(prefix + "rcvKernelSize"), channel.GetRcvKernelSize());
SetVarMapValue<int>(string(prefix + "linger"), channel.GetLinger());
SetVarMapValue<int>(string(prefix + "rateLogging"), channel.GetRateLogging());
SetVarMapValue<int>(string(prefix + "portRangeMin"), channel.GetPortRangeMin());
SetVarMapValue<int>(string(prefix + "portRangeMax"), channel.GetPortRangeMax());
SetVarMapValue<bool>(string(prefix + "autoBind"), channel.GetAutoBind());
}
void ProgOptions::PrintHelp()
{
cout << fAllOptions << endl;
}
void ProgOptions::PrintOptions()
{
map<string, ValInfo> mapinfo;
int maxLenKey = 0;
int maxLenValue = 0;
int maxLenType = 0;
int maxLenDefault = 0;
for (const auto& m : fVarMap) {
maxLenKey = max(maxLenKey, static_cast<int>(m.first.length()));
ValInfo valinfo = ConvertVarValToValInfo(m.second);
mapinfo[m.first] = valinfo;
maxLenValue = max(maxLenValue, static_cast<int>(valinfo.value.length()));
maxLenType = max(maxLenType, static_cast<int>(valinfo.type.length()));
maxLenDefault = max(maxLenDefault, static_cast<int>(valinfo.origin.length()));
}
if (maxLenValue > 100) {
maxLenValue = 100;
}
for (const auto& o : fUnregisteredOptions) {
LOG(debug) << "detected unregistered option: " << o;
}
stringstream ss;
ss << "Configuration: \n";
for (const auto& p : mapinfo) {
string type("<" + p.second.type + ">");
ss << setfill(' ') << left
<< setw(maxLenKey) << p.first << " = "
<< setw(maxLenValue) << p.second.value << " "
<< setw(maxLenType + 2) << type << " "
<< setw(maxLenDefault) << p.second.origin
<< "\n";
}
LOG(debug) << ss.str();
}
void ProgOptions::PrintOptionsRaw()
{
const vector<boost::shared_ptr<po::option_description>>& options = fAllOptions.options();
for (const auto& o : options) {
ValInfo value;
if (fVarMap.count(o->canonical_display_name())) {
value = ConvertVarValToValInfo(fVarMap[o->canonical_display_name()]);
}
string description = o->description();
replace(description.begin(), description.end(), '\n', ' ');
cout << o->long_name() << ":" << value.value << ":" << (value.type == "" ? "<unknown>" : value.type) << ":" << description << endl;
}
}
} // namespace mq
} // namespace fair

163
fairmq/ProgOptions.h Normal file
View File

@@ -0,0 +1,163 @@
/********************************************************************************
* Copyright (C) 2014-2018 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_PROGOPTIONS_H
#define FAIR_MQ_PROGOPTIONS_H
#include <fairmq/EventManager.h>
#include "FairMQLogger.h"
#include "FairMQChannel.h"
#include <fairmq/Properties.h>
#include <fairmq/Tools.h>
#include <fairmq/ProgOptionsFwd.h>
#include <boost/program_options.hpp>
#include <unordered_map>
#include <map>
#include <functional>
#include <string>
#include <vector>
#include <mutex>
#include <sstream>
#include <stdexcept>
namespace fair
{
namespace mq
{
class ProgOptions
{
public:
ProgOptions();
virtual ~ProgOptions() {}
void ParseAll(const std::vector<std::string>& cmdArgs, bool allowUnregistered);
void ParseAll(const int argc, char const* const* argv, bool allowUnregistered = true);
void Notify();
void AddToCmdLineOptions(const boost::program_options::options_description optDesc, bool visible = true);
boost::program_options::options_description& GetCmdLineOptions();
int Count(const std::string& key) const;
std::unordered_map<std::string, int> GetChannelInfo() const;
std::vector<std::string> GetPropertyKeys() const;
template<typename T>
T GetProperty(const std::string& key) const
{
std::lock_guard<std::mutex> lock(fMtx);
if (fVarMap.count(key)) {
return fVarMap[key].as<T>();
} else {
LOG(warn) << "Config has no key: " << key << ". Returning default constructed object.";
return T();
}
}
template<typename T>
T GetProperty(const std::string& key, const T& ifNotFound) const
{
std::lock_guard<std::mutex> lock(fMtx);
if (fVarMap.count(key)) {
return fVarMap[key].as<T>();
}
return ifNotFound;
}
std::string GetPropertyAsString(const std::string& key) const;
std::string GetPropertyAsString(const std::string& key, const std::string& ifNotFound) const;
fair::mq::Properties GetProperties(const std::string& q) const;
fair::mq::Properties GetPropertiesStartingWith(const std::string& q) const;
std::map<std::string, std::string> GetPropertiesAsString(const std::string& q) const;
std::map<std::string, std::string> GetPropertiesAsStringStartingWith(const std::string& q) const;
template<typename T>
void SetProperty(const std::string& key, T val)
{
std::unique_lock<std::mutex> lock(fMtx);
SetVarMapValue<typename std::decay<T>::type>(key, val);
lock.unlock();
fEvents.Emit<fair::mq::PropertyChange, typename std::decay<T>::type>(key, val);
fEvents.Emit<fair::mq::PropertyChangeAsString, std::string>(key, GetStringValue(key));
}
void SetProperties(const fair::mq::Properties& input);
void DeleteProperty(const std::string& key);
void AddChannel(const std::string& name, const FairMQChannel& channel);
template<typename T>
void Subscribe(const std::string& subscriber, std::function<void(typename fair::mq::PropertyChange::KeyType, T)> func) const
{
std::lock_guard<std::mutex> lock(fMtx);
static_assert(!std::is_same<T,const char*>::value || !std::is_same<T, char*>::value,
"In template member ProgOptions::Subscribe<T>(key,Lambda) the types const char* or char* for the calback signatures are not supported.");
fEvents.Subscribe<fair::mq::PropertyChange, T>(subscriber, func);
}
template<typename T>
void Unsubscribe(const std::string& subscriber) const
{
std::lock_guard<std::mutex> lock(fMtx);
fEvents.Unsubscribe<fair::mq::PropertyChange, T>(subscriber);
}
void SubscribeAsString(const std::string& subscriber, std::function<void(typename fair::mq::PropertyChange::KeyType, std::string)> func) const
{
std::lock_guard<std::mutex> lock(fMtx);
fEvents.Subscribe<fair::mq::PropertyChangeAsString, std::string>(subscriber, func);
}
void UnsubscribeAsString(const std::string& subscriber) const
{
std::lock_guard<std::mutex> lock(fMtx);
fEvents.Unsubscribe<fair::mq::PropertyChangeAsString, std::string>(subscriber);
}
void PrintHelp();
void PrintOptions();
void PrintOptionsRaw();
const boost::program_options::variables_map& GetVarMap() const { return fVarMap; }
template<typename T>
T GetValue(const std::string& key) const /* TODO: deprecate this */ { return GetProperty<T>(key); }
template<typename T>
int SetValue(const std::string& key, T val) /* TODO: deprecate this */ { SetProperty(key, val); return 0; }
std::string GetStringValue(const std::string& key) const /* TODO: deprecate this */ { return GetPropertyAsString(key); }
private:
void ParseDefaults();
std::unordered_map<std::string, int> GetChannelInfoImpl() const;
template<typename T>
void SetVarMapValue(const std::string& key, const T& val)
{
std::map<std::string, boost::program_options::variable_value>& vm = fVarMap;
vm[key].value() = boost::any(val);
}
boost::program_options::variables_map fVarMap; ///< options container
boost::program_options::options_description fAllOptions; ///< all options descriptions
std::vector<std::string> fUnregisteredOptions; ///< container with unregistered options
mutable fair::mq::EventManager fEvents;
mutable std::mutex fMtx;
};
} // namespace mq
} // namespace fair
#endif /* FAIR_MQ_PROGOPTIONS_H */

View File

@@ -5,24 +5,19 @@
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIR_MQ_PROPERTIES_H
#define FAIR_MQ_PROPERTIES_H
#include <boost/any.hpp>
#include <map>
#include <string>
#ifndef FAIR_MQ_PROGOPTIONSFWD_H
#define FAIR_MQ_PROGOPTIONSFWD_H
namespace fair
{
namespace mq
{
using Property = boost::any;
using Properties = std::map<std::string, Property>;
class ProgOptions;
}
}
#endif /* FAIR_MQ_PROPERTIES_H */
using FairMQProgOptions = fair::mq::ProgOptions;
#endif

125
fairmq/Properties.cxx Normal file
View File

@@ -0,0 +1,125 @@
/********************************************************************************
* Copyright (C) 2014-2018 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/Properties.h>
#include <fairmq/Tools.h>
#include <boost/filesystem.hpp>
using namespace std;
using namespace fair::mq::tools;
using boost::any_cast;
namespace fair
{
namespace mq
{
template<class T>
ostream& operator<<(ostream& os, const vector<T>& v)
{
for (unsigned int i = 0; i < v.size(); ++i) {
os << v[i];
if (i != v.size() - 1) {
os << ", ";
}
}
return os;
}
template<typename T>
pair<string, string> getString(const boost::any& v, const string& label)
{
return { to_string(any_cast<T>(v)), label };
}
template<typename T>
pair<string, string> getStringPair(const boost::any& v, const string& label)
{
stringstream ss;
ss << any_cast<T>(v);
return { ss.str(), label };
}
unordered_map<type_index, function<pair<string, string>(const Property&)>> PropertyHelper::fTypeInfos = {
{ type_index(typeid(char)), [](const Property& p) { return pair<string, string>{ string(1, any_cast<char>(p)), "char" }; } },
{ type_index(typeid(unsigned char)), [](const Property& p) { return pair<string, string>{ string(1, any_cast<unsigned char>(p)), "unsigned char" }; } },
{ type_index(typeid(string)), [](const Property& p) { return pair<string, string>{ any_cast<string>(p), "string" }; } },
{ type_index(typeid(int)), [](const Property& p) { return getString<int>(p, "int"); } },
{ type_index(typeid(size_t)), [](const Property& p) { return getString<size_t>(p, "size_t"); } },
{ type_index(typeid(uint32_t)), [](const Property& p) { return getString<uint32_t>(p, "uint32_t"); } },
{ type_index(typeid(uint64_t)), [](const Property& p) { return getString<uint64_t>(p, "uint64_t"); } },
{ type_index(typeid(long)), [](const Property& p) { return getString<long>(p, "long"); } },
{ type_index(typeid(long long)), [](const Property& p) { return getString<long long>(p, "long long"); } },
{ type_index(typeid(unsigned)), [](const Property& p) { return getString<unsigned>(p, "unsigned"); } },
{ type_index(typeid(unsigned long)), [](const Property& p) { return getString<unsigned long>(p, "unsigned long"); } },
{ type_index(typeid(unsigned long long)), [](const Property& p) { return getString<unsigned long long>(p, "unsigned long long"); } },
{ type_index(typeid(float)), [](const Property& p) { return getString<float>(p, "float"); } },
{ type_index(typeid(double)), [](const Property& p) { return getString<double>(p, "double"); } },
{ type_index(typeid(long double)), [](const Property& p) { return getString<long double>(p, "long double"); } },
{ type_index(typeid(bool)), [](const Property& p) { stringstream ss; ss << boolalpha << any_cast<bool>(p); return pair<string, string>{ ss.str(), "bool" }; } },
{ type_index(typeid(vector<bool>)), [](const Property& p) { stringstream ss; ss << boolalpha << any_cast<vector<bool>>(p); return pair<string, string>{ ss.str(), "vector<bool>>" }; } },
{ type_index(typeid(boost::filesystem::path)), [](const Property& p) { return getStringPair<boost::filesystem::path>(p, "boost::filesystem::path"); } },
{ type_index(typeid(vector<char>)), [](const Property& p) { return getStringPair<vector<char>>(p, "vector<char>"); } },
{ type_index(typeid(vector<unsigned char>)), [](const Property& p) { return getStringPair<vector<unsigned char>>(p, "vector<unsigned char>"); } },
{ type_index(typeid(vector<string>)), [](const Property& p) { return getStringPair<vector<string>>(p, "vector<string>"); } },
{ type_index(typeid(vector<int>)), [](const Property& p) { return getStringPair<vector<int>>(p, "vector<int>"); } },
{ type_index(typeid(vector<size_t>)), [](const Property& p) { return getStringPair<vector<size_t>>(p, "vector<size_t>"); } },
{ type_index(typeid(vector<uint32_t>)), [](const Property& p) { return getStringPair<vector<uint32_t>>(p, "vector<uint32_t>"); } },
{ type_index(typeid(vector<uint64_t>)), [](const Property& p) { return getStringPair<vector<uint64_t>>(p, "vector<uint64_t>"); } },
{ type_index(typeid(vector<long>)), [](const Property& p) { return getStringPair<vector<long>>(p, "vector<long>"); } },
{ type_index(typeid(vector<long long>)), [](const Property& p) { return getStringPair<vector<long long>>(p, "vector<long long>"); } },
{ type_index(typeid(vector<unsigned>)), [](const Property& p) { return getStringPair<vector<unsigned>>(p, "vector<unsigned>"); } },
{ type_index(typeid(vector<unsigned long>)), [](const Property& p) { return getStringPair<vector<unsigned long>>(p, "vector<unsigned long>"); } },
{ type_index(typeid(vector<unsigned long long>)), [](const Property& p) { return getStringPair<vector<unsigned long long>>(p, "vector<unsigned long long>"); } },
{ type_index(typeid(vector<float>)), [](const Property& p) { return getStringPair<vector<float>>(p, "vector<float>"); } },
{ type_index(typeid(vector<double>)), [](const Property& p) { return getStringPair<vector<double>>(p, "vector<double>"); } },
{ type_index(typeid(vector<long double>)), [](const Property& p) { return getStringPair<vector<long double>>(p, "vector<long double>"); } },
{ type_index(typeid(vector<boost::filesystem::path>)), [](const Property& p) { return getStringPair<vector<boost::filesystem::path>>(p, "vector<boost::filesystem::path>"); } },
};
unordered_map<type_index, void(*)(const EventManager&, const string&, const Property&)> PropertyHelper::fEventEmitters = {
{ type_index(typeid(char)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, char>(k, any_cast<char>(p)); } },
{ type_index(typeid(unsigned char)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, unsigned char>(k, any_cast<unsigned char>(p)); } },
{ type_index(typeid(string)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, string>(k, any_cast<string>(p)); } },
{ type_index(typeid(int)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, int>(k, any_cast<int>(p)); } },
{ type_index(typeid(size_t)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, size_t>(k, any_cast<size_t>(p)); } },
{ type_index(typeid(uint32_t)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, uint32_t>(k, any_cast<uint32_t>(p)); } },
{ type_index(typeid(uint64_t)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, uint64_t>(k, any_cast<uint64_t>(p)); } },
{ type_index(typeid(long)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, long>(k, any_cast<long>(p)); } },
{ type_index(typeid(long long)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, long long>(k, any_cast<long long>(p)); } },
{ type_index(typeid(unsigned)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, unsigned>(k, any_cast<unsigned>(p)); } },
{ type_index(typeid(unsigned long)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, unsigned long>(k, any_cast<unsigned long>(p)); } },
{ type_index(typeid(unsigned long long)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, unsigned long long>(k, any_cast<unsigned long long>(p)); } },
{ type_index(typeid(float)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, float>(k, any_cast<float>(p)); } },
{ type_index(typeid(double)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, double>(k, any_cast<double>(p)); } },
{ type_index(typeid(long double)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, long double>(k, any_cast<long double>(p)); } },
{ type_index(typeid(bool)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, bool>(k, any_cast<bool>(p)); } },
{ type_index(typeid(vector<bool>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<bool>>(k, any_cast<vector<bool>>(p)); } },
{ type_index(typeid(boost::filesystem::path)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, boost::filesystem::path>(k, any_cast<boost::filesystem::path>(p)); } },
{ type_index(typeid(vector<char>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<char>>(k, any_cast<vector<char>>(p)); } },
{ type_index(typeid(vector<unsigned char>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<unsigned char>>(k, any_cast<vector<unsigned char>>(p)); } },
{ type_index(typeid(vector<string>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<string>>(k, any_cast<vector<string>>(p)); } },
{ type_index(typeid(vector<int>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<int>>(k, any_cast<vector<int>>(p)); } },
{ type_index(typeid(vector<size_t>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<size_t>>(k, any_cast<vector<size_t>>(p)); } },
{ type_index(typeid(vector<uint32_t>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<uint32_t>>(k, any_cast<vector<uint32_t>>(p)); } },
{ type_index(typeid(vector<uint64_t>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<uint64_t>>(k, any_cast<vector<uint64_t>>(p)); } },
{ type_index(typeid(vector<long>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<long>>(k, any_cast<vector<long>>(p)); } },
{ type_index(typeid(vector<long long>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<long long>>(k, any_cast<vector<long long>>(p)); } },
{ type_index(typeid(vector<unsigned>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<unsigned>>(k, any_cast<vector<unsigned>>(p)); } },
{ type_index(typeid(vector<unsigned long>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<unsigned long>>(k, any_cast<vector<unsigned long>>(p)); } },
{ type_index(typeid(vector<unsigned long long>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<unsigned long long>>(k, any_cast<vector<unsigned long long>>(p)); } },
{ type_index(typeid(vector<float>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<float>>(k, any_cast<vector<float>>(p)); } },
{ type_index(typeid(vector<double>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<double>>(k, any_cast<vector<double>>(p)); } },
{ type_index(typeid(vector<long double>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<long double>>(k, any_cast<vector<long double>>(p)); } },
{ type_index(typeid(vector<boost::filesystem::path>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<boost::filesystem::path>>(k, any_cast<vector<boost::filesystem::path>>(p)); } },
};
} // namespace mq
} // namespace fair

74
fairmq/Properties.h Normal file
View File

@@ -0,0 +1,74 @@
/********************************************************************************
* Copyright (C) 2014-2018 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_PROPERTIES_H
#define FAIR_MQ_PROPERTIES_H
#include <fairmq/EventManager.h>
#include <boost/any.hpp>
#include <boost/core/demangle.hpp>
#include <functional>
#include <map>
#include <unordered_map>
#include <string>
#include <typeindex>
#include <typeinfo>
#include <utility> // pair
namespace fair
{
namespace mq
{
using Property = boost::any;
using Properties = std::map<std::string, Property>;
struct PropertyChange : Event<std::string> {};
struct PropertyChangeAsString : Event<std::string> {};
class PropertyHelper
{
public:
template<typename T>
static void AddType(std::string label = "")
{
if (label == "") {
label = boost::core::demangle(typeid(T).name());
}
fTypeInfos[std::type_index(typeid(T))] = [label](const fair::mq::Property& p) {
std::stringstream ss;
ss << boost::any_cast<T>(p);
return std::pair<std::string, std::string>{ss.str(), label};
};
}
static std::string ConvertPropertyToString(const Property& p)
{
return fTypeInfos.at(p.type())(p).first;
}
// returns <valueAsString, typenameAsString>
static std::pair<std::string, std::string> GetPropertyInfo(const Property& p)
{
try {
return fTypeInfos.at(p.type())(p);
} catch (std::out_of_range& oor) {
return {"[unidentified_type]", "[unidentified_type]"};
}
}
static std::unordered_map<std::type_index, void(*)(const fair::mq::EventManager&, const std::string&, const fair::mq::Property&)> fEventEmitters;
private:
static std::unordered_map<std::type_index, std::function<std::pair<std::string, std::string>(const fair::mq::Property&)>> fTypeInfos;
};
}
}
#endif /* FAIR_MQ_PROPERTIES_H */

20
fairmq/PropertyOutput.h Normal file
View File

@@ -0,0 +1,20 @@
/********************************************************************************
* Copyright (C) 2014-2018 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_PROPERTYOUT_H
#define FAIR_MQ_PROPERTYOUT_H
#include <fairmq/Properties.h>
#include <sstream>
inline std::ostream& operator<<(std::ostream& os, const fair::mq::Property& p)
{
return os << fair::mq::PropertyHelper::GetPropertyInfo(p).first;
}
#endif /* FAIR_MQ_PROPERTYOUT_H */

View File

@@ -6,12 +6,12 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/// @file FairMQSuboptParser.cxx
/// @file SuboptParser.cxx
/// @author Matthias.Richter@scieq.net
/// @since 2017-03-30
/// @brief Parser implementation for key-value subopt format
#include "FairMQSuboptParser.h"
#include <fairmq/SuboptParser.h>
#include <boost/property_tree/ptree.hpp>
#include <cstring>
@@ -24,12 +24,47 @@ namespace fair
{
namespace mq
{
namespace parser
enum channelOptionKeyIds
{
NAME = 0, // name of the channel
TYPE, // push, pull, publish, subscribe, etc
METHOD, // bind or connect
ADDRESS, // host, protocol and port address
TRANSPORT, //
SNDBUFSIZE, // size of the send queue
RCVBUFSIZE, // size of the receive queue
SNDKERNELSIZE,
RCVKERNELSIZE,
LINGER,
RATELOGGING, // logging rate
PORTRANGEMIN,
PORTRANGEMAX,
AUTOBIND,
NUMSOCKETS,
lastsocketkey
};
constexpr const char* SUBOPT::channelOptionKeys[];
constexpr static const char* channelOptionKeys[] = {
/*[NAME] = */ "name",
/*[TYPE] = */ "type",
/*[METHOD] = */ "method",
/*[ADDRESS] = */ "address",
/*[TRANSPORT] = */ "transport",
/*[SNDBUFSIZE] = */ "sndBufSize",
/*[RCVBUFSIZE] = */ "rcvBufSize",
/*[SNDKERNELSIZE] = */ "sndKernelSize",
/*[RCVKERNELSIZE] = */ "rcvKernelSize",
/*[LINGER] = */ "linger",
/*[RATELOGGING] = */ "rateLogging",
/*[PORTRANGEMIN] = */ "portRangeMin",
/*[PORTRANGEMAX] = */ "portRangeMax",
/*[AUTOBIND] = */ "autoBind",
/*[NUMSOCKETS] = */ "numSockets",
nullptr
};
fair::mq::Properties SUBOPT::UserParser(const vector<string>& channelConfig, const string& deviceId)
Properties SuboptParser(const vector<string>& channelConfig, const string& deviceId)
{
ptree pt;
@@ -78,9 +113,8 @@ fair::mq::Properties SUBOPT::UserParser(const vector<string>& channelConfig, con
pt.add_child("fairMQOptions.devices", devicesArray);
return ptreeToProperties(pt, deviceId);
return PtreeParser(pt, deviceId);
}
}
}
}

49
fairmq/SuboptParser.h Normal file
View File

@@ -0,0 +1,49 @@
/********************************************************************************
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public License (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/// @file FairMQSuboptParser.h
/// @author Matthias.Richter@scieq.net
/// @since 2017-03-30
/// @brief Parser implementation for key-value subopt format
#ifndef FAIR_MQ_SUBOPTPARSER_H
#define FAIR_MQ_SUBOPTPARSER_H
#include <fairmq/JSONParser.h>
#include <vector>
#include <string>
namespace fair
{
namespace mq
{
/**
* A parser implementation for FairMQ channel properties.
* The parser handles a comma separated key=value list format by using the
* getsubopt function of the standard library.
*
* The option key '--channel-config' can be used with the list of key/value
* pairs like e.g.
* <pre>
* --channel-config name=output,type=push,method=bind
* </pre>
*
* The FairMQ option parser defines a 'UserParser' function for different
* formats. Currently it is strictly parsing channel options, but in general
* the concept is extensible by renaming UserParser to ChannelPropertyParser
* and introducing additional parser functions.
*/
Properties SuboptParser(const std::vector<std::string>& channelConfig, const std::string& deviceId);
}
}
#endif /* FAIR_MQ_SUBOPTPARSER_H */

View File

@@ -10,7 +10,6 @@
#include <fairmq/Tools.h>
#include "../FairMQLogger.h"
#include "../options/FairMQProgOptions.h"
#include <vector>
#include <chrono>
@@ -34,12 +33,12 @@ FairMQBenchmarkSampler::~FairMQBenchmarkSampler()
void FairMQBenchmarkSampler::InitTask()
{
fMultipart = fConfig->GetValue<bool>("multipart");
fNumParts = fConfig->GetValue<size_t>("num-parts");
fMsgSize = fConfig->GetValue<size_t>("msg-size");
fMsgRate = fConfig->GetValue<float>("msg-rate");
fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
fOutChannelName = fConfig->GetValue<string>("out-channel");
fMultipart = fConfig->GetProperty<bool>("multipart");
fNumParts = fConfig->GetProperty<size_t>("num-parts");
fMsgSize = fConfig->GetProperty<size_t>("msg-size");
fMsgRate = fConfig->GetProperty<float>("msg-rate");
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
fOutChannelName = fConfig->GetProperty<string>("out-channel");
}
void FairMQBenchmarkSampler::Run()

View File

@@ -15,7 +15,6 @@
#include "FairMQMerger.h"
#include "../FairMQLogger.h"
#include "../FairMQPoller.h"
#include "../options/FairMQProgOptions.h"
using namespace std;
@@ -40,9 +39,9 @@ FairMQMerger::~FairMQMerger()
void FairMQMerger::InitTask()
{
fMultipart = fConfig->GetValue<bool>("multipart");
fInChannelName = fConfig->GetValue<string>("in-channel");
fOutChannelName = fConfig->GetValue<string>("out-channel");
fMultipart = fConfig->GetProperty<bool>("multipart");
fInChannelName = fConfig->GetProperty<string>("in-channel");
fOutChannelName = fConfig->GetProperty<string>("out-channel");
}
void FairMQMerger::Run()

View File

@@ -9,7 +9,6 @@
#include "FairMQMultiplier.h"
#include "../FairMQLogger.h"
#include "../options/FairMQProgOptions.h"
using namespace std;
@@ -27,9 +26,9 @@ FairMQMultiplier::~FairMQMultiplier()
void FairMQMultiplier::InitTask()
{
fMultipart = fConfig->GetValue<bool>("multipart");
fInChannelName = fConfig->GetValue<string>("in-channel");
fOutChannelNames = fConfig->GetValue<vector<string>>("out-channel");
fMultipart = fConfig->GetProperty<bool>("multipart");
fInChannelName = fConfig->GetProperty<string>("in-channel");
fOutChannelNames = fConfig->GetProperty<vector<string>>("out-channel");
fNumOutputs = fChannels.at(fOutChannelNames.at(0)).size();
if (fMultipart)

View File

@@ -15,7 +15,6 @@
#include "FairMQProxy.h"
#include "../FairMQLogger.h"
#include "../options/FairMQProgOptions.h"
using namespace std;
@@ -32,9 +31,9 @@ FairMQProxy::~FairMQProxy()
void FairMQProxy::InitTask()
{
fMultipart = fConfig->GetValue<bool>("multipart");
fInChannelName = fConfig->GetValue<string>("in-channel");
fOutChannelName = fConfig->GetValue<string>("out-channel");
fMultipart = fConfig->GetProperty<bool>("multipart");
fInChannelName = fConfig->GetProperty<string>("in-channel");
fOutChannelName = fConfig->GetProperty<string>("out-channel");
}
void FairMQProxy::Run()

View File

@@ -20,7 +20,6 @@
#include "../FairMQDevice.h"
#include "../FairMQLogger.h"
#include "../options/FairMQProgOptions.h"
// template<typename OutputPolicy>
class FairMQSink : public FairMQDevice//, public OutputPolicy
@@ -44,9 +43,9 @@ class FairMQSink : public FairMQDevice//, public OutputPolicy
virtual void InitTask()
{
fMultipart = fConfig->GetValue<bool>("multipart");
fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
fInChannelName = fConfig->GetValue<std::string>("in-channel");
fMultipart = fConfig->GetProperty<bool>("multipart");
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
fInChannelName = fConfig->GetProperty<std::string>("in-channel");
}
virtual void Run()

View File

@@ -15,7 +15,6 @@
#include "FairMQSplitter.h"
#include "../FairMQLogger.h"
#include "../options/FairMQProgOptions.h"
using namespace std;
@@ -34,9 +33,9 @@ FairMQSplitter::~FairMQSplitter()
void FairMQSplitter::InitTask()
{
fMultipart = fConfig->GetValue<bool>("multipart");
fInChannelName = fConfig->GetValue<string>("in-channel");
fOutChannelName = fConfig->GetValue<string>("out-channel");
fMultipart = fConfig->GetProperty<bool>("multipart");
fInChannelName = fConfig->GetProperty<string>("in-channel");
fOutChannelName = fConfig->GetProperty<string>("out-channel");
fNumOutputs = fChannels.at(fOutChannelName).size();
fDirection = 0;

View File

@@ -17,7 +17,7 @@ using namespace std;
fair::mq::Transport FairMQTransportFactoryNN::fTransportType = fair::mq::Transport::NN;
FairMQTransportFactoryNN::FairMQTransportFactoryNN(const string& id, const FairMQProgOptions* /*config*/)
FairMQTransportFactoryNN::FairMQTransportFactoryNN(const string& id, const fair::mq::ProgOptions* /*config*/)
: FairMQTransportFactory(id)
{
LOG(debug) << "Transport: Using nanomsg library";

View File

@@ -14,7 +14,7 @@
#include "FairMQSocketNN.h"
#include "FairMQPollerNN.h"
#include "FairMQUnmanagedRegionNN.h"
#include <options/FairMQProgOptions.h>
#include <fairmq/ProgOptions.h>
#include <vector>
#include <string>
@@ -22,7 +22,7 @@
class FairMQTransportFactoryNN final : public FairMQTransportFactory
{
public:
FairMQTransportFactoryNN(const std::string& id = "", const FairMQProgOptions* config = nullptr);
FairMQTransportFactoryNN(const std::string& id = "", const fair::mq::ProgOptions* config = nullptr);
~FairMQTransportFactoryNN() override;
FairMQMessagePtr CreateMessage() override;

View File

@@ -23,14 +23,14 @@ namespace ofi
using namespace std;
TransportFactory::TransportFactory(const string& id, const FairMQProgOptions* config)
TransportFactory::TransportFactory(const string& id, const fair::mq::ProgOptions* config)
try : FairMQTransportFactory(id)
, fContext(*this, *this, 1)
{
LOG(debug) << "OFI transport: asiofi (" << fContext.GetAsiofiVersion() << ")";
if (config) {
fContext.SetSizeHint(config->GetValue<size_t>("ofi-size-hint"));
fContext.SetSizeHint(config->GetProperty<size_t>("ofi-size-hint", 0));
}
} catch (ContextError& e) {
throw TransportFactoryError{e.what()};

View File

@@ -10,7 +10,7 @@
#define FAIR_MQ_OFI_TRANSPORTFACTORY_H
#include <FairMQTransportFactory.h>
#include <options/FairMQProgOptions.h>
#include <fairmq/ProgOptions.h>
#include <fairmq/ofi/Context.h>
#include <asiofi.hpp>
@@ -31,7 +31,7 @@ namespace ofi
class TransportFactory final : public FairMQTransportFactory
{
public:
TransportFactory(const std::string& id = "", const FairMQProgOptions* config = nullptr);
TransportFactory(const std::string& id = "", const fair::mq::ProgOptions* config = nullptr);
TransportFactory(const TransportFactory&) = delete;
TransportFactory operator=(const TransportFactory&) = delete;

View File

@@ -1,596 +0,0 @@
/********************************************************************************
* Copyright (C) 2014-2018 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" *
********************************************************************************/
/*
* File: FairMQProgOptions.cxx
* Author: winckler
*
* Created on March 11, 2015, 10:20 PM
*/
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
#include "FairMQParser.h"
#include "FairMQSuboptParser.h"
#include "tools/Unique.h"
#include <boost/filesystem.hpp>
#include <boost/any.hpp>
#include <boost/algorithm/string.hpp> // join/split
#include <boost/regex.hpp>
#include <algorithm>
#include <iomanip>
#include <iostream>
#include <exception>
using namespace std;
using namespace fair::mq;
using boost::any_cast;
namespace po = boost::program_options;
struct ValInfo
{
string value;
string type;
string origin;
};
template<class T>
ostream& operator<<(ostream& os, const vector<T>& v)
{
for (unsigned int i = 0; i < v.size(); ++i) {
os << v[i];
if (i != v.size() - 1) {
os << ", ";
}
}
return os;
}
template<typename T>
pair<string, string> getString(const boost::any& v, const string& label)
{
return { to_string(any_cast<T>(v)), label };
}
template<typename T>
pair<string, string> getStringPair(const boost::any& v, const string& label)
{
stringstream ss;
ss << any_cast<T>(v);
return { ss.str(), label };
}
unordered_map<type_index, function<pair<string, string>(const Property&)>> FairMQProgOptions::fTypeInfos = {
{ type_index(typeid(char)), [](const Property& p) { return pair<string, string>{ string(1, any_cast<char>(p)), "char" }; } },
{ type_index(typeid(unsigned char)), [](const Property& p) { return pair<string, string>{ string(1, any_cast<unsigned char>(p)), "unsigned char" }; } },
{ type_index(typeid(string)), [](const Property& p) { return pair<string, string>{ any_cast<string>(p), "string" }; } },
{ type_index(typeid(int)), [](const Property& p) { return getString<int>(p, "int"); } },
{ type_index(typeid(size_t)), [](const Property& p) { return getString<size_t>(p, "size_t"); } },
{ type_index(typeid(uint32_t)), [](const Property& p) { return getString<uint32_t>(p, "uint32_t"); } },
{ type_index(typeid(uint64_t)), [](const Property& p) { return getString<uint64_t>(p, "uint64_t"); } },
{ type_index(typeid(long)), [](const Property& p) { return getString<long>(p, "long"); } },
{ type_index(typeid(long long)), [](const Property& p) { return getString<long long>(p, "long long"); } },
{ type_index(typeid(unsigned)), [](const Property& p) { return getString<unsigned>(p, "unsigned"); } },
{ type_index(typeid(unsigned long)), [](const Property& p) { return getString<unsigned long>(p, "unsigned long"); } },
{ type_index(typeid(unsigned long long)), [](const Property& p) { return getString<unsigned long long>(p, "unsigned long long"); } },
{ type_index(typeid(float)), [](const Property& p) { return getString<float>(p, "float"); } },
{ type_index(typeid(double)), [](const Property& p) { return getString<double>(p, "double"); } },
{ type_index(typeid(long double)), [](const Property& p) { return getString<long double>(p, "long double"); } },
{ type_index(typeid(bool)), [](const Property& p) { stringstream ss; ss << boolalpha << any_cast<bool>(p); return pair<string, string>{ ss.str(), "bool" }; } },
{ type_index(typeid(vector<bool>)), [](const Property& p) { stringstream ss; ss << boolalpha << any_cast<vector<bool>>(p); return pair<string, string>{ ss.str(), "vector<bool>>" }; } },
{ type_index(typeid(boost::filesystem::path)), [](const Property& p) { return getStringPair<boost::filesystem::path>(p, "boost::filesystem::path"); } },
{ type_index(typeid(vector<char>)), [](const Property& p) { return getStringPair<vector<char>>(p, "vector<char>"); } },
{ type_index(typeid(vector<unsigned char>)), [](const Property& p) { return getStringPair<vector<unsigned char>>(p, "vector<unsigned char>"); } },
{ type_index(typeid(vector<string>)), [](const Property& p) { return getStringPair<vector<string>>(p, "vector<string>"); } },
{ type_index(typeid(vector<int>)), [](const Property& p) { return getStringPair<vector<int>>(p, "vector<int>"); } },
{ type_index(typeid(vector<size_t>)), [](const Property& p) { return getStringPair<vector<size_t>>(p, "vector<size_t>"); } },
{ type_index(typeid(vector<uint32_t>)), [](const Property& p) { return getStringPair<vector<uint32_t>>(p, "vector<uint32_t>"); } },
{ type_index(typeid(vector<uint64_t>)), [](const Property& p) { return getStringPair<vector<uint64_t>>(p, "vector<uint64_t>"); } },
{ type_index(typeid(vector<long>)), [](const Property& p) { return getStringPair<vector<long>>(p, "vector<long>"); } },
{ type_index(typeid(vector<long long>)), [](const Property& p) { return getStringPair<vector<long long>>(p, "vector<long long>"); } },
{ type_index(typeid(vector<unsigned>)), [](const Property& p) { return getStringPair<vector<unsigned>>(p, "vector<unsigned>"); } },
{ type_index(typeid(vector<unsigned long>)), [](const Property& p) { return getStringPair<vector<unsigned long>>(p, "vector<unsigned long>"); } },
{ type_index(typeid(vector<unsigned long long>)), [](const Property& p) { return getStringPair<vector<unsigned long long>>(p, "vector<unsigned long long>"); } },
{ type_index(typeid(vector<float>)), [](const Property& p) { return getStringPair<vector<float>>(p, "vector<float>"); } },
{ type_index(typeid(vector<double>)), [](const Property& p) { return getStringPair<vector<double>>(p, "vector<double>"); } },
{ type_index(typeid(vector<long double>)), [](const Property& p) { return getStringPair<vector<long double>>(p, "vector<long double>"); } },
{ type_index(typeid(vector<boost::filesystem::path>)), [](const Property& p) { return getStringPair<vector<boost::filesystem::path>>(p, "vector<boost::filesystem::path>"); } },
};
unordered_map<type_index, void(*)(const EventManager&, const string&, const Property&)> FairMQProgOptions::fEventEmitters = {
{ type_index(typeid(char)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, char>(k, any_cast<char>(p)); } },
{ type_index(typeid(unsigned char)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, unsigned char>(k, any_cast<unsigned char>(p)); } },
{ type_index(typeid(string)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, string>(k, any_cast<string>(p)); } },
{ type_index(typeid(int)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, int>(k, any_cast<int>(p)); } },
{ type_index(typeid(size_t)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, size_t>(k, any_cast<size_t>(p)); } },
{ type_index(typeid(uint32_t)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, uint32_t>(k, any_cast<uint32_t>(p)); } },
{ type_index(typeid(uint64_t)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, uint64_t>(k, any_cast<uint64_t>(p)); } },
{ type_index(typeid(long)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, long>(k, any_cast<long>(p)); } },
{ type_index(typeid(long long)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, long long>(k, any_cast<long long>(p)); } },
{ type_index(typeid(unsigned)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, unsigned>(k, any_cast<unsigned>(p)); } },
{ type_index(typeid(unsigned long)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, unsigned long>(k, any_cast<unsigned long>(p)); } },
{ type_index(typeid(unsigned long long)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, unsigned long long>(k, any_cast<unsigned long long>(p)); } },
{ type_index(typeid(float)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, float>(k, any_cast<float>(p)); } },
{ type_index(typeid(double)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, double>(k, any_cast<double>(p)); } },
{ type_index(typeid(long double)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, long double>(k, any_cast<long double>(p)); } },
{ type_index(typeid(bool)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, bool>(k, any_cast<bool>(p)); } },
{ type_index(typeid(vector<bool>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<bool>>(k, any_cast<vector<bool>>(p)); } },
{ type_index(typeid(boost::filesystem::path)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, boost::filesystem::path>(k, any_cast<boost::filesystem::path>(p)); } },
{ type_index(typeid(vector<char>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<char>>(k, any_cast<vector<char>>(p)); } },
{ type_index(typeid(vector<unsigned char>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<unsigned char>>(k, any_cast<vector<unsigned char>>(p)); } },
{ type_index(typeid(vector<string>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<string>>(k, any_cast<vector<string>>(p)); } },
{ type_index(typeid(vector<int>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<int>>(k, any_cast<vector<int>>(p)); } },
{ type_index(typeid(vector<size_t>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<size_t>>(k, any_cast<vector<size_t>>(p)); } },
{ type_index(typeid(vector<uint32_t>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<uint32_t>>(k, any_cast<vector<uint32_t>>(p)); } },
{ type_index(typeid(vector<uint64_t>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<uint64_t>>(k, any_cast<vector<uint64_t>>(p)); } },
{ type_index(typeid(vector<long>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<long>>(k, any_cast<vector<long>>(p)); } },
{ type_index(typeid(vector<long long>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<long long>>(k, any_cast<vector<long long>>(p)); } },
{ type_index(typeid(vector<unsigned>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<unsigned>>(k, any_cast<vector<unsigned>>(p)); } },
{ type_index(typeid(vector<unsigned long>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<unsigned long>>(k, any_cast<vector<unsigned long>>(p)); } },
{ type_index(typeid(vector<unsigned long long>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<unsigned long long>>(k, any_cast<vector<unsigned long long>>(p)); } },
{ type_index(typeid(vector<float>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<float>>(k, any_cast<vector<float>>(p)); } },
{ type_index(typeid(vector<double>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<double>>(k, any_cast<vector<double>>(p)); } },
{ type_index(typeid(vector<long double>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<long double>>(k, any_cast<vector<long double>>(p)); } },
{ type_index(typeid(vector<boost::filesystem::path>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<boost::filesystem::path>>(k, any_cast<vector<boost::filesystem::path>>(p)); } },
};
namespace fair
{
namespace mq
{
string ConvertPropertyToString(const Property& p)
{
pair<string, string> info = FairMQProgOptions::fTypeInfos.at(p.type())(p);
return info.first;
}
ValInfo ConvertVarValToValInfo(const po::variable_value& v)
{
string origin;
if (v.defaulted()) {
origin = "[default]";
} else if (v.empty()) {
origin = "[empty]";
} else {
origin = "[provided]";
}
try {
pair<string, string> info = FairMQProgOptions::fTypeInfos.at(v.value().type())(v.value());
return {info.first, info.second, origin};
} catch (out_of_range& oor) {
return {string("[unidentified_type]"), string("[unidentified_type]"), origin};
}
};
string ConvertVarValToString(const po::variable_value& v)
{
return ConvertVarValToValInfo(v).value;
}
} // namespace mq
} // namespace fair
FairMQProgOptions::FairMQProgOptions()
: fVarMap()
, fAllOptions("FairMQ Command Line Options")
, fGeneralOptions("General options")
, fMQOptions("FairMQ device options")
, fParserOptions("FairMQ channel config parser options")
, fMtx()
, fUnregisteredOptions()
, fEvents()
{
fGeneralOptions.add_options()
("help,h", "Print help")
("version,v", "Print version")
("severity", po::value<string>()->default_value("debug"), "Log severity level: trace, debug, info, state, warn, error, fatal, nolog")
("verbosity", po::value<string>()->default_value("medium"), "Log verbosity level: veryhigh, high, medium, low")
("color", po::value<bool >()->default_value(true), "Log color (true/false)")
("log-to-file", po::value<string>()->default_value(""), "Log output to a file.")
("print-options", po::value<bool >()->implicit_value(true), "Print options in machine-readable format (<option>:<computed-value>:<type>:<description>)");
fMQOptions.add_options()
("id", po::value<string >(), "Device ID (required argument).")
("io-threads", po::value<int >()->default_value(1), "Number of I/O threads.")
("transport", po::value<string >()->default_value("zeromq"), "Transport ('zeromq'/'nanomsg'/'shmem').")
("network-interface", po::value<string >()->default_value("default"), "Network interface to bind on (e.g. eth0, ib0..., default will try to detect the interface of the default route).")
("config-key", po::value<string >(), "Use provided value instead of device id for fetching the configuration from the config file.")
("initialization-timeout", po::value<int >()->default_value(120), "Timeout for the initialization in seconds (when expecting dynamic initialization).")
("max-run-time", po::value<uint64_t>()->default_value(0), "Maximum runtime for the Running state handler, after which state will change to Ready (in seconds, 0 for no limit).")
("print-channels", po::value<bool >()->implicit_value(true), "Print registered channel endpoints in a machine-readable format (<channel name>:<min num subchannels>:<max num subchannels>)")
("shm-segment-size", po::value<size_t >()->default_value(2000000000), "Shared memory: size of the shared memory segment (in bytes).")
("shm-monitor", po::value<bool >()->default_value(true), "Shared memory: run monitor daemon.")
("ofi-size-hint", po::value<size_t >()->default_value(0), "EXPERIMENTAL: OFI size hint for the allocator.")
("rate", po::value<float >()->default_value(0.), "Rate for conditional run loop (Hz).")
("session", po::value<string >()->default_value("default"), "Session name.");
fParserOptions.add_options()
("mq-config", po::value<string>(), "JSON input as file.")
("channel-config", po::value<vector<string>>()->multitoken()->composing(), "Configuration of single or multiple channel(s) by comma separated key=value list");
fAllOptions.add(fGeneralOptions);
fAllOptions.add(fMQOptions);
fAllOptions.add(fParserOptions);
ParseDefaults();
}
unordered_map<string, int> FairMQProgOptions::GetChannelInfo() const
{
lock_guard<mutex> lock(fMtx);
return GetChannelInfoImpl();
}
unordered_map<string, int> FairMQProgOptions::GetChannelInfoImpl() const
{
unordered_map<string, int> info;
boost::regex re("chans\\..*\\.type");
for (const auto& m : fVarMap) {
if (boost::regex_match(m.first, re)) {
string chan = m.first.substr(6);
string::size_type n = chan.find(".");
string chanName = chan.substr(0, n);
if (info.find(chanName) == info.end()) {
info.emplace(chanName, 1);
} else {
info[chanName] = info[chanName] + 1;
}
}
}
return info;
}
Properties FairMQProgOptions::GetProperties(const string& q) const
{
boost::regex re(q);
Properties result;
lock_guard<mutex> lock(fMtx);
for (const auto& m : fVarMap) {
if (boost::regex_match(m.first, re)) {
result.emplace(m.first, m.second.value());
}
}
if (result.size() == 0) {
LOG(warn) << "could not find anything with \"" << q << "\"";
}
return result;
}
map<string, string> FairMQProgOptions::GetPropertiesAsString(const string& q) const
{
boost::regex re(q);
map<string, string> result;
lock_guard<mutex> lock(fMtx);
for (const auto& m : fVarMap) {
if (boost::regex_match(m.first, re)) {
result.emplace(m.first, ConvertPropertyToString(m.second.value()));
}
}
if (result.size() == 0) {
LOG(warn) << "could not find anything with \"" << q << "\"";
}
return result;
}
Properties FairMQProgOptions::GetPropertiesStartingWith(const string& q) const
{
Properties result;
lock_guard<mutex> lock(fMtx);
for (const auto& m : fVarMap) {
if (m.first.compare(0, q.length(), q) == 0) {
result.emplace(m.first, m.second.value());
}
}
return result;
}
void FairMQProgOptions::SetProperties(const Properties& input)
{
unique_lock<mutex> lock(fMtx);
map<string, boost::program_options::variable_value>& vm = fVarMap;
for (const auto& m : input) {
vm[m.first].value() = m.second;
}
lock.unlock();
for (const auto& m : input) {
fEventEmitters.at(m.second.type())(fEvents, m.first, m.second);
fEvents.Emit<PropertyChangeAsString, string>(m.first, ConvertPropertyToString(m.second));
}
}
void FairMQProgOptions::AddChannel(const std::string& name, const FairMQChannel& channel)
{
lock_guard<mutex> lock(fMtx);
unordered_map<string, int> existingChannels = GetChannelInfoImpl();
int index = 0;
if (existingChannels.count(name) > 0) {
index = existingChannels.at(name);
}
string prefix = fair::mq::tools::ToString("chans.", name, ".", index, ".");
SetVarMapValue<string>(string(prefix + "type"), channel.GetType());
SetVarMapValue<string>(string(prefix + "method"), channel.GetMethod());
SetVarMapValue<string>(string(prefix + "address"), channel.GetAddress());
SetVarMapValue<string>(string(prefix + "transport"), channel.GetTransportName());
SetVarMapValue<int>(string(prefix + "sndBufSize"), channel.GetSndBufSize());
SetVarMapValue<int>(string(prefix + "rcvBufSize"), channel.GetRcvBufSize());
SetVarMapValue<int>(string(prefix + "sndKernelSize"), channel.GetSndKernelSize());
SetVarMapValue<int>(string(prefix + "rcvKernelSize"), channel.GetRcvKernelSize());
SetVarMapValue<int>(string(prefix + "linger"), channel.GetLinger());
SetVarMapValue<int>(string(prefix + "rateLogging"), channel.GetRateLogging());
SetVarMapValue<int>(string(prefix + "portRangeMin"), channel.GetPortRangeMin());
SetVarMapValue<int>(string(prefix + "portRangeMax"), channel.GetPortRangeMax());
SetVarMapValue<bool>(string(prefix + "autoBind"), channel.GetAutoBind());
}
void FairMQProgOptions::DeleteProperty(const string& key)
{
lock_guard<mutex> lock(fMtx);
map<string, boost::program_options::variable_value>& vm = fVarMap;
vm.erase(key);
}
int FairMQProgOptions::ParseAll(const vector<string>& cmdArgs, bool allowUnregistered)
{
vector<const char*> argv(cmdArgs.size());
transform(cmdArgs.begin(), cmdArgs.end(), argv.begin(), [](const string& str) {
return str.c_str();
});
return ParseAll(argv.size(), const_cast<char**>(argv.data()), allowUnregistered);
}
int FairMQProgOptions::ParseAll(const int argc, char const* const* argv, bool allowUnregistered)
{
ParseCmdLine(argc, argv, allowUnregistered);
// if this option is provided, handle them and return stop value
if (fVarMap.count("help")) {
cout << fAllOptions << endl;
return 1;
}
// if this option is provided, handle them and return stop value
if (fVarMap.count("print-options")) {
PrintOptionsRaw();
return 1;
}
// if these options are provided, do no further checks and let the device handle them
if (fVarMap.count("print-channels") || fVarMap.count("version")) {
fair::Logger::SetConsoleSeverity("nolog");
return 0;
}
string severity = GetValue<string>("severity");
string logFile = GetValue<string>("log-to-file");
bool color = GetValue<bool>("color");
string verbosity = GetValue<string>("verbosity");
fair::Logger::SetVerbosity(verbosity);
if (logFile != "") {
fair::Logger::InitFileSink(severity, logFile);
fair::Logger::SetConsoleSeverity("nolog");
} else {
fair::Logger::SetConsoleColor(color);
fair::Logger::SetConsoleSeverity(severity);
}
string idForParser;
// check if config-key for config parser is provided
if (fVarMap.count("config-key")) {
idForParser = fVarMap["config-key"].as<string>();
} else if (fVarMap.count("id")) {
idForParser = fVarMap["id"].as<string>();
}
// check if any config parser is selected
try {
if (fVarMap.count("mq-config")) {
LOG(debug) << "mq-config: Using default JSON parser";
SetProperties(parser::JSON().UserParser(fVarMap.at("mq-config").as<string>(), idForParser));
} else if (fVarMap.count("channel-config")) {
LOG(debug) << "channel-config: Parsing channel configuration";
ParseChannelsFromCmdLine();
} else {
LOG(warn) << "FairMQProgOptions: no channels configuration provided via neither of:";
for (const auto& p : fParserOptions.options()) {
LOG(warn) << "--" << p->canonical_display_name();
}
}
} catch (exception& e) {
LOG(error) << e.what();
return 1;
}
PrintOptions();
return 0;
}
void FairMQProgOptions::ParseChannelsFromCmdLine()
{
string idForParser;
// check if config-key for config parser is provided
if (fVarMap.count("config-key")) {
idForParser = fVarMap["config-key"].as<string>();
} else if (fVarMap.count("id")) {
idForParser = fVarMap["id"].as<string>();
}
SetProperties(parser::SUBOPT().UserParser(fVarMap.at("channel-config").as<vector<string>>(), idForParser));
}
void FairMQProgOptions::ParseCmdLine(const int argc, char const* const* argv, bool allowUnregistered)
{
// clear the container because it was filled with default values and subsequent calls to store() do not overwrite the existing values
fVarMap.clear();
if (allowUnregistered) {
po::command_line_parser parser{argc, argv};
parser.options(fAllOptions).allow_unregistered();
po::parsed_options parsed = parser.run();
fUnregisteredOptions = po::collect_unrecognized(parsed.options, po::include_positional);
po::store(parsed, fVarMap);
} else {
po::store(po::parse_command_line(argc, argv, fAllOptions), fVarMap);
}
po::notify(fVarMap);
}
void FairMQProgOptions::ParseDefaults()
{
vector<string> emptyArgs = {"dummy", "--id", tools::Uuid()};
vector<const char*> argv(emptyArgs.size());
transform(emptyArgs.begin(), emptyArgs.end(), argv.begin(), [](const string& str) {
return str.c_str();
});
po::store(po::parse_command_line(argv.size(), const_cast<char**>(argv.data()), fAllOptions), fVarMap);
}
vector<string> FairMQProgOptions::GetPropertyKeys() const
{
lock_guard<mutex> lock(fMtx);
vector<string> result;
for (const auto& it : fVarMap) {
result.push_back(it.first.c_str());
}
return result;
}
/// Add option descriptions
int FairMQProgOptions::AddToCmdLineOptions(const po::options_description optDesc, bool /* visible */)
{
fAllOptions.add(optDesc);
return 0;
}
po::options_description& FairMQProgOptions::GetCmdLineOptions()
{
return fAllOptions;
}
int FairMQProgOptions::PrintOptions()
{
map<string, ValInfo> mapinfo;
// get string length for formatting and convert varmap values into string
int maxLenKey = 0;
int maxLenValue = 0;
int maxLenType = 0;
int maxLenDefault = 0;
for (const auto& m : fVarMap) {
maxLenKey = max(maxLenKey, static_cast<int>(m.first.length()));
ValInfo valinfo = ConvertVarValToValInfo(m.second);
mapinfo[m.first] = valinfo;
maxLenValue = max(maxLenValue, static_cast<int>(valinfo.value.length()));
maxLenType = max(maxLenType, static_cast<int>(valinfo.type.length()));
maxLenDefault = max(maxLenDefault, static_cast<int>(valinfo.origin.length()));
}
if (maxLenValue > 100) {
maxLenValue = 100;
}
for (const auto& o : fUnregisteredOptions) {
LOG(debug) << "detected unregistered option: " << o;
}
stringstream ss;
ss << "Configuration: \n";
for (const auto& p : mapinfo) {
string type("<" + p.second.type + ">");
ss << setfill(' ') << left
<< setw(maxLenKey) << p.first << " = "
<< setw(maxLenValue) << p.second.value << " "
<< setw(maxLenType + 2) << type << " "
<< setw(maxLenDefault) << p.second.origin
<< "\n";
}
LOG(debug) << ss.str();
return 0;
}
int FairMQProgOptions::PrintOptionsRaw()
{
const vector<boost::shared_ptr<po::option_description>>& options = fAllOptions.options();
for (const auto& o : options) {
ValInfo value;
if (fVarMap.count(o->canonical_display_name())) {
value = ConvertVarValToValInfo(fVarMap[o->canonical_display_name()]);
}
string description = o->description();
replace(description.begin(), description.end(), '\n', ' ');
cout << o->long_name() << ":" << value.value << ":" << (value.type == "" ? "<unknown>" : value.type) << ":" << description << endl;
}
return 0;
}
string FairMQProgOptions::GetPropertyAsString(const string& key) const
{
lock_guard<mutex> lock(fMtx);
if (fVarMap.count(key)) {
return ConvertVarValToString(fVarMap.at(key));
}
throw PropertyNotFoundException(fair::mq::tools::ToString("Config has no key: ", key));
}
int FairMQProgOptions::Count(const string& key) const
{
lock_guard<mutex> lock(fMtx);
return fVarMap.count(key);
}

View File

@@ -9,203 +9,6 @@
#ifndef FAIRMQPROGOPTIONS_H
#define FAIRMQPROGOPTIONS_H
#include <fairmq/EventManager.h>
#include "FairMQLogger.h"
#include "FairMQChannel.h"
#include "Properties.h"
#include <fairmq/Tools.h>
#include <boost/program_options.hpp>
#include <boost/core/demangle.hpp>
#include <unordered_map>
#include <map>
#include <functional>
#include <string>
#include <vector>
#include <mutex>
#include <sstream>
#include <typeindex>
#include <typeinfo>
#include <utility> // pair
#include <stdexcept>
namespace fair
{
namespace mq
{
struct PropertyChange : Event<std::string> {};
struct PropertyChangeAsString : Event<std::string> {};
} /* namespace mq */
} /* namespace fair */
class FairMQProgOptions
{
public:
FairMQProgOptions();
virtual ~FairMQProgOptions() {}
struct PropertyNotFoundException : std::runtime_error { using std::runtime_error::runtime_error; };
int ParseAll(const std::vector<std::string>& cmdArgs, bool allowUnregistered);
int ParseAll(const int argc, char const* const* argv, bool allowUnregistered = true);
std::unordered_map<std::string, int> GetChannelInfo() const;
std::vector<std::string> GetPropertyKeys() const;
template<typename T>
T GetProperty(const std::string& key) const
{
std::lock_guard<std::mutex> lock(fMtx);
if (fVarMap.count(key)) {
return fVarMap[key].as<T>();
}
throw PropertyNotFoundException(fair::mq::tools::ToString("Config has no key: ", key));
}
template<typename T>
T GetProperty(const std::string& key, const T& ifNotFound) const
{
std::lock_guard<std::mutex> lock(fMtx);
if (fVarMap.count(key)) {
return fVarMap[key].as<T>();
}
return ifNotFound;
}
fair::mq::Properties GetProperties(const std::string& q) const;
std::map<std::string, std::string> GetPropertiesAsString(const std::string& q) const;
fair::mq::Properties GetPropertiesStartingWith(const std::string& q) const;
template<typename T>
T GetValue(const std::string& key) const // TODO: deprecate this
{
return GetProperty<T>(key);
}
std::string GetPropertyAsString(const std::string& key) const;
std::string GetStringValue(const std::string& key) const // TODO: deprecate this
{
return GetPropertyAsString(key);
}
template<typename T>
void SetProperty(const std::string& key, T val)
{
std::unique_lock<std::mutex> lock(fMtx);
SetVarMapValue<typename std::decay<T>::type>(key, val);
if (key == "channel-config") {
ParseChannelsFromCmdLine();
}
lock.unlock();
fEvents.Emit<fair::mq::PropertyChange, typename std::decay<T>::type>(key, val);
fEvents.Emit<fair::mq::PropertyChangeAsString, std::string>(key, GetStringValue(key));
}
template<typename T>
int SetValue(const std::string& key, T val) // TODO: deprecate this
{
SetProperty(key, val);
return 0;
}
void SetProperties(const fair::mq::Properties& input);
void DeleteProperty(const std::string& key);
template<typename T>
void Subscribe(const std::string& subscriber, std::function<void(typename fair::mq::PropertyChange::KeyType, T)> func) const
{
std::lock_guard<std::mutex> lock(fMtx);
static_assert(!std::is_same<T,const char*>::value || !std::is_same<T, char*>::value,
"In template member FairMQProgOptions::Subscribe<T>(key,Lambda) the types const char* or char* for the calback signatures are not supported.");
fEvents.Subscribe<fair::mq::PropertyChange, T>(subscriber, func);
}
template<typename T>
void Unsubscribe(const std::string& subscriber) const
{
std::lock_guard<std::mutex> lock(fMtx);
fEvents.Unsubscribe<fair::mq::PropertyChange, T>(subscriber);
}
void SubscribeAsString(const std::string& subscriber, std::function<void(typename fair::mq::PropertyChange::KeyType, std::string)> func) const
{
std::lock_guard<std::mutex> lock(fMtx);
fEvents.Subscribe<fair::mq::PropertyChangeAsString, std::string>(subscriber, func);
}
void UnsubscribeAsString(const std::string& subscriber) const
{
std::lock_guard<std::mutex> lock(fMtx);
fEvents.Unsubscribe<fair::mq::PropertyChangeAsString, std::string>(subscriber);
}
int Count(const std::string& key) const;
// add options_description
int AddToCmdLineOptions(const boost::program_options::options_description optDesc, bool visible = true);
boost::program_options::options_description& GetCmdLineOptions();
int PrintOptions();
int PrintOptionsRaw();
void AddChannel(const std::string& name, const FairMQChannel& channel);
template<typename T>
static void AddType(std::string label = "")
{
if (label == "") {
label = boost::core::demangle(typeid(T).name());
}
fTypeInfos[std::type_index(typeid(T))] = [label](const fair::mq::Property& p) {
std::stringstream ss;
ss << boost::any_cast<T>(p);
return std::pair<std::string, std::string>{ss.str(), label};
};
}
static std::unordered_map<std::type_index, std::function<std::pair<std::string, std::string>(const fair::mq::Property&)>> fTypeInfos;
static std::unordered_map<std::type_index, void(*)(const fair::mq::EventManager&, const std::string&, const fair::mq::Property&)> fEventEmitters;
private:
boost::program_options::variables_map fVarMap; ///< options container
boost::program_options::options_description fAllOptions; ///< all options descriptions
boost::program_options::options_description fGeneralOptions; ///< general options descriptions
boost::program_options::options_description fMQOptions; ///< MQ options descriptions
boost::program_options::options_description fParserOptions; ///< MQ Parser options descriptions
mutable std::mutex fMtx;
std::vector<std::string> fUnregisteredOptions; ///< container with unregistered options
mutable fair::mq::EventManager fEvents;
void ParseCmdLine(const int argc, char const* const* argv, bool allowUnregistered = true);
void ParseDefaults();
std::unordered_map<std::string, int> GetChannelInfoImpl() const;
// modify the value of variable map after calling boost::program_options::store
template<typename T>
void SetVarMapValue(const std::string& key, const T& val)
{
std::map<std::string, boost::program_options::variable_value>& vm = fVarMap;
vm[key].value() = boost::any(val);
}
void ParseChannelsFromCmdLine();
};
#include <fairmq/ProgOptions.h>
#endif /* FAIRMQPROGOPTIONS_H */

View File

@@ -1,94 +0,0 @@
/********************************************************************************
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public License (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/// @file FairMQSuboptParser.h
/// @author Matthias.Richter@scieq.net
/// @since 2017-03-30
/// @brief Parser implementation for key-value subopt format
#ifndef FAIRMQPARSER_SUBOPT_H
#define FAIRMQPARSER_SUBOPT_H
#include "FairMQParser.h" // for FairMQProperties
#include <boost/program_options.hpp>
#include <cstring>
#include <vector>
#include <string>
namespace fair
{
namespace mq
{
namespace parser
{
/**
* A parser implementation for FairMQ channel properties.
* The parser handles a comma separated key=value list format by using the
* getsubopt function of the standard library.
*
* The option key '--channel-config' can be used with the list of key/value
* pairs like e.g.
* <pre>
* --channel-config name=output,type=push,method=bind
* </pre>
*
* The FairMQ option parser defines a 'UserParser' function for different
* formats. Currently it is strictly parsing channel options, but in general
* the concept is extensible by renaming UserParser to ChannelPropertyParser
* and introducing additional parser functions.
*/
struct SUBOPT
{
enum channelOptionKeyIds
{
NAME = 0, // name of the channel
TYPE, // push, pull, publish, subscribe, etc
METHOD, // bind or connect
ADDRESS, // host, protocol and port address
TRANSPORT, //
SNDBUFSIZE, // size of the send queue
RCVBUFSIZE, // size of the receive queue
SNDKERNELSIZE,
RCVKERNELSIZE,
LINGER,
RATELOGGING, // logging rate
PORTRANGEMIN,
PORTRANGEMAX,
AUTOBIND,
NUMSOCKETS,
lastsocketkey
};
constexpr static const char* channelOptionKeys[] = {
/*[NAME] = */ "name",
/*[TYPE] = */ "type",
/*[METHOD] = */ "method",
/*[ADDRESS] = */ "address",
/*[TRANSPORT] = */ "transport",
/*[SNDBUFSIZE] = */ "sndBufSize",
/*[RCVBUFSIZE] = */ "rcvBufSize",
/*[SNDKERNELSIZE] = */ "sndKernelSize",
/*[RCVKERNELSIZE] = */ "rcvKernelSize",
/*[LINGER] = */ "linger",
/*[RATELOGGING] = */ "rateLogging",
/*[PORTRANGEMIN] = */ "portRangeMin",
/*[PORTRANGEMAX] = */ "portRangeMax",
/*[AUTOBIND] = */ "autoBind",
/*[NUMSOCKETS] = */ "numSockets",
nullptr
};
fair::mq::Properties UserParser(const std::vector<std::string>& channelConfig, const std::string& deviceId);
};
}
}
}
#endif /* FAIRMQPARSER_SUBOPT_H */

View File

@@ -32,12 +32,20 @@ The basic structure looks like this:
The top level key is `fairMQOptions`, followed by one or more devices (with their IDs), each containing one or more channels (with their names), each containing one or more sockets.
The socket parameters accept following values:
- `type` (default = ""): "push"/"pull", "pub"/"sub", "req"/"rep", "xsub"/"xpub", "dealer/router", "pair".
- `type` (default = ""): "push"/"pull", "pub"/"sub", "req"/"rep", "pair".
- `method` (default = ""): "bind"/"connect".
- `address` (default = ""): address to bind/connect.
- `sndBufSize` (default = 1000): socket send queue size in number of messages.
- `rcvBufSize` (default = 1000): socket receive queue size in number of messages.
- `sndKernelSize"` (default = ):
- `rcvKernelSize"` (default = ):
- `rateLogging` (default = 1): log socket transfer rates in seconds. 0 for no logging of this socket.
- `transport"` (default = ): override the default device transport for this channel.
- `linger"` (default = ):
- `portRangeMin"` (default = ):
- `portRangeMax"` (default = ):
- `autoBind"` (default = ):
- `numSockets"` (default = ):
If a parameter is not specified, its default value will be set.

View File

@@ -1,150 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 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" *
********************************************************************************/
/*
* File: runSamplerRoot.cxx
* Author: winckler
*/
// FairRoot - FairMQ
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
#include "FairMQDevice.h"
#include <exception>
#include <string>
#include <vector>
#include <unordered_map>
using namespace std;
using FairMQMap = unordered_map<string, vector<FairMQChannel>>;
class MyDevice : public FairMQDevice
{
public:
MyDevice()
: fRate(0.5)
{}
virtual ~MyDevice()
{}
void SetRate(double r)
{
fRate = r;
}
double GetRate()
{
return fRate;
}
void Print()
{
LOG(info) << "[MyDevice] rate = " << fRate;
}
private:
double fRate;
};
void MyCallBack(MyDevice& d, double val)
{
d.SetRate(val);
d.Print();
}
int main(int argc, char** argv)
{
try
{
FairMQProgOptions config;
config.GetCmdLineOptions().add_options()
("data-rate", boost::program_options::value<double>()->default_value(0.5), "Data rate");
config.ParseAll(argc, argv, true);
// // get FairMQMap
// auto map1 = config.GetFairMQMap();
// // update value in variable map, and propagate the update to the FairMQMap
// config.SetValue<string>("chans.data.0.address","tcp://localhost:1234");
// // get the updated FairMQMap
// auto map2 = config.GetFairMQMap();
// // modify one channel value
// map2.at("data").at(0).UpdateSndBufSize(500);
// // update the FairMQMap and propagate the change in variable map
// config.UpdateChannelMap(map2);
MyDevice device;
device.SetConfig(config);
LOG(info) << "Subscribing: <string>(chans.data.0.address)";
config.Subscribe<string>("test", [&device](const string& key, string value)
{
if (key == "chans.data.0.address")
{
LOG(info) << "[callback] Updating device parameter " << key << " = " << value;
device.fChannels.at("data").at(0).UpdateAddress(value);
}
});
LOG(info) << "Subscribing: <int>(chans.data.0.rcvBufSize)";
config.Subscribe<int>("test", [&device](const string& key, int value)
{
if(key == "chans.data.0.rcvBufSize")
{
LOG(info) << "[callback] Updating device parameter " << key << " = " << value;
device.fChannels.at("data").at(0).UpdateRcvBufSize(value);
}
});
LOG(info) << "Subscribing: <double>(data-rate)";
config.Subscribe<double>("test", [&device](const string& key, double value)
{
if (key == "data-rate")
{
LOG(info) << "[callback] Updating device parameter " << key << " = " << value;
device.SetRate(value);
}
});
LOG(info) << "Starting value updates...\n";
config.SetValue<string>("chans.data.0.address", "tcp://localhost:4321");
LOG(info) << "config: " << config.GetValue<string>("chans.data.0.address");
LOG(info) << "device: " << device.fChannels.at("data").at(0).GetAddress() << endl;
config.SetValue<int>("chans.data.0.rcvBufSize", 100);
LOG(info) << "config: " << config.GetValue<int>("chans.data.0.rcvBufSize");
LOG(info) << "device: " << device.fChannels.at("data").at(0).GetRcvBufSize() << endl;
config.SetValue<double>("data-rate", 0.9);
LOG(info) << "config: " << config.GetValue<double>("data-rate");
LOG(info) << "device: " << device.GetRate() << endl;
device.Print();
LOG(info) << "nase: " << config.GetValue<double>("nase");
config.Unsubscribe<string>("test");
config.Unsubscribe<int>("test");
config.Unsubscribe<double>("test");
device.ChangeState("END");
}
catch (exception& e)
{
LOG(error) << "Unhandled Exception reached the top of main: " << e.what() << ", application will now exit";
return 1;
}
return 0;
}

View File

@@ -1,5 +0,0 @@
#!/bin/bash
DEVICE="runConfigExample"
DEVICE+=" --id sampler1 --channel-config name=data,type=push,method=bind,address=tcp://*:5555,rateLogging=0"
@CMAKE_CURRENT_BINARY_DIR@/$DEVICE

View File

@@ -8,4 +8,5 @@
// List of all builtin plugin headers (the ones which call REGISTER_FAIRMQ_PLUGIN macro)
#include <fairmq/plugins/config/Config.h>
#include <fairmq/plugins/Control.h>

View File

@@ -0,0 +1,89 @@
/********************************************************************************
* Copyright (C) 2017-2018 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 "Config.h"
#include <fairmq/JSONParser.h>
#include <fairmq/SuboptParser.h>
#include <vector>
using namespace std;
namespace fair
{
namespace mq
{
namespace plugins
{
Config::Config(const string& name, const Plugin::Version version, const string& maintainer, const string& homepage, PluginServices* pluginServices)
: Plugin(name, version, maintainer, homepage, pluginServices)
{
SubscribeToDeviceStateChange([&](DeviceState newState) {
if (newState == DeviceState::InitializingDevice) {
string idForParser;
if (PropertyExists("config-key")) {
idForParser = GetProperty<string>("config-key");
} else if (PropertyExists("id")) {
idForParser = GetProperty<string>("id");
}
if (idForParser != "") {
try {
if (PropertyExists("mq-config")) {
LOG(debug) << "mq-config: Using default JSON parser";
SetProperties(JSONParser(GetProperty<string>("mq-config"), idForParser));
} else if (PropertyExists("channel-config")) {
LOG(debug) << "channel-config: Parsing channel configuration";
SetProperties(SuboptParser(GetProperty<vector<string>>("channel-config"), idForParser));
} else {
LOG(warn) << "fair::mq::plugins::Config: no channels configuration provided via --mq-config or --channel-config";
}
} catch (exception& e) {
LOG(error) << e.what();
}
} else {
LOG(warn) << "No device ID or config-key provided for the configuration parser.";
}
}
});
}
Plugin::ProgOptions ConfigPluginProgramOptions()
{
namespace po = boost::program_options;
auto pluginOptions = po::options_description{"FairMQ device options"};
pluginOptions.add_options()
("id", po::value<string >()->default_value(""), "Device ID (required argument).")
("io-threads", po::value<int >()->default_value(1), "Number of I/O threads.")
("transport", po::value<string >()->default_value("zeromq"), "Transport ('zeromq'/'nanomsg'/'shmem').")
("network-interface", po::value<string >()->default_value("default"), "Network interface to bind on (e.g. eth0, ib0..., default will try to detect the interface of the default route).")
("init-timeout", po::value<int >()->default_value(120), "Timeout for the initialization in seconds (when expecting dynamic initialization).")
("max-run-time", po::value<uint64_t >()->default_value(0), "Maximum runtime for the Running state handler, after which state will change to Ready (in seconds, 0 for no limit).")
("print-channels", po::value<bool >()->implicit_value(true), "Print registered channel endpoints in a machine-readable format (<channel name>:<min num subchannels>:<max num subchannels>)")
("shm-segment-size", po::value<size_t >()->default_value(2000000000), "Shared memory: size of the shared memory segment (in bytes).")
("shm-monitor", po::value<bool >()->default_value(true), "Shared memory: run monitor daemon.")
("ofi-size-hint", po::value<size_t >()->default_value(0), "EXPERIMENTAL: OFI size hint for the allocator.")
("rate", po::value<float >()->default_value(0.), "Rate for conditional run loop (Hz).")
("session", po::value<string >()->default_value("default"), "Session name.")
("config-key", po::value<string >(), "Use provided value instead of device id for fetching the configuration from JSON file.")
("mq-config", po::value<string >(), "JSON input as file.")
("channel-config", po::value<vector<string>>()->multitoken()->composing(), "Configuration of single or multiple channel(s) by comma separated key=value list");
return pluginOptions;
}
Config::~Config()
{
UnsubscribeFromDeviceStateChange();
}
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */

View File

@@ -0,0 +1,47 @@
/********************************************************************************
* 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_PLUGINS_CONFIG
#define FAIR_MQ_PLUGINS_CONFIG
#include <fairmq/Plugin.h>
#include <fairmq/Version.h>
#include <string>
namespace fair
{
namespace mq
{
namespace plugins
{
class Config : public Plugin
{
public:
Config(const std::string& name, const Plugin::Version version, const std::string& maintainer, const std::string& homepage, PluginServices* pluginServices);
~Config();
};
Plugin::ProgOptions ConfigPluginProgramOptions();
REGISTER_FAIRMQ_PLUGIN(
Config, // Class name
config, // Plugin name
(Plugin::Version{FAIRMQ_VERSION_MAJOR, FAIRMQ_VERSION_MINOR, FAIRMQ_VERSION_PATCH}),
"FairRootGroup <fairroot@gsi.de>",
"https://github.com/FairRootGroup/FairRoot",
ConfigPluginProgramOptions
)
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
#endif /* FAIR_MQ_PLUGINS_CONFIG */

View File

@@ -23,7 +23,7 @@ void addCustomOptions(bpo::options_description& options)
("msg-rate", bpo::value<float>()->default_value(0), "Msg rate limit in maximum number of messages per second");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /* config */)
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /* config */)
{
return new FairMQBenchmarkSampler();
}

View File

@@ -19,7 +19,7 @@ void addCustomOptions(bpo::options_description& options)
("multipart", bpo::value<bool>()->default_value(true), "Handle multipart payloads");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/)
{
return new FairMQMerger();
}

View File

@@ -19,7 +19,7 @@ void addCustomOptions(bpo::options_description& options)
("multipart", bpo::value<bool>()->default_value(true), "Handle multipart payloads");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/)
{
return new FairMQMultiplier();
}

View File

@@ -19,7 +19,7 @@ void addCustomOptions(bpo::options_description& options)
("multipart", bpo::value<bool>()->default_value(true), "Handle multipart payloads");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/)
{
return new FairMQProxy();
}

View File

@@ -19,7 +19,7 @@ void addCustomOptions(bpo::options_description& options)
("multipart", bpo::value<bool>()->default_value(false), "Handle multipart payloads");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/)
{
return new FairMQSink();
}

View File

@@ -19,7 +19,7 @@ void addCustomOptions(bpo::options_description& options)
("multipart", bpo::value<bool>()->default_value(true), "Handle multipart payloads");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/)
{
return new FairMQSplitter();
}

View File

@@ -14,7 +14,7 @@
using FairMQDevicePtr = FairMQDevice*;
// to be implemented by the user to return a child class of FairMQDevice
FairMQDevicePtr getDevice(const FairMQProgOptions& config);
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& 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&);
@@ -24,8 +24,7 @@ int main(int argc, char* argv[])
using namespace fair::mq;
using namespace fair::mq::hooks;
try
{
try {
fair::mq::DeviceRunner runner{argc, argv};
// runner.AddHook<LoadPlugins>([](DeviceRunner& r){
@@ -53,14 +52,10 @@ int main(int argc, char* argv[])
// Run with builtin catch all exception handler, just:
// return runner.RunWithExceptionHandlers();
}
catch (std::exception& e)
{
} catch (std::exception& e) {
LOG(error) << "Uncaught exception reached the top of main: " << e.what();
return 1;
}
catch (...)
{
} catch (...) {
LOG(error) << "Uncaught exception reached the top of main.";
return 1;
}

View File

@@ -33,7 +33,7 @@ namespace bipc = ::boost::interprocess;
fair::mq::Transport FairMQTransportFactorySHM::fTransportType = fair::mq::Transport::SHM;
FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const FairMQProgOptions* config)
FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const fair::mq::ProgOptions* config)
: FairMQTransportFactory(id)
, fDeviceId(id)
, fShmId()
@@ -57,12 +57,12 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
size_t segmentSize = 2000000000;
bool autolaunchMonitor = false;
if (config) {
numIoThreads = config->GetValue<int>("io-threads");
sessionName = config->GetValue<string>("session");
segmentSize = config->GetValue<size_t>("shm-segment-size");
autolaunchMonitor = config->GetValue<bool>("shm-monitor");
numIoThreads = config->GetProperty<int>("io-threads", numIoThreads);
sessionName = config->GetProperty<string>("session", sessionName);
segmentSize = config->GetProperty<size_t>("shm-segment-size", segmentSize);
autolaunchMonitor = config->GetProperty<bool>("shm-monitor", autolaunchMonitor);
} else {
LOG(debug) << "FairMQProgOptions not available! Using defaults.";
LOG(debug) << "fair::mq::ProgOptions not available! Using defaults.";
}
fShmId = buildShmIdFromSessionIdAndUserId(sessionName);

View File

@@ -17,7 +17,7 @@
#include "FairMQSocketSHM.h"
#include "FairMQPollerSHM.h"
#include "FairMQUnmanagedRegionSHM.h"
#include <options/FairMQProgOptions.h>
#include <fairmq/ProgOptions.h>
#include <vector>
#include <string>
@@ -27,7 +27,7 @@
class FairMQTransportFactorySHM final : public FairMQTransportFactory
{
public:
FairMQTransportFactorySHM(const std::string& id = "", const FairMQProgOptions* config = nullptr);
FairMQTransportFactorySHM(const std::string& id = "", const fair::mq::ProgOptions* config = nullptr);
FairMQTransportFactorySHM(const FairMQTransportFactorySHM&) = delete;
FairMQTransportFactorySHM operator=(const FairMQTransportFactorySHM&) = delete;

View File

@@ -13,7 +13,7 @@ using namespace std;
fair::mq::Transport FairMQTransportFactoryZMQ::fTransportType = fair::mq::Transport::ZMQ;
FairMQTransportFactoryZMQ::FairMQTransportFactoryZMQ(const string& id, const FairMQProgOptions* config)
FairMQTransportFactoryZMQ::FairMQTransportFactoryZMQ(const string& id, const fair::mq::ProgOptions* config)
: FairMQTransportFactory(id)
, fContext(zmq_ctx_new())
{
@@ -35,11 +35,11 @@ FairMQTransportFactoryZMQ::FairMQTransportFactoryZMQ(const string& id, const Fai
int numIoThreads = 1;
if (config)
{
numIoThreads = config->GetValue<int>("io-threads");
numIoThreads = config->GetProperty<int>("io-threads", numIoThreads);
}
else
{
LOG(debug) << "FairMQProgOptions not available! Using defaults.";
LOG(debug) << "fair::mq::ProgOptions not available! Using defaults.";
}
if (zmq_ctx_set(fContext, ZMQ_IO_THREADS, numIoThreads) != 0)

View File

@@ -23,12 +23,12 @@
#include "FairMQSocketZMQ.h"
#include "FairMQPollerZMQ.h"
#include "FairMQUnmanagedRegionZMQ.h"
#include <options/FairMQProgOptions.h>
#include <fairmq/ProgOptions.h>
class FairMQTransportFactoryZMQ final : public FairMQTransportFactory
{
public:
FairMQTransportFactoryZMQ(const std::string& id = "", const FairMQProgOptions* config = nullptr);
FairMQTransportFactoryZMQ(const std::string& id = "", const fair::mq::ProgOptions* config = nullptr);
FairMQTransportFactoryZMQ(const FairMQTransportFactoryZMQ&) = delete;
FairMQTransportFactoryZMQ operator=(const FairMQTransportFactoryZMQ&) = delete;