AthMpEvtLoopMgr Class Reference

#include <AthMpEvtLoopMgr.h>

Inheritance diagram for AthMpEvtLoopMgr:

Collaboration diagram for AthMpEvtLoopMgr:

Public Member Functions
	AthMpEvtLoopMgr (const std::string &name, ISvcLocator *svcLocator)
virtual	~AthMpEvtLoopMgr ()=default
	AthMpEvtLoopMgr ()=delete
	AthMpEvtLoopMgr (const AthMpEvtLoopMgr &)=delete
AthMpEvtLoopMgr &	operator= (const AthMpEvtLoopMgr &)=delete
virtual StatusCode	initialize () override
virtual StatusCode	finalize () override
virtual StatusCode	nextEvent (int maxevt) override
virtual StatusCode	executeEvent (EventContext &&ctx) override
virtual StatusCode	executeRun (int maxevt) override
virtual StatusCode	stopRun () override
virtual EventContext	createEventContext () override
virtual bool	stopScheduled () const override

Private Member Functions
StatusCode	wait ()
StatusCode	generateOutputReport ()
std::shared_ptr< AthenaInterprocess::FdsRegistry >	extractFds ()
StatusCode	updateSkipEvents (int skipEvents)

Private Attributes
ServiceHandle< IEventProcessor >	m_evtProcessor {this,"EventLoopManager","AthenaEventLoopMgr"}
SmartIF< IService >	m_evtSelector {nullptr}
SmartIF< IDataShare >	m_dataShare
Gaudi::Property< int >	m_nWorkers
Gaudi::Property< std::string >	m_workerTopDir
Gaudi::Property< std::string >	m_outputReportName
Gaudi::Property< std::string >	m_strategy
Gaudi::Property< bool >	m_isPileup
Gaudi::Property< bool >	m_collectSubprocessLogs
ToolHandleArray< IAthenaMPTool >	m_tools {this,"Tools", {}}
Gaudi::Property< int >	m_nPollingInterval
Gaudi::Property< int >	m_nMemSamplingInterval
Gaudi::Property< int >	m_nEventsBeforeFork
Gaudi::Property< unsigned int >	m_eventPrintoutInterval
StringArrayProperty	m_execAtPreFork
int	m_nChildProcesses {0}
pid_t	m_masterPid {}
bool	m_scheduledStop {false}
std::vector< unsigned long >	m_samplesRss
std::vector< unsigned long >	m_samplesPss
std::vector< unsigned long >	m_samplesSize
std::vector< unsigned long >	m_samplesSwap

Detailed Description

Definition at line 19 of file AthMpEvtLoopMgr.h.

Constructor & Destructor Documentation

AthMpEvtLoopMgr() [1/3]

AthMpEvtLoopMgr::AthMpEvtLoopMgr	(	const std::string &	name,
		ISvcLocator *	svcLocator )

Definition at line 39 of file AthMpEvtLoopMgr.cxx.

  : base_class(name,svcLocator)
  , m_masterPid(getpid())
{
}

~AthMpEvtLoopMgr()

virtual AthMpEvtLoopMgr::~AthMpEvtLoopMgr ( )

virtualdefault

AthMpEvtLoopMgr() [2/3]

AthMpEvtLoopMgr::AthMpEvtLoopMgr ( )

delete

AthMpEvtLoopMgr() [3/3]

AthMpEvtLoopMgr::AthMpEvtLoopMgr ( const AthMpEvtLoopMgr & )

delete

Member Function Documentation

createEventContext()

EventContext AthMpEvtLoopMgr::createEventContext ( )

overridevirtual

Definition at line 117 of file AthMpEvtLoopMgr.cxx.

                                                 {
  // return an invalid context - method should not be called
  return EventContext{};
}

executeEvent()

StatusCode AthMpEvtLoopMgr::executeEvent ( EventContext && ctx )

overridevirtual

Definition at line 122 of file AthMpEvtLoopMgr.cxx.

{
  // Perhaps there we should return StatusCode::FAILURE as this method shoud not be called directly
  return m_evtProcessor->executeEvent(std::move(ctx));
}

executeRun()

StatusCode AthMpEvtLoopMgr::executeRun ( int maxevt )

overridevirtual

Definition at line 128 of file AthMpEvtLoopMgr.cxx.

