Inheritance diagram for python.PyComps.MpEvtLoopMgr:

Collaboration diagram for python.PyComps.MpEvtLoopMgr:

Public Member Functions
	__init__ (self, name='AthMpEvtLoopMgr', isPileup=False, **kw)
None	configureStrategy (self, strategy, pileup, events_before_fork)
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

Public Attributes
	nThreads = theApp._opts.threads
	WorkerTopDir = jp.AthenaMPFlags.WorkerTopDir()
	OutputReportFile = jp.AthenaMPFlags.OutputReportFile()
	CollectSubprocessLogs = jp.AthenaMPFlags.CollectSubprocessLogs()
str	Strategy = jp.AthenaMPFlags.Strategy()
	PollingInterval = jp.AthenaMPFlags.PollingInterval()
	MemSamplingInterval = jp.AthenaMPFlags.MemSamplingInterval()
int	EventsBeforeFork = jp.AthenaMPFlags.EventsBeforeFork()
	IsPileup = isPileup

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 10 of file Control/AthenaMP/python/PyComps.py.

Constructor & Destructor Documentation

◆ init()

python.PyComps.MpEvtLoopMgr.__init__	(		self,
			name = 'AthMpEvtLoopMgr',
			isPileup = False,
		**	kw )

Definition at line 11 of file Control/AthenaMP/python/PyComps.py.

    def __init__(self, name='AthMpEvtLoopMgr', isPileup=False, **kw):
 
        from AthenaCommon.AppMgr import theApp
        self.nThreads = theApp._opts.threads
 
        
        kw['name'] = name
        super(MpEvtLoopMgr, self).__init__(**kw)
 
        os.putenv('XRD_ENABLEFORKHANDLERS','1')
        os.putenv('XRD_RUNFORKHANDLER','1')
 
        from .AthenaMPFlags import jobproperties as jp
        self.WorkerTopDir = jp.AthenaMPFlags.WorkerTopDir()
        self.OutputReportFile = jp.AthenaMPFlags.OutputReportFile()
        self.CollectSubprocessLogs = jp.AthenaMPFlags.CollectSubprocessLogs()
        self.Strategy = jp.AthenaMPFlags.Strategy()
        self.PollingInterval = jp.AthenaMPFlags.PollingInterval()
        self.MemSamplingInterval = jp.AthenaMPFlags.MemSamplingInterval()
        self.EventsBeforeFork = jp.AthenaMPFlags.EventsBeforeFork()
        self.IsPileup = isPileup
 
        if self.Strategy=='EventService':
            self.EventsBeforeFork = 0
 
        from AthenaCommon.AppMgr import theApp as app
        app.EventLoop = self.getFullJobOptName()
 
        # Enable FileMgr logging
        from GaudiSvc.GaudiSvcConf import FileMgr
        from AthenaCommon.AppMgr import ServiceMgr as svcMgr
        svcMgr+=FileMgr(LogFile="FileManagerLog")
 
        # Save PoolFileCatalog.xml if exists in the run directory
        if os.path.isfile('PoolFileCatalog.xml'):
            shutil.copyfile('PoolFileCatalog.xml','PoolFileCatalog.xml.AthenaMP-saved')
 
        self.configureStrategy(self.Strategy,self.IsPileup,self.EventsBeforeFork)
        

Member Function Documentation

◆ configureStrategy()

None python.PyComps.MpEvtLoopMgr.configureStrategy	(	self,
		strategy,
		pileup,
		events_before_fork )

