EvtRangeScatterer Class Reference

#include <EvtRangeScatterer.h>

Inheritance diagram for EvtRangeScatterer:

Collaboration diagram for EvtRangeScatterer:

Public Member Functions
	EvtRangeScatterer (const std::string &type, const std::string &name, const IInterface *parent)
virtual	~EvtRangeScatterer () override
virtual StatusCode	initialize () override
virtual StatusCode	finalize () override
virtual int makePool	ATLAS_NOT_THREAD_SAFE (int maxevt, int nprocs, const std::string &topdir) override
virtual StatusCode exec	ATLAS_NOT_THREAD_SAFE () override
virtual void	subProcessLogs (std::vector< std::string > &) override
virtual AthenaMP::AllWorkerOutputs_ptr	generateOutputReport () override
virtual std::unique_ptr< AthenaInterprocess::ScheduledWork >	bootstrap_func () override
virtual std::unique_ptr< AthenaInterprocess::ScheduledWork >	exec_func () override
virtual std::unique_ptr< AthenaInterprocess::ScheduledWork >	fin_func () override
virtual StatusCode wait_once	ATLAS_NOT_THREAD_SAFE (pid_t &pid) override
virtual void	reportSubprocessStatuses () override
virtual void	useFdsRegistry (std::shared_ptr< AthenaInterprocess::FdsRegistry >) override
virtual void	setRandString (const std::string &randStr) override
virtual void	setMaxEvt (int maxEvt) override
virtual void	setMPRunStop (const AthenaInterprocess::IMPRunStop *runStop) override
virtual void	killChildren () override
virtual std::unique_ptr< AthenaInterprocess::ScheduledWork > virtual operator() ATLAS_NOT_THREAD_SAFE(const AthenaInterprocess std::unique_ptr< AthenaInterprocess::ScheduledWork >	bootstrap_func ()=0
virtual std::unique_ptr< ScheduledWork >	operator () ATLAS_NOT_THREAD_SAFE(const ScheduledWork &)=0

Protected Types
enum	ESRange_Status {   ESRANGE_SUCCESS, ESRANGE_NOTFOUND, ESRANGE_SEEKFAILED, ESRANGE_PROCFAILED,   ESRANGE_FILENOTMADE, ESRANGE_BADINPFILE }
enum	Func_Flag { FUNC_BOOTSTRAP, FUNC_EXEC, FUNC_FIN }

Protected Member Functions
int mapAsyncFlag	ATLAS_NOT_THREAD_SAFE (Func_Flag flag, pid_t pid=0)
int	redirectLog (const std::string &rundir, bool addTimeStamp=true)
int	updateIoReg (const std::string &rundir)
std::string	fmterror (int errnum)
int	reopenFds ()
int	handleSavedPfc (const std::filesystem::path &dest_path)
void	waitForSignal ()
IEvtSelector *	evtSelector ()

Protected Attributes
int	m_nprocs {-1}
	Number of workers spawned by the master process. More...
int	m_maxEvt {-1}
	Maximum number of events assigned to the job. More...
std::string	m_subprocTopDir
	Top run directory for subprocesses. More...
std::string	m_subprocDirPrefix
	For ex. "worker__". More...
std::string	m_evtSelName
	Name of the event selector. More...
AthenaInterprocess::ProcessGroup *	m_processGroup {nullptr}
const AthenaInterprocess::IMPRunStop *	m_mpRunStop {nullptr}
ServiceHandle< IEventProcessor >	m_evtProcessor
ServiceHandle< IAppMgrUI >	m_appMgr
ServiceHandle< IFileMgr >	m_fileMgr
ServiceHandle< IIoComponentMgr >	m_ioMgr
SmartIF< IEvtSelector >	m_evtSelector
std::string	m_fileMgrLog
std::shared_ptr< AthenaInterprocess::FdsRegistry >	m_fdsRegistry
std::string	m_randStr
Gaudi::Property< bool >	m_isPileup {this, "IsPileup", false, "Flag for configuring PileUpEventLoopMgr"}

Private Member Functions
	EvtRangeScatterer ()
	EvtRangeScatterer (const EvtRangeScatterer &)
EvtRangeScatterer &	operator= (const EvtRangeScatterer &)
void	trimRangeStrings (std::string &)
std::string	getNewRangeRequest (yampl::ISocket *socket2Processor, yampl::ISocket *socket2Pilot, int &procReportPending)
pid_t	pollFailedPidQueue (AthenaInterprocess::SharedQueue *sharedFailedPidQueue, yampl::ISocket *socket2Pilot, int &procReportPending)
int	reopenFd (int fd, const std::string &name)

Private Attributes
Gaudi::Property< std::string >	m_processorChannel {this, "ProcessorChannel", ""}
Gaudi::Property< std::string >	m_eventRangeChannel {this, "EventRangeChannel", ""}
Gaudi::Property< bool >	m_doCaching {this, "DoCaching", false}
Pid2RangeID	m_pid2RangeID

Detailed Description

Definition at line 18 of file EvtRangeScatterer.h.

Member Enumeration Documentation

ESRange_Status

enum AthenaMPToolBase::ESRange_Status

protectedinherited

Enumerator
ESRANGE_SUCCESS
ESRANGE_NOTFOUND
ESRANGE_SEEKFAILED
ESRANGE_PROCFAILED
ESRANGE_FILENOTMADE
ESRANGE_BADINPFILE

Definition at line 58 of file AthenaMPToolBase.h.

                      {
    ESRANGE_SUCCESS
    , ESRANGE_NOTFOUND
    , ESRANGE_SEEKFAILED
    , ESRANGE_PROCFAILED
    , ESRANGE_FILENOTMADE
    , ESRANGE_BADINPFILE
  };

Func_Flag

enum AthenaMPToolBase::Func_Flag

protectedinherited

Enumerator
FUNC_BOOTSTRAP
FUNC_EXEC
FUNC_FIN

Definition at line 67 of file AthenaMPToolBase.h.

                 {
    FUNC_BOOTSTRAP
    , FUNC_EXEC
    , FUNC_FIN
  };

Constructor & Destructor Documentation

EvtRangeScatterer() [1/3]

EvtRangeScatterer::EvtRangeScatterer	(	const std::string &	type,
		const std::string &	name,
		const IInterface *	parent
	)

Definition at line 27 of file EvtRangeScatterer.cxx.

  : AthenaMPToolBase(type,name,parent)
{
  m_subprocDirPrefix = "range_scatterer";
}

~EvtRangeScatterer()

EvtRangeScatterer::~EvtRangeScatterer ( )

overridevirtual

Definition at line 35 of file EvtRangeScatterer.cxx.

{
}

EvtRangeScatterer() [2/3]

EvtRangeScatterer::EvtRangeScatterer ( )

private

EvtRangeScatterer() [3/3]

EvtRangeScatterer::EvtRangeScatterer ( const EvtRangeScatterer &  )

private

Member Function Documentation

ATLAS_NOT_THREAD_SAFE() [1/4]

virtual StatusCode exec EvtRangeScatterer::ATLAS_NOT_THREAD_SAFE ( )

overridevirtual

ATLAS_NOT_THREAD_SAFE() [2/4]

int mapAsyncFlag AthenaMPToolBase::ATLAS_NOT_THREAD_SAFE ( Func_Flag  flag, pid_t  pid = 0  )

protectedinherited

ATLAS_NOT_THREAD_SAFE() [3/4]

virtual int makePool EvtRangeScatterer::ATLAS_NOT_THREAD_SAFE ( int  maxevt, int  nprocs, const std::string &  topdir  )

overridevirtual

ATLAS_NOT_THREAD_SAFE() [4/4]

virtual StatusCode wait_once AthenaMPToolBase::ATLAS_NOT_THREAD_SAFE ( pid_t &  pid )

overridevirtualinherited

Reimplemented in EvtRangeProcessor, SharedEvtQueueConsumer, and SharedHiveEvtQueueConsumer.

bootstrap_func() [1/2]

std::unique_ptr< AthenaInterprocess::ScheduledWork > EvtRangeScatterer::bootstrap_func ( )

overridevirtual

Definition at line 109 of file EvtRangeScatterer.cxx.

{
  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
  outwork->data = CxxUtils::xmalloc(sizeof(int));
  *(int*)(outwork->data) = 1; // Error code: for now use 0 success, 1 failure
  outwork->size = sizeof(int);
 
  // ...
  // (possible) TODO: extend outwork with some error message, which will be eventually
  // reported in the master proces
  // ...
 
  // Reader dir: mkdir 
  std::filesystem::path reader_rundir(m_subprocTopDir);
  reader_rundir /= std::filesystem::path(m_subprocDirPrefix);
 
  if(mkdir(reader_rundir.string().c_str(),S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH)==-1) {
    ATH_MSG_ERROR("Unable to make reader run directory: " << reader_rundir.string() << ". " << fmterror(errno));
    return outwork;
  }
 
  // Redirect logs
  if(redirectLog(reader_rundir.string()))
    return outwork;
 
  ATH_MSG_INFO("Logs redirected in the AthenaMP Event Range Scatterer PID=" << getpid());
     
  // Update Io Registry   
  if(updateIoReg(reader_rundir.string()))
    return outwork;
 
  ATH_MSG_INFO("Io registry updated in the AthenaMP Event Range Scatterer PID=" << getpid());
 
  // _______________________ Handle saved PFC (if any) ______________________
  std::filesystem::path abs_reader_rundir = std::filesystem::absolute(reader_rundir);
  if(handleSavedPfc(abs_reader_rundir))
    return outwork;
 
  // Reopen file descriptors
  if(reopenFds())
    return outwork;
 
  ATH_MSG_INFO("File descriptors re-opened in the AthenaMP Event Range Scatterer PID=" << getpid());
 
  // ________________________ I/O reinit ________________________
  if(!m_ioMgr->io_reinitialize().isSuccess()) {
    ATH_MSG_ERROR("Failed to reinitialize I/O");
    return outwork;
  } else {
    ATH_MSG_DEBUG("Successfully reinitialized I/O");
  }
 
  // Start the event selector 
  SmartIF<IService> evtSelSvc(m_evtSelector);
  if(!evtSelSvc) {
    ATH_MSG_ERROR("Failed to dyncast event selector to IService");
    return outwork;
  }
  if(!evtSelSvc->start().isSuccess()) {
    ATH_MSG_ERROR("Failed to restart the event selector");
    return outwork;
  } else {
    ATH_MSG_DEBUG("Successfully restarted the event selector");
  }
 
  // Reader dir: chdir 
  if(chdir(reader_rundir.string().c_str())==-1) {
    ATH_MSG_ERROR("Failed to chdir to " << reader_rundir.string());
    return outwork;
  }
 
  // Declare success and return
  *(int*)(outwork->data) = 0;
  return outwork;
}

bootstrap_func() [2/2]

virtual std::unique_ptr<AthenaInterprocess::ScheduledWork> virtual operator () ATLAS_NOT_THREAD_SAFE ( const AthenaInterprocess std::unique_ptr<AthenaInterprocess::ScheduledWork> AthenaMPToolBase::bootstrap_func ( )

pure virtualinherited

evtSelector()

IEvtSelector* AthenaMPToolBase::evtSelector ( )

inlineprotectedinherited

Definition at line 83 of file AthenaMPToolBase.h.

{ return m_evtSelector; }

exec_func()

std::unique_ptr< AthenaInterprocess::ScheduledWork > EvtRangeScatterer::exec_func ( )

overridevirtual

Implements AthenaMPToolBase.

Definition at line 185 of file EvtRangeScatterer.cxx.

{
  ATH_MSG_INFO("Exec function in the AthenaMP Token Scatterer PID=" << getpid());
 
  yampl::ISocketFactory* socketFactory = new yampl::SocketFactory();
  // Create a socket to communicate with the Pilot
  yampl::ISocket* socket2Pilot = socketFactory->createClientSocket(yampl::Channel(m_eventRangeChannel.value(),yampl::LOCAL),yampl::MOVE_DATA);
  ATH_MSG_INFO("Created CLIENT socket for communicating Event Ranges with the Pilot");
  // Create a socket to communicate with EvtRangeProcessors
  std::string socket2ProcessorName = m_processorChannel.value() + std::string("_") + m_randStr;
  yampl::ISocket* socket2Processor = socketFactory->createServerSocket(yampl::Channel(socket2ProcessorName,yampl::LOCAL),yampl::MOVE_DATA);
  ATH_MSG_INFO("Created SERVER socket to token processors: " << socket2ProcessorName);
 
  bool all_ok=true;
  int procReportPending(0);  // Keep track of how many output files are yet to be reported by Token Processors
 
  AthenaInterprocess::SharedQueue*  sharedFailedPidQueue(0);
  if(detStore()->retrieve(sharedFailedPidQueue,"AthenaMPFailedPidQueue_"+m_randStr).isFailure()) {
    ATH_MSG_ERROR("Unable to retrieve the pointer to Shared Failed PID Queue");
    all_ok=false;
  }
 
  // Communication protocol with the Pilot
  std::string strReady("Ready for events");
  std::string strStopProcessing("No more events");
  std::string processorWaitRequest("");
  int workerPid{-1};
 
  ATH_MSG_INFO("Starting main loop");
 
  while(all_ok) {
    // NO CACHING MODE: first get a request from one of the processors and only after that request the next event range from the pilot
    if(!m_doCaching && processorWaitRequest.empty()) {
      ATH_MSG_DEBUG("Waiting for event range request from one of the processors ... ");
      while(processorWaitRequest.empty()) {
    processorWaitRequest = getNewRangeRequest(socket2Processor,socket2Pilot,procReportPending);
    pollFailedPidQueue(sharedFailedPidQueue,socket2Pilot,procReportPending);
    usleep(1000);
      }
      ATH_MSG_INFO("One of the processors is ready for the next range");
      // Get PID from the request and Update m_pid2RangeID
      workerPid = std::atoi(processorWaitRequest.c_str());
      auto it = m_pid2RangeID.find(workerPid);
      if(it!=m_pid2RangeID.end()) {
    m_pid2RangeID.erase(it);
      }
    }    
 
    // Signal the Pilot that AthenaMP is ready for event processing
    void* ready_message = CxxUtils::xmalloc(strReady.size());
    memcpy(ready_message,strReady.data(),strReady.size());
    socket2Pilot->send(ready_message,strReady.size());
    void* eventRangeMessage;
    std::string strPeerId;
    ssize_t eventRangeSize = socket2Pilot->recv(eventRangeMessage,strPeerId);
    std::string eventRange((const char*)eventRangeMessage,eventRangeSize);
    size_t carRet = eventRange.find('\n');
    if(carRet!=std::string::npos)
      eventRange.resize(carRet);
 
    // Break the loop if no more ranges are expected
    if(eventRange.find(strStopProcessing)!=std::string::npos) {
      ATH_MSG_INFO("Stopped the loop. Last message from the Event Range Channel: " << eventRange);
      break;
    }
    ATH_MSG_INFO("Got Event Range from the pilot: " << eventRange);
 
    // Parse the Event Range string 
    // Expected the following format: [{KEY:VALUE[,KEY:VALUE]}]
    // First get rid of the leading '[{' and the trailing '}]'
    if(eventRange.starts_with( "[{")) eventRange=eventRange.substr(2);
    if(eventRange.ends_with("}]")) eventRange.resize(eventRange.size()-2);
 
    std::map<std::string,std::string> eventRangeMap;
    size_t startpos(0);
    size_t endpos = eventRange.find(',');
    while(endpos!=std::string::npos) {
      // Get the Key-Value pair
      std::string keyValue(eventRange.substr(startpos,endpos-startpos));
      size_t colonPos = keyValue.find(':');
      std::string strKey = keyValue.substr(0,colonPos);
      std::string strVal = keyValue.substr(colonPos+1);
      trimRangeStrings(strKey);
      trimRangeStrings(strVal);
      eventRangeMap[strKey]=std::move(strVal);
      // Next iteration
      startpos = endpos+1;
      endpos = eventRange.find(',',startpos);
    }
    // Get the final Key-Value pair
    std::string keyValue(eventRange.substr(startpos));
    size_t colonPos = keyValue.find(':');
    std::string strKey = keyValue.substr(0,colonPos);
    std::string strVal = keyValue.substr(colonPos+1);
    trimRangeStrings(strKey);
    trimRangeStrings(strVal);
    eventRangeMap[strKey]=std::move(strVal);
             
    if(eventRangeMap.find("eventRangeID")==eventRangeMap.end()
       || eventRangeMap.find("startEvent")==eventRangeMap.end()
       || eventRangeMap.find("lastEvent")==eventRangeMap.end()
       || eventRangeMap.find("GUID")==eventRangeMap.end()) {
      std::string errorStr("ERR_ATHENAMP_PARSE \"" + eventRange + "\": Wrong format");
      ATH_MSG_ERROR(errorStr);
      ATH_MSG_INFO("Ignoring this event range ");
      void* errorMessage = CxxUtils::xmalloc(errorStr.size());
      memcpy(errorMessage,errorStr.data(),errorStr.size());
      socket2Pilot->send(errorMessage,errorStr.size());
      continue;
    }
    else {
      ATH_MSG_DEBUG("*** Decoded Event Range ***");
      std::map<std::string,std::string>::const_iterator it = eventRangeMap.begin(),
    itEnd = eventRangeMap.end();
      for(;it!=itEnd;++it)
    ATH_MSG_DEBUG(it->first << ":" << it->second);
      ATH_MSG_DEBUG("*** ***");
    }
 
    std::string rangeID = eventRangeMap["eventRangeID"];
    std::string guid = eventRangeMap["GUID"];
    int startEvent = std::atoi(eventRangeMap["startEvent"].c_str());
    int lastEvent = std::atoi(eventRangeMap["lastEvent"].c_str());
    if(rangeID.empty() 
       || guid.empty()
       || lastEvent < startEvent) {
      std::string errorStr("ERR_ATHENAMP_PARSE \"" + eventRange + "\": Wrong values of range fields");
      ATH_MSG_ERROR(errorStr);
      ATH_MSG_INFO("Ignoring this event range ");
      void* errorMessage = CxxUtils::xmalloc(errorStr.size());
      memcpy(errorMessage,errorStr.data(),errorStr.size());
      socket2Pilot->send(errorMessage,errorStr.size());
      continue;
    }
 
    std::string message2ProcessorStr;
    char* message2Processor(0);
 
    std::ostringstream ostr;
    ostr << rangeID;
    if(eventRangeMap.find("PFN")!=eventRangeMap.end()) {
      ostr << "," << "PFN:" << eventRangeMap["PFN"];
    }
    ostr << "," << eventRangeMap["startEvent"]
         << "," << eventRangeMap["lastEvent"];
    message2ProcessorStr = ostr.str();
 
    // CACHING MODE: first get an event range from the pilot, transform it into the tokens
    // and only after that wait for a new range request by one of the processors
    if(m_doCaching) {
      ATH_MSG_DEBUG("Waiting for event range request from one of the processors");
      while(processorWaitRequest.empty()) {
    processorWaitRequest = getNewRangeRequest(socket2Processor,socket2Pilot,procReportPending);
    pollFailedPidQueue(sharedFailedPidQueue,socket2Pilot,procReportPending);
    usleep(1000);
      }
      ATH_MSG_INFO("One of the processors is ready for the next range");
      // Get PID from the request and Update m_pid2RangeID
      workerPid = std::atoi(processorWaitRequest.c_str());
      auto it = m_pid2RangeID.find(workerPid);
      if(it!=m_pid2RangeID.end()) {
    m_pid2RangeID.erase(it);
      }
    }
    
    // Send to the Processor: RangeID,evtToken[,evtToken] 
    message2Processor = (char*)CxxUtils::xmalloc(message2ProcessorStr.size());
    memcpy(message2Processor,message2ProcessorStr.data(),message2ProcessorStr.size());
    socket2Processor->send(message2Processor,message2ProcessorStr.size());
    procReportPending++;
 
    // Get PID from the request and Update m_pid2RangeID
    m_pid2RangeID[workerPid] = std::move(rangeID);
    processorWaitRequest.clear();
 
    ATH_MSG_INFO("Sent response to the processor : " << message2ProcessorStr);
  }
 
  if(all_ok) {
    // We are done distributing event tokens. 
    // Tell the workers that the event processing is over
    // i.e. send out m_nprocs empty messages
    void* emptyMess4Processor(0);
    if(!processorWaitRequest.empty()) {
      // We already have one processor waiting for the answer
      emptyMess4Processor = CxxUtils::xmalloc(1);
      socket2Processor->send(emptyMess4Processor,1);
      ATH_MSG_INFO("Set worker PID=" << workerPid << " free");
      processorWaitRequest.clear();
    }
    bool endLoop{false};
    while(true) {
      ATH_MSG_DEBUG("Going to set another processor free");
      while(processorWaitRequest.empty()) {
    processorWaitRequest = getNewRangeRequest(socket2Processor,socket2Pilot,procReportPending);
    if(pollFailedPidQueue(sharedFailedPidQueue,socket2Pilot,procReportPending)==-1) {
      endLoop = true;
      break;
    }
    usleep(1000);
      }
      if(endLoop) break;
      // Remove worker from m_pid2RangeID
      workerPid = std::atoi(processorWaitRequest.c_str());
      auto it = m_pid2RangeID.find(workerPid);
      if(it!=m_pid2RangeID.end()) {
    m_pid2RangeID.erase(it);
      }
      emptyMess4Processor = CxxUtils::xmalloc(1);
      socket2Processor->send(emptyMess4Processor,1);
      ATH_MSG_INFO("Set worker PID=" << workerPid << " free");
      ATH_MSG_INFO("Still " << procReportPending << " pending reports");
      processorWaitRequest.clear();
    }
  }
 
  if(m_appMgr->stop().isFailure()) {
    ATH_MSG_ERROR("Unable to stop AppMgr"); 
    all_ok=false;
  }
  else { 
    if(m_appMgr->finalize().isFailure()) {
      std::cerr << "Unable to finalize AppMgr" << std::endl;
      all_ok=false;
    }
  }
  
  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
  outwork->data = CxxUtils::xmalloc(sizeof(int));
  *(int*)(outwork->data) = (all_ok?0:1); // Error code: for now use 0 success, 1 failure
  outwork->size = sizeof(int);
 
  // ...
  // (possible) TODO: extend outwork with some error message, which will be eventually
  // reported in the master proces
  // ...
  delete socket2Processor;
  delete socket2Pilot;
  delete socketFactory;
 
  return outwork;
}

fin_func()

std::unique_ptr< AthenaInterprocess::ScheduledWork > EvtRangeScatterer::fin_func ( )

overridevirtual

Implements AthenaMPToolBase.

Definition at line 428 of file EvtRangeScatterer.cxx.

{
  // Dummy
  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
  outwork->data = CxxUtils::xmalloc(sizeof(int));
  *(int*)(outwork->data) = 0; // Error code: for now use 0 success, 1 failure
  outwork->size = sizeof(int);
  return outwork;
}

finalize()

StatusCode EvtRangeScatterer::finalize ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 49 of file EvtRangeScatterer.cxx.

{
  return StatusCode::SUCCESS;
}

fmterror()

std::string AthenaMPToolBase::fmterror ( int  errnum )

protectedinherited

Definition at line 333 of file AthenaMPToolBase.cxx.

{
  char buf[256];
  strerror_r(errnum, buf, sizeof(buf));
  return std::string(buf);
}

generateOutputReport()

AthenaMP::AllWorkerOutputs_ptr EvtRangeScatterer::generateOutputReport ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 103 of file EvtRangeScatterer.cxx.

{
  AthenaMP::AllWorkerOutputs_ptr jobOutputs(new AthenaMP::AllWorkerOutputs());
  return jobOutputs;
}

getNewRangeRequest()

std::string EvtRangeScatterer::getNewRangeRequest ( yampl::ISocket *  socket2Processor, yampl::ISocket *  socket2Pilot, int &  procReportPending  )

private

Definition at line 471 of file EvtRangeScatterer.cxx.

{
  void* processor_request(0);
  std::string strPeerId;
  ssize_t processorRequestSize = socket2Processor->tryRecv(processor_request,0,strPeerId);
 
  if(processorRequestSize==-1) return std::string("");
  if(processorRequestSize==sizeof(pid_t)+sizeof(AthenaMPToolBase::ESRange_Status)) {
    ATH_MSG_INFO("Processor reported event range processing error");
    pid_t pid = *((pid_t*)processor_request);
    AthenaMPToolBase::ESRange_Status status = *reinterpret_cast<AthenaMPToolBase::ESRange_Status*>((pid_t*)processor_request+1);
    std::string errorStr("ERR_ATHENAMP_PROCESS "+ m_pid2RangeID[pid] + ": ");
    switch(status) {
    case AthenaMPToolBase::ESRANGE_NOTFOUND:
      errorStr+=std::string("Not found in the input file");
      break;
    case AthenaMPToolBase::ESRANGE_SEEKFAILED : 
      errorStr+=std::string("Seek failed");
      break;
    case AthenaMPToolBase::ESRANGE_PROCFAILED :
      errorStr+=std::string("Failed to process event range");
      break;
    case AthenaMPToolBase::ESRANGE_FILENOTMADE :
      errorStr+=std::string("Failed to make output file");
      break;
    case AthenaMPToolBase::ESRANGE_BADINPFILE :
      errorStr+=std::string("Failed to set input file");
      break;
    default:
      break;
    }
    void* errorMessage = CxxUtils::xmalloc(errorStr.size());
    memcpy(errorMessage,errorStr.data(),errorStr.size());
    socket2Pilot->send(errorMessage,errorStr.size());
    procReportPending--;
    ATH_MSG_INFO("Error reported to the pilot. Reports pending: " << procReportPending);
    return std::string("");
  }
  std::string strProcessorRequest((const char*)processor_request,processorRequestSize);
  ATH_MSG_INFO("Received request from a processor: " << strProcessorRequest);
  // Decode the request. If it contains output file name then pass it over to the pilot and return empty string
  if(strProcessorRequest.starts_with( "/")) {
    void* outpFileNameMessage = CxxUtils::xmalloc(strProcessorRequest.size());
    memcpy(outpFileNameMessage,strProcessorRequest.data(),strProcessorRequest.size());
    socket2Pilot->send(outpFileNameMessage,strProcessorRequest.size());
    procReportPending--;
    ATH_MSG_INFO("Output file reported to the pilot. Reports pending: " << procReportPending);
    return std::string("");
  }
  return strProcessorRequest;
}

handleSavedPfc()

int AthenaMPToolBase::handleSavedPfc ( const std::filesystem::path &  dest_path )

protectedinherited

Definition at line 396 of file AthenaMPToolBase.cxx.

{
  if(std::filesystem::is_regular_file("PoolFileCatalog.xml.AthenaMP-saved"))
    COPY_FILE_HACK("PoolFileCatalog.xml.AthenaMP-saved",dest_path.string()+"/PoolFileCatalog.xml");
  return 0;
}

initialize()

StatusCode EvtRangeScatterer::initialize ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 39 of file EvtRangeScatterer.cxx.

{
  ATH_MSG_DEBUG("In initialize");
 
  StatusCode sc = AthenaMPToolBase::initialize();
  if(!sc.isSuccess()) return sc;
 
  return StatusCode::SUCCESS;
}

killChildren()

void AthenaMPToolBase::killChildren ( )

overridevirtualinherited

Definition at line 201 of file AthenaMPToolBase.cxx.

{
  for(const AthenaInterprocess::Process& child : m_processGroup->getChildren()) {
    kill(child.getProcessID(),SIGKILL);
  }
}

operator()

virtual std::unique_ptr<ScheduledWork> AthenaInterprocess::IMessageDecoder::operator ( ) const &

pure virtualinherited

operator=()

EvtRangeScatterer& EvtRangeScatterer::operator= ( const EvtRangeScatterer &  )

private

pollFailedPidQueue()

pid_t EvtRangeScatterer::pollFailedPidQueue ( AthenaInterprocess::SharedQueue *  sharedFailedPidQueue, yampl::ISocket *  socket2Pilot, int &  procReportPending  )

private

Definition at line 525 of file EvtRangeScatterer.cxx.

{
  pid_t pid{0};
  if(sharedFailedPidQueue->try_receive_basic<pid_t>(pid)
     && pid!=-1) {
    ATH_MSG_INFO("Procesor with PID=" << pid << " has failed!");
    auto itPid = m_pid2RangeID.find(pid);
    if(itPid!=m_pid2RangeID.end()) {
      ATH_MSG_WARNING("The failed RangeID = " << m_pid2RangeID[pid] << " will be reported to Pilot");
 
      std::string errorStr("ERR_ATHENAMP_PROCESS " + m_pid2RangeID[pid] + ": Failed to process event range");
      void* errorMessage = CxxUtils::xmalloc(errorStr.size());
      memcpy(errorMessage,errorStr.data(),errorStr.size());
      socket2Pilot->send(errorMessage,errorStr.size());
      --procReportPending;
      m_pid2RangeID.erase(pid);
    }
    ATH_MSG_INFO("Reports pending: " << procReportPending);
  } 
  return pid;
}

redirectLog()

int AthenaMPToolBase::redirectLog ( const std::string &  rundir, bool  addTimeStamp = true  )

protectedinherited

Definition at line 269 of file AthenaMPToolBase.cxx.

{
  // Redirect both stdout and stderr to the same file AthenaMP.log
  int dup2result1(0), dup2result2(0);
 
  int newout = open(std::string(rundir+"/AthenaMP.log").c_str(),O_CREAT | O_RDWR, S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH);
  if(newout==-1) {
    ATH_MSG_ERROR("Unable to open log file in the run directory. " << fmterror(errno));
    return -1;
  }
  dup2result1 = dup2(newout, STDOUT_FILENO);
  dup2result2 = dup2(newout, STDERR_FILENO);
  TEMP_FAILURE_RETRY(close(newout));
  if(dup2result1==-1) {
    ATH_MSG_ERROR("Unable to redirect standard output. " << fmterror(errno));
    return -1;
  }
  if(dup2result2==-1) {
    ATH_MSG_ERROR("Unable to redirect standard error. " << fmterror(errno));
    return -1;
  }
 
  if(addTimeStamp) {
    SmartIF<IProperty> propertyServer(msgSvc());
    if(propertyServer==0) {
      ATH_MSG_ERROR("Unable to cast message svc to IProperty");
      return -1;
    }
 
    std::string propertyName("Format");
    std::string oldFormat("");
    StringProperty formatProp(propertyName,oldFormat);
    StatusCode sc = propertyServer->getProperty(&formatProp);
    if(sc.isFailure()) {
      ATH_MSG_WARNING("Message Service does not have Format property");
    }
    else {
      oldFormat = formatProp.value();
      if(oldFormat.find("%t")==std::string::npos) {
        // Add time stamps
        std::string newFormat("%t " + oldFormat);
        StringProperty newFormatProp(std::move(propertyName),newFormat);
        ATH_CHECK(propertyServer->setProperty(newFormatProp), -1);
      }
      else {
        ATH_MSG_DEBUG("MsgSvc format already contains timestamps. Nothing to be done");
      }
    }
  }
 
  return 0;
}

reopenFd()

int AthenaMPToolBase::reopenFd ( int  fd, const std::string &  name  )

privateinherited

Definition at line 419 of file AthenaMPToolBase.cxx.

{
  ATH_MSG_DEBUG("Attempting to reopen descriptor for " << name);
  int old_openflags = fcntl(fd,F_GETFL,0);
  switch(old_openflags & O_ACCMODE) {
  case O_RDONLY: {
    ATH_MSG_DEBUG("The File Access Mode is RDONLY");
    break;
  }
  case O_WRONLY: {
    ATH_MSG_DEBUG("The File Access Mode is WRONLY");
    break;
  }
  case O_RDWR: {
    ATH_MSG_DEBUG("The File Access Mode is RDWR");
    break;
  }
  }
   
  int old_descflags = fcntl(fd,F_GETFD,0);
  off_t oldpos = lseek(fd,0,SEEK_CUR);
  if(oldpos==-1) {
    if(errno==ESPIPE) {
      ATH_MSG_WARNING("Dealing with PIPE. Skipping ... (FIXME!)");
    }
    else {
      ATH_MSG_ERROR("When re-opening file descriptors lseek failed on " << name << ". " << fmterror(errno));
      return -1;
    }
  }
  else {
    Io::Fd newfd = open(name.c_str(),old_openflags);
    if(newfd==-1) {
      ATH_MSG_ERROR("When re-opening file descriptors unable to open " << name << " for reading. " << fmterror(errno));
      return -1;
    }
    if(lseek(newfd,oldpos,SEEK_SET)==-1){
      ATH_MSG_ERROR("When re-opening file descriptors lseek failed on the newly opened " << name << ". " << fmterror(errno));
      TEMP_FAILURE_RETRY(close(newfd));
      return -1;
    }
    TEMP_FAILURE_RETRY(close(fd));
    if(dup2(newfd,fd)==-1) {
      ATH_MSG_ERROR("When re-opening file descriptors unable to duplicate descriptor for " << name << ". " << fmterror(errno));
      TEMP_FAILURE_RETRY(close(newfd));
      return -1;
    }
    if(fcntl(fd,F_SETFD,old_descflags)==-1) {
      ATH_MSG_ERROR("When re-opening file descriptors unable to set descriptor flags for " << name << ". " << fmterror(errno));
      TEMP_FAILURE_RETRY(close(newfd));
      return -1;
    }
    TEMP_FAILURE_RETRY(close(newfd));
  }
  return 0;
}

reopenFds()

int AthenaMPToolBase::reopenFds ( )

protectedinherited

Definition at line 340 of file AthenaMPToolBase.cxx.

{
  // Reopen file descriptors.
  // First go over all open files, which have been registered with the FileMgr
  // Then also check the FdsRegistry, in case it contains some files not registered with the FileMgr
  std::set<int> fdLog;
  
  // Query the FileMgr contents
  std::vector<const Io::FileAttr*> filemgrFiles;
  std::vector<const Io::FileAttr*>::const_iterator itFile;
  unsigned filenum = m_fileMgr->getFiles(filemgrFiles); // Get attributes for open files only. We don't care about closed ones at this point
  if(filenum!=filemgrFiles.size())
    ATH_MSG_WARNING("getFiles returned " << filenum << " while vector size is " << filemgrFiles.size());
  
  for(itFile=filemgrFiles.begin();itFile!=filemgrFiles.end();++itFile) {
    ATH_MSG_DEBUG("* " << **itFile);
    const std::string& filename = (**itFile).name();
    Io::Fd fd = (**itFile).fd();
 
    if(fd==-1) {
      // It is legal to have fd=-1 for remote inputs
      // On the other hand, these inputs should not remain open after fork. The issue being tracked at ATEAM-434.
      // So, this hopefully is a temporary patch
      ATH_MSG_WARNING("FD=-1 detected on an open file retrieved from FileMgr. Skip FD reopening. File name: " << filename);
      continue;
    }
    
    if(reopenFd(fd,filename))
      return -1;
    
    fdLog.insert(fd);
  }
  
  // Check the FdsRegistry
  for(const AthenaInterprocess::FdsRegistryEntry& regEntry : *m_fdsRegistry) {
    if(fdLog.find(regEntry.fd)!=fdLog.end()) {
      ATH_MSG_DEBUG("The file from FdsRegistry " << regEntry.name << " was registered with FileMgr. Skip reopening");
    }
    else {
      ATH_MSG_WARNING("The file " << regEntry.name << " has not been registered with the FileMgr!");
      
      if(regEntry.fd==-1) {
      // Same protection as the one above
    ATH_MSG_WARNING("FD=-1 detected on an open file retrieved from FD Registry. Skip FD reopening. File name: " << regEntry.name);
    continue;
      }
 
      if(reopenFd(regEntry.fd,regEntry.name))
        return -1;
      
      fdLog.insert(regEntry.fd);
    }
  }
  return 0;
}

reportSubprocessStatuses()

void AthenaMPToolBase::reportSubprocessStatuses ( )

overridevirtualinherited

Reimplemented in EvtRangeProcessor, SharedEvtQueueConsumer, and SharedHiveEvtQueueConsumer.

Definition at line 111 of file AthenaMPToolBase.cxx.

{
  ATH_MSG_INFO("Statuses of sub-processes");
  const std::vector<AthenaInterprocess::ProcessStatus>& statuses = m_processGroup->getStatuses();
  for(size_t i=0; i<statuses.size(); ++i)
    ATH_MSG_INFO("*** Process PID=" << statuses[i].pid << ". Status " << ((statuses[i].exitcode)?"FAILURE":"SUCCESS"));
}

setMaxEvt()

virtual void AthenaMPToolBase::setMaxEvt ( int  maxEvt )

inlineoverridevirtualinherited

Definition at line 44 of file AthenaMPToolBase.h.

{m_maxEvt=maxEvt;}

setMPRunStop()

virtual void AthenaMPToolBase::setMPRunStop ( const AthenaInterprocess::IMPRunStop *  runStop )

inlineoverridevirtualinherited

Definition at line 45 of file AthenaMPToolBase.h.

{m_mpRunStop=runStop;}

setRandString()

void AthenaMPToolBase::setRandString ( const std::string &  randStr )

overridevirtualinherited

Definition at line 196 of file AthenaMPToolBase.cxx.

{
  m_randStr = randStr;
}

subProcessLogs()

void EvtRangeScatterer::subProcessLogs ( std::vector< std::string > &  filenames )

overridevirtual

Definition at line 95 of file EvtRangeScatterer.cxx.

{
  filenames.clear();
  std::filesystem::path reader_rundir(m_subprocTopDir);
  reader_rundir/= std::filesystem::path(m_subprocDirPrefix);
  filenames.push_back(reader_rundir.string()+std::string("/AthenaMP.log"));
}

trimRangeStrings()

void EvtRangeScatterer::trimRangeStrings ( std::string &  str )

private

Definition at line 438 of file EvtRangeScatterer.cxx.

{
  size_t i(0);
  // get rid of leading spaces
  while(i<str.size() && str[i]==' ') i++;
  if(i) str = str.substr(i);
 
  if(str.empty()) return;  // Corner case: string consists only of spaces
 
  // get rid of trailing spaces
  i=str.size()-1;
  while(str[i]==' ') i--;
  if(i) str.resize(i+1);
  
  // the string might be enclosed by either
  // "u\'" and "\'"
  // or
  // "\"" and "\""
  // Get rid of them!
  if(str.starts_with( "u\'")) {
    str = str.substr(2);
    if(str.ends_with( "\'")) {
      str.resize(str.size()-1);
    }
  }
  else if(str.starts_with( "\"")) {
    str = str.substr(1);
    if(str.ends_with( "\"")) {
      str.resize(str.size()-1);
    }
  }
}

updateIoReg()

int AthenaMPToolBase::updateIoReg ( const std::string &  rundir )

protectedinherited

Definition at line 322 of file AthenaMPToolBase.cxx.

{
  ATH_CHECK(m_ioMgr.retrieve(), -1);
 
  // update the IoRegistry for the new workdir - make sure we use absolute path
  std::filesystem::path abs_rundir = std::filesystem::absolute(rundir);
  ATH_CHECK(m_ioMgr->io_update_all(abs_rundir.string()), -1);
 
  return 0;
}

useFdsRegistry()

void AthenaMPToolBase::useFdsRegistry ( std::shared_ptr< AthenaInterprocess::FdsRegistry >  registry )

overridevirtualinherited

Definition at line 191 of file AthenaMPToolBase.cxx.

{
  m_fdsRegistry = std::move(registry);
}

waitForSignal()

void AthenaMPToolBase::waitForSignal ( )

protectedinherited

Definition at line 403 of file AthenaMPToolBase.cxx.

{
  ATH_MSG_INFO("Bootstrap worker PID " << getpid() << " - waiting for SIGUSR1");
  sigset_t mask, oldmask;
  
  signal(SIGUSR1, AthenaMPToolBase_d::pauseForDebug);
  
  sigemptyset (&mask);
  sigaddset (&mask, SIGUSR1);
  
  sigprocmask (SIG_BLOCK, &mask, &oldmask);
  while (!AthenaMPToolBase_d::sig_done)
    sigsuspend (&oldmask);
  sigprocmask (SIG_UNBLOCK, &mask, NULL);
}

Member Data Documentation

m_appMgr

ServiceHandle<IAppMgrUI> AthenaMPToolBase::m_appMgr

protectedinherited

Definition at line 95 of file AthenaMPToolBase.h.

m_doCaching

Gaudi::Property<bool> EvtRangeScatterer::m_doCaching {this, "DoCaching", false}

private

Definition at line 65 of file EvtRangeScatterer.h.

m_eventRangeChannel

Gaudi::Property<std::string> EvtRangeScatterer::m_eventRangeChannel {this, "EventRangeChannel", ""}

private

Definition at line 64 of file EvtRangeScatterer.h.

m_evtProcessor

ServiceHandle<IEventProcessor> AthenaMPToolBase::m_evtProcessor

protectedinherited

Definition at line 94 of file AthenaMPToolBase.h.

m_evtSelector

SmartIF<IEvtSelector> AthenaMPToolBase::m_evtSelector

protectedinherited

Definition at line 98 of file AthenaMPToolBase.h.

m_evtSelName

std::string AthenaMPToolBase::m_evtSelName

protectedinherited

Name of the event selector.

Definition at line 89 of file AthenaMPToolBase.h.

m_fdsRegistry

std::shared_ptr<AthenaInterprocess::FdsRegistry> AthenaMPToolBase::m_fdsRegistry

protectedinherited

Definition at line 100 of file AthenaMPToolBase.h.

m_fileMgr

ServiceHandle<IFileMgr> AthenaMPToolBase::m_fileMgr

protectedinherited

Definition at line 96 of file AthenaMPToolBase.h.

m_fileMgrLog

std::string AthenaMPToolBase::m_fileMgrLog

protectedinherited

Definition at line 99 of file AthenaMPToolBase.h.

m_ioMgr

ServiceHandle<IIoComponentMgr> AthenaMPToolBase::m_ioMgr

protectedinherited

Definition at line 97 of file AthenaMPToolBase.h.

m_isPileup

Gaudi::Property<bool> AthenaMPToolBase::m_isPileup {this, "IsPileup", false, "Flag for configuring PileUpEventLoopMgr"}

protectedinherited

Definition at line 103 of file AthenaMPToolBase.h.

m_maxEvt

int AthenaMPToolBase::m_maxEvt {-1}

protectedinherited

Maximum number of events assigned to the job.

Definition at line 86 of file AthenaMPToolBase.h.

m_mpRunStop

const AthenaInterprocess::IMPRunStop* AthenaMPToolBase::m_mpRunStop {nullptr}

protectedinherited

Definition at line 92 of file AthenaMPToolBase.h.

m_nprocs

int AthenaMPToolBase::m_nprocs {-1}

protectedinherited

Number of workers spawned by the master process.

Definition at line 85 of file AthenaMPToolBase.h.

m_pid2RangeID

Pid2RangeID EvtRangeScatterer::m_pid2RangeID

private

Definition at line 67 of file EvtRangeScatterer.h.

m_processGroup

AthenaInterprocess::ProcessGroup* AthenaMPToolBase::m_processGroup {nullptr}

protectedinherited

Definition at line 91 of file AthenaMPToolBase.h.

m_processorChannel

Gaudi::Property<std::string> EvtRangeScatterer::m_processorChannel {this, "ProcessorChannel", ""}

private

Definition at line 63 of file EvtRangeScatterer.h.

m_randStr

std::string AthenaMPToolBase::m_randStr

protectedinherited

Definition at line 101 of file AthenaMPToolBase.h.

m_subprocDirPrefix

std::string AthenaMPToolBase::m_subprocDirPrefix

protectedinherited

For ex. "worker__".

Definition at line 88 of file AthenaMPToolBase.h.

m_subprocTopDir

std::string AthenaMPToolBase::m_subprocTopDir

protectedinherited

Top run directory for subprocesses.

Definition at line 87 of file AthenaMPToolBase.h.

---

The documentation for this class was generated from the following files:

• EvtRangeScatterer.h
• EvtRangeScatterer.cxx