Add ParameterMQServer and example of its use

fairmq/CMakeLists.txt

@@ -18,6 +18,7 @@ EndIf(DDS_FOUND)
 add_subdirectory(examples/4-copypush)
 add_subdirectory(examples/5-req-rep)
 add_subdirectory(examples/6-multiple-channels)
+add_subdirectory(examples/7-parameters)
 
 add_subdirectory(test)
 

fairmq/examples/7-parameters/CMakeLists.txt

@@ -0,0 +1,96 @@
+ ################################################################################
+ #    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"                       #
+ ################################################################################
+
+configure_file(${CMAKE_SOURCE_DIR}/fairmq/examples/7-parameters/ex7-client.json ${CMAKE_BINARY_DIR}/bin/config/ex7-client.json)
+
+Set(INCLUDE_DIRECTORIES
+  ${BASE_INCLUDE_DIRECTORIES}
+  ${CMAKE_SOURCE_DIR}/fairmq
+  ${CMAKE_SOURCE_DIR}/fairmq/devices
+  ${CMAKE_SOURCE_DIR}/fairmq/tools
+  ${CMAKE_SOURCE_DIR}/fairmq/options
+  ${CMAKE_SOURCE_DIR}/fairmq/examples/7-parameters
+  ${CMAKE_CURRENT_BINARY_DIR}
+)
+
+Set(SYSTEM_INCLUDE_DIRECTORIES
+  ${SYSTEM_INCLUDE_DIRECTORIES}
+  ${Boost_INCLUDE_DIR}
+)
+
+If(NANOMSG_FOUND)
+  Set(INCLUDE_DIRECTORIES
+    ${INCLUDE_DIRECTORIES}
+    ${CMAKE_SOURCE_DIR}/fairmq/nanomsg
+  )
+  Set(SYSTEM_INCLUDE_DIRECTORIES
+    ${SYSTEM_INCLUDE_DIRECTORIES}
+    ${ZMQ_INCLUDE_DIR}
+  )
+Else(NANOMSG_FOUND)
+  Set(INCLUDE_DIRECTORIES
+    ${INCLUDE_DIRECTORIES}
+    ${CMAKE_SOURCE_DIR}/fairmq/zeromq
+  )
+  Set(SYSTEM_INCLUDE_DIRECTORIES
+    ${SYSTEM_INCLUDE_DIRECTORIES}
+    ${ZMQ_INCLUDE_DIR}
+  )
+EndIf(NANOMSG_FOUND)
+
+Include_Directories(${INCLUDE_DIRECTORIES})
+Include_Directories(SYSTEM ${SYSTEM_INCLUDE_DIRECTORIES})
+
+Set(LINK_DIRECTORIES
+  ${Boost_LIBRARY_DIRS}
+  ${ROOT_LIBRARY_DIR}
+)
+
+Link_Directories(${LINK_DIRECTORIES})
+
+set(SRCS
+  "FairMQExample7ParOne.cxx"
+  "FairMQExample7ContFact.cxx"
+)
+
+Set(NO_DICT_SRCS
+  "FairMQExample7Client.cxx"
+)
+
+set(DEPENDENCIES
+  ${DEPENDENCIES}
+  FairMQ
+  Base
+  MCStack
+)
+
+set(LINKDEF "FairMQExample7LinkDef.h")
+
+set(LIBRARY_NAME FairMQExample7)
+
+GENERATE_LIBRARY()
+
+set(Exe_Names
+  ex7-client
+)
+
+set(Exe_Source
+  runExample7Client.cxx
+)
+
+list(LENGTH Exe_Names _length)
+math(EXPR _length ${_length}-1)
+
+ForEach(_file RANGE 0 ${_length})
+  list(GET Exe_Names ${_file} _name)
+  list(GET Exe_Source ${_file} _src)
+  set(EXE_NAME ${_name})
+  set(SRCS ${_src})
+  set(DEPENDENCIES FairMQExample7)
+  GENERATE_EXECUTABLE()
+EndForEach(_file RANGE 0 ${_length})

fairmq/examples/7-parameters/FairMQExample7Client.cxx