Definition at line 50 of file Control/AthenaMP/python/PyComps.py.

    def configureStrategy(self,strategy,pileup,events_before_fork) -> None :
        from .AthenaMPFlags import jobproperties as jp
        import AthenaCommon.ConcurrencyFlags # noqa: F401
        event_range_channel = jp.AthenaMPFlags.EventRangeChannel()
 
        chunk_size = getChunkSize()
        if chunk_size < 1:
            msg.warning('Nonpositive ChunkSize (%i) caught, setting it to 1', chunk_size)
            chunk_size = 1
        
        debug_worker = jp.ConcurrencyFlags.DebugWorkers()
        use_shared_reader = jp.AthenaMPFlags.UseSharedReader()
        use_shared_writer = jp.AthenaMPFlags.UseSharedWriter()
        use_parallel_compression = jp.AthenaMPFlags.UseParallelCompression()
        unique_id = f"{str(os.getpid())}-{uuid.uuid4().hex}"
 
        # For e.g. event generation, if we use SharedQueue the job does not complete correctly
        if strategy == 'SharedQueue':
            from AthenaCommon.AthenaCommonFlags import jobproperties as ajp
            if (not ajp.AthenaCommonFlags.FilesInput.statusOn) or ajp.AthenaCommonFlags.FilesInput == []:
                msg.info('MP strategy "SharedQueue" will not work without input files when maxEvents=-1. Switching to "RoundRobin" just in case')
                strategy = 'RoundRobin'
 
        if strategy=='SharedQueue' or strategy=='RoundRobin':
            if use_shared_reader:
                from AthenaCommon.AppMgr import ServiceMgr as svcMgr
                svcMgr.PoolSvc.MaxFilesOpen = 2
                from AthenaIPCTools.AthenaIPCToolsConf import AthenaSharedMemoryTool
                svcMgr.EventSelector.SharedMemoryTool = AthenaSharedMemoryTool("EventStreamingTool", SharedMemoryName=f"EventStream{unique_id}")
                if 'AthenaPoolCnvSvc.ReadAthenaPool' in sys.modules:
                    svcMgr.AthenaPoolCnvSvc.InputStreamingTool = AthenaSharedMemoryTool("InputStreamingTool", SharedMemoryName=f"InputStream{unique_id}")
            if use_shared_writer:
                from AthenaCommon.AppMgr import ServiceMgr as svcMgr
                if 'AthenaPoolCnvSvc.WriteAthenaPool' in sys.modules:
                    from AthenaIPCTools.AthenaIPCToolsConf import AthenaSharedMemoryTool
                    svcMgr.AthenaPoolCnvSvc.OutputStreamingTool = AthenaSharedMemoryTool("OutputStreamingTool", SharedMemoryName=f"OutputStream{unique_id}")
                    svcMgr.AthenaPoolCnvSvc.ParallelCompression=use_parallel_compression
 
            if strategy=='SharedQueue':
                from AthenaMPTools.AthenaMPToolsConf import SharedEvtQueueProvider
                self.Tools += [ SharedEvtQueueProvider(UseSharedReader=use_shared_reader,
                                                       IsPileup=pileup,
                                                       EventsBeforeFork=events_before_fork,
                                                       ChunkSize=chunk_size) ]
 
            # In pure MP, self.nThreads may be set to None - we want the pure MP setup in that case
            if self.nThreads is not None and self.nThreads >= 1:
                if(pileup):
                    raise Exception('Running pileup digitization in mixed MP+MT currently not supported')
                from AthenaMPTools.AthenaMPToolsConf import SharedHiveEvtQueueConsumer
                self.Tools += [ SharedHiveEvtQueueConsumer(UseSharedWriter=use_shared_writer, 
                                                           EventsBeforeFork=events_before_fork,
                                                           Debug=debug_worker)   ]
            else:
                from AthenaMPTools.AthenaMPToolsConf import SharedEvtQueueConsumer
                self.Tools += [ SharedEvtQueueConsumer(UseSharedReader=use_shared_reader,
                                                       UseSharedWriter=use_shared_writer,
                                                       IsPileup=pileup,
                                                       IsRoundRobin=(strategy=='RoundRobin'),
                                                       EventsBeforeFork=events_before_fork,
                                                       ReadEventOrders=jp.AthenaMPFlags.ReadEventOrders(),
                                                       EventOrdersFile=jp.AthenaMPFlags.EventOrdersFile(),
                                                       Debug=debug_worker)   ]
            if use_shared_writer:
                from AthenaMPTools.AthenaMPToolsConf import SharedWriterTool
                self.Tools += [ SharedWriterTool(MotherProcess=(events_before_fork>0),
                                                 IsPileup=pileup,
                                                 Debug=debug_worker) ]
        elif strategy=='EventService':
            channelScatterer2Processor = "AthenaMP_Scatterer2Processor"
            channelProcessor2EvtSel = "AthenaMP_Processor2EvtSel"
 
            from AthenaMPTools.AthenaMPToolsConf import EvtRangeScatterer
            self.Tools += [ EvtRangeScatterer(ProcessorChannel = channelScatterer2Processor,
                                              EventRangeChannel = event_range_channel,
                                              DoCaching=jp.AthenaMPFlags.EvtRangeScattererCaching()) ]
 
            from AthenaMPTools.AthenaMPToolsConf import EvtRangeProcessor
            self.Tools += [ EvtRangeProcessor(IsPileup=pileup,
                                              Channel2Scatterer = channelScatterer2Processor,
                                              Channel2EvtSel = channelProcessor2EvtSel,
                                              Debug=debug_worker) ]
        else:
            msg.warning("Unknown strategy. No MP tools will be configured")
 

◆ createEventContext()

EventContext AthMpEvtLoopMgr::createEventContext ( )

