PndMvdMQTaskProcessor.cxx

Go to the documentation of this file.

/********************************************************************************
 *    Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
 *                                                                              *
 *              This software is distributed under the terms of the             * 
 *         GNU Lesser General Public Licence version 3 (LGPL) version 3,        *  
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/

#include <PndMvdMQTaskProcessor.h>
#include "FairMQParts.h"
#include "PixelFindHits.h"
#include "TMessage.h"
#include "PixelDigiPar.h"

// special class to expose protected TMessage constructor
class Ex9TMessage2 : public TMessage
{
  public:
  Ex9TMessage2(void* buf, Int_t len)
    : TMessage(buf, len)
  {
    ResetBit(kIsOwner);
  }
};

// helper function to clean up the object holding the data after it is transported.
void free_tmessage4(void *data, void *hint)
{
    delete (TMessage*)hint;
}

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

}

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


void PndMvdMQTaskProcessor::Init()
{
    //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();
}

void PndMvdMQTaskProcessor::Run()
{
    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!";
}




void PndMvdMQTaskProcessor::CustomCleanup(void *data, void *hint)
{
    delete (std::string*)hint;
}



void PndMvdMQTaskProcessor::SetProperty(const int key, const std::string& value)
{
    switch (key)
    {
        default:
            FairMQDevice::SetProperty(key, value);
            break;
    }
}



std::string PndMvdMQTaskProcessor::GetProperty(const int key, const std::string& default_)
{
    switch (key)
    {
        default:
            return FairMQDevice::GetProperty(key, default_);
    }
}


void PndMvdMQTaskProcessor::SetProperty(const int key, const int value)
{
    FairMQDevice::SetProperty(key, value);
}

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







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

FairParGenericSet* PndMvdMQTaskProcessor::UpdateParameter(FairParGenericSet* thisPar) {
  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;
}