{
  ATH_MSG_DEBUG("in executeRun()");
 
  // Generate random component of the Shared Memory and Shared Queue names
  srand(time(NULL));
  std::ostringstream randStream;
  randStream << getpid() << '_' << AthenaInterprocess::randString();
  ATH_MSG_INFO("Using random components for IPC object names: " << randStream.str());
 
  ServiceHandle<StoreGateSvc> pDetStore("DetectorStore",name());
  ATH_CHECK(pDetStore.retrieve());
 
  // Create Shared Event queue if necessary and make it available to the tools
  if(m_strategy=="SharedQueue" 
     || m_strategy=="RoundRobin") {
    AthenaInterprocess::SharedQueue* evtQueue = new AthenaInterprocess::SharedQueue("AthenaMPEventQueue_"+randStream.str(),2000,sizeof(long));
    if(pDetStore->record(evtQueue,"AthenaMPEventQueue_"+randStream.str()).isFailure()) {
      ATH_MSG_FATAL("Unable to record the pointer to the Shared Event queue into Detector Store");
      delete evtQueue;
      return StatusCode::FAILURE;
    }
  }
 
  // For the Event Service: create a queue for connecting EvtRangeProcessor in the master with EvtRangeScatterer subprocess
  // The TokenProcessor master will be sending pid-s of failed processes to Token Scatterer
  if(m_strategy=="EventService") {
    AthenaInterprocess::SharedQueue* failedPidQueue = new AthenaInterprocess::SharedQueue("AthenaMPFailedPidQueue_"+randStream.str(),100,sizeof(pid_t));
    if(pDetStore->record(failedPidQueue,"AthenaMPFailedPidQueue_"+randStream.str()).isFailure()) {
      ATH_MSG_FATAL("Unable to record the pointer to the Failed PID queue into Detector Store");
      delete failedPidQueue;
      return StatusCode::FAILURE;
    }
  }
 
  // Prepare work directory for sub-processes
  if(mkdir(m_workerTopDir.value().c_str(),S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH)!=0) {
    switch(errno) {
    case EEXIST:
      {
    // Top directory already exists, maybe a leftover from previous AthenaMP job in the same rundir
    // Rename it with m_workerTopDir+"-bak-rand"
 
    srand((unsigned)time(0));
    std::ostringstream randname;
    randname << rand();
    std::string backupDir = (m_workerTopDir.value().rfind('/')==(m_workerTopDir.value().size()-1)
                 ? m_workerTopDir.value().substr(0,m_workerTopDir.value().size()-1)
                 : m_workerTopDir.value())+std::string("-bak-")+randname.str(); 
 
    ATH_MSG_WARNING("The top directory " << m_workerTopDir << " already exists");
    ATH_MSG_WARNING("The job will attempt to save it with the name " << backupDir <<  " and create new top directory from scratch");
 
    if(rename(m_workerTopDir.value().c_str(),backupDir.c_str())!=0) {
      char buf[256];
      strerror_r(errno, buf, sizeof(buf));
      ATH_MSG_ERROR("Unable to make backup directory! " << buf);
      return StatusCode::FAILURE;
    }
 
    if(mkdir(m_workerTopDir.value().c_str(),S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH)==0) break;
      }
      /* FALLTHROUGH */
    default:
      {
    char buf[256];
    strerror_r(errno, buf, sizeof(buf));
    ATH_MSG_ERROR("Unable to make top directory " << m_workerTopDir << " for children processes! " << buf);
    return StatusCode::FAILURE;
      }
    }
  }
 
  // When forking before 1st event, fire BeforeFork incident in non-pileup jobs
  ServiceHandle<IIncidentSvc> incSvc("IncidentSvc",name());
  ATH_CHECK(incSvc.retrieve());
 
  if(m_nEventsBeforeFork==0 && !m_isPileup) {
    incSvc->fireIncident(Incident(name(),"BeforeFork"));
  }
 
  // Extract process file descriptors
  std::shared_ptr<AthenaInterprocess::FdsRegistry> registry = extractFds();
 
  ToolHandleArray<IAthenaMPTool>::iterator it = m_tools.begin(),
    itLast = m_tools.end();
 
  // When using SharedWriter in conjunction with fork-after-N-events
  // we have to make sure that mother process is a conversion service
  // client so that events before forking workers are captured...
 
  auto sharedWriterTool = m_tools["SharedWriterTool"];
  const bool sharedWriterWithFAFE = (m_nEventsBeforeFork!=0 && sharedWriterTool);
 
  if(sharedWriterWithFAFE) {
    m_dataShare = SmartIF<IDataShare>(serviceLocator()->service("AthenaPoolSharedIOCnvSvc"));
    ATH_CHECK(m_dataShare.isValid());
 
    (*sharedWriterTool)->useFdsRegistry(registry);
    (*sharedWriterTool)->setRandString(randStream.str());
 
    int nChildren = (*sharedWriterTool)->makePool(maxevt,m_nWorkers,m_workerTopDir);
    if(nChildren==-1) {
      ATH_MSG_FATAL("makePool failed for " << (*sharedWriterTool)->name());
      return StatusCode::FAILURE;
    }
    else {
      m_nChildProcesses+=nChildren;
    }
 
    // Execute the SharedWriterTool at this point
    StatusCode mySc = (*sharedWriterTool)->exec();
    if(!mySc.isSuccess()) {
      ATH_MSG_FATAL("Cannot Execute SharedWriter Tool");
      return StatusCode::FAILURE;
    }
 
    // Make the mother process a client
    if(!m_dataShare->makeClient(m_nWorkers+1).isSuccess()) {
      ATH_MSG_FATAL("Cannot make mother process a client for Conversion Service");
      return StatusCode::FAILURE;
    }
  }
 
  //
  // Try processing requested number of events here
  if(m_nEventsBeforeFork) {
    // Take into account a corner case: m_nEventsBeforeFork > maxevt
    int nEventsToProcess = (maxevt>-1 && m_nEventsBeforeFork>maxevt)
      ? maxevt
      : m_nEventsBeforeFork.value();
    StatusCode scEvtProc = m_evtProcessor->nextEvent(nEventsToProcess);
    if(!scEvtProc.isSuccess()) {
      if(nEventsToProcess) {
    ATH_MSG_FATAL("Unable to process first " << nEventsToProcess << " events in the master");
      }
      else {
    ATH_MSG_FATAL("Unable to process first event in the master");
      }
      return scEvtProc;
    }
  }
 
  // Finalize I/O (close input files) by IoComponents
  ServiceHandle<IIoComponentMgr> ioMgr("IoComponentMgr",name());
  ATH_CHECK(ioMgr.retrieve());
  ATH_CHECK(ioMgr->io_finalize());
  ATH_MSG_DEBUG("Successfully finalized I/O before forking");
 
  // Flush stream buffers
  fflush(NULL);
 
  // Make the mother process not client
  if(sharedWriterWithFAFE && !m_dataShare->makeClient(0).isSuccess()) {
    ATH_MSG_FATAL("Cannot make mother process not client for Conversion Service");
    return StatusCode::FAILURE;
  }
 
  int maxEvents(maxevt); // This can be modified after restart
 
  // Re-extract process file descriptors
  registry = extractFds();
 
  // Make worker pools
  it = m_tools.begin();
  for(; it!=itLast; ++it) {
    if(sharedWriterWithFAFE && (*it)->name() == "AthMpEvtLoopMgr.SharedWriterTool") continue;
    (*it)->useFdsRegistry(registry);
    (*it)->setRandString(randStream.str());
    (*it)->setMaxEvt(maxevt);
    (*it)->setMPRunStop(this);
    if(it==m_tools.begin()) {
      incSvc->fireIncident(Incident(name(),"PreFork")); // Do it only once
    }
    int nChildren = (*it)->makePool(maxEvents,m_nWorkers,m_workerTopDir);
    if(nChildren==-1) {
      ATH_MSG_FATAL("makePool failed for " << (*it)->name());
      return StatusCode::FAILURE;
    }
    else {
      m_nChildProcesses+=nChildren;
    }
  }
 
  if(m_nChildProcesses==0) {
    ATH_MSG_ERROR("No child processes were created");
    return StatusCode::FAILURE;
  }
 
  // Assign work to child processes
  for(it=m_tools.begin(); it!=itLast; ++it) {
    if(sharedWriterWithFAFE && (*it)->name() == "AthMpEvtLoopMgr.SharedWriterTool") continue;
    if((*it)->exec().isFailure()) {
      ATH_MSG_FATAL("Unable to submit work to the tool " << (*it)->name());
      return StatusCode::FAILURE;
    }
  }
 
  StatusCode sc = wait();
 
  if(m_nMemSamplingInterval>0) {
    ATH_MSG_INFO("*** *** Memory Usage *** ***");
    ATH_MSG_INFO("*** MAX PSS  "  << (*std::max_element(m_samplesPss.cbegin(),m_samplesPss.cend()))/1024 << "MB");
    ATH_MSG_INFO("*** MAX RSS  "  << (*std::max_element(m_samplesRss.cbegin(),m_samplesRss.cend()))/1024 << "MB");
    ATH_MSG_INFO("*** MAX SIZE " << (*std::max_element(m_samplesSize.cbegin(),m_samplesSize.cend()))/1024 << "MB");
    ATH_MSG_INFO("*** MAX SWAP " << (*std::max_element(m_samplesSwap.cbegin(),m_samplesSwap.cend()))/1024 << "MB");
    ATH_MSG_INFO("*** *** Memory Usage *** ***");
  }
 
  if(m_collectSubprocessLogs) {
    ATH_MSG_INFO("BEGIN collecting sub-process logs");
    std::vector<std::string> logs;
    for(it=m_tools.begin(); it!=itLast; ++it) {
      (*it)->subProcessLogs(logs);
      for(size_t i=0;i<logs.size();++i) {
    std::cout << "\n File: " << logs[i] << "\n" << std::endl;
    std::ifstream log;
    log.open(logs[i].c_str(),std::ifstream::in);
    std::string line;
    while(!log.eof()) {
      std::getline(log,line);
      std::cout << line << std::endl;
    }
    log.close();
      }
    }
    ATH_MSG_INFO("END collecting sub-process logs");
  }
 
  if(sc.isSuccess())
    return generateOutputReport();
  else 
    return sc;
}