overridevirtualinherited

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 )

overridevirtualinherited

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 )

overridevirtualinherited

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 ( )

privateinherited

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 ( )

overridevirtualinherited

Definition at line 106 of file AthMpEvtLoopMgr.cxx.

{
  return StatusCode::SUCCESS;
}

◆ generateOutputReport()

StatusCode AthMpEvtLoopMgr::generateOutputReport ( )

privateinherited

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 ( )

overridevirtualinherited

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 )

overridevirtualinherited

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

◆ stopRun()

StatusCode AthMpEvtLoopMgr::stopRun ( )

overridevirtualinherited

Definition at line 363 of file AthMpEvtLoopMgr.cxx.

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

◆ stopScheduled()

virtual bool AthMpEvtLoopMgr::stopScheduled ( ) const

inlineoverridevirtualinherited

Definition at line 42 of file AthMpEvtLoopMgr.h.

42{return m_scheduledStop;};

◆ updateSkipEvents()

StatusCode AthMpEvtLoopMgr::updateSkipEvents ( int skipEvents )

privateinherited

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 ( )

privateinherited

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

◆ CollectSubprocessLogs

python.PyComps.MpEvtLoopMgr.CollectSubprocessLogs = jp.AthenaMPFlags.CollectSubprocessLogs()

Definition at line 26 of file Control/AthenaMP/python/PyComps.py.

◆ EventsBeforeFork

python.PyComps.MpEvtLoopMgr.EventsBeforeFork = jp.AthenaMPFlags.EventsBeforeFork()

Definition at line 30 of file Control/AthenaMP/python/PyComps.py.

◆ IsPileup

python.PyComps.MpEvtLoopMgr.IsPileup = isPileup

Definition at line 31 of file Control/AthenaMP/python/PyComps.py.

◆ m_collectSubprocessLogs

Gaudi::Property<bool> AthMpEvtLoopMgr::m_collectSubprocessLogs

privateinherited

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.

64 {this, "CollectSubprocessLogs", false,

65 "Copy all workers' logs into the main log file at the end of the job?"};

◆ m_dataShare

SmartIF<IDataShare> AthMpEvtLoopMgr::m_dataShare

privateinherited

Definition at line 47 of file AthMpEvtLoopMgr.h.

◆ m_eventPrintoutInterval

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

privateinherited

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.

78 {this, "EventPrintoutInterval", 1,

79 "The value to be forwarded to the EventPrintoutInterval property of the AthenaEventLoopMgr"};

◆ m_evtProcessor

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

privateinherited

Definition at line 45 of file AthMpEvtLoopMgr.h.

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

◆ m_evtSelector

SmartIF<IService> AthMpEvtLoopMgr::m_evtSelector {nullptr}

privateinherited

Definition at line 46 of file AthMpEvtLoopMgr.h.

46{nullptr};

◆ m_execAtPreFork

StringArrayProperty AthMpEvtLoopMgr::m_execAtPreFork

privateinherited

Initial value:

