Add qc example

examples/CMakeLists.txt

@@ -15,7 +15,8 @@ add_subdirectory(multipart)
 add_subdirectory(multiple-channels)
 if(BUILD_NANOMSG_TRANSPORT)
   add_subdirectory(multiple-transports)
-endif()
+  endif()
+add_subdirectory(qc)
 add_subdirectory(readout)
 add_subdirectory(region)
 add_subdirectory(req-rep)

examples/qc/CMakeLists.txt

@@ -0,0 +1,71 @@
+################################################################################
+#    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"                       #
+################################################################################
+
+add_executable(fairmq-ex-qc-sampler runSampler.cxx)
+target_link_libraries(fairmq-ex-qc-sampler PRIVATE FairMQ)
+
+add_executable(fairmq-ex-qc-producer runQCProducer.cxx)
+target_link_libraries(fairmq-ex-qc-producer PRIVATE FairMQ)
+
+add_executable(fairmq-ex-qc-consumer runQCConsumer.cxx)
+target_link_libraries(fairmq-ex-qc-consumer PRIVATE FairMQ)
+
+add_executable(fairmq-ex-qc-sink runSink.cxx)
+target_link_libraries(fairmq-ex-qc-sink PRIVATE FairMQ)
+
+add_custom_target(ExampleQC DEPENDS fairmq-ex-qc-sampler fairmq-ex-qc-producer fairmq-ex-qc-consumer fairmq-ex-qc-sink)
+
+list(JOIN Boost_LIBRARY_DIRS ":" LIB_DIR)
+set(BIN_DIR ${CMAKE_CURRENT_BINARY_DIR}:${CMAKE_BINARY_DIR}/fairmq/plugins/DDS)
+set(DATA_DIR ${CMAKE_CURRENT_BINARY_DIR})
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/ex-qc-topology.xml ${CMAKE_CURRENT_BINARY_DIR}/ex-qc-topology.xml @ONLY)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-ex-qc-env.sh ${CMAKE_CURRENT_BINARY_DIR}/fairmq-ex-qc-env.sh @ONLY)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-qc.sh.in ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-qc.sh @ONLY)
+
+# test
+if(DDS_FOUND)
+add_test(NAME Example.QC.localhost COMMAND ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-qc.sh localhost)
+set_tests_properties(Example.QC.localhost PROPERTIES TIMEOUT 15 PASS_REGULAR_EXPRESSION "Example successful")
+endif()
+
+# install
+install(
+    TARGETS
+    fairmq-ex-qc-sampler
+    fairmq-ex-qc-producer
+    fairmq-ex-qc-consumer
+    fairmq-ex-qc-sink
+
+    LIBRARY DESTINATION ${PROJECT_INSTALL_LIBDIR}
+    RUNTIME DESTINATION ${PROJECT_INSTALL_BINDIR}
+)
+
+# configure run script with different executable paths for build and for install directories
+set(BIN_DIR ${CMAKE_INSTALL_PREFIX}/${PROJECT_INSTALL_BINDIR})
+set(DATA_DIR ${CMAKE_INSTALL_PREFIX}/${PROJECT_INSTALL_DATADIR})
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/ex-qc-topology.xml ${CMAKE_CURRENT_BINARY_DIR}/ex-qc-topology.xml_install @ONLY)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-ex-qc-env.sh ${CMAKE_CURRENT_BINARY_DIR}/fairmq-ex-qc-env.sh_install @ONLY)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-qc.sh.in ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-qc.sh_install @ONLY)
+
+install(
+    FILES ${CMAKE_CURRENT_BINARY_DIR}/ex-qc-topology.xml_install
+    DESTINATION ${PROJECT_INSTALL_DATADIR}
+    RENAME ex-qc-topology.xml
+)
+
+install(
+    PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/fairmq-ex-qc-env.sh_install
+    DESTINATION ${PROJECT_INSTALL_BINDIR}
+    RENAME fairmq-ex-qc-env.sh
+)
+
+install(
+    PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-qc.sh_install
+    DESTINATION ${PROJECT_INSTALL_BINDIR}
+    RENAME fairmq-start-ex-qc.sh
+)

examples/qc/ex-qc-topology.xml