extractFds()

std::shared_ptr< AthenaInterprocess::FdsRegistry > AthMpEvtLoopMgr::extractFds ( )

private

Definition at line 507 of file AthMpEvtLoopMgr.cxx.

{
  ATH_MSG_DEBUG("Extracting file descriptors");
  using namespace std::filesystem;
  std::shared_ptr<AthenaInterprocess::FdsRegistry> registry(new AthenaInterprocess::FdsRegistry());
 
  // Extract file descriptors associated with the current process
  // 1. Store only those regular files in the registry, which
  // don't contain substrings from the "exclusion pattern" set
  // 2. Skip also stdout and stderr
 
  std::vector<std::string> excludePatterns {
    "/root/etc/plugins/"
      ,"/root/cint/cint/"
      ,"/root/include/"
      ,"/var/tmp/"
      ,"/var/lock/"
      ,"/var/lib/"
      ,"/bin/python/"
      ,"/include/c++/"
      ,".confdb2"
      };
 
  path fdPath("/proc/self/fd");
  for(directory_iterator fdIt(fdPath); fdIt!=directory_iterator(); fdIt++) {
    if(is_symlink(fdIt->path())) {
      path realpath = read_symlink(fdIt->path());
      int fd = atoi(fdIt->path().filename().string().c_str());
 
      if (fd==1 || fd==2) // Skip stdout and stderr
        continue;
 
      if(exists(realpath)) {
        if(is_regular_file(realpath)) {
      // Check against the exclusion criteria
          bool exclude(false);
          for(size_t i=0;i<excludePatterns.size(); ++i) {
            if(realpath.string().find(excludePatterns[i])!=std::string::npos) {
              exclude = true;
              break;
            }
      }
          if(exclude) {
        ATH_MSG_DEBUG(realpath.string() << " Excluded from the registry by the pattern");
      }
          else {
            registry->push_back(AthenaInterprocess::FdsRegistryEntry(fd,realpath.string()));
      }
        }
        else {
      ATH_MSG_DEBUG(realpath.string() << " is not a regular file"); // TODO: deal with these?
    }
      } // File exists
    }
    else
      ATH_MSG_WARNING("UNEXPECTED. " << fdIt->path().string() << " Not a symlink");
  } // Directory iteration
 
  ATH_MSG_DEBUG("Fds Reistry created. Contents:");
  for(size_t ii(0); ii<registry->size(); ++ii)
    ATH_MSG_DEBUG((*registry)[ii].fd << " " << (*registry)[ii].name);
 
  return registry;
}