@@ -0,0 +1,130 @@
+/********************************************************************************
+ *    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"                       *
+ ********************************************************************************/
+/**
+ * FairMQExample7Client.cpp
+ *
+ * @since 2015-10-26
+ * @author A. Rybalchenko
+ */
+
+#include <boost/thread.hpp>
+#include <boost/bind.hpp>
+
+#include "FairMQLogger.h"
+#include "FairMQExample7Client.h"
+#include "FairMQExample7ParOne.h"
+
+#include "TMessage.h"
+#include "Rtypes.h"
+
+using namespace std;
+
+FairMQExample7Client::FairMQExample7Client() :
+    fRunId(0),
+    fParameterName()
+{
+}
+
+FairMQExample7Client::~FairMQExample7Client()
+{
+}
+
+void FairMQExample7Client::CustomCleanup(void *data, void *hint)
+{
+    delete (string*)hint;
+}
+
+// special class to expose protected TMessage constructor
+class FairMQExample7TMessage : public TMessage
+{
+  public:
+    FairMQExample7TMessage(void* buf, Int_t len)
+        : TMessage(buf, len)
+    {
+        ResetBit(kIsOwner);
+    }
+};
+
+void FairMQExample7Client::Run()
+{
+    int runId = 2001;
+
+    while (CheckCurrentState(RUNNING))
+    {
+        boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
+
+        string* reqStr = new string(fParameterName + "," + to_string(runId));
+
+        LOG(INFO) << "Requesting parameter \"" << fParameterName << "\" for Run ID " << runId << ".";
+
+        unique_ptr<FairMQMessage> req(fTransportFactory->CreateMessage(const_cast<char*>(reqStr->c_str()), reqStr->length(), CustomCleanup, reqStr));
+        unique_ptr<FairMQMessage> rep(fTransportFactory->CreateMessage());
+
+        if (fChannels.at("data").at(0).Send(req) > 0)
+        {
+            if (fChannels.at("data").at(0).Receive(rep) > 0)
+            {
+                FairMQExample7TMessage tmsg(rep->GetData(), rep->GetSize());
+                FairMQExample7ParOne* par = (FairMQExample7ParOne*)(tmsg.ReadObject(tmsg.GetClass()));
+                LOG(INFO) << "Received parameter from the server:";
+                par->print();
+            }
+        }
+
+        runId++;
+        if (runId == 2101)
+        {
+            runId = 2001;
+        }
+    }
+}
+
+
+void FairMQExample7Client::SetProperty(const int key, const string& value)
+{
+    switch (key)
+    {
+        case ParameterName:
+            fParameterName = value;
+            break;
+        default:
+            FairMQDevice::SetProperty(key, value);
+            break;
+    }
+}
+
+string FairMQExample7Client::GetProperty(const int key, const string& default_ /*= ""*/)
+{
+    switch (key)
+    {
+        case ParameterName:
+            return fParameterName;
+            break;
+        default:
+            return FairMQDevice::GetProperty(key, default_);
+    }
+}
+
+void FairMQExample7Client::SetProperty(const int key, const int value)
+{
+    switch (key)
+    {
+        default:
+            FairMQDevice::SetProperty(key, value);
+            break;
+    }
+}
+
+int FairMQExample7Client::GetProperty(const int key, const int default_ /*= 0*/)
+{
+    switch (key)
+    {
+        default:
+            return FairMQDevice::GetProperty(key, default_);
+    }
+}

fairmq/examples/7-parameters/FairMQExample7Client.h

@@ -0,0 +1,48 @@
+/********************************************************************************
+ *    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"                       *
+ ********************************************************************************/
+/**
+ * FairMQExample7Client.h
+ *
+ * @since 2015-10-26
+ * @author A. Rybalchenko
+ */
+
+#ifndef FAIRMQEXAMPLE7CLIENT_H_
+#define FAIRMQEXAMPLE7CLIENT_H_
+
+#include <string>
+
+#include "FairMQDevice.h"
+
+class FairMQExample7Client : public FairMQDevice
+{
+  public:
+    enum
+    {
+        ParameterName = FairMQDevice::Last,
+        Last
+    };
+    FairMQExample7Client();
+    virtual ~FairMQExample7Client();
+
+    static void CustomCleanup(void* data, void* hint);
+
+    virtual void SetProperty(const int key, const std::string& value);
+    virtual std::string GetProperty(const int key, const std::string& default_ = "");
+    virtual void SetProperty(const int key, const int value);
+    virtual int GetProperty(const int key, const int default_ = 0);
+
+  protected:
+    virtual void Run();
+
+  private:
+    int fRunId;
+    std::string fParameterName;
+};
+
+#endif /* FAIRMQEXAMPLE7CLIENT_H_ */

fairmq/examples/7-parameters/FairMQExample7ContFact.cxx

