[u/mrichter/AliRoot.git] / MONITOR / AliRecoServerThread.cxx

// Author: Mihai Niculescu 2013

/**************************************************************************
 * Copyright(c) 1998-2013, ALICE Experiment at CERN, all rights reserved. *
 * See http://aliceinfo.cern.ch/Offline/AliRoot/License.html for          *
 * full copyright notice.                                                 *
 **************************************************************************/

#include <RVersion.h>
#include <stdlib.h>

#include <TCondition.h>
#include <TBufferFile.h>
#include <TMessage.h>
#include <TObjArray.h>
#include <TStreamerInfo.h>
#include <TThread.h>

#include <AliESDEvent.h>
#include <AliESDfriend.h>
#include <AliRawReader.h>
#include <AliRunLoader.h>
#include <AliReconstruction.h>

#include <AliNetMessage.h>
#include <AliSocket.h>

#include <zmq.hpp>

#include "AliRecoServerThread.h"

ClassImp(AliRecoServerThread)
AliRecoServerThread::AliRecoServerThread(zmq::context_t *context, AliReconstruction* reco)
  : AliThreadedSocket(context, AliThreadedSocket::WRITE),
  	fReco(0),
    fCond(0)
{
	fReco = reco;
}

AliRecoServerThread::~AliRecoServerThread()
{
	Stop();
}

Bool_t AliRecoServerThread::Start(const char* endpoint)
{
	fHost = endpoint;
	
	return AliThreadedSocket::Start();
}

void* AliRecoServerThread::RunThrdWrite(void* arg)
{
	TThread::SetCancelAsynchronous();
	TThread::SetCancelOn();
	
	AliRecoServerThread* recoTh = (AliRecoServerThread*)arg;
	
	const char* host = recoTh->GetHost();
	zmq::context_t* context = recoTh->GetContext();
	AliReconstruction* reco = recoTh->GetReconstruction();

 // generate a publish socket
	AliSocket publisher(context, ZMQ_PUB);
	publisher.Bind(host);
	
  if(reco==0) return 0;
  
  AliESDEvent* event;
  
	reco->Begin(NULL);
  if (reco->GetAbort() != TSelector::kContinue) return 0;
  
  reco->SlaveBegin(NULL);
	if (reco->GetAbort() != TSelector::kContinue) return 0;
  
  //******* The loop over events
    Int_t iEvent = 0;
    while ( reco->HasNextEventAfter(iEvent) ) {
      // check if process has enough resources 
      if (!reco->HasEnoughResources(iEvent)) break;
      Bool_t status = reco->ProcessEvent(iEvent);
      
      if (status)
      {
		    event = reco->GetESDEvent();
		    
      	AliNetMessage tmess(kMESS_OBJECT);
  			tmess.Reset();
  			tmess.WriteObject(event);
  				
				publisher.Send(tmess);

   			sleep(1);
     }
      else {
        reco->Abort("ProcessEvent",TSelector::kAbortFile);
      }
      		
      reco->CleanProcessedEvent();
      if(recoTh->Condition()->TimedWaitRelative(500)==0){
				break;
			}			
      iEvent++;
    }
    reco->SlaveTerminate();
    if (reco->GetAbort() != TSelector::kContinue) return 0;
    reco->Terminate();
    if (reco->GetAbort() != TSelector::kContinue) return 0;
  
}
Commit	Line	Data
7e0cf530	1	// Author: Mihai Niculescu 2013
	2
	3	/**************************************************************************
	4	* Copyright(c) 1998-2013, ALICE Experiment at CERN, all rights reserved. *
	5	* See http://aliceinfo.cern.ch/Offline/AliRoot/License.html for *
	6	* full copyright notice. *
	7	**************************************************************************/
	8
bd2715e3	9	#include <RVersion.h>
7e0cf530	10	#include <stdlib.h>
7e0cf530	11
7e0cf530	12	#include <TCondition.h>
	13	#include <TBufferFile.h>
	14	#include <TMessage.h>
	15	#include <TObjArray.h>
	16	#include <TStreamerInfo.h>
	17	#include <TThread.h>
	18
7e0cf530	19	#include <AliESDEvent.h>
	20	#include <AliESDfriend.h>
	21	#include <AliRawReader.h>
	22	#include <AliRunLoader.h>
	23	#include <AliReconstruction.h>
	24
db352b46	25	#include <AliNetMessage.h>
	26	#include <AliSocket.h>
	27
	28	#include <zmq.hpp>
	29
7e0cf530	30	#include "AliRecoServerThread.h"
7e0cf530	31
db352b46	32	ClassImp(AliRecoServerThread)
7e0cf530	33	AliRecoServerThread::AliRecoServerThread(zmq::context_t context, AliReconstruction reco)
db352b46	34	: AliThreadedSocket(context, AliThreadedSocket::WRITE),
7e0cf530	35	fReco(0),
7e0cf530	36	fCond(0)
7e0cf530	37	{
7e0cf530	38	fReco = reco;
	39	}
	40
	41	AliRecoServerThread::~AliRecoServerThread()
	42	{
	43	Stop();
	44	}
	45
db352b46	46	Bool_t AliRecoServerThread::Start(const char* endpoint)
7e0cf530	47	{
db352b46	48	fHost = endpoint;
7e0cf530	49
db352b46	50	return AliThreadedSocket::Start();
7e0cf530	51	}
7e0cf530	52
db352b46	53	void* AliRecoServerThread::RunThrdWrite(void* arg)
7e0cf530	54	{
	55	TThread::SetCancelAsynchronous();
	56	TThread::SetCancelOn();
	57
	58	AliRecoServerThread* recoTh = (AliRecoServerThread*)arg;
	59
	60	const char* host = recoTh->GetHost();
	61	zmq::context_t* context = recoTh->GetContext();
db352b46	62	AliReconstruction* reco = recoTh->GetReconstruction();
7e0cf530	63
db352b46	64	// generate a publish socket
	65	AliSocket publisher(context, ZMQ_PUB);
	66	publisher.Bind(host);
7e0cf530	67
	68	if(reco==0) return 0;
	69
	70	AliESDEvent* event;
	71
	72	reco->Begin(NULL);
	73	if (reco->GetAbort() != TSelector::kContinue) return 0;
	74
	75	reco->SlaveBegin(NULL);
	76	if (reco->GetAbort() != TSelector::kContinue) return 0;
	77
	78	//******* The loop over events
	79	Int_t iEvent = 0;
	80	while ( reco->HasNextEventAfter(iEvent) ) {
	81	// check if process has enough resources
	82	if (!reco->HasEnoughResources(iEvent)) break;
	83	Bool_t status = reco->ProcessEvent(iEvent);
	84
	85	if (status)
	86	{
db352b46	87	event = reco->GetESDEvent();
	88
	89	AliNetMessage tmess(kMESS_OBJECT);
	90	tmess.Reset();
	91	tmess.WriteObject(event);
	92
	93	publisher.Send(tmess);
7e0cf530	94
	95	sleep(1);
	96	}
	97	else {
	98	reco->Abort("ProcessEvent",TSelector::kAbortFile);
	99	}
	100
	101	reco->CleanProcessedEvent();
	102	if(recoTh->Condition()->TimedWaitRelative(500)==0){
	103	break;
	104	}
	105	iEvent++;
	106	}
	107	reco->SlaveTerminate();
	108	if (reco->GetAbort() != TSelector::kContinue) return 0;
	109	reco->Terminate();
	110	if (reco->GetAbort() != TSelector::kContinue) return 0;
	111
	112	}