@@ -0,0 +1,48 @@
+<topology name="ExampleQC">
+
+    <property name="fmqchan_data1" />
+    <property name="fmqchan_data2" />
+    <property name="fmqchan_qc" />
+
+    <decltask name="Sampler">
+        <exe>fairmq-ex-qc-sampler --color false --channel-config name=data1,type=push,method=bind -P dds --max-iterations 1000</exe>
+        <env reachable="false">fairmq-ex-qc-env.sh</env>
+        <properties>
+            <name access="write">fmqchan_data1</name>
+        </properties>
+    </decltask>
+
+    <decltask name="QCProducer">
+        <exe>fairmq-ex-qc-producer --color false --channel-config name=data1,type=pull,method=connect name=data2,type=push,method=connect name=qc,type=push,method=connect -P dds</exe>
+        <env reachable="false">fairmq-ex-qc-env.sh</env>
+        <properties>
+            <name access="read">fmqchan_data1</name>
+            <name access="read">fmqchan_data2</name>
+            <name access="read">fmqchan_qc</name>
+        </properties>
+    </decltask>
+
+    <decltask name="QCConsumer">
+        <exe>fairmq-ex-qc-consumer --color false --channel-config name=qc,type=pull,method=bind -P dds</exe>
+        <env reachable="false">fairmq-ex-qc-env.sh</env>
+        <properties>
+            <name access="write">fmqchan_qc</name>
+        </properties>
+    </decltask>
+
+    <decltask name="Sink">
+        <exe>fairmq-ex-qc-sink --color false --channel-config name=data2,type=pull,method=bind -P dds --max-iterations 1000</exe>
+        <env reachable="false">fairmq-ex-qc-env.sh</env>
+        <properties>
+            <name access="write">fmqchan_data2</name>
+        </properties>
+    </decltask>
+
+    <main name="main">
+        <task>Sampler</task>
+        <task>QCProducer</task>
+        <task>QCConsumer</task>
+        <task>Sink</task>
+    </main>
+
+</topology>

examples/qc/fairmq-ex-qc-env.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+
+################################################################################
+#    Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    #
+#                                                                              #
+#              This software is distributed under the terms of the             #
+#              GNU Lesser General Public Licence (LGPL) version 3,             #
+#                  copied verbatim in the file "LICENSE"                       #
+################################################################################
+
+export PATH=@BIN_DIR@:$PATH
+
+OS=$(uname -s 2>&1)
+if [ "$OS" == "Darwin" ]; then
+   export DYLD_LIBRARY_PATH=@LIB_DIR@:$DYLD_LIBRARY_PATH
+fi

examples/qc/fairmq-start-ex-qc.sh.in

@@ -0,0 +1,79 @@
+#!/bin/bash
+
+################################################################################
+#    Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    #
+#                                                                              #
+#              This software is distributed under the terms of the             #
+#              GNU Lesser General Public Licence (LGPL) version 3,             #
+#                  copied verbatim in the file "LICENSE"                       #
+################################################################################
+
+# fairmq-start-ex-qc.sh -> submit agents with localhost plugin
+
+set -e
+
+cleanup() {
+  dds-session stop $1
+  echo "CLEANUP PERFORMED"
+}
+
+source @DDS_INSTALL_PREFIX@/DDS_env.sh
+export PATH=@BIN_DIR@:$PATH
+
+exec 5>&1
+output=$(dds-session start | tee >(cat - >&5))
+export FAIRMQ_DDS_SESSION_ID=$(echo ${output} | grep "DDS session ID: " | cut -d' ' -f4)
+echo "SESSION ID: ${FAIRMQ_DDS_SESSION_ID}"
+
+trap "cleanup ${FAIRMQ_DDS_SESSION_ID}" EXIT
+
+requiredNofSlots=4
+dds-submit -r localhost --slots ${requiredNofSlots}
+echo "...waiting for ${requiredNofSlots} idle slots..."
+dds-info --idle-count --wait ${requiredNofSlots}
+
+export FAIRMQ_DDS_TOPO_FILE=@DATA_DIR@/ex-qc-topology.xml
+echo "TOPOLOGY FILE: ${FAIRMQ_DDS_TOPO_FILE}"
+echo "TOPOLOGY NAME: $(dds-topology --disable-validation --topology-name ${FAIRMQ_DDS_TOPO_FILE})"
+
+dds-info --active-topology
+dds-topology --activate ${FAIRMQ_DDS_TOPO_FILE}
+dds-info --active-topology
+echo "...waiting for ${requiredNofSlots} executing slots..."
+dds-info --executing-count --wait ${requiredNofSlots}
+
+echo "------------------------"
+echo "...waiting for Topology to finish..."
+# TODO Retrieve number of devices from DDS topology API instead of having the user pass it explicitely
+fairmq-dds-command-ui -w "IDLE" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c i -w "INITIALIZING DEVICE" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c k -w "INITIALIZED" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c b -w "BOUND" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c x -w "DEVICE READY" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c j -w "READY" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c r
+sampler_and_sink="main/(Sampler|Sink)"
+# processors="main/ProcessorGroup/Processor"
+fairmq-dds-command-ui -p $sampler_and_sink -w "RUNNING->READY" -n 2
+echo "...$sampler_and_sink are READY, sending shutdown..."
+fairmq-dds-command-ui -c s -w "RUNNING->READY" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c t -w "DEVICE READY" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c d -w "IDLE" -n ${requiredNofSlots}
+fairmq-dds-command-ui -c q -w "EXITING" -n ${requiredNofSlots}
+echo "...waiting for ${requiredNofSlots} idle slots..."
+dds-info --idle-count --wait ${requiredNofSlots}
+echo "------------------------"
+
+dds-info --active-topology
+dds-topology --stop
+dds-info --active-topology
+
+dds-agent-cmd getlog -a
+logDir="${wrkDir}/logs"
+for file in $(find "${logDir}" -name "*.tar.gz"); do tar -xf ${file} -C "${logDir}" ; done
+echo "AGENT LOG FILES IN: ${logDir}"
+
+# This string is used by ctest to detect success
+echo "Example successful :)"
+
+# Cleanup function is called by EXIT trap