finalize()

StatusCode AthMpEvtLoopMgr::finalize ( )

overridevirtual

Definition at line 106 of file AthMpEvtLoopMgr.cxx.

{
  return StatusCode::SUCCESS;
}

generateOutputReport()

StatusCode AthMpEvtLoopMgr::generateOutputReport ( )

private

Definition at line 436 of file AthMpEvtLoopMgr.cxx.

{
  // Loop over tools, collect their output reports and put them all together into a single file.
  // If m_nEventsBeforeFork!=0 then take into account the outputs made by the master process too
 
  std::ofstream ofs;
  ofs.open(m_outputReportName.value().c_str());
  if(!ofs) {
    ATH_MSG_ERROR("Unable to open AthenaMPOutputs for writing!");
    return StatusCode::FAILURE;
  }
  else {
    std::vector<AthenaMP::AllWorkerOutputs_ptr> allptrs;
 
    ToolHandleArray<IAthenaMPTool>::iterator it = m_tools.begin(),
      itLast = m_tools.end();
    for(it=m_tools.begin(); it!=itLast; ++it)
      allptrs.push_back((*it)->generateOutputReport());
 
    // First collect keys=file_names from all tools
    std::set<std::string> allkeys;
    for(size_t i=0; i<allptrs.size(); ++i) {
      AthenaMP::AllWorkerOutputsIterator it_wos = allptrs[i]->begin(),
    it_wosLast = allptrs[i]->end();
      for(;it_wos!=it_wosLast;++it_wos)
    allkeys.insert(it_wos->first);
    }
 
    // Generate XML
    ofs << "<?xml version=\"1.0\" encoding=\"utf-8\"?>" << std::endl;
    ofs << "<athenaFileReport>" << std::endl;
    std::set<std::string>::const_iterator keys_it = allkeys.begin(),
      keys_itLast = allkeys.end();
    for(;keys_it!=keys_itLast;++keys_it) {
      ofs << "  <Files OriginalName=\"" << (*keys_it) << "\">" << std::endl;
      for(size_t i=0; i<allptrs.size(); ++i) {
    AthenaMP::AllWorkerOutputsIterator it_wos = (allptrs[i])->find(*keys_it);
    if(it_wos!=(allptrs[i])->end()) {
      for(size_t ii=0; ii<it_wos->second.size(); ++ii) {
        AthenaMP::WorkerOutput& outp = it_wos->second[ii];
        if(ii==0 && m_nEventsBeforeFork>0) {
          std::filesystem::path masterFile(std::filesystem::current_path());
          masterFile /= std::filesystem::path(*keys_it);
          if(std::filesystem::exists(masterFile) && std::filesystem::is_regular_file(masterFile))
        ofs << "    <File "
            << "description=\"" << outp.description
            << "\" mode=\"" << outp.access_mode
            << "\" name=\"" << masterFile.string()
            << "\" shared=\"" << (outp.shared?"True":"False")
            << "\" technology=\"" << outp.technology
            << "\"/>" << std::endl;
        }
        ofs << "    <File "
        << "description=\"" << outp.description
        << "\" mode=\"" << outp.access_mode
        << "\" name=\"" << outp.filename
        << "\" shared=\"" << (outp.shared?"True":"False")
        << "\" technology=\"" << outp.technology
        << "\"/>" << std::endl;
      }
    }
      }
      ofs << "  </Files>" << std::endl;
    }
    ofs << "</athenaFileReport>" << std::endl;
    ofs.close();
  }
 
  return StatusCode::SUCCESS;
}

