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

add typedef for policies in host class, and make the sampler task container optional.

Browse Source
This commit is contained in:
NicolasWinckler 2015-10-13 16:05:49 +02:00
parent 946f1c9d00
commit fe91aa1634
2 changed files with 44 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
fairmq/devices/GenericFileSink.h
View File

@ -27,24 +27,26 @@
* Function to define in (parent) policy classes : * Function to define in (parent) policy classes :
* *
* -------- INPUT POLICY -------- * -------- INPUT POLICY --------
* InputPolicy::InitContainer(...) * deserialization_type::InitContainer(...)
* CONTAINER_TYPE InputPolicy::DeSerializeMsg(FairMQMessage* msg) * CONTAINER_TYPE deserialization_type::DeSerializeMsg(FairMQMessage* msg)
* *
* *
* -------- OUTPUT POLICY -------- * -------- OUTPUT POLICY --------
* OutputPolicy::AddToFile(CONTAINER_TYPE); * sink_type::AddToFile(CONTAINER_TYPE);
* OutputPolicy::InitOutputFile() * sink_type::InitOutputFile()
**********************************************************************/ **********************************************************************/
#include "FairMQDevice.h" #include "FairMQDevice.h"
template <typename InputPolicy, typename OutputPolicy> template <typename T, typename U>
class GenericFileSink : public FairMQDevice, public InputPolicy, public OutputPolicy class GenericFileSink : public FairMQDevice, public T, public U
{ {
typedef T deserialization_type;
typedef U sink_type;
public: public:
GenericFileSink() GenericFileSink()
: InputPolicy() : deserialization_type()
, OutputPolicy() , sink_type()
{} {}
virtual ~GenericFileSink() virtual ~GenericFileSink()
@ -58,13 +60,13 @@ class GenericFileSink : public FairMQDevice, public InputPolicy, public OutputPo
template <typename... Args> template <typename... Args>
void InitInputContainer(Args... args) void InitInputContainer(Args... args)
{ {
InputPolicy::InitContainer(std::forward<Args>(args)...); deserialization_type::InitContainer(std::forward<Args>(args)...);
} }
protected: protected:
virtual void InitTask() virtual void InitTask()
{ {
OutputPolicy::InitOutputFile(); sink_type::InitOutputFile();
} }
virtual void Run() virtual void Run()
@ -80,7 +82,7 @@ class GenericFileSink : public FairMQDevice, public InputPolicy, public OutputPo
if (inputChannel.Receive(msg) > 0) if (inputChannel.Receive(msg) > 0)
{ {
OutputPolicy::AddToFile(InputPolicy::DeSerializeMsg(msg.get())); sink_type::AddToFile(deserialization_type::DeSerializeMsg(msg.get()));
receivedMsg++; receivedMsg++;
} }
} }

59
fairmq/devices/GenericProcessor.h
View File

