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

Update FairMQStateMachine & introduce FairMQChannels

Browse Source 
Organize sockets as a map of vectors of FairMQChannels.

Update FairMQStateMachine by removing SETTINGINPUT, SETTINGOUTPUT,
BIND and CONNECT states and by adding INITIALIZING_TASK, RESETTING_TASK
and RESETTING_DEVICE states. Run states functions in their own thread.
This commit is contained in:
Alexey Rybalchenko
2015-04-29 13:25:42 +02:00
parent a2ebbbe450
commit 7fda980710
54 changed files with 1674 additions and 1573 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
fairmq/devices/FairMQBenchmarkSampler.cxx
View File

@@ -33,27 +33,19 @@ FairMQBenchmarkSampler::~FairMQBenchmarkSampler()
{
}
void FairMQBenchmarkSampler::Init()
{
FairMQDevice::Init();
}
void FairMQBenchmarkSampler::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
boost::thread resetEventCounter(boost::bind(&FairMQBenchmarkSampler::ResetEventCounter, this));
void* buffer = operator new[](fEventSize);
FairMQMessage* base_msg = fTransportFactory->CreateMessage(buffer, fEventSize);
FairMQMessage* baseMsg = fTransportFactory->CreateMessage(buffer, fEventSize);
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
msg->Copy(base_msg);
msg->Copy(baseMsg);
fPayloadOutputs->at(0)->Send(msg);
fChannels["data-out"].at(0).Send(msg);
--fEventCounter;
@@ -65,28 +57,19 @@ void FairMQBenchmarkSampler::Run()
delete msg;
}
delete base_msg;
delete baseMsg;
try {
rateLogger.interrupt();
rateLogger.join();
resetEventCounter.interrupt();
resetEventCounter.join();
} catch(boost::thread_resource_error& e) {
LOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}
void FairMQBenchmarkSampler::ResetEventCounter()
{
while (true)
while (GetCurrentState() == RUNNING)
{
try
{
@@ -100,61 +83,26 @@ void FairMQBenchmarkSampler::ResetEventCounter()
}
}
void FairMQBenchmarkSampler::Log(int intervalInMs)
{
timestamp_t t0;
timestamp_t t1;
unsigned long bytes = fPayloadOutputs->at(0)->GetBytesTx();
unsigned long messages = fPayloadOutputs->at(0)->GetMessagesTx();
unsigned long bytesNew = 0;
unsigned long messagesNew = 0;
double megabytesPerSecond = 0;
double messagesPerSecond = 0;
t0 = get_timestamp();
while (true)
{
boost::this_thread::sleep(boost::posix_time::milliseconds(intervalInMs));
t1 = get_timestamp();
bytesNew = fPayloadOutputs->at(0)->GetBytesTx();
messagesNew = fPayloadOutputs->at(0)->GetMessagesTx();
timestamp_t timeSinceLastLog_ms = (t1 - t0) / 1000.0L;
megabytesPerSecond = ((double)(bytesNew - bytes) / (1024. * 1024.)) / (double)timeSinceLastLog_ms * 1000.;
messagesPerSecond = (double)(messagesNew - messages) / (double)timeSinceLastLog_ms * 1000.;
LOG(DEBUG) << "send " << messagesPerSecond << " msg/s, " << megabytesPerSecond << " MB/s";
bytes = bytesNew;
messages = messagesNew;
t0 = t1;
}
}
void FairMQBenchmarkSampler::SetProperty(const int key, const string& value, const int slot /*= 0*/)
void FairMQBenchmarkSampler::SetProperty(const int key, const string& value)
{
switch (key)
{
default:
FairMQDevice::SetProperty(key, value, slot);
FairMQDevice::SetProperty(key, value);
break;
}
}
string FairMQBenchmarkSampler::GetProperty(const int key, const string& default_ /*= ""*/, const int slot /*= 0*/)
string FairMQBenchmarkSampler::GetProperty(const int key, const string& default_ /*= ""*/)
{
switch (key)
{
default:
return FairMQDevice::GetProperty(key, default_, slot);
return FairMQDevice::GetProperty(key, default_);
}
}
void FairMQBenchmarkSampler::SetProperty(const int key, const int value, const int slot /*= 0*/)
void FairMQBenchmarkSampler::SetProperty(const int key, const int value)
{
switch (key)
{
@@ -165,12 +113,12 @@ void FairMQBenchmarkSampler::SetProperty(const int key, const int value, const i
fEventRate = value;
break;
default:
FairMQDevice::SetProperty(key, value, slot);
FairMQDevice::SetProperty(key, value);
break;
}
}
int FairMQBenchmarkSampler::GetProperty(const int key, const int default_ /*= 0*/, const int slot /*= 0*/)
int FairMQBenchmarkSampler::GetProperty(const int key, const int default_ /*= 0*/)
{
switch (key)
{
@@ -179,6 +127,6 @@ int FairMQBenchmarkSampler::GetProperty(const int key, const int default_ /*= 0*
case EventRate:
return fEventRate;
default:
return FairMQDevice::GetProperty(key, default_, slot);
return FairMQDevice::GetProperty(key, default_);
}
}

14
fairmq/devices/FairMQBenchmarkSampler.h
View File

@@ -29,25 +29,27 @@ class FairMQBenchmarkSampler : public FairMQDevice
enum
{
InputFile = FairMQDevice::Last,
EventRate,
EventSize,
EventRate,
Last
};
FairMQBenchmarkSampler();
virtual ~FairMQBenchmarkSampler();
void Log(int intervalInMs);
void ResetEventCounter();
virtual void SetProperty(const int key, const std::string& value, const int slot = 0);
virtual std::string GetProperty(const int key, const std::string& default_ = "", const int slot = 0);
virtual void SetProperty(const int key, const int value, const int slot = 0);
virtual int GetProperty(const int key, const int default_ = 0, const int slot = 0);
virtual void SetProperty(const int key, const std::string& value);
virtual std::string GetProperty(const int key, const std::string& default_ = "");
virtual void SetProperty(const int key, const int value);
virtual int GetProperty(const int key, const int default_ = 0);
protected:
int fEventSize;
int fEventRate;
int fEventCounter;
virtual void Init();
virtual void Run();
};

28
fairmq/devices/FairMQBuffer.cxx
View File

@@ -26,39 +26,17 @@ FairMQBuffer::FairMQBuffer()
void FairMQBuffer::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
int received = 0;
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
if (received > 0)
if (fChannels["data-in"].at(0).Receive(msg) > 0)
{
fPayloadOutputs->at(0)->Send(msg);
received = 0;
fChannels["data-out"].at(0).Send(msg);
}
delete msg;
}
try {
rateLogger.interrupt();
rateLogger.join();
} catch(boost::thread_resource_error& e) {
LOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}
FairMQBuffer::~FairMQBuffer()

36
fairmq/devices/FairMQMerger.cxx
View File

@@ -29,30 +29,22 @@ FairMQMerger::~FairMQMerger()
void FairMQMerger::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
FairMQPoller* poller = fTransportFactory->CreatePoller(fChannels["data-in"]);
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
FairMQPoller* poller = fTransportFactory->CreatePoller(*fPayloadInputs);
int received = 0;
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
poller->Poll(100);
for (int i = 0; i < fNumInputs; i++)
for (int i = 0; i < fChannels["data-in"].size(); ++i)
{
if (poller->CheckInput(i))
{
received = fPayloadInputs->at(i)->Receive(msg);
}
if (received > 0)
{
fPayloadOutputs->at(0)->Send(msg);
received = 0;
if (fChannels["data-in"].at(i).Receive(msg))
{
fChannels["data-out"].at(0).Send(msg);
}
}
}
@@ -60,18 +52,4 @@ void FairMQMerger::Run()
}
delete poller;
try {
rateLogger.interrupt();
rateLogger.join();
} catch(boost::thread_resource_error& e) {
LOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}

28
fairmq/devices/FairMQProxy.cxx
View File

@@ -28,37 +28,15 @@ FairMQProxy::~FairMQProxy()
void FairMQProxy::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
FairMQMessage* msg = fTransportFactory->CreateMessage();
int received = 0;
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
received = fPayloadInputs->at(0)->Receive(msg);
if (received > 0)
if (fChannels["data-in"].at(0).Receive(msg) > 0)
{
fPayloadOutputs->at(0)->Send(msg);
received = 0;
fChannels["data-out"].at(0).Send(msg);
}
}
delete msg;
try {
rateLogger.interrupt();
rateLogger.join();
} catch(boost::thread_resource_error& e) {
LOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}

24
fairmq/devices/FairMQSink.cxx
View File

@@ -24,34 +24,14 @@ FairMQSink::FairMQSink()
void FairMQSink::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
int received = 0;
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
fChannels["data-in"].at(0).Receive(msg);
delete msg;
}
try {
rateLogger.interrupt();
rateLogger.join();
} catch(boost::thread_resource_error& e) {
LOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}
FairMQSink::~FairMQSink()

30
fairmq/devices/FairMQSplitter.cxx
View File

@@ -28,44 +28,22 @@ FairMQSplitter::~FairMQSplitter()
void FairMQSplitter::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
int received = 0;
int direction = 0;
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
if (received > 0)
if (fChannels["data-in"].at(0).Receive(msg) > 0)
{
fPayloadOutputs->at(direction)->Send(msg);
fChannels["data-out"].at(direction).Send(msg);
direction++;
if (direction >= fNumOutputs)
if (direction >= fChannels["data-out"].size())
{
direction = 0;
}
received = 0;
}
delete msg;
}
try {
rateLogger.interrupt();
rateLogger.join();
} catch(boost::thread_resource_error& e) {
LOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}

74
fairmq/devices/GenericFileSink.h
View File

@@ -21,8 +21,6 @@
#include <boost/bind.hpp>
#include "FairMQLogger.h"
/*********************************************************************
* -------------- NOTES -----------------------
* All policies must have a default constructor
@@ -38,87 +36,57 @@
* OutputPolicy::InitOutputFile()
**********************************************************************/
#include "FairMQDevice.h"
template < typename InputPolicy,
typename OutputPolicy>
class GenericFileSink : public FairMQDevice,
public InputPolicy,
public OutputPolicy
{
public:
GenericFileSink():
InputPolicy(),
OutputPolicy()
template <typename InputPolicy, typename OutputPolicy>
class GenericFileSink : public FairMQDevice, public InputPolicy, public OutputPolicy
{
public:
GenericFileSink()
: InputPolicy()
, OutputPolicy()
{}
virtual ~GenericFileSink()
{}
void SetTransport(FairMQTransportFactory* transport)
{
FairMQDevice::SetTransport(transport);
}
template <typename... Args>
void InitInputContainer(Args... args)
{
InputPolicy::InitContainer(std::forward<Args>(args)...);
}
protected:
virtual void Init()
void InitInputContainer(Args... args)
{
FairMQDevice::Init();
OutputPolicy::InitOutputFile();
InputPolicy::InitContainer(std::forward<Args>(args)...);
}
protected:
virtual void InitTask()
{
OutputPolicy::InitOutputFile();
}
virtual void Run()
{
MQLOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
int received = 0;
int receivedMsg = 0;
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
if(received>0)
if (fChannels["data-in"].at(0).Receive(msg) > 0)
{
OutputPolicy::AddToFile(InputPolicy::DeSerializeMsg(msg));
receivedMsg++;
}
delete msg;
}
MQLOG(INFO) << "Received " << receivedMsg << " messages!";
try
{
rateLogger.interrupt();
rateLogger.join();
}
catch(boost::thread_resource_error& e)
{
MQLOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}
};
#endif /* GENERICFILESINK_H */
#endif /* GENERICFILESINK_H */

41
fairmq/devices/GenericFileSink.tpl
View File

@@ -6,9 +6,9 @@
*/
template <typename InputPolicy, typename OutputPolicy>
GenericFileSink<InputPolicy, OutputPolicy>::GenericFileSink() :
InputPolicy(),
OutputPolicy()
GenericFileSink<InputPolicy, OutputPolicy>::GenericFileSink()
: InputPolicy()
, OutputPolicy()
{
}
@@ -21,17 +21,16 @@ template <typename InputPolicy, typename OutputPolicy>
void GenericFileSink<InputPolicy, OutputPolicy>::SetTransport(FairMQTransportFactory* transport)
{
FairMQDevice::SetTransport(transport);
//InputPolicy::SetTransport(transport);
// InputPolicy::SetTransport(transport);
}
template <typename InputPolicy, typename OutputPolicy>
void GenericFileSink<InputPolicy, OutputPolicy>::Init()
void GenericFileSink<InputPolicy, OutputPolicy>::InitTask()
{
FairMQDevice::Init();
InitOutputFile();
//InputPolicy::Init();
//OutputPolicy::Init();
// InputPolicy::Init();
// OutputPolicy::Init();
}
template <typename InputPolicy, typename OutputPolicy>
@@ -43,18 +42,12 @@ void GenericFileSink<InputPolicy, OutputPolicy>::InitOutputFile()
template <typename InputPolicy, typename OutputPolicy>
void GenericFileSink<InputPolicy, OutputPolicy>::Run()
{
MQLOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
int received = 0;
int receivedMsg = 0;
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
if(received>0)
if (fChannels["data-in"].at(0).Receive(msg) > 0)
{
OutputPolicy::AddToFile(InputPolicy::DeSerializeMsg(msg));
receivedMsg++;
@@ -63,20 +56,4 @@ void GenericFileSink<InputPolicy, OutputPolicy>::Run()
}
MQLOG(INFO) << "Received " << receivedMsg << " messages!";
try
{
rateLogger.interrupt();
rateLogger.join();
}
catch(boost::thread_resource_error& e)
{
MQLOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}

65
fairmq/devices/GenericMerger.h
View File

@@ -17,19 +17,14 @@
#include "FairMQPoller.h"
template < typename MergerPolicy,
typename InputPolicy,
typename OutputPolicy
>
class GenericMerger : public FairMQDevice,
public MergerPolicy,
public InputPolicy,
public OutputPolicy
template <typename MergerPolicy, typename InputPolicy, typename OutputPolicy>
class GenericMerger : public FairMQDevice, public MergerPolicy, public InputPolicy, public OutputPolicy
{
public:
GenericMerger() : fBlockingTime(100)
GenericMerger()
: fBlockingTime(100)
{}
virtual ~GenericMerger()
{}
@@ -37,44 +32,35 @@ class GenericMerger : public FairMQDevice,
{
FairMQDevice::SetTransport(transport);
}
protected:
int fBlockingTime;
virtual void Run()
{
MQLOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
FairMQPoller* poller = fTransportFactory->CreatePoller(*fPayloadInputs);
FairMQPoller* poller = fTransportFactory->CreatePoller(fChannels["data-in"]);
int received = 0;
while (fState == RUNNING)
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
//MergerPolicy::
// MergerPolicy::
poller->Poll(fBlockingTime);
for (int i = 0; i < fNumInputs; i++)
for (int i = 0; i < fChannels["datain"].size(); i++)
{
if (poller->CheckInput(i))
{
received = fPayloadInputs->at(i)->Receive(msg);
received = fChannels["data-in"].at(i).Receive(msg)
MergerPolicy::Merge(InputPolicy::DeSerializeMsg(msg));
}
OutputPolicy::SetMessage(msg);
if ( received > 0 && MergerPolicy::ReadyToSend() )
if (received > 0 && MergerPolicy::ReadyToSend())
{
fPayloadOutputs->at(0)->Send(OutputPolicy::SerializeMsg(MergerPolicy::GetOutputData()));
fChannels["data-out"].at(0).Send(OutputPolicy::SerializeMsg(MergerPolicy::GetOutputData()));
received = 0;
}
}
@@ -83,25 +69,8 @@ class GenericMerger : public FairMQDevice,
}
delete poller;
try
{
rateLogger.interrupt();
rateLogger.join();
}
catch(boost::thread_resource_error& e)
{
MQLOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}
};
#endif /* GENERICMERGER_H */
#endif /* GENERICMERGER_H */

108
fairmq/devices/GenericProcessor.h
View File

@@ -13,11 +13,10 @@
*/
#ifndef GENERICPROCESSOR_H
#define GENERICPROCESSOR_H
#define GENERICPROCESSOR_H
#include "FairMQDevice.h"
/*********************************************************************
* -------------- NOTES -----------------------
* All policies must have a default constructor
@@ -41,73 +40,67 @@
*
**********************************************************************/
template < typename InputPolicy,
typename OutputPolicy,
typename TaskPolicy>
class GenericProcessor: public FairMQDevice,
public InputPolicy,
public OutputPolicy,
public TaskPolicy
template <typename InputPolicy, typename OutputPolicy, typename TaskPolicy>
class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputPolicy, public TaskPolicy
{
public:
GenericProcessor() : InputPolicy(), OutputPolicy(), TaskPolicy()
GenericProcessor()
: InputPolicy()
, OutputPolicy()
, TaskPolicy()
{}
virtual ~GenericProcessor()
{}
// the four following methods ensure
// that the correct policy method is called
void SetTransport(FairMQTransportFactory* transport)
{
FairMQDevice::SetTransport(transport);
}
template <typename... Args>
void InitTask(Args... args)
{
TaskPolicy::InitTask(std::forward<Args>(args)...);
}
template <typename... Args>
void InitInputContainer(Args... args)
{
InputPolicy::InitContainer(std::forward<Args>(args)...);
}
void InitTask(Args... args)
{
TaskPolicy::InitTask(std::forward<Args>(args)...);
}
template <typename... Args>
void InitOutputContainer(Args... args)
{
OutputPolicy::InitContainer(std::forward<Args>(args)...);
}
void InitInputContainer(Args... args)
{
InputPolicy::InitContainer(std::forward<Args>(args)...);
}
template <typename... Args>
void InitOutputContainer(Args... args)
{
OutputPolicy::InitContainer(std::forward<Args>(args)...);
}
/*
*
// *********************** TODO: implement multipart features
void SendPart()
{
fPayloadOutputs->at(0)->Send(OutputPolicy::SerializeMsg(TaskPolicy::GetData()), "snd-more");
fChannels["data-out"].at(0).Send(OutputPolicy::SerializeMsg(TaskPolicy::GetData()), "snd-more");
OutputPolicy::CloseMessage();
}
//void SendPart();
//bool ReceivePart();
// void SendPart();
// bool ReceivePart();
bool ReceivePart()
{
int64_t more = 0;
size_t more_size = sizeof(more);
fPayloadInputs->at(0)->GetOption("rcv-more", &more, &more_size);
if(more)
fChannels["data-in"].at(0).fSocket->GetOption("rcv-more", &more, &more_size);
if (more)
{
InputPolicy::CloseMessage();
//fProcessorTask->GetPayload()->CloseMessage();
// fProcessorTask->GetPayload()->CloseMessage();
fProcessorTask->SetPayload(fTransportFactory->CreateMessage());
return fPayloadInputs->at(0)->Receive(fProcessorTask->GetPayload());
return fChannels["data-in"].at(0).Receive(fProcessorTask->GetPayload());
}
else
{
@@ -117,29 +110,23 @@ class GenericProcessor: public FairMQDevice,
*/
protected:
virtual void Init()
virtual void InitTask()
{
FairMQDevice::Init();
// TODO: implement multipart features
//fProcessorTask->InitTask();
//fProcessorTask->SetSendPart(boost::bind(&FairMQProcessor::SendPart, this));
//fProcessorTask->SetReceivePart(boost::bind(&FairMQProcessor::ReceivePart, this));
// fProcessorTask->InitTask();
// fProcessorTask->SetSendPart(boost::bind(&FairMQProcessor::SendPart, this));
// fProcessorTask->SetReceivePart(boost::bind(&FairMQProcessor::ReceivePart, this));
}
virtual void Run()
{
MQLOG(INFO) << ">>>>>>> Run <<<<<<<";
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
int receivedMsgs = 0;
int sentMsgs = 0;
int received = 0;
while ( fState == RUNNING )
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
receivedMsgs++;
// InputPolicy::DeSerializeMsg(msg) --> deserialize data of msg and fill output container
@@ -148,37 +135,22 @@ class GenericProcessor: public FairMQDevice,
// OutputPolicy::fMessage point to msg
OutputPolicy::SetMessage(msg);
if (received > 0)
if (fChannels["data-in"].at(0).Receive(msg) > 0)
{
// TaskPolicy::GetOutputData() --> Get processed output container
// OutputPolicy::message(...) --> Serialize output container and fill fMessage
fPayloadOutputs->at(0)->Send(OutputPolicy::SerializeMsg(TaskPolicy::GetOutputData()));
fChannels["data-out"].at(0).Send(OutputPolicy::SerializeMsg(TaskPolicy::GetOutputData()));
sentMsgs++;
received = 0;
}
if(msg)
if (msg)
{
msg->CloseMessage();
}
}
MQLOG(INFO) << "Received " << receivedMsgs << " and sent " << sentMsgs << " messages!";
try
{
rateLogger.interrupt();
rateLogger.join();
}
catch(boost::thread_resource_error& e)
{
MQLOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
}
};

58
fairmq/devices/GenericSampler.h
View File

@@ -12,7 +12,6 @@
* Created on November 24, 2014, 3:30 PM
*/
#ifndef GENERICSAMPLER_H
#define GENERICSAMPLER_H
@@ -45,38 +44,34 @@
*
**********************************************************************/
template <typename SamplerPolicy,
typename OutputPolicy>
class GenericSampler: public FairMQDevice,
public SamplerPolicy,
public OutputPolicy
{
template <typename SamplerPolicy, typename OutputPolicy>
class GenericSampler : public FairMQDevice, public SamplerPolicy, public OutputPolicy
{
public:
enum
enum
{
InputFile = FairMQDevice::Last,
Branch,
ParFile,
EventRate
};
GenericSampler();
virtual ~GenericSampler();
virtual void SetTransport(FairMQTransportFactory* factory);
void ResetEventCounter();
virtual void ListenToCommands();
template <typename... Args>
void SetFileProperties(Args&... args)
{
SamplerPolicy::SetFileProperties(args...);
}
void SetFileProperties(Args&... args)
{
SamplerPolicy::SetFileProperties(args...);
}
virtual void SetProperty(const int key, const std::string& value, const int slot = 0);
virtual std::string GetProperty(const int key, const std::string& default_ = "", const int slot = 0);
virtual void SetProperty(const int key, const int value, const int slot = 0);
virtual int GetProperty(const int key, const int default_ = 0, const int slot = 0);
virtual void SetProperty(const int key, const std::string& value);
virtual std::string GetProperty(const int key, const std::string& default_ = "");
virtual void SetProperty(const int key, const int value);
virtual int GetProperty(const int key, const int default_ = 0);
/**
* Sends the currently available output of the Sampler Task as part of a multipart message
@@ -87,23 +82,22 @@ class GenericSampler: public FairMQDevice,
// temporary disabled
//void SendPart();
void SetContinuous(bool flag) { fContinuous = flag; }
void SetContinuous(bool flag);
protected:
virtual void Init();
virtual void Run();
protected:
virtual void InitTask();
virtual void Run();
protected:
std::string fInputFile; // Filename of a root file containing the simulated digis.
std::string fParFile;
std::string fBranch; // The name of the sub-detector branch to stream the digis from.
int64_t fNumEvents;
int fEventRate;
int fEventCounter;
bool fContinuous;
int64_t fNumEvents;
int fEventRate;
int fEventCounter;
bool fContinuous;
std::string fInputFile; // Filename of a root file containing the simulated digis.
std::string fParFile;
std::string fBranch; // The name of the sub-detector branch to stream the digis from.
};
#include "GenericSampler.tpl"
#endif /* GENERICSAMPLER_H */
#endif /* GENERICSAMPLER_H */

251
fairmq/devices/GenericSampler.tpl
View File

@@ -6,11 +6,14 @@
*/
template <typename SamplerPolicy, typename OutputPolicy>
GenericSampler<SamplerPolicy,OutputPolicy>::GenericSampler() :
fNumEvents(0),
fEventRate(1),
fEventCounter(0),
fContinuous(false)
GenericSampler<SamplerPolicy,OutputPolicy>::GenericSampler()
: fNumEvents(0)
, fEventRate(1)
, fEventCounter(0)
, fContinuous(false)
, fInputFile()
, fParFile()
, fBranch()
{
}
@@ -23,202 +26,152 @@ template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::SetTransport(FairMQTransportFactory* factory)
{
FairMQDevice::SetTransport(factory);
//OutputPolicy::SetTransport(factory);
// OutputPolicy::SetTransport(factory);
}
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::Init()
void GenericSampler<SamplerPolicy,OutputPolicy>::InitTask()
{
FairMQDevice::Init();
SamplerPolicy::InitSampler();
fNumEvents=SamplerPolicy::GetNumberOfEvent();
SamplerPolicy::InitSampler();
fNumEvents = SamplerPolicy::GetNumberOfEvent();
}
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::Run()
{
LOG(INFO) << ">>>>>>> Run <<<<<<<";
// boost::thread resetEventCounter(boost::bind(&GenericSampler::ResetEventCounter, this));
boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));
// boost::thread resetEventCounter(boost::bind(&GenericSampler::ResetEventCounter, this));
// boost::thread commandListener(boost::bind(&GenericSampler::ListenToCommands, this));
int sentMsgs = 0;
int sentMsgs = 0;
boost::timer::auto_cpu_timer timer;
boost::timer::auto_cpu_timer timer;
LOG(INFO) << "Number of events to process: " << fNumEvents;
LOG(INFO) << "Number of events to process: " << fNumEvents;
// while ( fState == RUNNING ) {
do
{
for ( int64_t eventNr = 0 ; eventNr < fNumEvents; ++eventNr )
do {
for (int64_t eventNr = 0; eventNr < fNumEvents; ++eventNr)
{
//fSamplerTask->SetEventIndex(eventNr);
FairMQMessage* msg = fTransportFactory->CreateMessage();
OutputPolicy::SetMessage(msg);
fPayloadOutputs->at(0)->Send(OutputPolicy::SerializeMsg(
SamplerPolicy::GetDataBranch(eventNr)));
fChannels["data-out"].at(0).Send(OutputPolicy::SerializeMsg(SamplerPolicy::GetDataBranch(eventNr)));
++sentMsgs;
if(msg)
if (msg)
{
msg->CloseMessage();
}
// Optional event rate limiting
// --fEventCounter;
// while (fEventCounter == 0) {
// boost::this_thread::sleep(boost::posix_time::milliseconds(1));
// }
if( fState != RUNNING ) { break; }
if (GetCurrentState() != RUNNING)
{
break;
}
}
}
while ( fState == RUNNING && fContinuous );
}
while ( GetCurrentState() == RUNNING && fContinuous );
// }
boost::timer::cpu_times const elapsed_time(timer.elapsed());
LOG(INFO) << "Sent everything in:\n" << boost::timer::format(elapsed_time, 2);
LOG(INFO) << "Sent " << sentMsgs << " messages!";
try {
rateLogger.interrupt();
rateLogger.join();
// resetEventCounter.interrupt();
// resetEventCounter.join();
// commandListener.interrupt();
// commandListener.join();
}
catch (boost::thread_resource_error &e) {
LOG(ERROR) << e.what();
}
FairMQDevice::Shutdown();
// notify parent thread about end of processing.
boost::lock_guard<boost::mutex> lock(fRunningMutex);
fRunningFinished = true;
fRunningCondition.notify_one();
boost::timer::cpu_times const elapsed_time(timer.elapsed());
LOG(INFO) << "Sent everything in:\n" << boost::timer::format(elapsed_time, 2);
LOG(INFO) << "Sent " << sentMsgs << " messages!";
}
/*
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::SendPart()
{
fPayloadOutputs->at(0)->Send(OutputPolicy::GetMessage(), "snd-more");
OutputPolicy::CloseMessage();
fChannels["data-out"].at(0).Send(OutputPolicy::GetMessage(), "snd-more");
OutputPolicy::CloseMessage();
}
*/
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::SetContinuous(bool flag)
{
fContinuous = flag;
}
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::ResetEventCounter()
{
while (true)
{
try
while (GetCurrentState() == RUNNING)
{
fEventCounter = fEventRate / 100;
boost::this_thread::sleep(boost::posix_time::milliseconds(10));
try
{
fEventCounter = fEventRate / 100;
boost::this_thread::sleep(boost::posix_time::milliseconds(10));
}
catch (boost::thread_interrupted &)
{
LOG(DEBUG) << "resetEventCounter interrupted";
break;
}
}
catch (boost::thread_interrupted &)
LOG(DEBUG) << ">>>>>>> stopping resetEventCounter <<<<<<<";
}
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::SetProperty(const int key, const std::string& value)
{
switch (key)
{
LOG(DEBUG) << "resetEventCounter interrupted";
break;
case InputFile:
fInputFile = value;
break;
case ParFile:
fParFile = value;
break;
case Branch:
fBranch = value;
break;
default:
FairMQDevice::SetProperty(key, value);
break;
}
}
LOG(DEBUG) << ">>>>>>> stopping resetEventCounter <<<<<<<";
}
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::ListenToCommands()
std::string GenericSampler<SamplerPolicy,OutputPolicy>::GetProperty(const int key, const std::string& default_/*= ""*/)
{
LOG(INFO) << ">>>>>>> ListenToCommands <<<<<<<";
int received = 0;
while (true) {
try {
FairMQMessage *msg = fTransportFactory->CreateMessage();
received = fPayloadInputs->at(0)->Receive(msg);
if (received > 0) {
// command handling goes here.
LOG(INFO) << "> received command <";
received = 0;
}
delete msg;
boost::this_thread::interruption_point();
switch (key)
{
case InputFile:
return fInputFile;
case ParFile:
return fParFile;
case Branch:
return fBranch;
default:
return FairMQDevice::GetProperty(key, default_);
}
catch (boost::thread_interrupted &) {
LOG(DEBUG) << "commandListener interrupted";
break;
}
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::SetProperty(const int key, const int value)
{
switch (key)
{
case EventRate:
fEventRate = value;
break;
default:
FairMQDevice::SetProperty(key, value);
break;
}
}
LOG(DEBUG) << ">>>>>>> stopping commandListener <<<<<<<";
}
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::SetProperty(const int key, const std::string& value, const int slot/*= 0*/)
int GenericSampler<SamplerPolicy,OutputPolicy>::GetProperty(const int key, const int default_/*= 0*/)
{
switch (key)
{
case InputFile:
fInputFile = value;
break;
case ParFile:
fParFile = value;
break;
case Branch:
fBranch = value;
break;
default:
FairMQDevice::SetProperty(key, value, slot);
break;
}
}
template <typename SamplerPolicy, typename OutputPolicy>
std::string GenericSampler<SamplerPolicy,OutputPolicy>::GetProperty(const int key, const std::string& default_/*= ""*/, const int slot/*= 0*/)
{
switch (key)
{
case InputFile:
return fInputFile;
case ParFile:
return fParFile;
case Branch:
return fBranch;
default:
return FairMQDevice::GetProperty(key, default_, slot);
}
}
template <typename SamplerPolicy, typename OutputPolicy>
void GenericSampler<SamplerPolicy,OutputPolicy>::SetProperty(const int key, const int value, const int slot/*= 0*/)
{
switch (key)
{
case EventRate:
fEventRate = value;
break;
default:
FairMQDevice::SetProperty(key, value, slot);
break;
}
}
template <typename SamplerPolicy, typename OutputPolicy>
int GenericSampler<SamplerPolicy,OutputPolicy>::GetProperty(const int key, const int default_/*= 0*/, const int slot/*= 0*/)
{
switch (key)
{
case EventRate:
return fEventRate;
default:
return FairMQDevice::GetProperty(key, default_, slot);
}
switch (key)
{
case EventRate:
return fEventRate;
default:
return FairMQDevice::GetProperty(key, default_);
}
}