initialize()

StatusCode AthMpEvtLoopMgr::initialize ( )

overridevirtual

Definition at line 46 of file AthMpEvtLoopMgr.cxx.

{
  ATH_MSG_DEBUG("in initialize() ... ");
 
  Gaudi::Concurrency::ConcurrencyFlags::setNumProcs(m_nWorkers);
 
  SmartIF<IProperty> prpMgr(serviceLocator());
  if(!prpMgr.isValid()) {
    ATH_MSG_ERROR("Failed to get hold of the Property Manager");
    return StatusCode::FAILURE;
  }
 
  {
    std::string evtSelName = prpMgr->getProperty("EvtSel").toString();
    m_evtSelector = serviceLocator()->service(std::move(evtSelName));
  }
  ATH_CHECK(m_evtSelector.isValid());
 
  if(m_strategy=="EventService") {
    // ES with non-zero events before forking makes no sense
    if(m_nEventsBeforeFork!=0) {
      ATH_MSG_ERROR("The EventService strategy cannot run with non-zero value for EventsBeforeFork");
      return StatusCode::FAILURE;
    }
 
    // We need to ignore SkipEvents in ES
    if(updateSkipEvents(0).isFailure()) {
      ATH_MSG_ERROR("Failed to set skipEvents=0 in Event Service");
      return StatusCode::FAILURE;
    }
  }
 
  if(m_isPileup) {
    m_evtProcessor = ServiceHandle<IEventProcessor>("PileUpEventLoopMgr",name());
    ATH_MSG_INFO("ELM: The job running in pileup mode");
  }
  else {
    ATH_MSG_INFO("ELM: The job running in non-pileup mode");
  }
 
  ATH_CHECK(m_evtProcessor.retrieve());
  if(!m_isPileup) {
    SmartIF<IProperty> propertyServer(m_evtProcessor.get());
    if(propertyServer) {
      if(propertyServer->setProperty("EventPrintoutInterval",m_eventPrintoutInterval).isFailure()) {
        ATH_MSG_WARNING("Could not set AthenaEventLoopMgr EventPrintoutInterval to " << m_eventPrintoutInterval);
      }
      if(propertyServer->setProperty("ExecAtPreFork",m_execAtPreFork).isFailure()) {
        ATH_MSG_WARNING("Could not set AthenaEventLoopMgr ExecAtPreFork property, memory usage might get affected!");
      }
    }
    else {
      ATH_MSG_WARNING("Could not cast AthenaEventLoopMgr to IProperty");
    }
  }
  ATH_CHECK(m_tools.retrieve());
 
  return StatusCode::SUCCESS;
}