@ -23,31 +23,34 @@
* Function to define in (parent) policy classes : * Function to define in (parent) policy classes :
* *
* -------- INPUT POLICY -------- * -------- INPUT POLICY --------
* InputPolicy::InitContainer(...) * deserialization_type::InitContainer(...)
* CONTAINER_TYPE InputPolicy::DeSerializeMsg(FairMQMessage* msg) * CONTAINER_TYPE deserialization_type::DeSerializeMsg(FairMQMessage* msg)
* InputPolicy::InitContainer(...) // if GenericProcessor::InitInputContainer(...) is used * deserialization_type::InitContainer(...) // if GenericProcessor::InitInputContainer(...) is used
* *
* *
* -------- OUTPUT POLICY -------- * -------- OUTPUT POLICY --------
* OutputPolicy::SerializeMsg(CONTAINER_TYPE) * serialization_type::SerializeMsg(CONTAINER_TYPE)
* OutputPolicy::SetMessage(FairMQMessage* msg) * serialization_type::SetMessage(FairMQMessage* msg)
* OutputPolicy::InitContainer(...) // if GenericProcessor::InitOutputContainer(...) is used * serialization_type::InitContainer(...) // if GenericProcessor::InitOutputContainer(...) is used
* *
* -------- TASK POLICY -------- * -------- TASK POLICY --------
* CONTAINER_TYPE TaskPolicy::GetOutputData() * CONTAINER_TYPE proc_task_type::GetOutputData()
* TaskPolicy::ExecuteTask(CONTAINER_TYPE container) * proc_task_type::ExecuteTask(CONTAINER_TYPE container)
* TaskPolicy::InitTask(...) // if GenericProcessor::InitTask(...) is used * proc_task_type::InitTask(...) // if GenericProcessor::InitTask(...) is used
* *
**********************************************************************/ **********************************************************************/
template <typename InputPolicy, typename OutputPolicy, typename TaskPolicy> template <typename T, typename U, typename V>
class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputPolicy, public TaskPolicy class GenericProcessor : public FairMQDevice, public T, public U, public V
{ {
typedef T deserialization_type;
typedef U serialization_type;
typedef V proc_task_type;
public: public:
GenericProcessor() GenericProcessor()
: InputPolicy() : deserialization_type()
, OutputPolicy() , serialization_type()
, TaskPolicy() , proc_task_type()
{} {}
virtual ~GenericProcessor() virtual ~GenericProcessor()
@ -64,19 +67,19 @@ class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputP
template <typename... Args> template <typename... Args>
void InitTask(Args... args) void InitTask(Args... args)
{ {
TaskPolicy::InitTask(std::forward<Args>(args)...); proc_task_type::InitTask(std::forward<Args>(args)...);
} }
template <typename... Args> template <typename... Args>
void InitInputContainer(Args... args) void InitInputContainer(Args... args)
{ {
InputPolicy::InitContainer(std::forward<Args>(args)...); deserialization_type::InitContainer(std::forward<Args>(args)...);
} }
template <typename... Args> template <typename... Args>
void InitOutputContainer(Args... args) void InitOutputContainer(Args... args)
{ {
OutputPolicy::InitContainer(std::forward<Args>(args)...); serialization_type::InitContainer(std::forward<Args>(args)...);
} }
/* /*
@ -84,8 +87,8 @@ class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputP
// *********************** TODO: implement multipart features // *********************** TODO: implement multipart features
void SendPart() void SendPart()
{ {
fChannels["data-out"].at(0).Send(OutputPolicy::SerializeMsg(TaskPolicy::GetData()), "snd-more"); fChannels["data-out"].at(0).Send(serialization_type::SerializeMsg(proc_task_type::GetData()), "snd-more");
OutputPolicy::CloseMessage(); serialization_type::CloseMessage();
} }
// void SendPart(); // void SendPart();
@ -94,7 +97,7 @@ class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputP
{ {
if (fChannels["data-in"].at(0).ExpectsAnotherPart()) if (fChannels["data-in"].at(0).ExpectsAnotherPart())
{ {
InputPolicy::CloseMessage(); deserialization_type::CloseMessage();
// fProcessorTask->GetPayload()->CloseMessage(); // fProcessorTask->GetPayload()->CloseMessage();
fProcessorTask->SetPayload(fTransportFactory->CreateMessage()); fProcessorTask->SetPayload(fTransportFactory->CreateMessage());
return fChannels["data-in"].at(0).Receive(fProcessorTask->GetPayload()); return fChannels["data-in"].at(0).Receive(fProcessorTask->GetPayload());
@ -131,16 +134,16 @@ class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputP
if (inputChannel.Receive(msg) > 0) if (inputChannel.Receive(msg) > 0)
{ {
// InputPolicy::DeSerializeMsg(msg) --> deserialize data of msg and fill output container // deserialization_type::DeSerializeMsg(msg) --> deserialize data of msg and fill output container
// TaskPolicy::ExecuteTask( ... ) --> process output container // proc_task_type::ExecuteTask( ... ) --> process output container
TaskPolicy::ExecuteTask(InputPolicy::DeSerializeMsg(msg.get())); proc_task_type::ExecuteTask(deserialization_type::DeSerializeMsg(msg.get()));
// OutputPolicy::fMessage point to msg // serialization_type::fMessage point to msg
OutputPolicy::SetMessage(msg.get()); serialization_type::SetMessage(msg.get());
// TaskPolicy::GetOutputData() --> Get processed output container // proc_task_type::GetOutputData() --> Get processed output container
// OutputPolicy::message(...) --> Serialize output container and fill fMessage // serialization_type::message(...) --> Serialize output container and fill fMessage
outputChannel.Send(OutputPolicy::SerializeMsg(TaskPolicy::GetOutputData())); outputChannel.Send(serialization_type::SerializeMsg(proc_task_type::GetOutputData()));
sentMsgs++; sentMsgs++;
} }
} }