@@ -0,0 +1,46 @@
+/********************************************************************************
+ *    Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
+ *                                                                              *
+ *              This software is distributed under the terms of the             * 
+ *         GNU Lesser General Public Licence version 3 (LGPL) version 3,        *  
+ *                  copied verbatim in the file "LICENSE"                       *
+ ********************************************************************************/
+#include <iostream>
+
+#include "FairRuntimeDb.h"
+
+#include "FairMQExample7ContFact.h"
+#include "FairMQExample7ParOne.h"
+
+static FairMQExample7ContFact gFairMQExample7ContFact;
+
+FairMQExample7ContFact::FairMQExample7ContFact()
+{
+    fName = "FairMQExample7ContFact";
+    fTitle = "Factory for parameter containers in FairMQ Example 7";
+    setAllContainers();
+    FairRuntimeDb::instance()->addContFactory(this);
+}
+
+void FairMQExample7ContFact::setAllContainers()
+{
+    FairContainer* container = new FairContainer("FairMQExample7ParOne", "FairMQExample7ParOne Parameters", "TestDefaultContext");
+    container->addContext("TestNonDefaultContext");
+
+    containers->Add(container);
+}
+
+FairParSet* FairMQExample7ContFact::createContainer(FairContainer* container)
+{
+    const char* name = container->GetName();
+    FairParSet* p = NULL;
+
+    if (strcmp(name, "FairMQExample7ParOne") == 0)
+    {
+        p = new FairMQExample7ParOne(container->getConcatName().Data(), container->GetTitle(), container->getContext());
+    }
+
+    return p;
+}
+
+ClassImp(FairMQExample7ContFact);

fairmq/examples/7-parameters/FairMQExample7ContFact.h

@@ -0,0 +1,28 @@
+/********************************************************************************
+ *    Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
+ *                                                                              *
+ *              This software is distributed under the terms of the             * 
+ *         GNU Lesser General Public Licence version 3 (LGPL) version 3,        *  
+ *                  copied verbatim in the file "LICENSE"                       *
+ ********************************************************************************/
+#ifndef FAIRMQEXAMPLE7CONTFACT_H_
+#define FAIRMQEXAMPLE7CONTFACT_H_
+
+#include "FairContFact.h"
+
+class FairContainer;
+
+class FairMQExample7ContFact : public FairContFact
+{
+  private:
+    void setAllContainers();
+
+  public:
+    FairMQExample7ContFact();
+    ~FairMQExample7ContFact() {}
+
+    FairParSet* createContainer(FairContainer*);
+    ClassDef(FairMQExample7ContFact,0)
+};
+
+#endif  /* FAIRMQEXAMPLE7CONTFACT_H_ */

fairmq/examples/7-parameters/FairMQExample7LinkDef.h

@@ -0,0 +1,17 @@
+/********************************************************************************
+ *    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"                       *
+ ********************************************************************************/
+#ifdef __CINT__
+
+#pragma link off all globals;
+#pragma link off all classes;
+#pragma link off all functions;
+
+#pragma link C++ class FairMQExample7ContFact+;
+#pragma link C++ class FairMQExample7ParOne+;
+
+#endif

fairmq/examples/7-parameters/FairMQExample7ParOne.cxx

@@ -0,0 +1,70 @@
+/********************************************************************************
+ *    Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
+ *                                                                              *
+ *              This software is distributed under the terms of the             * 
+ *         GNU Lesser General Public Licence version 3 (LGPL) version 3,        *  
+ *                  copied verbatim in the file "LICENSE"                       *
+ ********************************************************************************/
+#include "FairMQExample7ParOne.h"
+
+#include "FairParamList.h"
+#include "FairDetParIo.h"
+
+#include "FairLogger.h"
+#include "TString.h"
+
+FairMQExample7ParOne::FairMQExample7ParOne(const char* name, const char* title, const char* context) :
+    FairParGenericSet(name, title, context),
+    fParameterValue(0)
+{
+    detName = "TutorialDet";
+}
+
+FairMQExample7ParOne::~FairMQExample7ParOne()
+{
+    clear();
+}
+
+void FairMQExample7ParOne::clear()
+{
+    status = kFALSE;
+    resetInputVersions();
+}
+
+void FairMQExample7ParOne::print()
+{
+    LOG(INFO) << "Print" << FairLogger::endl;
+    LOG(INFO) << "fParameterValue: " << fParameterValue << FairLogger::endl;
+}
+
+void FairMQExample7ParOne::putParams(FairParamList* list)
+{
+    LOG(INFO) << "FairMQExample7ParOne::putParams()" << FairLogger::endl;
+
+    if (!list)
+    {
+        return;
+    }
+
+    list->add("Example7ParameterValue", fParameterValue);
+}
+
+
+Bool_t FairMQExample7ParOne::getParams(FairParamList* list)
+{
+    LOG(INFO) << "FairMQExample7ParOne::getParams()" << FairLogger::endl;
+
+    if (!list)
+    {
+        return kFALSE;
+    }
+
+    if (!list->fill("Example7ParameterValue", &fParameterValue))
+    {
+        return kFALSE;
+    }
+
+    return kTRUE;
+}
+
+ClassImp(FairMQExample7ParOne)

