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 ()
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
	AthMpEvtLoopMgr ()
	AthMpEvtLoopMgr (const AthMpEvtLoopMgr &)
AthMpEvtLoopMgr &	operator= (const AthMpEvtLoopMgr &)
StatusCode	wait ()
StatusCode	generateOutputReport ()
std::shared_ptr< AthenaInterprocess::FdsRegistry >	extractFds ()
StatusCode	updateSkipEvents (int skipEvents)

Private Attributes
ServiceHandle< IEventProcessor >	m_evtProcessor
SmartIF< IService >	m_evtSelector
int	m_nWorkers
std::string	m_workerTopDir
std::string	m_outputReportName
std::string	m_strategy
bool	m_isPileup
bool	m_collectSubprocessLogs
ToolHandleArray< IAthenaMPTool >	m_tools
int	m_nChildProcesses
int	m_nPollingInterval
int	m_nMemSamplingInterval
int	m_nEventsBeforeFork
unsigned int	m_eventPrintoutInterval
StringArrayProperty	m_execAtPreFork
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 18 of file AthMpEvtLoopMgr.h.

Constructor & Destructor Documentation

AthMpEvtLoopMgr() [1/3]

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

Definition at line 41 of file AthMpEvtLoopMgr.cxx.

  : base_class(name,svcLocator)
  , m_evtProcessor("AthenaEventLoopMgr", name)
  , m_evtSelector(nullptr)
  , m_nWorkers(0)
  , m_workerTopDir("athenaMP_workers")
  , m_outputReportName("AthenaMPOutputs")
  , m_strategy("")
  , m_isPileup(false)
  , m_collectSubprocessLogs(false)
  , m_tools(this)
  , m_nChildProcesses(0)
  , m_nPollingInterval(100) // 0.1 second
  , m_nMemSamplingInterval(0) // no sampling by default
  , m_nEventsBeforeFork(0)
  , m_eventPrintoutInterval(1)
  , m_execAtPreFork()
  , m_masterPid(getpid())
{
  declareProperty("NWorkers",m_nWorkers);
  declareProperty("WorkerTopDir",m_workerTopDir);
  declareProperty("OutputReportFile",m_outputReportName);
  declareProperty("Strategy",m_strategy);
  declareProperty("IsPileup",m_isPileup);
  declareProperty("CollectSubprocessLogs",m_collectSubprocessLogs);
  declareProperty("Tools",m_tools);
  declareProperty("PollingInterval",m_nPollingInterval);
  declareProperty("MemSamplingInterval",m_nMemSamplingInterval);
  declareProperty("EventsBeforeFork",m_nEventsBeforeFork);
  declareProperty("EventPrintoutInterval",m_eventPrintoutInterval);
  declareProperty("ExecAtPreFork", m_execAtPreFork);
}

~AthMpEvtLoopMgr()

AthMpEvtLoopMgr::~AthMpEvtLoopMgr ( )

virtual

Definition at line 75 of file AthMpEvtLoopMgr.cxx.

{
}

AthMpEvtLoopMgr() [2/3]

AthMpEvtLoopMgr::AthMpEvtLoopMgr ( )

private

AthMpEvtLoopMgr() [3/3]

AthMpEvtLoopMgr::AthMpEvtLoopMgr ( const AthMpEvtLoopMgr &  )

private

Member Function Documentation

createEventContext()

EventContext AthMpEvtLoopMgr::createEventContext ( )

overridevirtual

