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

QC example: add setting of the property, test if its successful

Browse Source
This commit is contained in:
Alexey Rybalchenko
2020-02-20 16:33:51 +01:00
committed by Dennis Klein
parent beff0af51b
commit 8123a6ecab
8 changed files with 47 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
examples/qc/fairmq-start-ex-qc.sh.in
View File

@@ -52,9 +52,11 @@ fairmq-dds-command-ui -c b
fairmq-dds-command-ui -c x
fairmq-dds-command-ui -c j
fairmq-dds-command-ui -c r
sampler_and_sink="main/(Sampler|Sink).*"
fairmq-dds-command-ui -w "RUNNING->READY" -p $sampler_and_sink
echo "...$sampler_and_sink are READY, sending shutdown..."
qcconsumer="main/QCConsumer.*"
qcproducer="main/QCProducer.*"
fairmq-dds-command-ui -c p --property-key qc --property-value active -p $qcproducer
fairmq-dds-command-ui -w "RUNNING->READY" -p $qcconsumer
echo "...$qcconsumer received data and transitioned to READY, sending shutdown..."
fairmq-dds-command-ui -c s
fairmq-dds-command-ui -c t
fairmq-dds-command-ui -c d

4
examples/qc/runQCConsumer.cxx
View File

@@ -14,9 +14,9 @@ class QCConsumer : public FairMQDevice
public:
QCConsumer()
{
OnData("qc", [](FairMQMessagePtr& /*msg*/, int){
OnData("qc", [](FairMQMessagePtr& /*msg*/, int) {
LOG(info) << "received data";
return true;
return false;
});
}
};

19
examples/qc/runQCProducer.cxx
View File

@@ -14,15 +14,13 @@ 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) {
GetConfig()->Subscribe<std::string>("qcdevice", [&](const std::string& key, std::string value) {
if (key == "qc") {
if (value == "active") {
fDoQC.store(true);
@@ -37,13 +35,10 @@ class QCProducer : public FairMQDevice
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;
}
FairMQMessagePtr msgCopy(NewMessage());
msgCopy->Copy(*msg);
if (Send(msg, "qc") < 0) {
return false;
}
}
@@ -54,12 +49,10 @@ class QCProducer : public FairMQDevice
return true;
}
void ResetTask() override { GetConfig()->Unsubscribe<std::string>("qc"); }
void ResetTask() override { GetConfig()->Unsubscribe<std::string>("qcdevice"); }
private:
std::atomic<bool> fDoQC;
int fCounter;
int fInterval;
};
void addCustomOptions(boost::program_options::options_description& /*options*/) {}

25
examples/qc/runSampler.cxx
View File

@@ -15,41 +15,22 @@
class Sampler : public FairMQDevice
{
public:
Sampler()
: fMaxIterations(0)
, fNumIterations(0)
{}
Sampler() {}
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));
std::this_thread::sleep_for(std::chrono::milliseconds(10));
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)");
}
void addCustomOptions(bpo::options_description& options) {}
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/) { return new Sampler(); }

32
examples/qc/runSink.cxx
View File

@@ -14,38 +14,12 @@
class Sink : public FairMQDevice
{
public:
Sink()
: fMaxIterations(0)
, fNumIterations(0)
{
OnData("data2", &Sink::HandleData);
}
Sink() { 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;
bool HandleData(FairMQMessagePtr& /*msg*/, int /*index*/) { 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)");
}
void addCustomOptions(bpo::options_description& options) {}
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/) { return new Sink(); }