fairmq/examples/7-parameters/FairMQExample7ParOne.h

@@ -0,0 +1,47 @@
+/********************************************************************************
+ *    Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
+ *                                                                              *
+ *              This software is distributed under the terms of the             * 
+ *         GNU Lesser General Public Licence version 3 (LGPL) version 3,        *  
+ *                  copied verbatim in the file "LICENSE"                       *
+ ********************************************************************************/
+#ifndef FAIRMQEXAMPLE7PARONE_H_
+#define FAIRMQEXAMPLE7PARONE_H_
+
+#include "FairParGenericSet.h"
+
+#include "TObject.h"
+
+class FairParamList;
+
+class FairMQExample7ParOne : public FairParGenericSet
+{
+  public:
+    /** Standard constructor **/
+    FairMQExample7ParOne(const char* name    = "FairMQExample7ParOne",
+                         const char* title   = "FairMQ Example 7 Parameter One",
+                         const char* context = "Default");
+
+    /** Destructor **/
+    virtual ~FairMQExample7ParOne();
+
+    virtual void print();
+
+    /** Reset all parameters **/
+    virtual void clear();
+
+    void putParams(FairParamList* list);
+    Bool_t getParams(FairParamList* list);
+
+    inline void SetValue(const Int_t& val) { fParameterValue = val; }
+
+  private:
+    Int_t fParameterValue; //
+
+    FairMQExample7ParOne(const FairMQExample7ParOne&);
+    FairMQExample7ParOne& operator=(const FairMQExample7ParOne&);
+
+    ClassDef(FairMQExample7ParOne,1);
+};
+
+#endif // FAIRMQEXAMPLE7PARONE_H_

fairmq/examples/7-parameters/README.md

@@ -0,0 +1,10 @@
+Example 7: Communicating with ParameterMQServer
+===============
+
+This example shows how to communicate with the ParameterMQServer, that retrieves parameters from FairRuntimeDb.
+
+The `fill_parameters.C` ROOT macro can be used to generate the parameter file for the server to read from. The generated file will contain paramters with the name `FairMQExample7ParOne` for run IDs 2000-2099.
+
+FairMQExample7Client device requests parameter data from ParameterMQServer via REQ-REP pattern. The request contains parameter name and run ID.
+
+The parameter name can be configured via command line (`--parameter-name`, default is `FairMQExample7ParOne`) and the run ID is hardcoded for this example.

fairmq/examples/7-parameters/ex7-client.json

@@ -0,0 +1,23 @@
+{
+    "fairMQOptions":
+    {
+        "device":
+        {
+            "id": "parmq-client",
+            "channel":
+            {
+                "name": "data",
+                "socket":
+                {
+                    "type": "req",
+                    "method": "connect",
+                    "address": "tcp://localhost:5005",
+                    "sndBufSize": "1000",
+                    "rcvBufSize": "1000",
+                    "rateLogging": "0"
+                }
+            }
+        }
+    }
+}
+

fairmq/examples/7-parameters/fill_parameters.C

@@ -0,0 +1,24 @@
+{
+  FairRuntimeDb *rtdb = FairRuntimeDb::instance();
+
+  Bool_t kParameterMerged = kTRUE;
+  FairParRootFileIo* parOut = new FairParRootFileIo(kParameterMerged);
+  parOut->open("mqexample7_param.root");
+  rtdb->setOutput(parOut);
+  //rtdb->saveOutput();
+  //rtdb->print();
+
+  FairMQExample7ParOne *par = rtdb->getContainer("FairMQExample7ParOne");
+
+  for(Int_t i = 0; i < 100; i++)
+  {
+    rtdb->addRun(2000 + i);
+
+    par->SetValue(1983 + i);
+    par->setChanged();
+
+    rtdb->saveOutput();
+  }
+
+  rtdb->print();
+}

fairmq/examples/7-parameters/read_parameters.C

@@ -0,0 +1,19 @@
+{
+  FairRuntimeDb *rtdb = FairRuntimeDb::instance();
+
+  Bool_t kParameterMerged = kTRUE;
+  FairParRootFileIo* parOut = new FairParRootFileIo(kParameterMerged);
+  parOut->open("mqexample7_param.root");
+  rtdb->setFirstInput(parOut);
+  //rtdb->saveOutput();
+  //rtdb->print();
+
+  FairMQExample7ParOne *par = rtdb->getContainer("FairMQExample7ParOne");
+
+  for(Int_t i = 0; i < 100; i++)
+  {
+    rtdb->initContainers(2000+i);
+
+    par->print();
+  }
+}

