Add example/test for built-in devices

examples/CMakeLists.txt

@@ -8,6 +8,7 @@
 
 add_subdirectory(1-1)
 add_subdirectory(1-n-1)
+add_subdirectory(builtin-devices)
 add_subdirectory(copypush)
 add_subdirectory(dds)
 add_subdirectory(multipart)

examples/builtin-devices/CMakeLists.txt

@@ -0,0 +1,50 @@
+ ################################################################################
+ #    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"                       #
+ ################################################################################
+
+set(EX_BIN_DIR ${CMAKE_CURRENT_BINARY_DIR})
+set(FAIRMQ_BIN_DIR ${CMAKE_BINARY_DIR}/fairmq)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-builtin-devices.sh.in ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-builtin-devices.sh)
+
+# test
+
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/test-ex-builtin-devices.sh.in ${CMAKE_CURRENT_BINARY_DIR}/test-ex-builtin-devices.sh)
+
+add_test(NAME Example-Builtin-Devices-zeromq COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-builtin-devices.sh zeromq)
+set_tests_properties(Example-Builtin-Devices-zeromq PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Configured maximum number of iterations reached")
+
+if(BUILD_NANOMSG_TRANSPORT)
+  add_test(NAME Example-Builtin-Devices-nanomsg COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-builtin-devices.sh nanomsg)
+  set_tests_properties(Example-Builtin-Devices-nanomsg PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Configured maximum number of iterations reached")
+endif()
+
+add_test(NAME Example-Builtin-Devices-shmem COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-builtin-devices.sh shmem)
+set_tests_properties(Example-Builtin-Devices-shmem PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Configured maximum number of iterations reached")
+
+add_test(NAME Example-Builtin-Devices-zeromq-multipart COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-builtin-devices.sh zeromq true 2)
+set_tests_properties(Example-Builtin-Devices-zeromq-multipart PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Configured maximum number of iterations reached")
+
+if(BUILD_NANOMSG_TRANSPORT)
+  add_test(NAME Example-Builtin-Devices-nanomsg-multipart COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-builtin-devices.sh nanomsg true 2)
+  set_tests_properties(Example-Builtin-Devices-nanomsg-multipart PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Configured maximum number of iterations reached")
+endif()
+
+add_test(NAME Example-Builtin-Devices-shmem-multipart COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-builtin-devices.sh shmem true 2)
+set_tests_properties(Example-Builtin-Devices-shmem-multipart PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Configured maximum number of iterations reached")
+
+# install
+
+# configure run script with different executable paths for build and for install directories
+set(EX_BIN_DIR ${CMAKE_INSTALL_PREFIX}/${PROJECT_INSTALL_BINDIR})
+set(FAIRMQ_BIN_DIR ${CMAKE_INSTALL_PREFIX}/${PROJECT_INSTALL_BINDIR}/fairmq)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-builtin-devices.sh.in ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-builtin-devices.sh_install)
+
+install(
+    PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-builtin-devices.sh_install
+    DESTINATION ${PROJECT_INSTALL_BINDIR}
+    RENAME fairmq-start-ex-builtin-devices.sh
+)

examples/builtin-devices/README.md

@@ -0,0 +1,4 @@
+Built-in devices
+==========================
+
+This example demonstrates use of generic devices that are provided with FairMQ - BenchmarkSampler, Merger, Multiplier, Proxy, Sink, Splitter. They are all connected in one topology and transfer some dummy buffers generated by the BenchmarkSampler.

examples/builtin-devices/fairmq-start-ex-builtin-devices.sh.in

@@ -0,0 +1,93 @@
+#!/bin/bash
+
+export FAIRMQ_PATH=@FAIRMQ_BIN_DIR@
+
+transport="zeromq"
+multipart="false"
+numParts="1"
+
+if [[ $1 =~ ^[a-z]+$ ]]; then
+    transport=$1
+fi
+
+if [[ $2 =~ ^[a-z]+$ ]]; then
+    multipart=$2
+fi
+
+if [[ $3 =~ ^[0-9]+$ ]]; then
+    numParts=$3
+fi
+
+SAMPLER="fairmq-bsampler"
+SAMPLER+=" --id bsampler1"
+SAMPLER+=" --transport $transport"
+SAMPLER+=" --control interactive"
+SAMPLER+=" --severity debug"
+SAMPLER+=" --msg-size 100000"
+SAMPLER+=" --multipart $multipart"
+SAMPLER+=" --num-parts $numParts"
+SAMPLER+=" --msg-rate 100"
+SAMPLER+=" --max-iterations 0"
+SAMPLER+=" --out-channel data1"
+SAMPLER+=" --channel-config name=data1,type=push,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5555"
+xterm -geometry 90x20+0+175 -hold -e @FAIRMQ_BIN_DIR@/$SAMPLER &
+
+SPLITTER="fairmq-splitter"
+SPLITTER+=" --id splitter"
+SPLITTER+=" --transport $transport"
+SPLITTER+=" --multipart $multipart"
+SPLITTER+=" --in-channel data1"
+SPLITTER+=" --out-channel data2"
+SPLITTER+=" --channel-config name=data1,type=pull,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5555"
+SPLITTER+="                  name=data2,type=push,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5556,address=tcp://localhost:5557"
+xterm -geometry 90x20+0+475 -hold -e @FAIRMQ_BIN_DIR@/$SPLITTER &
+
+PROXY1="fairmq-proxy"
+PROXY1+=" --id proxy1"
+PROXY1+=" --transport $transport"
+PROXY1+=" --multipart $multipart"
+PROXY1+=" --in-channel data2"
+PROXY1+=" --out-channel data3"
+PROXY1+=" --channel-config name=data2,type=pull,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5556"
+PROXY1+="                  name=data3,type=push,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5558"
+xterm -geometry 90x20+550+175 -hold -e @FAIRMQ_BIN_DIR@/$PROXY1 &
+
+PROXY2="fairmq-proxy"
+PROXY2+=" --id proxy2"
+PROXY2+=" --transport $transport"
+PROXY2+=" --multipart $multipart"
+PROXY2+=" --in-channel data2"
+PROXY2+=" --out-channel data3"
+PROXY2+=" --channel-config name=data2,type=pull,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5557"
+PROXY2+="                  name=data3,type=push,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5559"
+xterm -geometry 90x20+550+475 -hold -e @FAIRMQ_BIN_DIR@/$PROXY2 &
+
+MERGER="fairmq-merger"
+MERGER+=" --id merger"
+MERGER+=" --transport $transport"
+MERGER+=" --multipart $multipart"
+MERGER+=" --in-channel data3"
+MERGER+=" --out-channel data4"
+MERGER+=" --channel-config name=data3,type=pull,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5558,address=tcp://localhost:5559"
+MERGER+="                  name=data4,type=push,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5560"
+xterm -geometry 90x20+1100+50 -hold -e @FAIRMQ_BIN_DIR@/$MERGER &
+
+MULTIPLIER="fairmq-multiplier"
+MULTIPLIER+=" --id multiplier"
+MULTIPLIER+=" --transport $transport"
+MULTIPLIER+=" --multipart $multipart"
+MULTIPLIER+=" --in-channel data4"
+MULTIPLIER+=" --out-channel data5"
+MULTIPLIER+=" --channel-config name=data4,type=pull,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5560"
+MULTIPLIER+="                  name=data5,type=push,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5561,address=tcp://localhost:5561"
+xterm -geometry 90x20+1100+350 -hold -e @FAIRMQ_BIN_DIR@/$MULTIPLIER &
+
+SINK="fairmq-sink"
+SINK+=" --id sink1"
+SINK+=" --transport $transport"
+SINK+=" --severity debug"
+SINK+=" --multipart $multipart"
+SINK+=" --max-iterations 0"
+SINK+=" --in-channel data5"
+SINK+=" --channel-config name=data5,type=pull,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5561"
+xterm -geometry 90x20+1100+650 -hold -e @FAIRMQ_BIN_DIR@/$SINK &

examples/builtin-devices/test-ex-builtin-devices.sh.in

@@ -0,0 +1,148 @@
+#!/bin/bash
+
+export FAIRMQ_PATH=@FAIRMQ_BIN_DIR@
+
+SESSION="$(@CMAKE_BINARY_DIR@/fairmq/fairmq-uuid-gen -h)"
+
+transport="zeromq"
+multipart="false"
+numParts="1"
+
+if [[ $1 =~ ^[a-z]+$ ]]; then
+    transport=$1
+fi
+
+if [[ $2 =~ ^[a-z]+$ ]]; then
+    multipart=$2
+fi
+
+if [[ $3 =~ ^[0-9]+$ ]]; then
+    numParts=$3
+fi
+
+# setup a trap to kill everything if the test fails/timeouts
+trap 'kill -TERM $SAMPLER_PID; kill -TERM $SPLITTER_PID; kill -TERM $PROXY1_PID; kill -TERM $PROXY2_PID; kill -TERM $MERGER_PID; kill -TERM $MULTIPLIER_PID; kill -TERM $SINK_PID;' TERM
+
+SAMPLER="fairmq-bsampler"
+SAMPLER+=" --id bsampler1"
+SAMPLER+=" --session $SESSION"
+SAMPLER+=" --transport $transport"
+SAMPLER+=" --color false"
+SAMPLER+=" --control static"
+SAMPLER+=" --verbosity veryhigh"
+SAMPLER+=" --severity debug"
+SAMPLER+=" --msg-size 100000"
+SAMPLER+=" --multipart $multipart"
+SAMPLER+=" --num-parts $numParts"
+SAMPLER+=" --msg-rate 1"
+SAMPLER+=" --max-iterations 0"
+SAMPLER+=" --out-channel data1"
+SAMPLER+=" --channel-config name=data1,type=push,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5555"
+@FAIRMQ_BIN_DIR@/$SAMPLER &
+SAMPLER_PID=$!
+
+SPLITTER="fairmq-splitter"
+SPLITTER+=" --id splitter"
+SPLITTER+=" --session $SESSION"
+SPLITTER+=" --transport $transport"
+SPLITTER+=" --color false"
+SPLITTER+=" --control static"
+SPLITTER+=" --verbosity veryhigh"
+SPLITTER+=" --multipart $multipart"
+SPLITTER+=" --in-channel data1"
+SPLITTER+=" --out-channel data2"
+SPLITTER+=" --channel-config name=data1,type=pull,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5555"
+SPLITTER+="                  name=data2,type=push,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5556,address=tcp://localhost:5557"
+@FAIRMQ_BIN_DIR@/$SPLITTER &
+SPLITTER_PID=$!
+
+PROXY1="fairmq-proxy"
+PROXY1+=" --id proxy1"
+PROXY1+=" --session $SESSION"
+PROXY1+=" --transport $transport"
+PROXY1+=" --color false"
+PROXY1+=" --control static"
+PROXY1+=" --verbosity veryhigh"
+PROXY1+=" --multipart $multipart"
+PROXY1+=" --in-channel data2"
+PROXY1+=" --out-channel data3"
+PROXY1+=" --channel-config name=data2,type=pull,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5556"
+PROXY1+="                  name=data3,type=push,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5558"
+@FAIRMQ_BIN_DIR@/$PROXY1 &
+PROXY1_PID=$!
+
+PROXY2="fairmq-proxy"
+PROXY2+=" --id proxy2"
+PROXY2+=" --session $SESSION"
+PROXY2+=" --transport $transport"
+PROXY2+=" --color false"
+PROXY2+=" --control static"
+PROXY2+=" --verbosity veryhigh"
+PROXY2+=" --multipart $multipart"
+PROXY2+=" --in-channel data2"
+PROXY2+=" --out-channel data3"
+PROXY2+=" --channel-config name=data2,type=pull,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5557"
+PROXY2+="                  name=data3,type=push,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5559"
+@FAIRMQ_BIN_DIR@/$PROXY2 &
+PROXY2_PID=$!
+
+MERGER="fairmq-merger"
+MERGER+=" --id merger"
+MERGER+=" --session $SESSION"
+MERGER+=" --transport $transport"
+MERGER+=" --color false"
+MERGER+=" --control static"
+MERGER+=" --verbosity veryhigh"
+MERGER+=" --multipart $multipart"
+MERGER+=" --in-channel data3"
+MERGER+=" --out-channel data4"
+MERGER+=" --channel-config name=data3,type=pull,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5558,address=tcp://localhost:5559"
+MERGER+="                  name=data4,type=push,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5560"
+@FAIRMQ_BIN_DIR@/$MERGER &
+MERGER_PID=$!
+
+MULTIPLIER="fairmq-multiplier"
+MULTIPLIER+=" --id multiplier"
+MULTIPLIER+=" --session $SESSION"
+MULTIPLIER+=" --transport $transport"
+MULTIPLIER+=" --color false"
+MULTIPLIER+=" --control static"
+MULTIPLIER+=" --verbosity veryhigh"
+MULTIPLIER+=" --multipart $multipart"
+MULTIPLIER+=" --in-channel data4"
+MULTIPLIER+=" --out-channel data5"
+MULTIPLIER+=" --channel-config name=data4,type=pull,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5560"
+MULTIPLIER+="                  name=data5,type=push,method=connect,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5561,address=tcp://localhost:5561"
+@FAIRMQ_BIN_DIR@/$MULTIPLIER &
+MULTIPLIER_PID=$!
+
+SINK="fairmq-sink"
+SINK+=" --id sink1"
+SINK+=" --session $SESSION"
+SINK+=" --transport $transport"
+SINK+=" --color false"
+SINK+=" --control static"
+SINK+=" --verbosity veryhigh"
+SINK+=" --severity debug"
+SINK+=" --multipart $multipart"
+SINK+=" --max-iterations 2"
+SINK+=" --in-channel data5"
+SINK+=" --channel-config name=data5,type=pull,method=bind,sndBufSize=50,rcvBufSize=50,address=tcp://localhost:5561"
+@FAIRMQ_BIN_DIR@/$SINK &
+SINK_PID=$!
+
+wait $SINK_PID
+
+kill -SIGINT $SAMPLER_PID
+kill -SIGINT $SPLITTER_PID
+kill -SIGINT $PROXY1_PID
+kill -SIGINT $PROXY2_PID
+kill -SIGINT $MERGER_PID
+kill -SIGINT $MULTIPLIER_PID
+
+wait $SAMPLER_PID
+wait $SPLITTER_PID
+wait $PROXY1_PID
+wait $PROXY2_PID
+wait $MERGER_PID
+wait $MULTIPLIER_PID

fairmq/devices/FairMQBenchmarkSampler.cxx

@@ -18,7 +18,8 @@
 using namespace std;
 
 FairMQBenchmarkSampler::FairMQBenchmarkSampler()
-    : fSameMessage(true)
+    : fMultipart(false)
+    , fNumParts(1)
     , fMsgSize(10000)
     , fMsgRate(0)
     , fNumIterations(0)
@@ -33,8 +34,9 @@ FairMQBenchmarkSampler::~FairMQBenchmarkSampler()
 
 void FairMQBenchmarkSampler::InitTask()
 {
-    fSameMessage = fConfig->GetValue<bool>("same-msg");
-    fMsgSize = fConfig->GetValue<int>("msg-size");
+    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");
@@ -54,12 +56,16 @@ void FairMQBenchmarkSampler::Run()
 
     while (CheckCurrentState(RUNNING))
     {
-        if (fSameMessage)
+        if (fMultipart)
         {
-            FairMQMessagePtr msg(dataOutChannel.NewMessage());
-            msg->Copy(*baseMsg);
+            FairMQParts parts;
 
-            if (dataOutChannel.Send(msg) >= 0)
+            for (size_t i = 0; i < fNumParts; ++i)
+            {
+                parts.AddPart(dataOutChannel.NewMessage(fMsgSize));
+            }
+
+            if (dataOutChannel.Send(parts) >= 0)
             {
                 if (fMaxIterations > 0)
                 {

fairmq/devices/FairMQBenchmarkSampler.h

@@ -26,8 +26,9 @@ class FairMQBenchmarkSampler : public FairMQDevice
     virtual ~FairMQBenchmarkSampler();
 
   protected:
-    bool fSameMessage;
-    int fMsgSize;
+    bool fMultipart;
+    size_t fNumParts;
+    size_t fMsgSize;
     std::atomic<int> fMsgCounter;
     float fMsgRate;
     uint64_t fNumIterations;

fairmq/devices/FairMQMerger.cxx

@@ -20,7 +20,7 @@
 using namespace std;
 
 FairMQMerger::FairMQMerger()
-    : fMultipart(1)
+    : fMultipart(true)
     , fInChannelName("data-in")
     , fOutChannelName("data-out")
 {
@@ -30,6 +30,8 @@ void FairMQMerger::RegisterChannelEndpoints()
 {
     RegisterChannelEndpoint(fInChannelName, 1, 10000);
     RegisterChannelEndpoint(fOutChannelName, 1, 1);
+
+    PrintRegisteredChannels();
 }
 
 FairMQMerger::~FairMQMerger()
@@ -38,7 +40,7 @@ FairMQMerger::~FairMQMerger()
 
 void FairMQMerger::InitTask()
 {
-    fMultipart = fConfig->GetValue<int>("multipart");
+    fMultipart = fConfig->GetValue<bool>("multipart");
     fInChannelName = fConfig->GetValue<string>("in-channel");
     fOutChannelName = fConfig->GetValue<string>("out-channel");
 }

fairmq/devices/FairMQMerger.h

@@ -26,7 +26,7 @@ class FairMQMerger : public FairMQDevice
     virtual ~FairMQMerger();
 
   protected:
-    int fMultipart;
+    bool fMultipart;
     std::string fInChannelName;
     std::string fOutChannelName;