Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-15 09:31:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Tests for MQ examples

Browse Source
This commit is contained in:
Alexey Rybalchenko
2017-08-23 11:12:29 +02:00
committed by Mohammad Al-Turany
parent 984eed1a89
commit 319bdc91a1
57 changed files with 658 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
examples/MQ/multiple-transports/CMakeLists.txt
View File

@@ -6,10 +6,9 @@
# copied verbatim in the file "LICENSE" #
################################################################################
configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/multiple-transports/ex-multiple-transports.json
${CMAKE_BINARY_DIR}/bin/config/ex-multiple-transports.json)
configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/multiple-transports/startMTEx.sh.in
${CMAKE_BINARY_DIR}/bin/examples/MQ/multiple-transports/startMTEx.sh)
configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/multiple-transports/ex-multiple-transports.json ${CMAKE_BINARY_DIR}/bin/config/ex-multiple-transports.json)
configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/multiple-transports/startMTEx.sh.in ${CMAKE_BINARY_DIR}/bin/examples/MQ/multiple-transports/startMTEx.sh)
configure_file(${CMAKE_SOURCE_DIR}/examples/MQ/multiple-transports/testMTEx.sh.in ${CMAKE_BINARY_DIR}/bin/examples/MQ/multiple-transports/testMTEx.sh)
Set(INCLUDE_DIRECTORIES
${CMAKE_SOURCE_DIR}/fairmq
@@ -76,6 +75,11 @@ ForEach(_file RANGE 0 ${_length})
GENERATE_EXECUTABLE()
EndForEach(_file RANGE 0 ${_length})
add_test(NAME MQ.ex-multiple-transports COMMAND ${CMAKE_BINARY_DIR}/bin/examples/MQ/multiple-transports/testMTEx.sh)
set_tests_properties(MQ.ex-multiple-transports PROPERTIES TIMEOUT "30")
set_tests_properties(MQ.ex-multiple-transports PROPERTIES RUN_SERIAL true)
set_tests_properties(MQ.ex-multiple-transports PROPERTIES PASS_REGULAR_EXPRESSION "Received messages from both sources.")
Install(
FILES ex-multiple-transports.json
DESTINATION share/fairbase/examples/MQ/multiple-transports/config/

9
examples/MQ/multiple-transports/FairMQExampleMTSampler1.cxx
View File

@@ -14,11 +14,14 @@ using namespace std;
FairMQExampleMTSampler1::FairMQExampleMTSampler1()
: fAckListener()
, fMaxIterations(0)
, fNumIterations(0)
{
}
void FairMQExampleMTSampler1::InitTask()
{
fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
}
void FairMQExampleMTSampler1::PreRun()
@@ -38,6 +41,12 @@ bool FairMQExampleMTSampler1::ConditionalRun()
return false;
}
if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations)
{
LOG(INFO) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
return false;
}
return true;
}

2
examples/MQ/multiple-transports/FairMQExampleMTSampler1.h
View File

@@ -28,6 +28,8 @@ class FairMQExampleMTSampler1 : public FairMQDevice
void ListenForAcks();
std::thread fAckListener;
uint64_t fMaxIterations;
uint64_t fNumIterations;
};
#endif /* FAIRMQEXAMPLEMTSAMPLER1_H_ */

13
examples/MQ/multiple-transports/FairMQExampleMTSampler2.cxx
View File

@@ -13,9 +13,16 @@
using namespace std;
FairMQExampleMTSampler2::FairMQExampleMTSampler2()
: fMaxIterations(0)
, fNumIterations(0)
{
}
void FairMQExampleMTSampler2::InitTask()
{
fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
}
bool FairMQExampleMTSampler2::ConditionalRun()
{
FairMQMessagePtr msg(NewMessage(1000));
@@ -27,6 +34,12 @@ bool FairMQExampleMTSampler2::ConditionalRun()
return false;
}
if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations)
{
LOG(INFO) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
return false;
}
return true;
}

5
examples/MQ/multiple-transports/FairMQExampleMTSampler2.h
View File

@@ -21,7 +21,12 @@ class FairMQExampleMTSampler2 : public FairMQDevice
virtual ~FairMQExampleMTSampler2();
protected:
virtual void InitTask();
virtual bool ConditionalRun();
private:
uint64_t fMaxIterations;
uint64_t fNumIterations;
};
#endif /* FAIRMQEXAMPLEMTSAMPLER2_H_ */

29
examples/MQ/multiple-transports/FairMQExampleMTSink.cxx
View File

