diff --git a/fairmq/CMakeLists.txt b/fairmq/CMakeLists.txt index 1f63620f..787980da 100644 --- a/fairmq/CMakeLists.txt +++ b/fairmq/CMakeLists.txt @@ -1,12 +1,11 @@ ################################################################################ # Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH # # # - # This software is distributed under the terms of the # - # GNU Lesser General Public Licence version 3 (LGPL) version 3, # + # This software is distributed under the terms of the # + # GNU Lesser General Public Licence version 3 (LGPL) version 3, # # copied verbatim in the file "LICENSE" # ################################################################################ -#Temporary for test -#add_subdirectory (options/ProgOptionTest) + configure_file(${CMAKE_SOURCE_DIR}/fairmq/options/ProgOptionTest/macro/bsampler-sink.json ${CMAKE_BINARY_DIR}/bin/bsampler-sink.json) Set(INCLUDE_DIRECTORIES diff --git a/fairmq/FairMQDevice.cxx b/fairmq/FairMQDevice.cxx index ec362582..accf78e7 100644 --- a/fairmq/FairMQDevice.cxx +++ b/fairmq/FairMQDevice.cxx @@ -108,7 +108,7 @@ void FairMQDevice::InitWrapper() fInitializingFinished = true; fInitializingCondition.notify_one(); - ChangeState(DEVICE_READY); + ChangeState(internal_DEVICE_READY); } void FairMQDevice::Init() @@ -178,7 +178,7 @@ void FairMQDevice::InitTaskWrapper() fInitializingTaskFinished = true; fInitializingTaskCondition.notify_one(); - ChangeState(READY); + ChangeState(internal_READY); } void FairMQDevice::InitTask() @@ -283,7 +283,7 @@ void FairMQDevice::ResetTaskWrapper() fResetTaskFinished = true; fResetTaskCondition.notify_one(); - ChangeState(DEVICE_READY); + ChangeState(internal_DEVICE_READY); } void FairMQDevice::ResetTask() @@ -299,7 +299,7 @@ void FairMQDevice::ResetWrapper() fResetFinished = true; fResetCondition.notify_one(); - ChangeState(IDLE); + ChangeState(internal_IDLE); } void FairMQDevice::Reset() diff --git a/fairmq/FairMQStateMachine.cxx b/fairmq/FairMQStateMachine.cxx index edf13147..e216520c 100644 --- a/fairmq/FairMQStateMachine.cxx +++ b/fairmq/FairMQStateMachine.cxx @@ -48,14 +48,14 @@ bool FairMQStateMachine::ChangeState(int event) case INIT_DEVICE: process_event(FairMQFSM::INIT_DEVICE()); return true; - case DEVICE_READY: - process_event(FairMQFSM::DEVICE_READY()); + case internal_DEVICE_READY: + process_event(FairMQFSM::internal_DEVICE_READY()); return true; case INIT_TASK: process_event(FairMQFSM::INIT_TASK()); return true; - case READY: - process_event(FairMQFSM::READY()); + case internal_READY: + process_event(FairMQFSM::internal_READY()); return true; case RUN: process_event(FairMQFSM::RUN()); @@ -75,8 +75,8 @@ bool FairMQStateMachine::ChangeState(int event) case RESET_TASK: process_event(FairMQFSM::RESET_TASK()); return true; - case IDLE: - process_event(FairMQFSM::IDLE()); + case internal_IDLE: + process_event(FairMQFSM::internal_IDLE()); return true; case END: process_event(FairMQFSM::END()); @@ -99,18 +99,10 @@ bool FairMQStateMachine::ChangeState(std::string event) { return ChangeState(INIT_DEVICE); } - if (event == "DEVICE_READY") - { - return ChangeState(DEVICE_READY); - } if (event == "INIT_TASK") { return ChangeState(INIT_TASK); } - if (event == "READY") - { - return ChangeState(READY); - } else if (event == "RUN") { return ChangeState(RUN); @@ -135,10 +127,6 @@ bool FairMQStateMachine::ChangeState(std::string event) { return ChangeState(RESET_TASK); } - else if (event == "IDLE") - { - return ChangeState(IDLE); - } else if (event == "END") { return ChangeState(END); @@ -212,18 +200,10 @@ void FairMQStateMachine::WaitForEndOfState(std::string event) { return WaitForEndOfState(INIT_DEVICE); } - if (event == "DEVICE_READY") - { - return WaitForEndOfState(DEVICE_READY); - } if (event == "INIT_TASK") { return WaitForEndOfState(INIT_TASK); } - if (event == "READY") - { - return WaitForEndOfState(READY); - } else if (event == "RUN") { return WaitForEndOfState(RUN); @@ -248,10 +228,6 @@ void FairMQStateMachine::WaitForEndOfState(std::string event) { return WaitForEndOfState(RESET_TASK); } - else if (event == "IDLE") - { - return WaitForEndOfState(IDLE); - } else if (event == "END") { return WaitForEndOfState(END); diff --git a/fairmq/FairMQStateMachine.h b/fairmq/FairMQStateMachine.h index 85f92cd7..035f892d 100644 --- a/fairmq/FairMQStateMachine.h +++ b/fairmq/FairMQStateMachine.h @@ -40,16 +40,16 @@ namespace FairMQFSM // defining events for the boost MSM state machine struct INIT_DEVICE {}; -struct DEVICE_READY {}; +struct internal_DEVICE_READY {}; struct INIT_TASK {}; -struct READY {}; +struct internal_READY {}; struct RUN {}; struct PAUSE {}; struct RESUME {}; struct STOP {}; struct RESET_TASK {}; struct RESET_DEVICE {}; -struct IDLE {}; +struct internal_IDLE {}; struct END {}; // defining the boost MSM state machine @@ -258,22 +258,22 @@ struct FairMQFSM_ : public msm::front::state_machine_def // Transition table for FairMQFMS struct transition_table : mpl::vector< - // Start Event Next Action Guard - // +-------------------------+-------------+------------------------+---------------+---------+ - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, - msmf::Row, // temporary - msmf::Row > + // Start Event Next Action Guard + // +-------------------------+----------------------+------------------------+---------------+---------+ + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, + msmf::Row, // temporary + msmf::Row > {}; // Replaces the default no-transition response. @@ -349,16 +349,16 @@ class FairMQStateMachine : public FairMQFSM::FairMQFSM enum Event { INIT_DEVICE, - DEVICE_READY, + internal_DEVICE_READY, INIT_TASK, - READY, + internal_READY, RUN, PAUSE, RESUME, STOP, RESET_TASK, RESET_DEVICE, - IDLE, + internal_IDLE, END }; FairMQStateMachine(); diff --git a/fairmq/devices/GenericProcessor.h b/fairmq/devices/GenericProcessor.h index 746076d1..ebbf0a60 100644 --- a/fairmq/devices/GenericProcessor.h +++ b/fairmq/devices/GenericProcessor.h @@ -126,18 +126,18 @@ class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputP while (GetCurrentState() == RUNNING) { FairMQMessage* msg = fTransportFactory->CreateMessage(); - - receivedMsgs++; - // InputPolicy::DeSerializeMsg(msg) --> deserialize data of msg and fill output container - // TaskPolicy::ExecuteTask( ... ) --> process output container - TaskPolicy::ExecuteTask(InputPolicy::DeSerializeMsg(msg)); - - // OutputPolicy::fMessage point to msg - OutputPolicy::SetMessage(msg); + ++receivedMsgs; if (fChannels["data-in"].at(0).Receive(msg) > 0) { + // InputPolicy::DeSerializeMsg(msg) --> deserialize data of msg and fill output container + // TaskPolicy::ExecuteTask( ... ) --> process output container + TaskPolicy::ExecuteTask(InputPolicy::DeSerializeMsg(msg)); + + // OutputPolicy::fMessage point to msg + OutputPolicy::SetMessage(msg); + // TaskPolicy::GetOutputData() --> Get processed output container // OutputPolicy::message(...) --> Serialize output container and fill fMessage fChannels["data-out"].at(0).Send(OutputPolicy::SerializeMsg(TaskPolicy::GetOutputData())); @@ -155,5 +155,5 @@ class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputP }; -#endif /* GENERICPROCESSOR_H */ +#endif /* GENERICPROCESSOR_H */ diff --git a/fairmq/options/FairMQChannel.cxx b/fairmq/options/FairMQChannel.cxx deleted file mode 100644 index 27faa60b..00000000 --- a/fairmq/options/FairMQChannel.cxx +++ /dev/null @@ -1,143 +0,0 @@ -/******************************************************************************** - * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH * - * * - * This software is distributed under the terms of the * - * GNU Lesser General Public Licence version 3 (LGPL) version 3, * - * copied verbatim in the file "LICENSE" * - ********************************************************************************/ -/** - * FairMQChannel.cxx - * - * @since 2015-06-02 - * @author A. Rybalchenko - */ - -#include - -#include "FairMQChannel.h" -#include "FairMQLogger.h" - -using namespace std; - -FairMQChannel::FairMQChannel() - : fType("unspecified") - , fMethod("unspecified") - , fProtocol("unspecified") - , fAddress("unspecified") - , fPort("unspecified") - , fSndBufSize(1000) - , fRcvBufSize(1000) - , fRateLogging(1) - , fSocket() - , fIsValid(false) - , fChannelName("") -{ -} - -FairMQChannel::FairMQChannel(string type, string method, string protocol, string address, string port) - : fType(type) - , fMethod(method) - , fProtocol(protocol) - , fAddress(address) - , fPort(port) - , fSndBufSize(1000) - , fRcvBufSize(1000) - , fRateLogging(1) - , fSocket() - , fIsValid(false) - , fChannelName("") -{ -} - -bool FairMQChannel::ValidateChannel() -{ - LOG(DEBUG) << "Validating channel " << fChannelName << "... "; - if (fIsValid) - { - LOG(DEBUG) << "Channel is already valid"; - return true; - } - - const string socketTypeNames[] = { "sub", "pub", "pull", "push", "req", "rep", "xsub", "xpub", "dealer", "router", "pair" }; - const set socketTypes(socketTypeNames, socketTypeNames + sizeof(socketTypeNames) / sizeof(string)); - if (socketTypes.find(fType) == socketTypes.end()) - { - LOG(DEBUG) << "Invalid channel type: " << fType; - fIsValid = false; - return false; - } - - const string socketMethodNames[] = { "bind", "connect" }; - const set socketMethods(socketMethodNames, socketMethodNames + sizeof(socketMethodNames) / sizeof(string)); - if (socketMethods.find(fMethod) == socketMethods.end()) - { - LOG(DEBUG) << "Invalid channel method: " << fMethod; - fIsValid = false; - return false; - } - - const string socketProtocolNames[] = { "tcp", "ipc", "inproc" }; - const set socketProtocols(socketProtocolNames, socketProtocolNames + sizeof(socketProtocolNames) / sizeof(string)); - if (socketProtocols.find(fProtocol) == socketProtocols.end()) - { - LOG(DEBUG) << "Invalid channel protocol: " << fProtocol; - fIsValid = false; - return false; - } - - if (fAddress == "unspecified" && fAddress == "") - { - LOG(DEBUG) << "invalid channel address: " << fAddress; - fIsValid = false; - return false; - } - - if (fPort == "unspecified" && fPort == "") - { - LOG(DEBUG) << "invalid channel port: " << fPort; - fIsValid = false; - return false; - } - - if (fSndBufSize < 0) - { - LOG(DEBUG) << "invalid channel send buffer size: " << fSndBufSize; - fIsValid = false; - return false; - } - - if (fRcvBufSize < 0) - { - LOG(DEBUG) << "invalid channel receive buffer size: " << fRcvBufSize; - fIsValid = false; - return false; - } - - LOG(DEBUG) << "Channel is valid"; - fIsValid = true; - return true; -} - -int FairMQChannel::Send(FairMQMessage* msg, const string& flag) -{ - return fSocket->Send(msg, flag); -} - -int FairMQChannel::Send(FairMQMessage* msg, const int flags) -{ - return fSocket->Send(msg, flags); -} - -int FairMQChannel::Receive(FairMQMessage* msg, const string& flag) -{ - return fSocket->Receive(msg, flag); -} - -int FairMQChannel::Receive(FairMQMessage* msg, const int flags) -{ - return fSocket->Receive(msg, flags); -} - -FairMQChannel::~FairMQChannel() -{ -} diff --git a/fairmq/options/FairMQChannel.h b/fairmq/options/FairMQChannel.h deleted file mode 100644 index 56f86fd3..00000000 --- a/fairmq/options/FairMQChannel.h +++ /dev/null @@ -1,54 +0,0 @@ -/******************************************************************************** - * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH * - * * - * This software is distributed under the terms of the * - * GNU Lesser General Public Licence version 3 (LGPL) version 3, * - * copied verbatim in the file "LICENSE" * - ********************************************************************************/ -/** - * FairMQChannel.h - * - * @since 2015-06-02 - * @author A. Rybalchenko - */ - -#ifndef FAIRMQCHANNEL_H_ -#define FAIRMQCHANNEL_H_ - -#include - -#include "FairMQSocket.h" - -class FairMQChannel -{ - friend class FairMQDevice; - - public: - FairMQChannel(); - FairMQChannel(std::string type, std::string method, std::string protocol, std::string address, std::string port); - virtual ~FairMQChannel(); - - bool ValidateChannel(); - - // Wrappers for the socket methods to simplify the usage of channels - int Send(FairMQMessage* msg, const std::string& flag=""); - int Send(FairMQMessage* msg, const int flags); - int Receive(FairMQMessage* msg, const std::string& flag=""); - int Receive(FairMQMessage* msg, const int flags); - - std::string fType; - std::string fMethod; - std::string fProtocol; - std::string fAddress; - std::string fPort; - int fSndBufSize; - int fRcvBufSize; - int fRateLogging; - - private: - FairMQSocket* fSocket; - bool fIsValid; - std::string fChannelName; -}; - -#endif /* FAIRMQCHANNEL_H_ */ \ No newline at end of file diff --git a/fairmq/options/FairMQParser.cxx b/fairmq/options/FairMQParser.cxx index b3b8ef5d..719fdb8b 100644 --- a/fairmq/options/FairMQParser.cxx +++ b/fairmq/options/FairMQParser.cxx @@ -17,152 +17,168 @@ namespace FairMQParser { - - // TODO : add key-value map parameter for replacing/updating values from keys - // function that convert property tree (given the xml or json structure) to FairMQMap - FairMQMap ptreeToMQMap(const boost::property_tree::ptree& pt, const std::string& device_id, const std::string& root_node, const std::string& format_flag) + +// TODO : add key-value map parameter for replacing/updating values from keys +// function that convert property tree (given the xml or json structure) to FairMQMap +FairMQMap ptreeToMQMap(const boost::property_tree::ptree& pt, const std::string& deviceId, const std::string& rootNode, const std::string& formatFlag) +{ + // Create fair mq map + FairMQMap MQChannelMap; + + // variables to create key for the mq map. Note: maybe device name and id useless here + std::string deviceIdKey; + std::string channelKey; + + // do a first loop just to print the device-id in xml/json input + for(const auto& p : pt.get_child(rootNode)) { - // Create fair mq map - FairMQMap MQChannelMap; - - // variables to create key for the mq map. Note: maybe device name and id useless here - std::string kdevice_id; - std::string kchannel; - - // do a first loop just to print the device-id in xml/json input - for(const auto& p : pt.get_child(root_node)) + if (p.first != "device") { - if (p.first != "device") - continue; - - //get id attribute to choose the device - if(format_flag=="xml") - { - kdevice_id=p.second.get(".id"); - MQLOG(DEBUG)<<"Found device id '"<< kdevice_id <<"' in XML input"; - } - - if(format_flag=="json") - { - kdevice_id=p.second.get("id"); - MQLOG(DEBUG)<<"Found device id '"<< kdevice_id <<"' in JSON input"; - } + continue; } - - // Extract value from boost::property_tree - // For each device in fairMQOptions - for(const auto& p : pt.get_child(root_node)) - { - if (p.first != "device") - continue; - //get id attribute to choose the device - if(format_flag=="xml") - kdevice_id=p.second.get(".id"); - - if(format_flag=="json") - kdevice_id=p.second.get("id"); - // if not correct device id, do not fill MQMap - if(device_id != kdevice_id) + //get id attribute to choose the device + if (formatFlag == "xml") + { + deviceIdKey = p.second.get(".id"); + MQLOG(DEBUG) << "Found device id '" << deviceIdKey << "' in XML input"; + } + + if (formatFlag == "json") + { + deviceIdKey = p.second.get("id"); + MQLOG(DEBUG) << "Found device id '"<< deviceIdKey << "' in JSON input"; + } + } + + // Extract value from boost::property_tree + // For each device in fairMQOptions + for(const auto& p : pt.get_child(rootNode)) + { + if (p.first != "device") + { + continue; + } + + //get id attribute to choose the device + if (formatFlag == "xml") + { + deviceIdKey = p.second.get(".id"); + } + + if (formatFlag == "json") + { + deviceIdKey = p.second.get("id"); + } + + // if not correct device id, do not fill MQMap + if (deviceId != deviceIdKey) + { + continue; + } + + // print if DEBUG log level set + std::stringstream deviceStream; + deviceStream << "[node = " << p.first + << "] id = " << deviceIdKey; + MQLOG(DEBUG) << deviceStream.str(); + + // for each channel in device + for(const auto& q : p.second.get_child("")) + { + if (q.first != "channel") + { continue; - + } + + //get name attribute to form key + if (formatFlag == "xml") + { + channelKey = q.second.get(".name"); + } + + if (formatFlag=="json") + { + channelKey = q.second.get("name"); + } + // print if DEBUG log level set - std::stringstream ss_device; - ss_device << "[node = " << p.first - << "] id = " << kdevice_id; - MQLOG(DEBUG)< channelList; + + int socketCounter = 0; + // for each socket in channel + for (const auto& r : q.second.get_child("")) { - if (q.first != "channel") - continue; - - //get name attribute to form key - if(format_flag=="xml") - kchannel=q.second.get(".name"); - - if(format_flag=="json") - kchannel=q.second.get("name"); - - // print if DEBUG log level set - std::stringstream ss_chan; - ss_chan << "\t [node = " << q.first - << "] name = " << kchannel; - MQLOG(DEBUG)< channel_list; - - int count_socket=0; - // for each socket in channel - for(const auto& r : q.second.get_child("")) + if (r.first != "socket") { - if (r.first != "socket") - continue; - - count_socket++; - FairMQChannel channel; - - // print if DEBUG log level set - std::stringstream ss_sock; - ss_sock << "\t \t [node = " << r.first - << "] socket index = " << count_socket; - MQLOG(DEBUG)<("type",channel.fType); - channel.fMethod = r.second.get("method",channel.fMethod); - channel.fAddress = r.second.get("address",channel.fAddress); - channel.fSndBufSize = r.second.get("sndBufSize",channel.fSndBufSize);//int - channel.fRcvBufSize = r.second.get("rcvBufSize",channel.fRcvBufSize);//int - channel.fRateLogging = r.second.get("rateLogging",channel.fRateLogging);//int + ++socketCounter; + FairMQChannel channel; - channel_list.push_back(channel); - }// end socket loop - - //fill mq map option - MQChannelMap.insert(std::make_pair(kchannel,std::move(channel_list))); - } + // print if DEBUG log level set + std::stringstream socket; + socket << "\t \t [node = " << r.first + << "] socket index = " << socketCounter; + MQLOG(DEBUG) << socket.str(); + MQLOG(DEBUG) << "\t \t \t type = " << r.second.get("type", channel.fType); + MQLOG(DEBUG) << "\t \t \t method = " << r.second.get("method", channel.fMethod); + MQLOG(DEBUG) << "\t \t \t address = " << r.second.get("address", channel.fAddress); + MQLOG(DEBUG) << "\t \t \t sndBufSize = " << r.second.get("sndBufSize", channel.fSndBufSize); + MQLOG(DEBUG) << "\t \t \t rcvBufSize = " << r.second.get("rcvBufSize", channel.fRcvBufSize); + MQLOG(DEBUG) << "\t \t \t rateLogging = " << r.second.get("rateLogging", channel.fRateLogging); + + channel.fType = r.second.get("type", channel.fType); + channel.fMethod = r.second.get("method", channel.fMethod); + channel.fAddress = r.second.get("address", channel.fAddress); + channel.fSndBufSize = r.second.get("sndBufSize", channel.fSndBufSize); // int + channel.fRcvBufSize = r.second.get("rcvBufSize", channel.fRcvBufSize); // int + channel.fRateLogging = r.second.get("rateLogging", channel.fRateLogging); // int + + channelList.push_back(channel); + }// end socket loop + + //fill mq map option + MQChannelMap.insert(std::make_pair(channelKey,std::move(channelList))); } - - if(MQChannelMap.size()>0) - { - MQLOG(DEBUG)<<"---- Channel-keys found are :"; - for(const auto& p : MQChannelMap) - MQLOG(DEBUG)< 0) { - boost::property_tree::ptree pt; - boost::property_tree::read_json(filename, pt); - return ptreeToMQMap(pt,device_id,root_node,"json"); + MQLOG(DEBUG) << "---- Channel-keys found are :"; + for (const auto& p : MQChannelMap) + { + MQLOG(DEBUG) << p.first; + } } - - FairMQMap JSON::UserParser(std::stringstream& input_ss, const std::string& device_id, const std::string& root_node) + else { - boost::property_tree::ptree pt; - boost::property_tree::read_json(input_ss, pt); - return ptreeToMQMap(pt,device_id,root_node,"json"); + MQLOG(WARN) << "---- No channel-keys found for device-id " << deviceId; + MQLOG(WARN) << "---- Check the "<< formatFlag << " inputs and/or command line inputs"; } - + return MQChannelMap; +} + +//////////////////////////////////////////////////////////////////////////// +FairMQMap JSON::UserParser(const std::string& filename, const std::string& deviceId, const std::string& rootNode) +{ + boost::property_tree::ptree pt; + boost::property_tree::read_json(filename, pt); + return ptreeToMQMap(pt, deviceId, rootNode,"json"); +} + +FairMQMap JSON::UserParser(std::stringstream& input, const std::string& deviceId, const std::string& rootNode) +{ + boost::property_tree::ptree pt; + boost::property_tree::read_json(input, pt); + return ptreeToMQMap(pt, deviceId, rootNode,"json"); +} + } // end FairMQParser namespace \ No newline at end of file diff --git a/fairmq/options/FairMQParser.h b/fairmq/options/FairMQParser.h index 92f9a7d8..71cda2a5 100644 --- a/fairmq/options/FairMQParser.h +++ b/fairmq/options/FairMQParser.h @@ -8,31 +8,30 @@ #ifndef FAIRMQPARSER_H #define FAIRMQPARSER_H -// FairRoot -#include "FairMQChannel.h" - -// Boost -#include - // std #include #include #include +// Boost +#include + +// FairMQ +#include "FairMQChannel.h" namespace FairMQParser { - - typedef std::map > FairMQMap; - FairMQMap ptreeToMQMap(const boost::property_tree::ptree& pt, const std::string& device_id, const std::string& root_node, const std::string& format_flag="json"); - struct JSON - { - FairMQMap UserParser(const std::string& filename, const std::string& device_id, const std::string& root_node="fairMQOptions"); - FairMQMap UserParser(std::stringstream& input_ss, const std::string& device_id, const std::string& root_node="fairMQOptions"); - }; - - -} // end FairMQParser namespace -#endif /* FAIRMQPARSER_H */ +typedef std::map< std::string,std::vector > FairMQMap; + +FairMQMap ptreeToMQMap(const boost::property_tree::ptree& pt, const std::string& deviceId, const std::string& rootNode, const std::string& formatFlag = "json"); + +struct JSON +{ + FairMQMap UserParser(const std::string& filename, const std::string& deviceId, const std::string& rootNode = "fairMQOptions"); + FairMQMap UserParser(std::stringstream& input, const std::string& deviceId, const std::string& rootNode = "fairMQOptions"); +}; + +} // FairMQParser namespace +#endif /* FAIRMQPARSER_H */ diff --git a/fairmq/options/FairMQProgOptions.cxx b/fairmq/options/FairMQProgOptions.cxx index 30aa6ff1..9895a313 100644 --- a/fairmq/options/FairMQProgOptions.cxx +++ b/fairmq/options/FairMQProgOptions.cxx @@ -8,11 +8,11 @@ #include "FairMQProgOptions.h" #include -FairMQProgOptions::FairMQProgOptions() : - FairProgOptions(), - fMQParserOptions("MQ-Device parser options"), - fMQtree(), - fFairMQmap() +FairMQProgOptions::FairMQProgOptions() + : FairProgOptions() + , fMQParserOptions("MQ-Device parser options") + , fMQtree() + , fFairMQmap() { InitOptionDescription(); } @@ -21,18 +21,16 @@ FairMQProgOptions::~FairMQProgOptions() { } - - int FairMQProgOptions::ParseAll(const int argc, char** argv, bool AllowUnregistered) { // before parsing, define cmdline and optionally cfgfile description, // and also what is visible for the user AddToCmdLineOptions(fGenericDesc); AddToCmdLineOptions(fMQParserOptions); - + // if config file option enabled then id non required in cmdline but required in configfile // else required in cmdline - if(fUseConfigFile) + if (fUseConfigFile) { fCmdline_options.add_options() ("device-id", po::value< std::string >(), "Device ID"); @@ -40,39 +38,40 @@ int FairMQProgOptions::ParseAll(const int argc, char** argv, bool AllowUnregiste ("device-id", po::value< std::string >()->required(), "Device ID"); } else + { fCmdline_options.add_options() ("device-id", po::value< std::string >()->required(), "Device ID"); - + } + fVisible_options.add_options() ("device-id", po::value< std::string >()->required(), "Device ID (required value)"); - - + // parse command line - if(ParseCmdLine(argc,argv,fCmdline_options,fvarmap,AllowUnregistered)) + if (ParseCmdLine(argc,argv,fCmdline_options,fvarmap,AllowUnregistered)) + { return 1; - + } + // if txt/INI configuration file enabled then parse it - if(fUseConfigFile && !fConfigFile.empty()) + if (fUseConfigFile && !fConfigFile.empty()) { AddToCfgFileOptions(fMQParserOptions,false); - - if(ParseCfgFile(fConfigFile,fConfig_file_options,fvarmap,AllowUnregistered)) + + if (ParseCfgFile(fConfigFile, fConfig_file_options, fvarmap, AllowUnregistered)) + { return 1; + } } - - + // set log level before printing (default is 0 = DEBUG level) int verbose=GetValue("verbose"); SET_LOGGER_LEVEL(verbose); - + PrintOptions(); - + return 0; } - - - int FairMQProgOptions::NotifySwitchOption() { if ( fvarmap.count("help") ) @@ -99,8 +98,7 @@ void FairMQProgOptions::InitOptionDescription() ("config-json-string", po::value< std::vector >()->multitoken(), "JSON input as command line string.") ("config-json-filename", po::value< std::string >(), "JSON input as file.") - //("ini.config.string", po::value< std::vector >()->multitoken(), "INI input as command line string.") - //("ini.config.filename", po::value< std::string >(), "INI input as file.") + // ("ini.config.string", po::value< std::vector >()->multitoken(), "INI input as command line string.") + // ("ini.config.filename", po::value< std::string >(), "INI input as file.") ; - } \ No newline at end of file diff --git a/fairmq/options/FairMQProgOptions.h b/fairmq/options/FairMQProgOptions.h index c0f40ab6..cd72404f 100644 --- a/fairmq/options/FairMQProgOptions.h +++ b/fairmq/options/FairMQProgOptions.h @@ -17,21 +17,20 @@ namespace pt = boost::property_tree; class FairMQProgOptions : public FairProgOptions { protected: - typedef std::map > FairMQMap; - + typedef std::map< std::string,std::vector > FairMQMap; + public: FairMQProgOptions(); virtual ~FairMQProgOptions(); - virtual int ParseAll(const int argc, char** argv, bool AllowUnregistered=false); - - + virtual int ParseAll(const int argc, char** argv, bool AllowUnregistered = false); + // external parser, store function - template < typename T, typename ...Args> + template int UserParser(Args &&... args) { try { - Store( T().UserParser(std::forward(args)...) ); + Store(T().UserParser(std::forward(args)...)); } catch (std::exception& e) { @@ -40,38 +39,36 @@ public: } return 0; } - + int Store(const po::variables_map& vm) { fvarmap = vm; return 0; } - + int Store(const pt::ptree& tree) { fMQtree = tree; return 0; } - + int Store(const FairMQMap& channels) { fFairMQmap = channels; return 0; } - - + FairMQMap GetFairMQMap() { return fFairMQmap; } - - + protected: po::options_description fMQParserOptions; pt::ptree fMQtree; FairMQMap fFairMQmap; - - virtual int NotifySwitchOption();// for custom help & version printing + + virtual int NotifySwitchOption(); // for custom help & version printing void InitOptionDescription(); }; diff --git a/fairmq/options/FairProgOptions.h b/fairmq/options/FairProgOptions.h index 811d8dcc..0d45463b 100644 --- a/fairmq/options/FairProgOptions.h +++ b/fairmq/options/FairProgOptions.h @@ -35,7 +35,7 @@ * fVisible_options.add(fCmdline_options); * } * virtual ~MyOptions(){} - * virtual int ParseAll(const int argc, char** argv, bool AllowUnregistered=false) + * virtual int ParseAll(const int argc, char** argv, bool AllowUnregistered = false) * { * if(ParseCmdLine(argc,argv,fCmdline_options,fvarmap,AllowUnregistered)) * return 1; @@ -47,50 +47,53 @@ */ template - std::ostream& operator<<(std::ostream& os, const std::vector& v) - { - std::copy(v.begin(), v.end(), std::ostream_iterator(os, " ")); - return os; - } - +std::ostream& operator<<(std::ostream& os, const std::vector& v) +{ + std::copy(v.begin(), v.end(), std::ostream_iterator(os, " ")); + return os; +} namespace po = boost::program_options; - -class FairProgOptions +class FairProgOptions { public: FairProgOptions(); virtual ~FairProgOptions(); // add options_description - int AddToCmdLineOptions(const po::options_description& optdesc, bool visible=true); - int AddToCfgFileOptions(const po::options_description& optdesc, bool visible=true); + int AddToCmdLineOptions(const po::options_description& optdesc, bool visible = true); + int AddToCfgFileOptions(const po::options_description& optdesc, bool visible = true); int AddToEnvironmentOptions(const po::options_description& optdesc); - void EnableCfgFile(const std::string& filename="") + void EnableCfgFile(const std::string& filename = "") { - fUseConfigFile=true; - if(filename.empty()) + fUseConfigFile = true; + if (filename.empty()) + { fCmdline_options.add_options() ("config,c", po::value(&fConfigFile)->required(), "Path to configuration file"); + } else - fConfigFile=filename; - + { + fConfigFile = filename; + } } - - - void UseConfigFile(const std::string& filename="") + + void UseConfigFile(const std::string& filename = "") { - fUseConfigFile=true; - if(filename.empty()) + fUseConfigFile = true; + if (filename.empty()) + { fCmdline_options.add_options() ("config,c", po::value(&fConfigFile)->required(), "Path to configuration file"); + } else - fConfigFile=filename; - + { + fConfigFile = filename; + } } - + // set value corresponding to the key template T GetValue(const std::string& key) const @@ -98,11 +101,13 @@ public: T val; try { - if ( fvarmap.count(key) ) - val=fvarmap[key].as(); + if (fvarmap.count(key)) + { + val = fvarmap[key].as(); + } else { - LOG(ERROR) <<"Key '"<< key <<"' not found in boost variable map"; + LOG(ERROR) << "Key '"<< key <<"' not found in boost variable map"; LOG(INFO) << "Command line / txt config file options are the following : "; this->PrintHelp(); } @@ -113,7 +118,7 @@ public: LOG(ERROR) << e.what(); this->PrintHelp(); } - + return val; } @@ -122,24 +127,22 @@ public: const po::variables_map& GetVarMap() const {return fvarmap;} // boost prog options parsers - int ParseCmdLine(const int argc, char** argv, const po::options_description& desc, po::variables_map& varmap, bool AllowUnregistered=false); - int ParseCmdLine(const int argc, char** argv, const po::options_description& desc, bool AllowUnregistered=false); - - int ParseCfgFile(const std::string& filename, const po::options_description& desc, po::variables_map& varmap, bool AllowUnregistered=false); - int ParseCfgFile(const std::string& filename, const po::options_description& desc, bool AllowUnregistered=false); - int ParseCfgFile(std::ifstream& ifs, const po::options_description& desc, po::variables_map& varmap, bool AllowUnregistered=false); - int ParseCfgFile(std::ifstream& ifs, const po::options_description& desc, bool AllowUnregistered=false); - + int ParseCmdLine(const int argc, char** argv, const po::options_description& desc, po::variables_map& varmap, bool AllowUnregistered = false); + int ParseCmdLine(const int argc, char** argv, const po::options_description& desc, bool AllowUnregistered = false); + int ParseCfgFile(const std::string& filename, const po::options_description& desc, po::variables_map& varmap, bool AllowUnregistered = false); + int ParseCfgFile(const std::string& filename, const po::options_description& desc, bool AllowUnregistered = false); + int ParseCfgFile(std::ifstream& ifs, const po::options_description& desc, po::variables_map& varmap, bool AllowUnregistered = false); + int ParseCfgFile(std::ifstream& ifs, const po::options_description& desc, bool AllowUnregistered = false); + int ParseEnvironment(const std::function&); - - virtual int ParseAll(const int argc, char** argv, bool AllowUnregistered=false)=0; + + virtual int ParseAll(const int argc, char** argv, bool AllowUnregistered = false) = 0; virtual int PrintOptions(); int PrintHelp() const; protected: - // options container po::variables_map fvarmap; @@ -152,7 +155,7 @@ protected: // Description of cmd line and simple configuration file (configuration file like txt, but not like xml json ini) po::options_description fCmdline_options; po::options_description fConfig_file_options; - + // Description which is printed in help command line po::options_description fVisible_options; @@ -165,17 +168,16 @@ protected: template void UpadateVarMap(const std::string& key, const T& val) { - replace(fvarmap,key,val); + replace(fvarmap, key, val); } - + template - void replace( std::map& vm, const std::string& opt, const T& val) + void replace(std::map& vm, const std::string& opt, const T& val) { vm[opt].value() = boost::any(val); } - -private: +private: // ///////////////////////////////////////////// // Methods below are helper functions used in the PrintOptions method typedef std::tuple VarValInfo_t; @@ -184,20 +186,24 @@ private: VarValInfo_t Get_variable_value_info(const po::variable_value& var_val); template - std::string variable_value_to_string(const po::variable_value& var_val) + std::string variable_value_to_string(const po::variable_value& var_val) + { + auto& value = var_val.value(); + std::ostringstream ostr; + if (auto q = boost::any_cast(&value)) { - auto& value = var_val.value(); - std::ostringstream ostr; - if(auto q = boost::any_cast< T >(&value)) - ostr<<*q; - std::string val_str=ostr.str(); - return val_str; + ostr << *q; } + std::string val_str = ostr.str(); + return val_str; + } static void Max(int &val, const int &comp) { - if (comp>val) - val=comp; + if (comp > val) + { + val = comp; + } } }; diff --git a/fairmq/run/runBenchmarkSampler.cxx b/fairmq/run/runBenchmarkSampler.cxx index 337d73ac..a30a4f07 100644 --- a/fairmq/run/runBenchmarkSampler.cxx +++ b/fairmq/run/runBenchmarkSampler.cxx @@ -18,6 +18,8 @@ #include "boost/program_options.hpp" #include "FairMQLogger.h" +#include "FairMQParser.h" +#include "FairMQProgOptions.h" #include "FairMQBenchmarkSampler.h" #ifdef NANOMSG @@ -27,6 +29,8 @@ #endif using namespace std; +using namespace FairMQParser; +using namespace boost::program_options; FairMQBenchmarkSampler sampler; @@ -36,7 +40,7 @@ static void s_signal_handler(int signal) sampler.ChangeState(FairMQBenchmarkSampler::END); - LOG(INFO) << "Shutdown complete"; + LOG(INFO) << "Shutdown complete."; exit(1); } @@ -50,137 +54,80 @@ static void s_catch_signals(void) sigaction(SIGTERM, &action, NULL); } -typedef struct DeviceOptions -{ - DeviceOptions() : - id(), eventSize(0), eventRate(0), ioThreads(0), - outputSocketType(), outputBufSize(0), outputMethod(), outputAddress() - {} - - string id; - int eventSize; - int eventRate; - int ioThreads; - string outputSocketType; - int outputBufSize; - string outputMethod; - string outputAddress; -} DeviceOptions_t; - -inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) -{ - if (_options == NULL) - throw runtime_error("Internal error: options' container is empty."); - - namespace bpo = boost::program_options; - bpo::options_description desc("Options"); - desc.add_options() - ("id", bpo::value()->required(), "Device ID") - ("event-size", bpo::value()->default_value(1000), "Event size in bytes") - ("event-rate", bpo::value()->default_value(0), "Event rate limit in maximum number of events per second") - ("io-threads", bpo::value()->default_value(1), "Number of I/O threads") - ("output-socket-type", bpo::value()->required(), "Output socket type: pub/push") - ("output-buff-size", bpo::value()->default_value(1000), "Output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("output-method", bpo::value()->required(), "Output method: bind/connect") - ("output-address", bpo::value()->required(), "Output address, e.g.: \"tcp://*:5555\"") - ("help", "Print help messages"); - - bpo::variables_map vm; - bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm); - - if (vm.count("help")) - { - LOG(INFO) << "FairMQ Benchmark Sampler" << endl << desc; - return false; - } - - bpo::notify(vm); - - if (vm.count("id")) - _options->id = vm["id"].as(); - - if (vm.count("event-size")) - _options->eventSize = vm["event-size"].as(); - - if (vm.count("event-rate")) - _options->eventRate = vm["event-rate"].as(); - - if (vm.count("io-threads")) - _options->ioThreads = vm["io-threads"].as(); - - if (vm.count("output-socket-type")) - _options->outputSocketType = vm["output-socket-type"].as(); - - if (vm.count("output-buff-size")) - _options->outputBufSize = vm["output-buff-size"].as(); - - if (vm.count("output-method")) - _options->outputMethod = vm["output-method"].as(); - - if (vm.count("output-address")) - _options->outputAddress = vm["output-address"].as(); - - return true; -} - int main(int argc, char** argv) { s_catch_signals(); - DeviceOptions_t options; + FairMQProgOptions config; + try { - if (!parse_cmd_line(argc, argv, &options)) + int eventSize; + int eventRate; + int ioThreads; + + options_description sampler_options("Sampler options"); + sampler_options.add_options() + ("event-size", value(&eventSize)->default_value(1000), "Event size in bytes") + ("event-rate", value(&eventRate)->default_value(0), "Event rate limit in maximum number of events per second") + ("io-threads", value(&ioThreads)->default_value(1), "Number of I/O threads"); + + config.AddToCmdLineOptions(sampler_options); + + if (config.ParseAll(argc, argv)) + { return 0; + } + + string filename = config.GetValue("config-json-filename"); + string id = config.GetValue("device-id"); + + config.UserParser(filename, id); + + sampler.fChannels = config.GetFairMQMap(); + + LOG(INFO) << "PID: " << getpid(); + +#ifdef NANOMSG + FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); +#else + FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); +#endif + + sampler.SetTransport(transportFactory); + + sampler.SetProperty(FairMQBenchmarkSampler::Id, id); + sampler.SetProperty(FairMQBenchmarkSampler::EventSize, eventSize); + sampler.SetProperty(FairMQBenchmarkSampler::EventRate, eventRate); + sampler.SetProperty(FairMQBenchmarkSampler::NumIoThreads, ioThreads); + + sampler.ChangeState(FairMQBenchmarkSampler::INIT_DEVICE); + sampler.WaitForEndOfState(FairMQBenchmarkSampler::INIT_DEVICE); + + sampler.ChangeState(FairMQBenchmarkSampler::INIT_TASK); + sampler.WaitForEndOfState(FairMQBenchmarkSampler::INIT_TASK); + + sampler.ChangeState(FairMQBenchmarkSampler::RUN); + sampler.WaitForEndOfState(FairMQBenchmarkSampler::RUN); + + sampler.ChangeState(FairMQBenchmarkSampler::STOP); + + sampler.ChangeState(FairMQBenchmarkSampler::RESET_TASK); + sampler.WaitForEndOfState(FairMQBenchmarkSampler::RESET_TASK); + + sampler.ChangeState(FairMQBenchmarkSampler::RESET_DEVICE); + sampler.WaitForEndOfState(FairMQBenchmarkSampler::RESET_DEVICE); + + sampler.ChangeState(FairMQBenchmarkSampler::END); + } catch (exception& e) { LOG(ERROR) << e.what(); + LOG(INFO) << "Command line options are the following : "; + config.PrintHelp(); return 1; } - LOG(INFO) << "PID: " << getpid(); - LOG(INFO) << "CONFIG: " << "id: " << options.id << ", event size: " << options.eventSize << ", event rate: " << options.eventRate << ", I/O threads: " << options.ioThreads; - LOG(INFO) << "OUTPUT: " << options.outputSocketType << " " << options.outputBufSize << " " << options.outputMethod << " " << options.outputAddress; - -#ifdef NANOMSG - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); -#else - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); -#endif - - sampler.SetTransport(transportFactory); - - FairMQChannel channel(options.outputSocketType, options.outputMethod, options.outputAddress); - channel.fSndBufSize = options.outputBufSize; - channel.fRcvBufSize = options.outputBufSize; - channel.fRateLogging = 1; - - sampler.fChannels["data-out"].push_back(channel); - - sampler.SetProperty(FairMQBenchmarkSampler::Id, options.id); - sampler.SetProperty(FairMQBenchmarkSampler::EventSize, options.eventSize); - sampler.SetProperty(FairMQBenchmarkSampler::EventRate, options.eventRate); - sampler.SetProperty(FairMQBenchmarkSampler::NumIoThreads, options.ioThreads); - - sampler.ChangeState(FairMQBenchmarkSampler::INIT_DEVICE); - sampler.WaitForEndOfState(FairMQBenchmarkSampler::INIT_DEVICE); - - sampler.ChangeState(FairMQBenchmarkSampler::INIT_TASK); - sampler.WaitForEndOfState(FairMQBenchmarkSampler::INIT_TASK); - - sampler.ChangeState(FairMQBenchmarkSampler::RUN); - sampler.WaitForEndOfState(FairMQBenchmarkSampler::RUN); - - sampler.ChangeState(FairMQBenchmarkSampler::STOP); - - sampler.ChangeState(FairMQBenchmarkSampler::RESET_TASK); - sampler.WaitForEndOfState(FairMQBenchmarkSampler::RESET_TASK); - - sampler.ChangeState(FairMQBenchmarkSampler::RESET_DEVICE); - sampler.WaitForEndOfState(FairMQBenchmarkSampler::RESET_DEVICE); - - sampler.ChangeState(FairMQBenchmarkSampler::END); - return 0; } diff --git a/fairmq/run/runSink.cxx b/fairmq/run/runSink.cxx index ec8b99e4..3494d72f 100644 --- a/fairmq/run/runSink.cxx +++ b/fairmq/run/runSink.cxx @@ -9,7 +9,7 @@ * runSink.cxx * * @since 2013-01-21 - * @author D. Klein, A. Rybalchenko + * @author: D. Klein, A. Rybalchenko */ #include @@ -18,6 +18,8 @@ #include "boost/program_options.hpp" #include "FairMQLogger.h" +#include "FairMQParser.h" +#include "FairMQProgOptions.h" #include "FairMQSink.h" #ifdef NANOMSG @@ -27,6 +29,8 @@ #endif using namespace std; +using namespace FairMQParser; +using namespace boost::program_options; FairMQSink sink; @@ -50,123 +54,74 @@ static void s_catch_signals(void) sigaction(SIGTERM, &action, NULL); } -typedef struct DeviceOptions -{ - DeviceOptions() : - id(), ioThreads(0), - inputSocketType(), inputBufSize(0), inputMethod(), inputAddress() - {} - - string id; - int ioThreads; - string inputSocketType; - int inputBufSize; - string inputMethod; - string inputAddress; -} DeviceOptions_t; - -inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) -{ - if (_options == NULL) - throw runtime_error("Internal error: options' container is empty."); - - namespace bpo = boost::program_options; - bpo::options_description desc("Options"); - desc.add_options() - ("id", bpo::value()->required(), "Device ID") - ("io-threads", bpo::value()->default_value(1), "Number of I/O threads") - ("input-socket-type", bpo::value()->required(), "Input socket type: sub/pull") - ("input-buff-size", bpo::value()->default_value(1000), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("input-method", bpo::value()->required(), "Input method: bind/connect") - ("input-address", bpo::value()->required(), "Input address, e.g.: \"tcp://*:5555\"") - ("help", "Print help messages"); - - bpo::variables_map vm; - bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm); - - if ( vm.count("help") ) - { - LOG(INFO) << "FairMQ Sink" << endl << desc; - return false; - } - - bpo::notify(vm); - - if ( vm.count("id") ) - _options->id = vm["id"].as(); - - if ( vm.count("io-threads") ) - _options->ioThreads = vm["io-threads"].as(); - - if ( vm.count("input-socket-type") ) - _options->inputSocketType = vm["input-socket-type"].as(); - - if ( vm.count("input-buff-size") ) - _options->inputBufSize = vm["input-buff-size"].as(); - - if ( vm.count("input-method") ) - _options->inputMethod = vm["input-method"].as(); - - if ( vm.count("input-address") ) - _options->inputAddress = vm["input-address"].as(); - - return true; -} - int main(int argc, char** argv) { s_catch_signals(); - DeviceOptions_t options; + FairMQProgOptions config; + try { - if (!parse_cmd_line(argc, argv, &options)) + int ioThreads; + + options_description sink_options("Sink options"); + sink_options.add_options() + ("io-threads", value(&ioThreads)->default_value(1), "Number of I/O threads"); + + config.AddToCmdLineOptions(sink_options); + + if (config.ParseAll(argc, argv)) + { return 0; + } + + string filename = config.GetValue("config-json-filename"); + string id = config.GetValue("device-id"); + + config.UserParser(filename, id); + + sink.fChannels = config.GetFairMQMap(); + + LOG(INFO) << "PID: " << getpid(); + +#ifdef NANOMSG + FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); +#else + FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); +#endif + + sink.SetTransport(transportFactory); + + sink.SetProperty(FairMQSink::Id, id); + sink.SetProperty(FairMQSink::NumIoThreads, ioThreads); + + sink.ChangeState(FairMQSink::INIT_DEVICE); + sink.WaitForEndOfState(FairMQSink::INIT_DEVICE); + + sink.ChangeState(FairMQSink::INIT_TASK); + sink.WaitForEndOfState(FairMQSink::INIT_TASK); + + sink.ChangeState(FairMQSink::RUN); + sink.WaitForEndOfState(FairMQSink::RUN); + + sink.ChangeState(FairMQSink::STOP); + + sink.ChangeState(FairMQSink::RESET_TASK); + sink.WaitForEndOfState(FairMQSink::RESET_TASK); + + sink.ChangeState(FairMQSink::RESET_DEVICE); + sink.WaitForEndOfState(FairMQSink::RESET_DEVICE); + + sink.ChangeState(FairMQSink::END); + } catch (exception& e) { LOG(ERROR) << e.what(); + LOG(INFO) << "Started with: "; + config.PrintHelp(); return 1; } - LOG(INFO) << "PID: " << getpid(); - -#ifdef NANOMSG - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); -#else - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); -#endif - - sink.SetTransport(transportFactory); - - FairMQChannel channel(options.inputSocketType, options.inputMethod, options.inputAddress); - channel.fSndBufSize = options.inputBufSize; - channel.fRcvBufSize = options.inputBufSize; - channel.fRateLogging = 1; - - sink.fChannels["data-in"].push_back(channel); - - sink.SetProperty(FairMQSink::Id, options.id); - sink.SetProperty(FairMQSink::NumIoThreads, options.ioThreads); - - sink.ChangeState(FairMQSink::INIT_DEVICE); - sink.WaitForEndOfState(FairMQSink::INIT_DEVICE); - - sink.ChangeState(FairMQSink::INIT_TASK); - sink.WaitForEndOfState(FairMQSink::INIT_TASK); - - sink.ChangeState(FairMQSink::RUN); - sink.WaitForEndOfState(FairMQSink::RUN); - - sink.ChangeState(FairMQSink::STOP); - - sink.ChangeState(FairMQSink::RESET_TASK); - sink.WaitForEndOfState(FairMQSink::RESET_TASK); - - sink.ChangeState(FairMQSink::RESET_DEVICE); - sink.WaitForEndOfState(FairMQSink::RESET_DEVICE); - - sink.ChangeState(FairMQSink::END); - return 0; }