nextEvent()

StatusCode AthMpEvtLoopMgr::nextEvent ( int maxevt )

overridevirtual

Definition at line 111 of file AthMpEvtLoopMgr.cxx.

{
  // Perhaps there we should return StatusCode::FAILURE as this method shoud not be called directly
  return m_evtProcessor->nextEvent(maxevt);
}

operator=()

AthMpEvtLoopMgr & AthMpEvtLoopMgr::operator= ( const AthMpEvtLoopMgr & )

delete

stopRun()

StatusCode AthMpEvtLoopMgr::stopRun ( )

overridevirtual

Definition at line 363 of file AthMpEvtLoopMgr.cxx.

{
  m_scheduledStop = true;
  return m_evtProcessor->stopRun();
}

stopScheduled()

virtual bool AthMpEvtLoopMgr::stopScheduled ( ) const

inlineoverridevirtual

Definition at line 42 of file AthMpEvtLoopMgr.h.

{return m_scheduledStop;};

updateSkipEvents()

StatusCode AthMpEvtLoopMgr::updateSkipEvents ( int skipEvents )

private

Definition at line 572 of file AthMpEvtLoopMgr.cxx.

{
  SmartIF<IProperty> propertyServer(m_evtSelector);
  if(!propertyServer) {
    ATH_MSG_ERROR("Unable to dyn-cast the event selector to IProperty");
    return StatusCode::FAILURE;
  }
 
  IntegerProperty skipEventsProperty("SkipEvents", skipEvents);
  if(propertyServer->setProperty(skipEventsProperty).isFailure()) {
    ATH_MSG_ERROR("Unable to update " << skipEventsProperty.name() << " property on the Event Selector");
    return StatusCode::FAILURE;
  }
  ATH_MSG_INFO("Updated the SkipEvents property of the event selector. New value: " << skipEvents);
  
  return StatusCode::SUCCESS;
}

wait()

StatusCode AthMpEvtLoopMgr::wait ( )

private

Definition at line 379 of file AthMpEvtLoopMgr.cxx.

