#include <PndMvdMQTaskProcessor.h>

Inheritance diagram for PndMvdMQTaskProcessor< T >:

Public Types
enum	{ Last }

Public Member Functions
	PndMvdMQTaskProcessor ()

virtual	~PndMvdMQTaskProcessor ()

void	SetProperty (const int key, const std::string &value)

std::string	GetProperty (const int key, const std::string &default_="")

void	SetProperty (const int key, const int value)

int	GetProperty (const int key, const int default_=0)

void	SetDataToKeep (std::string tStr)

Protected Member Functions
virtual void	Run ()

virtual void	Init ()

void	UpdateParameters ()

FairParGenericSet *	UpdateParameter (FairParGenericSet *thisPar)

Static Protected Member Functions
static void	CustomCleanup (void data, void hint)

Protected Attributes
FairEventHeader *	fEventHeader

TList *	fInput

TList *	fOutput

int	fNewRunId

int	fCurrentRunId

std::string	fDataToKeep

T *	fFairTask

TList *	fParCList

FairGeoParSet *	fGeoPar

Detailed Description

template<typename T>
class PndMvdMQTaskProcessor< T >

Definition at line 26 of file PndMvdMQTaskProcessor.h.

Member Enumeration Documentation

template<typename T >

anonymous enum

Enumerator
Last

Definition at line 29 of file PndMvdMQTaskProcessor.h.

     {
         Last
     };

Constructor & Destructor Documentation

template<typename T >

PndMvdMQTaskProcessor< T >::PndMvdMQTaskProcessor ( )

Definition at line 32 of file PndMvdMQTaskProcessor.cxx.

   : FairMQDevice()
   , fNewRunId(1)
   , fCurrentRunId(-1)
   , fInput(NULL)
   , fOutput(NULL)
   , fEventHeader(NULL)
   , fFairTask(NULL)
   , fGeoPar(nullptr)
   , fParCList(NULL)
 {
 
 }

template<typename T >

PndMvdMQTaskProcessor< T >::~PndMvdMQTaskProcessor ( )

virtual

Definition at line 46 of file PndMvdMQTaskProcessor.cxx.

References PndMvdMQTaskProcessor< T >::fGeoPar, PndMvdMQTaskProcessor< T >::fInput, and PndMvdMQTaskProcessor< T >::fOutput.

 {
   if(fGeoPar)
     {
       delete fGeoPar;
       fGeoPar=nullptr;
     }
   
   if(fInput)
     {
       delete fInput;
       fInput=nullptr;
     }
   
   if(fOutput)
     {
       delete fOutput;
       fOutput=nullptr;
     }
 }

Member Function Documentation

template<typename T >

void PndMvdMQTaskProcessor< T >::CustomCleanup	(	void *	data,
		void *	hint
	)

staticprotected

Definition at line 147 of file PndMvdMQTaskProcessor.cxx.

Referenced by PndMvdMQTaskProcessor< T >::UpdateParameter().

 {
     delete (std::string*)hint;
 }

template<typename T >

std::string PndMvdMQTaskProcessor< T >::GetProperty	(	const int	key,
		const std::string &	default_ = `""`
	)

Definition at line 166 of file PndMvdMQTaskProcessor.cxx.

 {
     switch (key)
     {
         default:
             return FairMQDevice::GetProperty(key, default_);
     }
 }

template<typename T >

int PndMvdMQTaskProcessor< T >::GetProperty	(	const int	key,
		const int	default_ = `0`
	)

Definition at line 181 of file PndMvdMQTaskProcessor.cxx.

 {
     return FairMQDevice::GetProperty(key, value);
 }

template<typename T >

void PndMvdMQTaskProcessor< T >::Init ( )

protectedvirtual

Definition at line 68 of file PndMvdMQTaskProcessor.cxx.

References PndMvdMQTaskProcessor< T >::fFairTask, PndMvdMQTaskProcessor< T >::fGeoPar, PndMvdMQTaskProcessor< T >::fInput, PndMvdMQTaskProcessor< T >::fOutput, and PndMvdMQTaskProcessor< T >::fParCList.

 {
     //fHitFinder->InitMQ(fRootParFileName,fAsciiParFileName);
   fFairTask = new PixelFindHits();
   fGeoPar = new FairGeoParSet("FairGeoParSet");
   fParCList = new TList();
   fParCList->Add(fGeoPar);
   fFairTask->GetParList(fParCList);
   
   fOutput = new TList();
   fInput = new TList();
 }

template<typename T >

void PndMvdMQTaskProcessor< T >::Run ( )

protectedvirtual

Definition at line 81 of file PndMvdMQTaskProcessor.cxx.

