FairMQSink.h
/********************************************************************************
* 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" *
********************************************************************************/
#ifndef FAIRMQSINK_H_
#define FAIRMQSINK_H_
#include <string>
#include <chrono>
#include "../FairMQDevice.h"
#include "../FairMQLogger.h"
#include "../options/FairMQProgOptions.h"
// template<typename OutputPolicy>
class FairMQSink : public FairMQDevice //, public OutputPolicy
{
public:
FairMQSink()
: fMultipart(false)
, fMaxIterations(0)
, fNumIterations(0)
, fInChannelName()
{}
virtual ~FairMQSink()
{}
protected:
bool fMultipart;
uint64_t fMaxIterations;
uint64_t fNumIterations;
std::string fInChannelName;
virtual void InitTask()
{
fMultipart = fConfig->GetValue<bool>("multipart");
fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
fInChannelName = fConfig->GetValue<std::string>("in-channel");
}
virtual void Run()
{
// store the channel reference to avoid traversing the map on every loop iteration
FairMQChannel& dataInChannel = fChannels.at(fInChannelName).at(0);
LOG(info) << "Starting the benchmark and expecting to receive " << fMaxIterations << " messages.";
auto tStart = std::chrono::high_resolution_clock::now();
while (CheckCurrentState(RUNNING))
{
if (fMultipart)
{
FairMQParts parts;
if (dataInChannel.Receive(parts) >= 0)
{
if (fMaxIterations > 0)
{
if (fNumIterations >= fMaxIterations)
{
LOG(info) << "Configured maximum number of iterations reached.";
break;
}
}
fNumIterations++;
}
}
else
{
FairMQMessagePtr msg(dataInChannel.NewMessage());
if (dataInChannel.Receive(msg) >= 0)
{
if (fMaxIterations > 0)
{
if (fNumIterations >= fMaxIterations)
{
LOG(info) << "Configured maximum number of iterations reached.";
break;
}
}
fNumIterations++;
}
}
}
auto tEnd = std::chrono::high_resolution_clock::now();
LOG(info) << "Leaving RUNNING state. Received " << fNumIterations << " messages in " << std::chrono::duration<double, std::milli>(tEnd - tStart).count() << "ms.";
}
};
#endif /* FAIRMQSINK_H_ */
FairMQSink::InitTask
virtual void InitTask()
Task initialization (can be overloaded in child classes)
Definition: FairMQSink.h:45
FairMQChannel::Receive
int Receive(FairMQMessagePtr &msg, int rcvTimeoutInMs=-1)
Definition: FairMQChannel.h:244
FairMQDevice::fConfig
FairMQProgOptions * fConfig
Pointer to config (internal or external)
Definition: FairMQDevice.h:413
FairMQChannel
Definition: FairMQChannel.h:27
FairMQSink
Definition: FairMQSink.h:26
FairMQDevice::fChannels
std::unordered_map< std::string, std::vector< FairMQChannel > > fChannels
Device channels.
Definition: FairMQDevice.h:411
FairMQParts
FairMQParts is a lightweight convenience wrapper around a vector of unique pointers to FairMQMessage,...
Definition: FairMQParts.h:20
FairMQDevice
Definition: FairMQDevice.h:46
FairMQSink::Run
virtual void Run()
Runs the device (to be overloaded in child classes)
Definition: FairMQSink.h:52

privacy