examples/qc/runQCConsumer.cxx

@@ -0,0 +1,26 @@
+/********************************************************************************
+ *    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"                       *
+ ********************************************************************************/
+
+#include "runFairMQDevice.h"
+#include "FairMQDevice.h"
+
+class QCConsumer : public FairMQDevice
+{
+  public:
+    QCConsumer()
+    {
+        OnData("qc", [](FairMQMessagePtr& /*msg*/, int){
+            LOG(info) << "received data";
+            return true;
+        });
+    }
+};
+
+namespace bpo = boost::program_options;
+void addCustomOptions(bpo::options_description& /*options*/) {}
+FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/) { return new QCConsumer(); }

examples/qc/runQCProducer.cxx

@@ -0,0 +1,66 @@
+/********************************************************************************
+ *    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"                       *
+ ********************************************************************************/
+
+#include "runFairMQDevice.h"
+#include "FairMQDevice.h"
+
+class QCProducer : public FairMQDevice
+{
+  public:
+    QCProducer()
+        : fDoQC(false)
+        , fCounter(0)
+        , fInterval(100)
+    {
+        OnData("data1", &QCProducer::HandleData);
+    }
+
+    void InitTask() override
+    {
+        GetConfig()->Subscribe<std::string>("qc", [&](const std::string& key, std::string value) {
+            if (key == "qc") {
+                if (value == "active") {
+                    fDoQC.store(true);
+                } else if (value == "inactive") {
+                    fDoQC.store(false);
+                }
+            }
+        });
+    }
+
+  protected:
+    bool HandleData(FairMQMessagePtr& msg, int)
+    {
+        if (fDoQC.load() == true) {
+            if (++fCounter == fInterval) {
+                fCounter = 0;
+                FairMQMessagePtr msgCopy(NewMessage());
+                msgCopy->Copy(*msg);
+                if (Send(msg, "qc") < 0) {
+                    return false;
+                }
+            }
+        }
+
+        if (Send(msg, "data2") < 0) {
+            return false;
+        }
+
+        return true;
+    }
+
+    void ResetTask() override { GetConfig()->Unsubscribe<std::string>("qc"); }
+
+  private:
+    std::atomic<bool> fDoQC;
+    int fCounter;
+    int fInterval;
+};
+
+void addCustomOptions(boost::program_options::options_description& /*options*/) {}
+FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/) { return new QCProducer(); }

examples/qc/runSampler.cxx

@@ -0,0 +1,55 @@
+/********************************************************************************
+ *    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"                       *
+ ********************************************************************************/
+
+#include "runFairMQDevice.h"
+#include "FairMQDevice.h"
+
+#include <thread> // this_thread::sleep_for
+#include <chrono>
+
+class Sampler : public FairMQDevice
+{
+  public:
+    Sampler()
+        : fMaxIterations(0)
+        , fNumIterations(0)
+    {}
+
+  protected:
+    uint64_t fMaxIterations;
+    uint64_t fNumIterations;
+
+    virtual void InitTask()
+    {
+        fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
+    }
+
+    virtual bool ConditionalRun()
+    {
+        FairMQMessagePtr msg(NewMessage(1000));
+
+        if (Send(msg, "data1") < 0) {
+            return false;
+        } else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
+            LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
+            return false;
+        }
+
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+
+        return true;
+    }
+};
+
+namespace bpo = boost::program_options;
+void addCustomOptions(bpo::options_description& options)
+{
+    options.add_options()
+        ("max-iterations", bpo::value<uint64_t>()->default_value(0), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)");
+}
+FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/) { return new Sampler(); }

examples/qc/runSink.cxx

@@ -0,0 +1,51 @@
+/********************************************************************************
+ *    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"                       *
+ ********************************************************************************/
+
+#include "runFairMQDevice.h"
+#include "FairMQDevice.h"
+
+#include <string>
+
+class Sink : public FairMQDevice
+{
+  public:
+    Sink()
+        : fMaxIterations(0)
+        , fNumIterations(0)
+    {
+        OnData("data2", &Sink::HandleData);
+    }
+
+  protected:
+    virtual void InitTask()
+    {
+        fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
+    }
+
+    bool HandleData(FairMQMessagePtr& /*msg*/, int /*index*/)
+    {
+        if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
+            LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
+            return false;
+        }
+
+        return true;
+    }
+
+  private:
+    uint64_t fMaxIterations;
+    uint64_t fNumIterations;
+};
+
+namespace bpo = boost::program_options;
+void addCustomOptions(bpo::options_description& options)
+{
+    options.add_options()
+        ("max-iterations", bpo::value<uint64_t>()->default_value(0), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)");
+}
+FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/) { return new Sink(); }