@@ -14,19 +14,31 @@
#include "FairMQExampleMTSink.h"
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
using namespace std;
FairMQExampleMTSink::FairMQExampleMTSink()
: fMaxIterations(0)
, fNumIterations1(0)
, fNumIterations2(0)
, fReceived1(false)
, fReceived2(false)
{
// register a handler for data arriving on "data" channel
OnData("data1", &FairMQExampleMTSink::HandleData1);
OnData("data2", &FairMQExampleMTSink::HandleData2);
}
void FairMQExampleMTSink::InitTask()
{
fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
}
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool FairMQExampleMTSink::HandleData1(FairMQMessagePtr& /*msg*/, int /*index*/)
{
fNumIterations1++;
// Creates a message using the transport of channel ack
FairMQMessagePtr ack(NewMessageFor("ack", 0));
if (Send(ack, "ack") < 0)
@@ -35,13 +47,28 @@ bool FairMQExampleMTSink::HandleData1(FairMQMessagePtr& /*msg*/, int /*index*/)
}
// return true if want to be called again (otherwise go to IDLE state)
return true;
return CheckIterations();
}
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool FairMQExampleMTSink::HandleData2(FairMQMessagePtr& /*msg*/, int /*index*/)
{
fNumIterations2++;
// return true if want to be called again (otherwise go to IDLE state)
return CheckIterations();
}
bool FairMQExampleMTSink::CheckIterations()
{
if (fMaxIterations > 0)
{
if (fNumIterations1 >= fMaxIterations && fNumIterations2 >= fMaxIterations)
{
LOG(INFO) << "Configured maximum number of iterations reached & Received messages from both sources. Leaving RUNNING state.";
return false;
}
}
return true;
}

9
examples/MQ/multiple-transports/FairMQExampleMTSink.h
View File

@@ -24,8 +24,17 @@ class FairMQExampleMTSink : public FairMQDevice
virtual ~FairMQExampleMTSink();
protected:
virtual void InitTask();
bool HandleData1(FairMQMessagePtr&, int);
bool HandleData2(FairMQMessagePtr&, int);
bool CheckIterations();
private:
uint64_t fMaxIterations;
uint64_t fNumIterations1;
uint64_t fNumIterations2;
bool fReceived1;
bool fReceived2;
};
#endif /* FAIRMQEXAMPLEMTSINK_H_ */

2
examples/MQ/multiple-transports/runExampleMTSampler1.cxx
View File

@@ -14,7 +14,7 @@ namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& options)
{
options.add_options()
("text", bpo::value<std::string>()->default_value("Hello"), "Text to send out");
("max-iterations", bpo::value<uint64_t>()->default_value(5), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)

2
examples/MQ/multiple-transports/runExampleMTSampler2.cxx
View File

@@ -14,7 +14,7 @@ namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& options)
{
options.add_options()
("text", bpo::value<std::string>()->default_value("Hello"), "Text to send out");
("max-iterations", bpo::value<uint64_t>()->default_value(5), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)

4
examples/MQ/multiple-transports/runExampleMTSink.cxx
View File

@@ -11,8 +11,10 @@
namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& /*options*/)
void addCustomOptions(bpo::options_description& options)
{
options.add_options()
("max-iterations", bpo::value<uint64_t>()->default_value(5), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)");
}
FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)

35
examples/MQ/multiple-transports/testMTEx.sh.in Executable file
View File

@@ -0,0 +1,35 @@
#!/bin/bash
config="@CMAKE_BINARY_DIR@/bin/config/ex-multiple-transports.json"
trap 'kill -TERM $SAMPLER1_PID; kill -TERM $SAMPLER2_PID; kill -TERM $SINK_PID; wait $SAMPLER1_PID; wait $SAMPLER2_PID; wait $SINK_PID; @CMAKE_BINARY_DIR@/bin/shmmonitor --cleanup;' TERM
SINK="ex-mt-sink"
SINK+=" --id sink1"
SINK+=" --max-iterations 1"
SINK+=" --control static --log-color false"
SINK+=" --transport shmem"
SINK+=" --mq-config $config"
@CMAKE_BINARY_DIR@/bin/examples/MQ/multiple-transports/$SINK &
SINK_PID=$!
SAMPLER1="ex-mt-sampler1"
SAMPLER1+=" --id sampler1"
SAMPLER1+=" --max-iterations 1"
SAMPLER1+=" --control static --log-color false"
SAMPLER1+=" --transport shmem"
SAMPLER1+=" --mq-config $config"
@CMAKE_BINARY_DIR@/bin/examples/MQ/multiple-transports/$SAMPLER1 &
SAMPLER1_PID=$!
SAMPLER2="ex-mt-sampler2"
SAMPLER2+=" --id sampler2"
SAMPLER2+=" --max-iterations 1"
SAMPLER2+=" --control static --log-color false"
SAMPLER2+=" --transport nanomsg"
SAMPLER2+=" --mq-config $config"
@CMAKE_BINARY_DIR@/bin/examples/MQ/multiple-transports/$SAMPLER2 &
SAMPLER2_PID=$!
wait $SAMPLER1_PID
wait $SAMPLER2_PID
wait $SINK_PID