References PndMvdMQTaskProcessor< T >::fCurrentRunId, PndMvdMQTaskProcessor< T >::fEventHeader, PndMvdMQTaskProcessor< T >::fFairTask, PndMvdMQTaskProcessor< T >::fInput, PndMvdMQTaskProcessor< T >::fNewRunId, PndMvdMQTaskProcessor< T >::fOutput, PndMvdMQTaskProcessor< T >::fParCList, free_tmessage4(), and PndMvdMQTaskProcessor< T >::UpdateParameters().

 {
     int receivedMsgs = 0;
     int sentMsgs = 0;
 
     while (CheckCurrentState(RUNNING))
     {
       FairMQParts parts;
       
       if ( Receive(parts,"data-in") >= 0 )
         {
           LOG(INFO)<<"message received";
           receivedMsgs++;
           
           TObject* tempObjects[10];
           for ( int ipart = 0 ; ipart < parts.Size() ; ipart++ ) 
             {
               Ex9TMessage2 tm(parts.At(ipart)->GetData(), parts.At(ipart)->GetSize());
               tempObjects[ipart] = (TObject*)tm.ReadObject(tm.GetClass());
               if ( strcmp(tempObjects[ipart]->GetName(),"EventHeader.") == 0 ) 
                 fEventHeader = (FairEventHeader*)tempObjects[ipart];
               else {
                 fInput->Add(tempObjects[ipart]);
               }
             }
           
           fNewRunId = fEventHeader->GetRunId();
           if(fNewRunId!=fCurrentRunId)
             {
               fCurrentRunId=fNewRunId;
               UpdateParameters();
               PixelDigiPar* tempDP = (PixelDigiPar*)fParCList->FindObject("PixelDigiParameters");
               fFairTask->InitMQ(fParCList);
             }
             
           
           // Execute hit finder task
           fOutput->Clear();
           LOG(INFO) << " analyzing event " << fEventHeader->GetMCEntryNumber();
           fFairTask->ExecMQ(fInput,fOutput);
 
           TMessage* messageFEH;
           TMessage* messageTCA[10];
           FairMQParts partsOut;
           
           messageFEH = new TMessage(kMESS_OBJECT);
           messageFEH->WriteObject(fEventHeader);
           partsOut.AddPart(NewMessage(messageFEH->Buffer(), messageFEH->BufferSize(), free_tmessage4, messageFEH));
           for ( int iobj = 0 ; iobj < fOutput->GetEntries() ; iobj++ ) {
             messageTCA[iobj] = new TMessage(kMESS_OBJECT);
             messageTCA[iobj]->WriteObject(fOutput->At(iobj));
             partsOut.AddPart(NewMessage(messageTCA[iobj]->Buffer(), messageTCA[iobj]->BufferSize(), free_tmessage4, messageTCA[iobj]));
             Send(partsOut, "data-out");
           }
           
           sentMsgs++;
         }
       fInput->Clear();
     }
     
     MQLOG(INFO) << "Received " << receivedMsgs << " and sent " << sentMsgs << " messages!";
 }

template<typename T >

void PndMvdMQTaskProcessor< T >::SetDataToKeep ( std::string tStr )

inline

Definition at line 42 of file PndMvdMQTaskProcessor.h.

References PndMvdMQTaskProcessor< T >::fDataToKeep.

Referenced by main().

42 { fDataToKeep = tStr;}

PndMvdMQTaskProcessor::fDataToKeep

std::string fDataToKeep

Definition: PndMvdMQTaskProcessor.h:60

template<typename T >

void PndMvdMQTaskProcessor< T >::SetProperty	(	const int	key,
		const std::string &	value
	)

Definition at line 154 of file PndMvdMQTaskProcessor.cxx.

 {
     switch (key)
     {
         default:
             FairMQDevice::SetProperty(key, value);
             break;
     }
 }

template<typename T >

void PndMvdMQTaskProcessor< T >::SetProperty	(	const int	key,
		const int	value
	)

Definition at line 176 of file PndMvdMQTaskProcessor.cxx.

 {
     FairMQDevice::SetProperty(key, value);
 }

template<typename T >

FairParGenericSet * PndMvdMQTaskProcessor< T >::UpdateParameter ( FairParGenericSet * thisPar )

protected

Definition at line 200 of file PndMvdMQTaskProcessor.cxx.

References PndMvdMQTaskProcessor< T >::CustomCleanup(), and PndMvdMQTaskProcessor< T >::fCurrentRunId.

Referenced by PndMvdMQTaskProcessor< T >::UpdateParameters().

                                                                                     {
   std::string paramName = thisPar->GetName();
   //  boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
   std::string* reqStr = new std::string(paramName + "," + std::to_string(fCurrentRunId));
   LOG(WARN) << "Requesting parameter \"" << paramName << "\" for Run ID " << fCurrentRunId << " (" << thisPar << ")";
   std::unique_ptr<FairMQMessage> req(NewMessage(const_cast<char*>(reqStr->c_str()), reqStr->length(), CustomCleanup, reqStr));
   std::unique_ptr<FairMQMessage> rep(NewMessage());
   
   if (Send(req,"param") > 0)
     {
       if (Receive(rep,"param") > 0)
         {
           Ex9TMessage2 tm(rep->GetData(), rep->GetSize());
           thisPar = (FairParGenericSet*)tm.ReadObject(tm.GetClass());
           LOG(WARN) << "Received parameter"<< paramName <<" from the server (" << thisPar << ")";
           return thisPar;
         }
     } 
   return NULL;
 }

template<typename T >

void PndMvdMQTaskProcessor< T >::UpdateParameters ( )

protected

Definition at line 192 of file PndMvdMQTaskProcessor.cxx.

References PndMvdMQTaskProcessor< T >::fParCList, and PndMvdMQTaskProcessor< T >::UpdateParameter().

Referenced by PndMvdMQTaskProcessor< T >::Run().

                                              {
   for ( int iparC = 0 ; iparC < fParCList->GetEntries() ; iparC++ ) {
     FairParGenericSet* tempObj = (FairParGenericSet*)(fParCList->At(iparC));
     fParCList->Remove(tempObj);
     fParCList->AddAt(UpdateParameter(tempObj),iparC);
   }
 }