EvtRangeProcessor Class Reference

#include <EvtRangeProcessor.h>

Inheritance diagram for EvtRangeProcessor:

Collaboration diagram for EvtRangeProcessor:

Public Member Functions
	EvtRangeProcessor (const std::string &type, const std::string &name, const IInterface *parent)
virtual	~EvtRangeProcessor () override
virtual StatusCode	initialize () 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 StatusCode wait_once	ATLAS_NOT_THREAD_SAFE (pid_t &pid) override
virtual void	reportSubprocessStatuses () 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	finalize () 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 Types
enum	ProcessState { PROC_STATE_INIT, PROC_STATE_EXEC, PROC_STATE_FIN, PROC_STATE_STOP }

Private Member Functions
StatusCode startProcess	ATLAS_NOT_THREAD_SAFE ()
StatusCode	setNewInputFile (const std::string &newFile)
void	reportError (yampl::ISocket *socket, AthenaMPToolBase::ESRange_Status status)
int	reopenFd (int fd, const std::string &name)

Private Attributes
Gaudi::Property< int >	m_nEventsBeforeFork {this, "EventsBeforeFork", 0, "Number of events before forking"}
Gaudi::Property< std::string >	m_channel2Scatterer {this, "Channel2Scatterer", {}}
Gaudi::Property< std::string >	m_channel2EvtSel {this, "Channel2EvtSel", {}}
Gaudi::Property< bool >	m_debug {this, "Debug", false}
int	m_rankId {-1}
	Each worker has its own unique RankID from the range (0,...,m_nprocs-1) More...
int	m_activeWorkers {0}
	Keep track of the number of workers. More...
std::string	m_inpFile
	Cached name of the input file. To avoid reopening. More...
ServiceHandle< IChronoStatSvc >	m_chronoStatSvc
ServiceHandle< IIncidentSvc >	m_incidentSvc
SmartIF< IEvtSelectorSeek >	m_evtSeek
std::unique_ptr< AthenaInterprocess::SharedQueue >	m_sharedRankQueue
AthenaInterprocess::SharedQueue *	m_sharedFailedPidQueue {nullptr}
std::map< pid_t, int >	m_nProcessedEvents
std::deque< pid_t >	m_finQueue
std::map< pid_t, ProcessState >	m_procStates

Detailed Description

Definition at line 25 of file EvtRangeProcessor.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
  };

ProcessState

enum EvtRangeProcessor::ProcessState

private

Enumerator
PROC_STATE_INIT
PROC_STATE_EXEC
PROC_STATE_FIN
PROC_STATE_STOP

Definition at line 56 of file EvtRangeProcessor.h.

                    {
    PROC_STATE_INIT
    , PROC_STATE_EXEC
    , PROC_STATE_FIN
    , PROC_STATE_STOP
  };

Constructor & Destructor Documentation

EvtRangeProcessor()

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

Definition at line 34 of file EvtRangeProcessor.cxx.

  : AthenaMPToolBase(type,name,parent)
  , m_chronoStatSvc("ChronoStatSvc", name)
  , m_incidentSvc("IncidentSvc", name)
{
  m_subprocDirPrefix = "worker_";
}

~EvtRangeProcessor()

EvtRangeProcessor::~EvtRangeProcessor ( )

overridevirtual

Definition at line 44 of file EvtRangeProcessor.cxx.

{
}

Member Function Documentation

ATLAS_NOT_THREAD_SAFE() [1/5]

StatusCode startProcess EvtRangeProcessor::ATLAS_NOT_THREAD_SAFE ( )

private

ATLAS_NOT_THREAD_SAFE() [2/5]

virtual StatusCode exec EvtRangeProcessor::ATLAS_NOT_THREAD_SAFE ( )

overridevirtual

ATLAS_NOT_THREAD_SAFE() [3/5]

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

protectedinherited

ATLAS_NOT_THREAD_SAFE() [4/5]

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

overridevirtual

ATLAS_NOT_THREAD_SAFE() [5/5]

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

overridevirtual

Reimplemented from AthenaMPToolBase.

bootstrap_func() [1/2]

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

overridevirtual

Definition at line 368 of file EvtRangeProcessor.cxx.