{
  ATH_MSG_INFO("Waiting for sub-processes");
  ToolHandleArray<IAthenaMPTool>::iterator it = m_tools.begin(),
    itLast = m_tools.end();
  pid_t pid(0);
  bool all_ok(true);
 
  auto memMonTime = std::chrono::system_clock::now();
 
  while(m_nChildProcesses>0) {
    for(it = m_tools.begin(); it!=itLast; ++it) {
      if((*it)->wait_once(pid).isFailure()) {
    all_ok = false;
    ATH_MSG_ERROR("Failure in waiting or sub-process finished abnormally");
    break;
      }
      else {
    if(pid>0) m_nChildProcesses -= 1;
      }
    }
    if(!all_ok) break;
 
    usleep(m_nPollingInterval*1000);
 
    if(m_nMemSamplingInterval>0) {
      auto currTime = std::chrono::system_clock::now();
      if(std::chrono::duration<double,std::ratio<1,1>>(currTime-memMonTime).count()>m_nMemSamplingInterval) {
    unsigned long size(0);
    unsigned long rss(0);
    unsigned long pss(0);
    unsigned long swap(0);
 
    if(athenaMP_MemHelper::getPss(getpid(), pss, swap, rss, size, msgLvl(MSG::DEBUG)))
      ATH_MSG_WARNING("Unable to get memory sample");
    else {
      m_samplesRss.push_back(rss);
      m_samplesPss.push_back(pss);
      m_samplesSize.push_back(size);
      m_samplesSwap.push_back(swap);
    }
    memMonTime=currTime;
      }
    }
  }
 
  for(it=m_tools.begin(); it!=itLast; ++it) 
    (*it)->reportSubprocessStatuses();
 
  if(!all_ok) {
    for(it=m_tools.begin(); it!=itLast; ++it)
      (*it)->killChildren();
  }
 
  return (all_ok?StatusCode::SUCCESS:StatusCode::FAILURE);
}

Member Data Documentation

m_collectSubprocessLogs

Gaudi::Property<bool> AthMpEvtLoopMgr::m_collectSubprocessLogs

private

Initial value:

{this, "CollectSubprocessLogs", false,
      "Copy all workers' logs into the main log file at the end of the job?"}

Definition at line 64 of file AthMpEvtLoopMgr.h.

                                             {this, "CollectSubprocessLogs", false,
      "Copy all workers' logs into the main log file at the end of the job?"};

m_dataShare

SmartIF<IDataShare> AthMpEvtLoopMgr::m_dataShare

private

Definition at line 47 of file AthMpEvtLoopMgr.h.

m_eventPrintoutInterval

Gaudi::Property<unsigned int> AthMpEvtLoopMgr::m_eventPrintoutInterval

private

Initial value:

{this, "EventPrintoutInterval", 1,
      "The value to be forwarded to the EventPrintoutInterval property of the AthenaEventLoopMgr"}

Definition at line 78 of file AthMpEvtLoopMgr.h.

                                                     {this, "EventPrintoutInterval", 1,
      "The value to be forwarded to the EventPrintoutInterval property of the AthenaEventLoopMgr"};

m_evtProcessor

ServiceHandle<IEventProcessor> AthMpEvtLoopMgr::m_evtProcessor {this,"EventLoopManager","AthenaEventLoopMgr"}

private

Definition at line 45 of file AthMpEvtLoopMgr.h.

{this,"EventLoopManager","AthenaEventLoopMgr"};

m_evtSelector

SmartIF<IService> AthMpEvtLoopMgr::m_evtSelector {nullptr}

private

Definition at line 46 of file AthMpEvtLoopMgr.h.

{nullptr};

m_execAtPreFork

StringArrayProperty AthMpEvtLoopMgr::m_execAtPreFork

private

Initial value:

{this, "ExecAtPreFork", {},
      "The value to be forwarded to the ExecAtPreFork property of the AthenaEventLoopMgr"}

Definition at line 81 of file AthMpEvtLoopMgr.h.

                                     {this, "ExecAtPreFork", {},
      "The value to be forwarded to the ExecAtPreFork property of the AthenaEventLoopMgr"};

m_isPileup

Gaudi::Property<bool> AthMpEvtLoopMgr::m_isPileup

private

Initial value:

{this, "IsPileup", false,
      "Is AthenaMP running a PileUp Digitization job?"}

Definition at line 61 of file AthMpEvtLoopMgr.h.

                                {this, "IsPileup", false,
      "Is AthenaMP running a PileUp Digitization job?"};

m_masterPid

pid_t AthMpEvtLoopMgr::m_masterPid {}

private

Definition at line 85 of file AthMpEvtLoopMgr.h.

{};           // PID of the main process

m_nChildProcesses

int AthMpEvtLoopMgr::m_nChildProcesses {0}

private

Definition at line 84 of file AthMpEvtLoopMgr.h.

{0};    // Total number of child processes

m_nEventsBeforeFork

Gaudi::Property<int> AthMpEvtLoopMgr::m_nEventsBeforeFork