Definition at line 148 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 153 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 159 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.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.rfind('/')==(m_workerTopDir.size()-1)?m_workerTopDir.substr(0,m_workerTopDir.size()-1):m_workerTopDir)+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.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.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...
 
  SmartIF<IDataShare> dataShare{serviceLocator()->service("AthenaPoolCnvSvc")};
  ATH_CHECK(dataShare.isValid());
 
  auto sharedWriterTool = m_tools["SharedWriterTool"];
  const bool sharedWriterWithFAFE = (m_nEventsBeforeFork!=0 && sharedWriterTool);
 
  if(sharedWriterWithFAFE) {
    (*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(!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);
    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 && !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 533 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 137 of file AthMpEvtLoopMgr.cxx.

{
  return StatusCode::SUCCESS;
}

generateOutputReport()

StatusCode AthMpEvtLoopMgr::generateOutputReport ( )

private

Definition at line 462 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.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 79 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(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 142 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 &  )

private

stopRun()

StatusCode AthMpEvtLoopMgr::stopRun ( )

overridevirtual

Definition at line 389 of file AthMpEvtLoopMgr.cxx.

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

stopScheduled()

virtual bool AthMpEvtLoopMgr::stopScheduled ( ) const

inlineoverridevirtual

Definition at line 36 of file AthMpEvtLoopMgr.h.

{return m_scheduledStop;};

updateSkipEvents()

StatusCode AthMpEvtLoopMgr::updateSkipEvents ( int  skipEvents )

private

Definition at line 598 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 405 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

bool AthMpEvtLoopMgr::m_collectSubprocessLogs

private

Definition at line 46 of file AthMpEvtLoopMgr.h.

m_eventPrintoutInterval

unsigned int AthMpEvtLoopMgr::m_eventPrintoutInterval

private

Definition at line 52 of file AthMpEvtLoopMgr.h.

m_evtProcessor

ServiceHandle<IEventProcessor> AthMpEvtLoopMgr::m_evtProcessor

private

Definition at line 39 of file AthMpEvtLoopMgr.h.

m_evtSelector

SmartIF<IService> AthMpEvtLoopMgr::m_evtSelector

private

Definition at line 40 of file AthMpEvtLoopMgr.h.

m_execAtPreFork

StringArrayProperty AthMpEvtLoopMgr::m_execAtPreFork

private

Definition at line 53 of file AthMpEvtLoopMgr.h.

m_isPileup

bool AthMpEvtLoopMgr::m_isPileup

private

Definition at line 45 of file AthMpEvtLoopMgr.h.

m_masterPid

pid_t AthMpEvtLoopMgr::m_masterPid

private

Definition at line 54 of file AthMpEvtLoopMgr.h.

m_nChildProcesses

int AthMpEvtLoopMgr::m_nChildProcesses

private

Definition at line 48 of file AthMpEvtLoopMgr.h.

m_nEventsBeforeFork

int AthMpEvtLoopMgr::m_nEventsBeforeFork

private

Definition at line 51 of file AthMpEvtLoopMgr.h.

m_nMemSamplingInterval

int AthMpEvtLoopMgr::m_nMemSamplingInterval

private

Definition at line 50 of file AthMpEvtLoopMgr.h.

m_nPollingInterval

int AthMpEvtLoopMgr::m_nPollingInterval

private

Definition at line 49 of file AthMpEvtLoopMgr.h.

m_nWorkers

int AthMpEvtLoopMgr::m_nWorkers

private

Definition at line 41 of file AthMpEvtLoopMgr.h.

m_outputReportName

std::string AthMpEvtLoopMgr::m_outputReportName

private

Definition at line 43 of file AthMpEvtLoopMgr.h.

m_samplesPss

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

private

Definition at line 59 of file AthMpEvtLoopMgr.h.

m_samplesRss

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

private

Definition at line 58 of file AthMpEvtLoopMgr.h.

m_samplesSize

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

private

Definition at line 60 of file AthMpEvtLoopMgr.h.

m_samplesSwap

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

private

Definition at line 61 of file AthMpEvtLoopMgr.h.

m_scheduledStop

bool AthMpEvtLoopMgr::m_scheduledStop {false}

private

Definition at line 55 of file AthMpEvtLoopMgr.h.

m_strategy

std::string AthMpEvtLoopMgr::m_strategy

private

Definition at line 44 of file AthMpEvtLoopMgr.h.

m_tools

ToolHandleArray<IAthenaMPTool> AthMpEvtLoopMgr::m_tools

private

Definition at line 47 of file AthMpEvtLoopMgr.h.

m_workerTopDir

std::string AthMpEvtLoopMgr::m_workerTopDir

private

Definition at line 42 of file AthMpEvtLoopMgr.h.

---

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

• AthMpEvtLoopMgr.h
• AthMpEvtLoopMgr.cxx