{
  if(m_debug) waitForSignal();
 
  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
  // ...
 
  // ________________________ Get RankID ________________________
  //
  if(!m_sharedRankQueue->receive_basic<int>(m_rankId)) {
    ATH_MSG_ERROR("Unable to get rank ID!");
    return outwork;
  }
  std::ostringstream workindex;
  workindex<<m_rankId;
 
  // ________________________ Worker dir: mkdir ________________________
  std::filesystem::path worker_rundir(m_subprocTopDir);
  worker_rundir /= std::filesystem::path(m_subprocDirPrefix+workindex.str());
  // TODO: this "worker_" can be made configurable too
 
  if(mkdir(worker_rundir.string().c_str(),S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH)==-1) {
    ATH_MSG_ERROR("Unable to make worker run directory: " << worker_rundir.string() << ". " << fmterror(errno));
    return outwork;
  }
 
  // ________________________ Redirect logs ________________________
  if(!m_debug) {
    if(redirectLog(worker_rundir.string()))
      return outwork;
    
    ATH_MSG_INFO("Logs redirected in the AthenaMP event worker PID=" << getpid());
  }
 
  // ________________________ Update Io Registry ____________________________
  if(updateIoReg(worker_rundir.string()))
    return outwork;
 
  ATH_MSG_INFO("Io registry updated in the AthenaMP event worker PID=" << getpid());
 
  // ________________________ SimParams & DigiParams & PDGTABLE.MeV ____________________________
  std::filesystem::path abs_worker_rundir = std::filesystem::absolute(worker_rundir);
  if(std::filesystem::is_regular_file("SimParams.db"))
    COPY_FILE_HACK("SimParams.db", abs_worker_rundir.string()+"/SimParams.db");
  if(std::filesystem::is_regular_file("DigitParams.db"))
    COPY_FILE_HACK("DigitParams.db", abs_worker_rundir.string()+"/DigitParams.db");
  if(std::filesystem::is_regular_file("PDGTABLE.MeV"))
    COPY_FILE_HACK("PDGTABLE.MeV", abs_worker_rundir.string()+"/PDGTABLE.MeV");
 
  // _______________________ Handle saved PFC (if any) ______________________
  if(handleSavedPfc(abs_worker_rundir))
    return outwork;
 
  // ________________________  reopen descriptors ____________________________
  if(reopenFds())
    return outwork;
 
  ATH_MSG_INFO("File descriptors re-opened in the AthenaMP event worker 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");
  }
 
  // ________________________ Event selector restart ________________________
  IService* evtSelSvc = dynamic_cast<IService*>(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");
  }
 
  // ________________________ Restart background event selectors in pileup jobs ________________________
  if(m_isPileup) {
    const std::list<IService*>& service_list = serviceLocator()->getServices();
    std::list<IService*>::const_iterator itSvc = service_list.begin(),
      itSvcLast = service_list.end();
    for(;itSvc!=itSvcLast;++itSvc) {
      IEvtSelector* evtsel = dynamic_cast<IEvtSelector*>(*itSvc);
      if(evtsel && (evtsel != evtSelector())) {
    if((*itSvc)->start().isSuccess())
      ATH_MSG_DEBUG("Restarted event selector " << (*itSvc)->name());
    else {
      ATH_MSG_ERROR("Failed to restart event selector " << (*itSvc)->name());
      return outwork;
    }
      }
    }
  }
 
  // ________________________ Worker dir: chdir ________________________
  if(chdir(worker_rundir.string().c_str())==-1) {
    ATH_MSG_ERROR("Failed to chdir to " << worker_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 > EvtRangeProcessor::exec_func ( )

overridevirtual

Implements AthenaMPToolBase.

Definition at line 484 of file EvtRangeProcessor.cxx.

{
  ATH_MSG_INFO("Exec function in the AthenaMP worker PID=" << getpid());
 
  int nEvt(1);
  int nEventsProcessed(0);
 
  std::queue<std::string> queueTokens;
 
  // Get the yampl connection channels
  yampl::ISocketFactory* socketFactory = new yampl::SocketFactory();
  std::string socket2ScattererName = m_channel2Scatterer.value() + std::string("_") + m_randStr;
  yampl::ISocket* socket2Scatterer = socketFactory->createClientSocket(yampl::Channel(socket2ScattererName,yampl::LOCAL),yampl::MOVE_DATA);
  ATH_MSG_INFO("Created CLIENT socket to the Scatterer: " << socket2ScattererName);
  std::ostringstream pidstr;
  pidstr << getpid();
 
  // Construct a "welcome" message to be sent to the EvtRangeScatterer
  std::string ping = pidstr.str() + std::string(" ready for event processing");
 
  while(true) {
    void* message2scatterer = CxxUtils::xmalloc(ping.size());
    memcpy(message2scatterer,ping.data(),ping.size());   
    socket2Scatterer->send(message2scatterer,ping.size());
    ATH_MSG_INFO("Sent a welcome message to the Scatterer");
 
    // Get the response - list of tokens - from the scatterer. 
    // The format of the response: | ResponseSize | RangeID, | evtEvtRange[,evtToken] |
    char *responseBuffer(0);
    std::string strPeerId;
    ssize_t responseSize = socket2Scatterer->recv(responseBuffer,strPeerId);
    // If response size is 0 then break the loop
    if(responseSize==1) {
      ATH_MSG_INFO("Empty range received. Terminating the loop");
      break;
    }
 
    std::string responseStr(responseBuffer,responseSize);
    ATH_MSG_INFO("Received response from the Scatterer : " << responseStr);
 
    // Start timing
    System::ProcessTime time_start = System::getProcessTime();
 
    size_t startpos(0);
    size_t endpos = responseStr.find(',');
    while(endpos!=std::string::npos) {
      queueTokens.push(responseStr.substr(startpos,endpos-startpos));
      startpos = endpos+1;
      endpos = responseStr.find(',',startpos);
    }
    queueTokens.push(responseStr.substr(startpos));
    // Actually the first element in the tokens queue is the RangeID. Get it
    std::string rangeID = queueTokens.front();
    queueTokens.pop();
    ATH_MSG_INFO("Received RangeID=" << rangeID);
    // Fire an incident
    m_incidentSvc->fireIncident(FileIncident(name(),"NextEventRange",rangeID));
 
    // Here we need to support two formats of the responseStr
    // Format 1. RangeID,startEvent,endEvent
    // Format 2. RangeID,fileName,startEvent,endEvent
    //
    // The difference between these two is that for Format 2 we first
    // need to update InputCollections property on the Event Selector
    // and only after that proceed with seeking
    //
    // The seeking part is identical for Format 1 and 2
 
    AthenaMPToolBase::ESRange_Status rangeStatus(AthenaMPToolBase::ESRANGE_SUCCESS);
 
    // Determine the format
    std::string filename("");
    if(queueTokens.front().find("PFN:")==0) {
      // We have Format 2
      // Update InputCollections property of the Event Selector with the file name from Event Range
      filename = queueTokens.front().substr(4);
      if(setNewInputFile(filename).isFailure()) {
    ATH_MSG_WARNING("Failed to set input file for the range: " << rangeID);
    rangeStatus = AthenaMPToolBase::ESRANGE_BADINPFILE;
    reportError(socket2Scatterer,rangeStatus);
    m_incidentSvc->fireIncident(FileIncident(name(),"NextEventRange","dummy"));
    continue;
      }
      queueTokens.pop();
    }
 
    // Get the number of events to process
    int startEvent = std::atoi(queueTokens.front().c_str());
    queueTokens.pop();
    int endEvent = std::atoi(queueTokens.front().c_str());
    queueTokens.pop();
    ATH_MSG_INFO("Range fields. File Name: " << (filename.empty()?"N/A":filename)
         << ", First Event:" << startEvent
         << ", Last Event:" << endEvent);
 
    // Process the events
    IEvtSelector::Context* ctx = nullptr;
    if (evtSelector()->createContext (ctx).isFailure()) {
      ATH_MSG_WARNING("Failed to create IEventSelector context.");
      rangeStatus = AthenaMPToolBase::ESRANGE_SEEKFAILED;
    }
    else {
      for(int i(startEvent-1); i<endEvent; ++i) {
        StatusCode sc = m_evtSeek->seek(*ctx, i);
        if(sc.isRecoverable()) {
          ATH_MSG_WARNING("Event " << i << " from range: " << rangeID << " not in the input file");
          rangeStatus = AthenaMPToolBase::ESRANGE_NOTFOUND;
          break;
        }
        else if(sc.isFailure()) {
          ATH_MSG_WARNING("Failed to seek to " << i << " in range: " << rangeID);
          rangeStatus = AthenaMPToolBase::ESRANGE_SEEKFAILED;
          break;
        }
        ATH_MSG_INFO("Seek to " << i << " succeeded");
        m_chronoStatSvc->chronoStart("AthenaMP_nextEvent");
        sc = m_evtProcessor->nextEvent(nEvt++);
 
        m_chronoStatSvc->chronoStop("AthenaMP_nextEvent");
        if(sc.isFailure()){
          ATH_MSG_WARNING("Failed to process the event " << i << " in range:" << rangeID);
          rangeStatus = AthenaMPToolBase::ESRANGE_PROCFAILED;
          break;
        }
        else {
          ATH_MSG_DEBUG("Event processed");
          nEventsProcessed++;
        }
      }
    }
    if (evtSelector()->releaseContext (ctx).isFailure()) {
      ATH_MSG_WARNING("Failed to release IEventSelector context.");
    }
 
    // Fire dummy NextEventRange incident in order to cut the previous output and report it
    m_incidentSvc->fireIncident(FileIncident(name(),"NextEventRange","dummy"));
    if(rangeStatus!=AthenaMPToolBase::ESRANGE_SUCCESS) {
      reportError(socket2Scatterer,rangeStatus);
      continue;
    }
 
    // Event range successfully processed
    std::string strOutpFile;
    // Get the full path of the event range output file
    for(std::filesystem::directory_iterator fdIt(std::filesystem::current_path()); fdIt!=std::filesystem::directory_iterator(); fdIt++) {
      if(fdIt->path().string().rfind(rangeID) == fdIt->path().string().size()-rangeID.size()) {
    if(strOutpFile.empty()) {
      strOutpFile = fdIt->path().string();
    }
    else {
      strOutpFile += (std::string(",")+fdIt->path().string());
    }   
      }
    }
 
    // Stop timing
    System::ProcessTime time_delta = System::getProcessTime() - time_start;
    
    // Prepare the output report
    if(!strOutpFile.empty()) {
      // We need to combine the output file name with
      // 1. RangeID (requested by JEDI)
      // 2. CPU time
      // 3. Wall time
      std::ostringstream outputReportStream;
      outputReportStream << strOutpFile
             << ",ID:" << rangeID
             << ",CPU:" << time_delta.cpuTime<System::Sec>()
             << ",WALL:" << time_delta.elapsedTime<System::Sec>();
      std::string outputFileReport = outputReportStream.str();
      
      // Report the output
      message2scatterer = CxxUtils::xmalloc(outputFileReport.size());
      memcpy(message2scatterer,outputFileReport.data(),outputFileReport.size());
      socket2Scatterer->send(message2scatterer,outputFileReport.size());
      ATH_MSG_INFO("Reported the output " << outputFileReport);
    }
    else {
      // This is an error: range successfully processed but no outputs were made
      ATH_MSG_WARNING("Failed to make an output file for range: " << rangeID);
      reportError(socket2Scatterer,AthenaMPToolBase::ESRANGE_FILENOTMADE);
    }
  } // Main "event loop"
 
  if(m_evtProcessor->executeRun(0).isFailure()) {
    ATH_MSG_WARNING("Could not finalize the Run");
  }
 
  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
 
  // Return value: "ERRCODE|Func_Flag|NEvt"
  int outsize = 2*sizeof(int)+sizeof(AthenaMPToolBase::Func_Flag);
  void* outdata = CxxUtils::xmalloc(outsize);
  *(int*)(outdata) = 0; // Error code: for now use 0 success, 1 failure
  AthenaMPToolBase::Func_Flag func = AthenaMPToolBase::FUNC_EXEC;
  memcpy((char*)outdata+sizeof(int),&func,sizeof(func));
  memcpy((char*)outdata+sizeof(int)+sizeof(func),&nEventsProcessed,sizeof(int));
 
  outwork->data = outdata;
  outwork->size = outsize;
  // ...
  // (possible) TODO: extend outwork with some error message, which will be eventually
  // reported in the master proces
  // ...
 
  delete socket2Scatterer;
  delete socketFactory;
 
  return outwork;
}

fin_func()

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

overridevirtual

Implements AthenaMPToolBase.

Definition at line 695 of file EvtRangeProcessor.cxx.

{
  ATH_MSG_INFO("Fin function in the AthenaMP worker PID=" << getpid());
 
  if(m_appMgr->stop().isFailure()) {
    ATH_MSG_WARNING("Unable to stop AppMgr"); 
  }
  else { 
    if(m_appMgr->finalize().isFailure()) {
      std::cout << "Unable to finalize AppMgr" << std::endl;
    }
  }
 
  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
 
  // Return value: "ERRCODE|Func_Flag|NEvt"  (Here NEvt=-1)
  int outsize = 2*sizeof(int)+sizeof(AthenaMPToolBase::Func_Flag);
  void* outdata = CxxUtils::xmalloc(outsize);
  *(int*)(outdata) = 0; // Error code: for now use 0 success, 1 failure
  AthenaMPToolBase::Func_Flag func = AthenaMPToolBase::FUNC_FIN;
  memcpy((char*)outdata+sizeof(int),&func,sizeof(func));
  int nEvt = -1;
  memcpy((char*)outdata+sizeof(int)+sizeof(func),&nEvt,sizeof(int));
 
  outwork->data = outdata;
  outwork->size = outsize;
 
  return outwork;
}

finalize()

StatusCode AthenaMPToolBase::finalize ( )

overridevirtualinherited

Reimplemented in SharedEvtQueueConsumer, SharedHiveEvtQueueConsumer, EvtRangeScatterer, and SharedWriterTool.

Definition at line 87 of file AthenaMPToolBase.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 EvtRangeProcessor::generateOutputReport ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 362 of file EvtRangeProcessor.cxx.

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

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 EvtRangeProcessor::initialize ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 48 of file EvtRangeProcessor.cxx.

{
  ATH_MSG_DEBUG("In initialize");
 
  ATH_CHECK(AthenaMPToolBase::initialize());
  m_evtSeek = serviceLocator()->service(m_evtSelName);
  ATH_CHECK(m_evtSeek.isValid());
  ATH_CHECK(m_chronoStatSvc.retrieve());
  ATH_CHECK(m_incidentSvc.retrieve());
  
  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

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

reportError()

void EvtRangeProcessor::reportError ( yampl::ISocket *  socket, AthenaMPToolBase::ESRange_Status  status  )

private

Definition at line 784 of file EvtRangeProcessor.cxx.

{
  pid_t pid = getpid();
  size_t messageSize = sizeof(pid_t)+sizeof(AthenaMPToolBase::ESRange_Status);
  void* message2scatterer = CxxUtils::xmalloc(messageSize);
  memcpy(message2scatterer,&pid,sizeof(pid_t));
  memcpy((pid_t*)message2scatterer+1,&status,sizeof(AthenaMPToolBase::ESRange_Status));
  socket->send(message2scatterer,messageSize);
}

reportSubprocessStatuses()

void EvtRangeProcessor::reportSubprocessStatuses ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 332 of file EvtRangeProcessor.cxx.

{
  ATH_MSG_INFO("Statuses of event processors");
  const std::vector<AthenaInterprocess::ProcessStatus>& statuses = m_processGroup->getStatuses();
  for(size_t i=0; i<statuses.size(); ++i) {
    // Get the number of events processed by this worker
    std::map<pid_t,int>::const_iterator it = m_nProcessedEvents.find(statuses[i].pid);
    std::ostringstream ostr;
    if(it==m_nProcessedEvents.end())
      ostr << "N/A";
    else
      ostr << it->second;
    ATH_MSG_INFO("*** Process PID=" << statuses[i].pid 
         << ". Status " << ((statuses[i].exitcode)?"FAILURE":"SUCCESS") 
         << ". Number of events processed: " << ostr.str());
  }
}

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

setNewInputFile()

StatusCode EvtRangeProcessor::setNewInputFile ( const std::string &  newFile )

private

Definition at line 750 of file EvtRangeProcessor.cxx.

{
  if(m_inpFile == newFile) return StatusCode::SUCCESS;
 
  // Get Property Server
  IProperty* propertyServer = dynamic_cast<IProperty*>(evtSelector());
  if(!propertyServer) {
    ATH_MSG_ERROR("Unable to dyn-cast the event selector to IProperty");
    return StatusCode::FAILURE;
  }
 
  std::string propertyName("InputCollections");
  if(m_inpFile.empty()) {
    std::vector<std::string> vect;
    StringArrayProperty inputFileList(propertyName, vect);
    if(propertyServer->getProperty(&inputFileList).isFailure()) {
      ATH_MSG_ERROR("Failed to get InputCollections property value of the Event Selector");
      return StatusCode::FAILURE;
    }
    if(newFile==inputFileList.value()[0]) {
      m_inpFile = newFile;
      return StatusCode::SUCCESS;
    }
  }
  std::vector<std::string> vect{newFile,};
  StringArrayProperty newInputFileList(std::move(propertyName), vect);
  if(propertyServer->setProperty(newInputFileList).isFailure()) {
    ATH_MSG_ERROR("Unable to update " << newInputFileList.name() << " property on the Event Selector");
    return StatusCode::FAILURE;
  }
  m_inpFile=newFile;
  return StatusCode::SUCCESS;
}

setRandString()

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

overridevirtualinherited

Definition at line 196 of file AthenaMPToolBase.cxx.

{
  m_randStr = randStr;
}

subProcessLogs()

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

overridevirtual

Definition at line 350 of file EvtRangeProcessor.cxx.

{
  filenames.clear();
  for(int i=0; i<m_nprocs; ++i) {
    std::ostringstream workerIndex;
    workerIndex << i;
    std::filesystem::path worker_rundir(m_subprocTopDir);
    worker_rundir /= std::filesystem::path(m_subprocDirPrefix+workerIndex.str());
    filenames.push_back(worker_rundir.string()+std::string("/AthenaMP.log"));
  }
}

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_activeWorkers

int EvtRangeProcessor::m_activeWorkers {0}

private

Keep track of the number of workers.

Definition at line 69 of file EvtRangeProcessor.h.

m_appMgr

ServiceHandle<IAppMgrUI> AthenaMPToolBase::m_appMgr

protectedinherited

Definition at line 95 of file AthenaMPToolBase.h.

m_channel2EvtSel

Gaudi::Property<std::string> EvtRangeProcessor::m_channel2EvtSel {this, "Channel2EvtSel", {}}

private

Definition at line 65 of file EvtRangeProcessor.h.

m_channel2Scatterer

Gaudi::Property<std::string> EvtRangeProcessor::m_channel2Scatterer {this, "Channel2Scatterer", {}}

private

Definition at line 64 of file EvtRangeProcessor.h.

m_chronoStatSvc

ServiceHandle<IChronoStatSvc> EvtRangeProcessor::m_chronoStatSvc

private

Definition at line 72 of file EvtRangeProcessor.h.

m_debug

Gaudi::Property<bool> EvtRangeProcessor::m_debug {this, "Debug", false}

private

Definition at line 66 of file EvtRangeProcessor.h.

m_evtProcessor

ServiceHandle<IEventProcessor> AthenaMPToolBase::m_evtProcessor

protectedinherited

Definition at line 94 of file AthenaMPToolBase.h.

m_evtSeek

SmartIF<IEvtSelectorSeek> EvtRangeProcessor::m_evtSeek

private

Definition at line 74 of file EvtRangeProcessor.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_finQueue

std::deque<pid_t> EvtRangeProcessor::m_finQueue

private

Definition at line 80 of file EvtRangeProcessor.h.

m_incidentSvc

ServiceHandle<IIncidentSvc> EvtRangeProcessor::m_incidentSvc

private

Definition at line 73 of file EvtRangeProcessor.h.

m_inpFile

std::string EvtRangeProcessor::m_inpFile

private

Cached name of the input file. To avoid reopening.

Definition at line 70 of file EvtRangeProcessor.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_nEventsBeforeFork

Gaudi::Property<int> EvtRangeProcessor::m_nEventsBeforeFork {this, "EventsBeforeFork", 0, "Number of events before forking"}

private

Definition at line 63 of file EvtRangeProcessor.h.

m_nProcessedEvents

std::map<pid_t,int> EvtRangeProcessor::m_nProcessedEvents

private

Definition at line 79 of file EvtRangeProcessor.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_processGroup

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

protectedinherited

Definition at line 91 of file AthenaMPToolBase.h.

m_procStates

std::map<pid_t,ProcessState> EvtRangeProcessor::m_procStates

private

Definition at line 81 of file EvtRangeProcessor.h.

m_randStr

std::string AthenaMPToolBase::m_randStr

protectedinherited

Definition at line 101 of file AthenaMPToolBase.h.

m_rankId

int EvtRangeProcessor::m_rankId {-1}

private

Each worker has its own unique RankID from the range (0,...,m_nprocs-1)

Definition at line 68 of file EvtRangeProcessor.h.

m_sharedFailedPidQueue

AthenaInterprocess::SharedQueue* EvtRangeProcessor::m_sharedFailedPidQueue {nullptr}

private

Definition at line 77 of file EvtRangeProcessor.h.

m_sharedRankQueue

std::unique_ptr<AthenaInterprocess::SharedQueue> EvtRangeProcessor::m_sharedRankQueue

private

Definition at line 76 of file EvtRangeProcessor.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:

• EvtRangeProcessor.h
• EvtRangeProcessor.cxx