fairmq/examples/7-parameters/runExample7Client.cxx

@@ -0,0 +1,99 @@
+/********************************************************************************
+ *    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"                       *
+ ********************************************************************************/
+/**
+ * runExample7Client.cxx
+ *
+ * @since 2013-04-23
+ * @author D. Klein, A. Rybalchenko
+ */
+
+#include <iostream>
+
+#include "FairMQLogger.h"
+#include "FairMQParser.h"
+#include "FairMQProgOptions.h"
+#include "FairMQExample7Client.h"
+
+#ifdef NANOMSG
+#include "FairMQTransportFactoryNN.h"
+#else
+#include "FairMQTransportFactoryZMQ.h"
+#endif
+
+using namespace std;
+using namespace boost::program_options;
+
+int main(int argc, char** argv)
+{
+    FairMQExample7Client client;
+    client.CatchSignals();
+
+    FairMQProgOptions config;
+
+    try
+    {
+        string parameterName;
+
+        options_description clientOptions("Parameter Client options");
+        clientOptions.add_options()
+            ("parameter-name", value<string>(&parameterName)->default_value("FairMQExample7ParOne"), "Parameter Name");
+
+        config.AddToCmdLineOptions(clientOptions);
+
+        if (config.ParseAll(argc, argv))
+        {
+            return 0;
+        }
+
+        string file = config.GetValue<string>("config-json-file");
+        string id = config.GetValue<string>("id");
+
+        config.UserParser<FairMQParser::JSON>(file, id);
+
+        client.fChannels = config.GetFairMQMap();
+
+        LOG(INFO) << "PID: " << getpid();
+
+#ifdef NANOMSG
+        FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+#else
+        FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
+#endif
+
+        client.SetTransport(transportFactory);
+
+        client.SetProperty(FairMQExample7Client::Id, "client");
+        client.SetProperty(FairMQExample7Client::ParameterName, parameterName);
+        client.SetProperty(FairMQExample7Client::NumIoThreads, 1);
+
+        FairMQChannel requestChannel("req", "connect", "tcp://localhost:5005");
+        requestChannel.UpdateSndBufSize(10000);
+        requestChannel.UpdateRcvBufSize(10000);
+        requestChannel.UpdateRateLogging(0);
+
+        client.fChannels["data"].push_back(requestChannel);
+
+        client.ChangeState("INIT_DEVICE");
+        client.WaitForEndOfState("INIT_DEVICE");
+
+        client.ChangeState("INIT_TASK");
+        client.WaitForEndOfState("INIT_TASK");
+
+        client.ChangeState("RUN");
+        client.InteractiveStateLoop();
+    }
+    catch (exception& e)
+    {
+        LOG(ERROR) << e.what();
+        LOG(INFO) << "Command line options are the following: ";
+        config.PrintHelp();
+        return 1;
+    }
+
+    return 0;
+}

fairmq/run/runBenchmarkSampler.cxx

@@ -58,7 +58,6 @@ int main(int argc, char** argv)
 
         string filename = config.GetValue<string>("config-json-file");
         string id = config.GetValue<string>("id");
-        int ioThreads = config.GetValue<int>("io-threads");
 
         config.UserParser<JSON>(filename, id);
 
@@ -75,7 +74,7 @@ int main(int argc, char** argv)
         sampler.SetProperty(FairMQBenchmarkSampler::Id, id);
         sampler.SetProperty(FairMQBenchmarkSampler::EventSize, eventSize);
         sampler.SetProperty(FairMQBenchmarkSampler::EventRate, eventRate);
-        sampler.SetProperty(FairMQBenchmarkSampler::NumIoThreads, ioThreads);
+        sampler.SetProperty(FairMQBenchmarkSampler::NumIoThreads, config.GetValue<int>("io-threads"));
 
         sampler.ChangeState("INIT_DEVICE");
         sampler.WaitForEndOfState("INIT_DEVICE");

fairmq/run/runSink.cxx

@@ -64,7 +64,7 @@ int main(int argc, char** argv)
         sink.SetTransport(transportFactory);
 
         sink.SetProperty(FairMQSink::Id, id);
-        sink.SetProperty(FairMQSink::NumIoThreads, ioThreads);
+        sink.SetProperty(FairMQSink::NumIoThreads, config.GetValue<int>("io-threads"));
 
         sink.ChangeState("INIT_DEVICE");
         sink.WaitForEndOfState("INIT_DEVICE");