<commit_before>// Author: <NAME> 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 <zmq.hpp>

#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 "AliRecoServerThread.h"

ClassImp(AliRecoServerThread);
AliRecoServerThread::AliRecoServerThread(zmq::context_t *context, AliReconstruction* reco)
  : TQObject(),
		fContext(0),
  	fReco(0),
  	fHost("tcp://*:5051"),
    fThread(0),
    fCond(0)
{
	fContext = context;
	fReco = reco;
}

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

Bool_t AliRecoServerThread::Start(const char* host)
{
	if(!fThread){
  	fHost = host;
  	fCond = new TCondition(0);
  	fThread = new TThread("AliRecoServerThread", (void(*) (void *) ) &RunThreaded, (void*)  this );
  	fThread->Run();
  
  	return kTRUE;
	}
	
	return kFALSE;	
}

Int_t AliRecoServerThread::Stop()
{
	fCond->Signal();
 
  return 0;
}

Bool_t AliRecoServerThread::ForceStop()
{
	if(fThread){
		fThread->Kill();
		fThread->Delete();
		fThread=0;
		
		return kTRUE;
	}
	
	return kFALSE;
}

void AliRecoServerThread::Finished(Int_t status)
{
  Emit("Finished(Int_t)", status);
}

void AliRecoServerThread::SendStreamerInfos(TMessage* mess, zmq::socket_t *sock)
{
	//printf("Sending Streamer Infos....\n");

	// Check if TStreamerInfo must be sent. The list of TStreamerInfo of classes
   // in the object in the message is in the fInfos list of the message.
   // We send only the TStreamerInfos not yet sent on this socket.
	TList* infos = mess->GetStreamerInfos();
   
      TIter next(infos);
      TStreamerInfo *info;
      TList *minilist = 0;
      while ((info = (TStreamerInfo*)next())) {
         Int_t uid = info->GetNumber();
         if (!minilist) minilist = new TList();
         
         minilist->Add(info);
      }
      
      if (minilist) {
         TMessage messinfo(kMESS_STREAMERINFO);
         messinfo.WriteObject(minilist);
         delete minilist;
         if (messinfo.GetStreamerInfos())
            messinfo.GetStreamerInfos()->Clear();
#if ROOT_VERSION_CODE < ROOT_VERSION(5,99,0)           
           messinfo.SetLength();
#endif
           
          int bufsize = messinfo.Length();
        	char* buf = (char*) malloc(bufsize * sizeof(char));
          memcpy(buf, messinfo.Buffer(), bufsize);

        	// send!
          zmq::message_t message((void*)buf, bufsize, 0, 0);
                     
         if (sock->send(message, ZMQ_SNDMORE))
            Warning("SendStreamerInfos", "problems sending TStreamerInfo's ...");
      }

   return;
}

void AliRecoServerThread::SendEvent(AliESDEvent* event, zmq::socket_t* socket)
{
  if(!event) return;

  TMessage tmess(kMESS_OBJECT);
  tmess.Reset();
  tmess.WriteObject(event);

  TMessage::EnableSchemaEvolutionForAll(kTRUE);
  SendStreamerInfos(&tmess, socket);

#if ROOT_VERSION_CODE < ROOT_VERSION(5,99,0)           
  tmess.SetLength();
#endif
  int bufsize = tmess.Length();
  char* buf = (char*) malloc(bufsize * sizeof(char));
  memcpy(buf, tmess.Buffer(), bufsize);

  // send!
  zmq::message_t message((void*)buf, bufsize, 0, 0);
  socket->send(message);

}


void* AliRecoServerThread::RunThreaded(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();

	zmq::socket_t 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();
				SendEvent(event, &publisher);

   			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_msg>TMessage::SetLength() is private in Root 5.34/10<commit_after>// Author: <NAME> 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 <zmq.hpp>

#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 "AliRecoServerThread.h"

ClassImp(AliRecoServerThread);
AliRecoServerThread::AliRecoServerThread(zmq::context_t *context, AliReconstruction* reco)
  : TQObject(),
		fContext(0),
  	fReco(0),
  	fHost("tcp://*:5051"),
    fThread(0),
    fCond(0)
{
	fContext = context;
	fReco = reco;
}

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

Bool_t AliRecoServerThread::Start(const char* host)
{
	if(!fThread){
  	fHost = host;
  	fCond = new TCondition(0);
  	fThread = new TThread("AliRecoServerThread", (void(*) (void *) ) &RunThreaded, (void*)  this );
  	fThread->Run();
  
  	return kTRUE;
	}
	
	return kFALSE;	
}

Int_t AliRecoServerThread::Stop()
{
	fCond->Signal();
 
  return 0;
}

Bool_t AliRecoServerThread::ForceStop()
{
	if(fThread){
		fThread->Kill();
		fThread->Delete();
		fThread=0;
		
		return kTRUE;
	}
	
	return kFALSE;
}

void AliRecoServerThread::Finished(Int_t status)
{
  Emit("Finished(Int_t)", status);
}

void AliRecoServerThread::SendStreamerInfos(TMessage* mess, zmq::socket_t *sock)
{
	//printf("Sending Streamer Infos....\n");

	// Check if TStreamerInfo must be sent. The list of TStreamerInfo of classes
   // in the object in the message is in the fInfos list of the message.
   // We send only the TStreamerInfos not yet sent on this socket.
	TList* infos = mess->GetStreamerInfos();
   
      TIter next(infos);
      TStreamerInfo *info;
      TList *minilist = 0;
      while ((info = (TStreamerInfo*)next())) {
         Int_t uid = info->GetNumber();
         if (!minilist) minilist = new TList();
         
         minilist->Add(info);
      }
      
      if (minilist) {
         TMessage messinfo(kMESS_STREAMERINFO);
         messinfo.WriteObject(minilist);
         delete minilist;
         if (messinfo.GetStreamerInfos())
            messinfo.GetStreamerInfos()->Clear();
#if ROOT_VERSION_CODE < ROOT_VERSION(5,34,10)           
           messinfo.SetLength();
#endif
           
          int bufsize = messinfo.Length();
        	char* buf = (char*) malloc(bufsize * sizeof(char));
          memcpy(buf, messinfo.Buffer(), bufsize);

        	// send!
          zmq::message_t message((void*)buf, bufsize, 0, 0);
                     
         if (sock->send(message, ZMQ_SNDMORE))
            Warning("SendStreamerInfos", "problems sending TStreamerInfo's ...");
      }

   return;
}

void AliRecoServerThread::SendEvent(AliESDEvent* event, zmq::socket_t* socket)
{
  if(!event) return;

  TMessage tmess(kMESS_OBJECT);
  tmess.Reset();
  tmess.WriteObject(event);

  TMessage::EnableSchemaEvolutionForAll(kTRUE);
  SendStreamerInfos(&tmess, socket);

#if ROOT_VERSION_CODE < ROOT_VERSION(5,34,10)           
  tmess.SetLength();
#endif
  int bufsize = tmess.Length();
  char* buf = (char*) malloc(bufsize * sizeof(char));
  memcpy(buf, tmess.Buffer(), bufsize);

  // send!
  zmq::message_t message((void*)buf, bufsize, 0, 0);
  socket->send(message);

}


void* AliRecoServerThread::RunThreaded(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();

	zmq::socket_t 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();
				SendEvent(event, &publisher);

   			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;
  
}

{'fraction_non_alphanumeric': 0.17027530126383855, 'fraction_numerical': 0.010091113941412756, 'mean_word_length': 3.33276740237691, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5754432', 'n_tokens_mistral': 3698, 'n_tokens_neox': 3492, 'n_words': 681}