private

Initial value:

{this, "EventsBeforeFork", 0,
      "Number of events to be processed by the main process before forking the workers. 0 - fork after BeginRun incident"}

Definition at line 75 of file AthMpEvtLoopMgr.h.

                                        {this, "EventsBeforeFork", 0,
      "Number of events to be processed by the main process before forking the workers. 0 - fork after BeginRun incident"};

m_nMemSamplingInterval

Gaudi::Property<int> AthMpEvtLoopMgr::m_nMemSamplingInterval

private

Initial value:

{this, "MemSamplingInterval", 0,
      "Interval in seconds between taking memory usage samples. 0 - no sampling"}

Definition at line 72 of file AthMpEvtLoopMgr.h.

                                           {this, "MemSamplingInterval", 0,
      "Interval in seconds between taking memory usage samples. 0 - no sampling"};

m_nPollingInterval

Gaudi::Property<int> AthMpEvtLoopMgr::m_nPollingInterval

private

Initial value:

{this, "PollingInterval", 100,
      "Interval in milliseconds between checks of sub-processes statuses"}

Definition at line 69 of file AthMpEvtLoopMgr.h.

                                       {this, "PollingInterval", 100,
      "Interval in milliseconds between checks of sub-processes statuses"};

m_nWorkers

Gaudi::Property<int> AthMpEvtLoopMgr::m_nWorkers

private

Initial value:

{this, "NWorkers", 0,
      "Number of AthenaMP worker processes"}

Definition at line 49 of file AthMpEvtLoopMgr.h.

                               {this, "NWorkers", 0,
      "Number of AthenaMP worker processes"};

m_outputReportName

Gaudi::Property<std::string> AthMpEvtLoopMgr::m_outputReportName

private

Initial value:

{this, "OutputReportFile", "AthenaMPOutputs",
      "ASCII file in the main run directory that lists outputs of all workers. Used by Job Transform"}

Definition at line 55 of file AthMpEvtLoopMgr.h.

                                             {this, "OutputReportFile", "AthenaMPOutputs",
      "ASCII file in the main run directory that lists outputs of all workers. Used by Job Transform"};

m_samplesPss

std::vector<unsigned long> AthMpEvtLoopMgr::m_samplesPss

private

Definition at line 90 of file AthMpEvtLoopMgr.h.

m_samplesRss

std::vector<unsigned long> AthMpEvtLoopMgr::m_samplesRss

private

Definition at line 89 of file AthMpEvtLoopMgr.h.

m_samplesSize

std::vector<unsigned long> AthMpEvtLoopMgr::m_samplesSize

private

Definition at line 91 of file AthMpEvtLoopMgr.h.

m_samplesSwap

std::vector<unsigned long> AthMpEvtLoopMgr::m_samplesSwap

private

Definition at line 92 of file AthMpEvtLoopMgr.h.

m_scheduledStop

bool AthMpEvtLoopMgr::m_scheduledStop {false}

private

Definition at line 86 of file AthMpEvtLoopMgr.h.

{false};  // Flag for early termination of the event loop (for the generators use-case)

m_strategy

Gaudi::Property<std::string> AthMpEvtLoopMgr::m_strategy

private

Initial value:

{this, "Strategy", "",
      "Event processing strategy used by AthenaMP workers. E.g, Shared Queue, Round Robin"}

Definition at line 58 of file AthMpEvtLoopMgr.h.

                                     {this, "Strategy", "",
      "Event processing strategy used by AthenaMP workers. E.g, Shared Queue, Round Robin"};

m_tools

ToolHandleArray<IAthenaMPTool> AthMpEvtLoopMgr::m_tools {this,"Tools", {}}

private

Definition at line 67 of file AthMpEvtLoopMgr.h.

{this,"Tools", {}};

m_workerTopDir

Gaudi::Property<std::string> AthMpEvtLoopMgr::m_workerTopDir

private

Initial value:

{this, "WorkerTopDir", "athenaMP_workers",
      "Sub-directory of the main run directory that contains run directories of all workers"}

Definition at line 52 of file AthMpEvtLoopMgr.h.

                                         {this, "WorkerTopDir", "athenaMP_workers",
      "Sub-directory of the main run directory that contains run directories of all workers"};

---

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

• AthMpEvtLoopMgr.h
• AthMpEvtLoopMgr.cxx