{this, "ExecAtPreFork", {},

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

Definition at line 81 of file AthMpEvtLoopMgr.h.

81 {this, "ExecAtPreFork", {},

82 "The value to be forwarded to the ExecAtPreFork property of the AthenaEventLoopMgr"};

◆ m_isPileup

Gaudi::Property<bool> AthMpEvtLoopMgr::m_isPileup

privateinherited

Initial value:

{this, "IsPileup", false,

"Is AthenaMP running a PileUp Digitization job?"}

Definition at line 61 of file AthMpEvtLoopMgr.h.

61 {this, "IsPileup", false,

62 "Is AthenaMP running a PileUp Digitization job?"};

◆ m_masterPid

pid_t AthMpEvtLoopMgr::m_masterPid {}

privateinherited

Definition at line 85 of file AthMpEvtLoopMgr.h.

85{}; // PID of the main process

◆ m_nChildProcesses

int AthMpEvtLoopMgr::m_nChildProcesses {0}

privateinherited

Definition at line 84 of file AthMpEvtLoopMgr.h.

84{0}; // Total number of child processes

◆ m_nEventsBeforeFork

Gaudi::Property<int> AthMpEvtLoopMgr::m_nEventsBeforeFork

privateinherited

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.

75 {this, "EventsBeforeFork", 0,

76 "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

privateinherited

Initial value:

{this, "MemSamplingInterval", 0,

"Interval in seconds between taking memory usage samples. 0 - no sampling"}

Definition at line 72 of file AthMpEvtLoopMgr.h.

72 {this, "MemSamplingInterval", 0,

73 "Interval in seconds between taking memory usage samples. 0 - no sampling"};

◆ m_nPollingInterval

Gaudi::Property<int> AthMpEvtLoopMgr::m_nPollingInterval

privateinherited

Initial value:

{this, "PollingInterval", 100,

"Interval in milliseconds between checks of sub-processes statuses"}

Definition at line 69 of file AthMpEvtLoopMgr.h.

69 {this, "PollingInterval", 100,

70 "Interval in milliseconds between checks of sub-processes statuses"};

◆ m_nWorkers

Gaudi::Property<int> AthMpEvtLoopMgr::m_nWorkers

privateinherited

Initial value:

{this, "NWorkers", 0,

"Number of AthenaMP worker processes"}

Definition at line 49 of file AthMpEvtLoopMgr.h.

49 {this, "NWorkers", 0,

50 "Number of AthenaMP worker processes"};

◆ m_outputReportName

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

privateinherited

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.

55 {this, "OutputReportFile", "AthenaMPOutputs",

56 "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

privateinherited

Definition at line 90 of file AthMpEvtLoopMgr.h.

◆ m_samplesRss

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

privateinherited

Definition at line 89 of file AthMpEvtLoopMgr.h.

◆ m_samplesSize

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

privateinherited

Definition at line 91 of file AthMpEvtLoopMgr.h.

◆ m_samplesSwap

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

privateinherited

Definition at line 92 of file AthMpEvtLoopMgr.h.

◆ m_scheduledStop

bool AthMpEvtLoopMgr::m_scheduledStop {false}

privateinherited

Definition at line 86 of file AthMpEvtLoopMgr.h.

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

◆ m_strategy

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

privateinherited

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.

58 {this, "Strategy", "",

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

◆ m_tools

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

privateinherited

Definition at line 67 of file AthMpEvtLoopMgr.h.

67{this,"Tools", {}};

◆ m_workerTopDir

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

privateinherited

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.

52 {this, "WorkerTopDir", "athenaMP_workers",

53 "Sub-directory of the main run directory that contains run directories of all workers"};

◆ MemSamplingInterval

python.PyComps.MpEvtLoopMgr.MemSamplingInterval = jp.AthenaMPFlags.MemSamplingInterval()

Definition at line 29 of file Control/AthenaMP/python/PyComps.py.

◆ nThreads

python.PyComps.MpEvtLoopMgr.nThreads = theApp._opts.threads

Definition at line 14 of file Control/AthenaMP/python/PyComps.py.

◆ OutputReportFile

python.PyComps.MpEvtLoopMgr.OutputReportFile = jp.AthenaMPFlags.OutputReportFile()

Definition at line 25 of file Control/AthenaMP/python/PyComps.py.

◆ PollingInterval

python.PyComps.MpEvtLoopMgr.PollingInterval = jp.AthenaMPFlags.PollingInterval()

Definition at line 28 of file Control/AthenaMP/python/PyComps.py.

◆ Strategy

python.PyComps.MpEvtLoopMgr.Strategy = jp.AthenaMPFlags.Strategy()

Definition at line 27 of file Control/AthenaMP/python/PyComps.py.

◆ WorkerTopDir

python.PyComps.MpEvtLoopMgr.WorkerTopDir = jp.AthenaMPFlags.WorkerTopDir()

Definition at line 24 of file Control/AthenaMP/python/PyComps.py.

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

Control/AthenaMP/python/PyComps.py

Public Member Functions

Public Attributes

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ configureStrategy()

◆ createEventContext()

◆ executeEvent()

◆ executeRun()

◆ extractFds()

◆ finalize()

◆ generateOutputReport()

◆ initialize()

◆ nextEvent()

◆ stopRun()

◆ stopScheduled()

◆ updateSkipEvents()

◆ wait()

Member Data Documentation

◆ CollectSubprocessLogs

◆ EventsBeforeFork

◆ IsPileup

◆ m_collectSubprocessLogs

◆ m_dataShare

◆ m_eventPrintoutInterval

◆ m_evtProcessor

◆ m_evtSelector

◆ m_execAtPreFork

◆ m_isPileup

◆ m_masterPid

◆ m_nChildProcesses

◆ m_nEventsBeforeFork

◆ m_nMemSamplingInterval

◆ m_nPollingInterval

◆ m_nWorkers

◆ m_outputReportName

◆ m_samplesPss

◆ m_samplesRss

◆ m_samplesSize

◆ m_samplesSwap

◆ m_scheduledStop

◆ m_strategy

◆ m_tools

◆ m_workerTopDir

◆ MemSamplingInterval

◆ nThreads

◆ OutputReportFile

◆ PollingInterval

◆ Strategy

◆ WorkerTopDir

◆ init()