#include <SharedEvtQueueConsumer.h>

Inheritance diagram for SharedEvtQueueConsumer:

Collaboration diagram for SharedEvtQueueConsumer:

Public Member Functions
	SharedEvtQueueConsumer (const std::string &type, const std::string &name, const IInterface *parent)

virtual	~SharedEvtQueueConsumer () override

virtual StatusCode	initialize () override

virtual StatusCode	finalize () override

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

virtual StatusCode exec	ATLAS_NOT_THREAD_SAFE () override

virtual StatusCode wait_once	ATLAS_NOT_THREAD_SAFE (pid_t &pid) override

virtual void	reportSubprocessStatuses () override

virtual void	subProcessLogs (std::vector< std::string > &) override

virtual std::unique_ptr< AthenaInterprocess::ScheduledWork >	bootstrap_func () override

virtual std::unique_ptr< AthenaInterprocess::ScheduledWork >	exec_func () override

virtual std::unique_ptr< AthenaInterprocess::ScheduledWork >	fin_func () override

virtual AthenaMP::AllWorkerOutputs_ptr	generateOutputReport () override

virtual void	useFdsRegistry (std::shared_ptr< AthenaInterprocess::FdsRegistry >) override

virtual void	setRandString (const std::string &randStr) override

virtual void	setMaxEvt (int maxEvt) override

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

virtual void	killChildren () override

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

virtual std::unique_ptr< ScheduledWork >	operator () ATLAS_NOT_THREAD_SAFE(const ScheduledWork &)=0

Protected Types
enum	ESRange_Status { ESRANGE_SUCCESS, ESRANGE_NOTFOUND, ESRANGE_SEEKFAILED, ESRANGE_PROCFAILED, ESRANGE_FILENOTMADE, ESRANGE_BADINPFILE }

enum	Func_Flag { FUNC_BOOTSTRAP, FUNC_EXEC, FUNC_FIN }

Protected Member Functions
int mapAsyncFlag	ATLAS_NOT_THREAD_SAFE (Func_Flag flag, pid_t pid=0)

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

int	updateIoReg (const std::string &rundir)

std::string	fmterror (int errnum)

int	reopenFds ()

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

void	waitForSignal ()

IEvtSelector *	evtSelector ()

Protected Attributes
int	m_nprocs {-1}
	Number of workers spawned by the master process. More...

int	m_maxEvt {-1}
	Maximum number of events assigned to the job. More...

std::string	m_subprocTopDir
	Top run directory for subprocesses. More...

std::string	m_subprocDirPrefix
	For ex. "worker__". More...

std::string	m_evtSelName
	Name of the event selector. More...

AthenaInterprocess::ProcessGroup *	m_processGroup {nullptr}

const AthenaInterprocess::IMPRunStop *	m_mpRunStop {nullptr}

ServiceHandle< IEventProcessor >	m_evtProcessor

ServiceHandle< IAppMgrUI >	m_appMgr

ServiceHandle< IFileMgr >	m_fileMgr

ServiceHandle< IIoComponentMgr >	m_ioMgr

SmartIF< IEvtSelector >	m_evtSelector

std::string	m_fileMgrLog

std::shared_ptr< AthenaInterprocess::FdsRegistry >	m_fdsRegistry

std::string	m_randStr

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

Private Types
typedef System::ProcessTime::TimeValueType	TimeValType

Private Member Functions
int decodeProcessResult	ATLAS_NOT_THREAD_SAFE (const AthenaInterprocess::ProcessResult *presult, bool doFinalize)

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

Private Attributes
Gaudi::Property< bool >	m_useSharedReader {this, "UseSharedReader", false, "Work in pair with a SharedReader"}

Gaudi::Property< bool >	m_useSharedWriter {this, "UseSharedWriter", false, "Work in pair with a SharedWriter"}

Gaudi::Property< bool >	m_isRoundRobin {this, "IsRoundRobin", false, "Are we running in the 'reproducible mode'?"}

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

Gaudi::Property< bool >	m_readEventOrders {this, "ReadEventOrders", false}

Gaudi::Property< int >	m_nEventsBeforeFork {this, "EventsBeforeFork", 0}

Gaudi::Property< std::string >	m_eventOrdersFile {this, "EventOrdersFile", "athenamp_eventorders.txt"}

int	m_rankId {-1}

int	m_nSkipEvents {0}

ServiceHandle< IChronoStatSvc >	m_chronoStatSvc

SmartIF< IEventSeek >	m_evtSeek

SmartIF< IEvtSelectorSeek >	m_evtSelSeek

IEvtSelector::Context *	m_evtContext {nullptr}

SmartIF< IEventShare >	m_evtShare

SmartIF< IDataShare >	m_dataShare

AthenaInterprocess::SharedQueue *	m_sharedEventQueue {nullptr}

std::unique_ptr< AthenaInterprocess::SharedQueue >	m_sharedRankQueue

std::map< pid_t, std::pair< int, TimeValType > >	m_eventStat

std::queue< pid_t >	m_finQueue

std::vector< int >	m_eventOrders

pid_t	m_masterPid

Detailed Description

Definition at line 23 of file SharedEvtQueueConsumer.h.

Member Typedef Documentation

◆ TimeValType

typedef System::ProcessTime::TimeValueType SharedEvtQueueConsumer::TimeValType

private

Definition at line 76 of file SharedEvtQueueConsumer.h.

Member Enumeration Documentation

◆ ESRange_Status

enum AthenaMPToolBase::ESRange_Status

protectedinherited

Enumerator
ESRANGE_SUCCESS
ESRANGE_NOTFOUND
ESRANGE_SEEKFAILED
ESRANGE_PROCFAILED
ESRANGE_FILENOTMADE
ESRANGE_BADINPFILE

Definition at line 58 of file AthenaMPToolBase.h.

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

◆ Func_Flag

enum AthenaMPToolBase::Func_Flag

protectedinherited

Enumerator
FUNC_BOOTSTRAP
FUNC_EXEC
FUNC_FIN

Definition at line 67 of file AthenaMPToolBase.h.

                  {
     FUNC_BOOTSTRAP
     , FUNC_EXEC
     , FUNC_FIN
   };

Constructor & Destructor Documentation

◆ SharedEvtQueueConsumer()

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

Definition at line 34 of file SharedEvtQueueConsumer.cxx.

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

◆ ~SharedEvtQueueConsumer()

SharedEvtQueueConsumer::~SharedEvtQueueConsumer ( )

overridevirtual

Definition at line 44 of file SharedEvtQueueConsumer.cxx.

45 {

46 }

Member Function Documentation

◆ ATLAS_NOT_THREAD_SAFE() [1/5]

virtual StatusCode exec SharedEvtQueueConsumer::ATLAS_NOT_THREAD_SAFE ( )

overridevirtual

◆ ATLAS_NOT_THREAD_SAFE() [2/5]

int decodeProcessResult SharedEvtQueueConsumer::ATLAS_NOT_THREAD_SAFE	(	const AthenaInterprocess::ProcessResult *	presult,
		bool	doFinalize
	)

private

◆ ATLAS_NOT_THREAD_SAFE() [3/5]

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

protectedinherited

◆ ATLAS_NOT_THREAD_SAFE() [4/5]

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

overridevirtual

◆ ATLAS_NOT_THREAD_SAFE() [5/5]

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

overridevirtual

Reimplemented from AthenaMPToolBase.

◆ bootstrap_func() [1/2]

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

overridevirtual

Definition at line 254 of file SharedEvtQueueConsumer.cxx.

 {
   if(m_debug) waitForSignal();
  
   std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
   outwork->data = CxxUtils::xmalloc(sizeof(int));
   *(int*)(outwork->data) = 1; // Error code: for now use 0 success, 1 failure
   outwork->size = sizeof(int);
  
   // For PileUp Digi Fork-After-N-Events >>>>
   // Retrieve cuEvent-s for all background event selectors, if we forked after N events
   std::map<IService*,int> bkgEvtSelectors;
  
   if(m_isPileup) {
     for(IService* ptrSvc : serviceLocator()->getServices()) {
       IEvtSelector* evtsel = dynamic_cast<IEvtSelector*>(ptrSvc);
       if(evtsel && (evtsel != m_evtSelector)) {
         if(m_nEventsBeforeFork>0) {
           IEvtSelectorSeek* evtselseek = dynamic_cast<IEvtSelectorSeek*>(evtsel);
           if(evtselseek) {
             bkgEvtSelectors.emplace(ptrSvc,evtselseek->curEvent(*m_evtContext));
           }
           else {
             ATH_MSG_ERROR("Failed to cast IEvtSelector* onto IEvtSelectorSeek* for " << (ptrSvc)->name());
             return outwork;
           }
         }
         else {
           bkgEvtSelectors.emplace(ptrSvc,0);
         }
       }
     }
   }
   // <<<< For PileUp Digi Fork-After-N-Events
  
   // ...
   // (possible) TODO: extend outwork with some error message, which will be eventually
   // reported in the master proces
   // ...
  
   // ________________________ Get IncidentSvc and fire PostFork ________________________
   SmartIF<IIncidentSvc> p_incidentSvc(serviceLocator()->service("IncidentSvc"));
   if(!p_incidentSvc) {
     ATH_MSG_ERROR("Unable to retrieve IncidentSvc");
     return outwork;
   }
   p_incidentSvc->fireIncident(Incident(name(),"PostFork"));
  
  
   // ________________________ Get RankID ________________________
   //
   if(!m_sharedRankQueue->receive_basic<int>(m_rankId)) {
     ATH_MSG_ERROR("Unable to get rank ID!");
     return outwork;
   }
   std::ostringstream workindex;
   workindex<<m_rankId;
  
   // ________________________ Worker dir: mkdir ________________________
   std::filesystem::path worker_rundir(m_subprocTopDir);
   worker_rundir /= std::filesystem::path(m_subprocDirPrefix+workindex.str());
   // TODO: this "worker_" can be made configurable too
  
   if(mkdir(worker_rundir.string().c_str(),S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH)==-1) {
     ATH_MSG_ERROR("Unable to make worker run directory: " << worker_rundir.string() << ". " << fmterror(errno));
     return outwork;
   }
  
   // __________ Redirect logs unless we want to attach debugger ____________
   if(!m_debug) {
     if(redirectLog(worker_rundir.string()))
       return outwork;
  
     ATH_MSG_INFO("Logs redirected in the AthenaMP event worker PID=" << getpid());
   }
  
   // ________________________ Update Io Registry ____________________________
   if(updateIoReg(worker_rundir.string()))
     return outwork;
  
   ATH_MSG_INFO("Io registry updated in the AthenaMP event worker PID=" << getpid());
  
   // ________________________ SimParams & DigiParams & PDGTABLE.MeV ____________________________
   std::filesystem::path abs_worker_rundir = std::filesystem::absolute(worker_rundir);
   if(std::filesystem::is_regular_file("SimParams.db"))
     COPY_FILE_HACK("SimParams.db", abs_worker_rundir.string()+"/SimParams.db");
   if(std::filesystem::is_regular_file("DigitParams.db"))
     COPY_FILE_HACK("DigitParams.db", abs_worker_rundir.string()+"/DigitParams.db");
   if(std::filesystem::is_regular_file("PDGTABLE.MeV"))
     COPY_FILE_HACK("PDGTABLE.MeV", abs_worker_rundir.string()+"/PDGTABLE.MeV");
  
   // _______________________ Handle saved PFC (if any) ______________________
   if(handleSavedPfc(abs_worker_rundir))
     return outwork;
  
   // ________________________  reopen descriptors ____________________________
   if(reopenFds())
     return outwork;
  
   ATH_MSG_INFO("File descriptors re-opened in the AthenaMP event worker PID=" << getpid());
  
   
   // ________________________ Make Shared Reader/Writer Client ________________________
   if(m_useSharedReader && m_evtShare) {
     ATH_CHECK( m_evtShare->makeClient(m_rankId), outwork);
   }
  
   if(m_useSharedWriter && m_dataShare) {
     SmartIF<IProperty> propertyServer(m_dataShare);
     if (!propertyServer || propertyServer->setProperty("MakeStreamingToolClient", m_rankId + 1).isFailure()) {
       ATH_MSG_ERROR("Could not change AthenaPoolSharedIOCnvSvc MakeClient Property");
       return outwork;
     } 
     else {
       ATH_MSG_DEBUG("Successfully made the conversion service a share client");
     }
   }
  
   // ________________________ I/O reinit ________________________
   ATH_CHECK( m_ioMgr->io_reinitialize(), outwork );
  
   // _______________ Get the value of SkipEvent ________________________
   if(m_evtSelector) {
     SmartIF<IProperty> propertyServer(m_evtSelector);
     ATH_CHECK( propertyServer.isValid(), outwork);
  
     IntegerProperty skipEventsProp("SkipEvents", m_nSkipEvents);
     if(propertyServer->getProperty(&skipEventsProp).isFailure()) {
       ATH_MSG_INFO("Event Selector does not have SkipEvents property");
     }
     else {
       m_nSkipEvents = skipEventsProp.value();
     }
  
     // ________________________ Event selector restart ________________________
     SmartIF<IService> evtSelSvc(m_evtSelector);
     ATH_CHECK( evtSelSvc.isValid(), outwork);
     ATH_CHECK( evtSelSvc->start(), outwork);
   } 
   // For PileUp jobs >>>>
   // Main event selector: advance it if we either forked after N events, or skipEvents!=0
   // Background event selectors: restart, and advance if we forked after N events
   if(m_isPileup) {
     // Deal with the main event selector first
     m_evtSelSeek = serviceLocator()->service(m_evtSelName);
     if(!m_evtSelSeek) {
       ATH_MSG_ERROR("Error retrieving Event Selector with IEvtSelectorSeek interface for PileUp job");
       return outwork;
     }
  
     if((m_nEventsBeforeFork + m_nSkipEvents)
        && m_evtSelSeek->seek(*m_evtContext, m_nEventsBeforeFork+m_nSkipEvents).isFailure()) {
       ATH_MSG_ERROR("Failed to seek to " << m_nEventsBeforeFork+m_nSkipEvents);
       return outwork;
     }
  
     // Deal with background event selectors
     for(auto [evtsel,curEvt] : bkgEvtSelectors) {
       if(evtsel->start().isSuccess()) {
     if (m_nEventsBeforeFork>0) {
       SmartIF<IEvtSelectorSeek> evtselseek(evtsel);
       if(evtselseek->seek(*m_evtContext,curEvt).isFailure()) {
         ATH_MSG_ERROR("Failed to seek to " << curEvt << " in the BKG Event Selector " << evtsel->name());
         return outwork;
       }
     }
       }
       else {
     ATH_MSG_ERROR("Failed to restart BKG Event Selector " << evtsel->name());
     return outwork;
       }
     }
   }
   // <<<< For PileUp jobs
  
   // _______________________ Event orders for debugging ________________________________
   if(m_readEventOrders) {
     std::fstream fs(m_eventOrdersFile,std::fstream::in);
     if(fs.good()) {
       ATH_MSG_INFO("Reading predefined event orders from " << m_eventOrdersFile);
       while(fs.good()){
     std::string line;
     std::getline(fs,line);
     if(line.empty())continue;
  
     // Parse the string
     size_t idx(0);
     int rank = std::stoi(line,&idx);
     if(rank==m_rankId) {
       msg(MSG::INFO) << "This worker will proces the following events #";
       while(idx<line.size()-1) {
         line = line.substr(idx+1);
         int evtnum = std::stoi(line,&idx);
         m_eventOrders.push_back(evtnum);
         msg(MSG::INFO) << " " << evtnum;
       }
       msg(MSG::INFO) << endmsg;
     }
       }
       if(m_eventOrders.empty()) {
     ATH_MSG_ERROR("Could not read event orders for the rank " << m_rankId);
     return outwork;
       }
       fs.close();
     }
     else {
       ATH_MSG_ERROR("Unable to read predefined event orders from " << m_eventOrdersFile);
       return outwork;
     }
   }
  
   // ________________________ Worker dir: chdir ________________________
   if(chdir(worker_rundir.string().c_str())==-1) {
     ATH_MSG_ERROR("Failed to chdir to " << worker_rundir.string());
     return outwork;
   }
  
   // ___________________ Fire UpdateAfterFork incident _________________
   p_incidentSvc->fireIncident(AthenaInterprocess::UpdateAfterFork(m_rankId,getpid(),name()));
  
   // Declare success and return
   *(int*)(outwork->data) = 0;
   return outwork;
 }

◆ bootstrap_func() [2/2]

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

pure virtualinherited

◆ evtSelector()

IEvtSelector* AthenaMPToolBase::evtSelector ( )

inlineprotectedinherited

Definition at line 83 of file AthenaMPToolBase.h.

83 { return m_evtSelector; }

◆ exec_func()

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

overridevirtual

Implements AthenaMPToolBase.

Definition at line 479 of file SharedEvtQueueConsumer.cxx.

 {
   ATH_MSG_INFO("Exec function in the AthenaMP worker PID=" << getpid());
  
   bool all_ok(true);
  
   long intmask = pow(0x100,sizeof(int))-1; // Mask for decoding event number from the value posted to the queue
   int nEvt(m_nEventsBeforeFork);
   int nEventsProcessed(0);
   long evtnumAndChunk(0);
  
   unsigned evtCounter(0);
   int evtnum(0), chunkSize(1);
   auto predefinedEvt = m_eventOrders.cbegin();
  
   // If the event orders file already exists in worker's run directory, then it's an unexpected error!
   std::filesystem::path ordersFile(m_eventOrdersFile.value());
   if(std::filesystem::exists(ordersFile)) {
     ATH_MSG_ERROR(m_eventOrdersFile << " already exists in the worker's run directory!");
     all_ok = false;
   }
  
   System::ProcessTime time_start = System::getProcessTime();
   if(all_ok) {
     std::fstream fs(m_eventOrdersFile,std::fstream::out);
     fs << m_rankId;
     bool firstOrder(true);
     while(true) {
       if(m_isRoundRobin) {
     evtnum = m_nSkipEvents + m_nprocs*evtCounter + m_rankId; 
     if(m_maxEvt!=-1 && evtnum>=m_maxEvt+m_nSkipEvents) {
       break;
     }
     evtCounter++;
       }
       else {
     if(m_readEventOrders) {
       if(predefinedEvt==m_eventOrders.cend()) break;
       evtnum = *predefinedEvt;
       predefinedEvt++;
       fs << (firstOrder?":":",") << evtnum;
       fs.flush();
       firstOrder=false;
       ATH_MSG_INFO("Read event number from the orders file: " << evtnum);
     }
     else {
       if(!m_sharedEventQueue->try_receive_basic<long>(evtnumAndChunk)) {
         // The event queue is empty, but we should check whether there are more events to come or not
         ATH_MSG_DEBUG("Event queue is empty"); 
         usleep(1000);
         continue;
       }
       if(evtnumAndChunk<=0) {
         evtnumAndChunk *= -1;
         ATH_MSG_DEBUG("No more events are expected. The total number of events for this job = " << evtnumAndChunk);
         break;
       }
       ATH_MSG_DEBUG("Received value from the queue 0x" << std::hex << evtnumAndChunk << std::dec);
       chunkSize = evtnumAndChunk >> (sizeof(int)*8);
       evtnum = evtnumAndChunk & intmask;
       ATH_MSG_INFO("Received from the queue: event num=" << evtnum << " chunk size=" << chunkSize);
  
       // Save event order
       for(int i(0);i<chunkSize;++i) {
         fs << (firstOrder?":":",") << evtnum+i;
         firstOrder=false;
       }
       fs.flush();
     } // Get event numbers from the shared queue
       } // Not RoundRobin
       nEvt+=chunkSize;
       StatusCode sc;
       if(m_useSharedReader) {
     sc = m_evtShare->share(evtnum);
     if(sc.isFailure()){
       ATH_MSG_ERROR("Unable to share " << evtnum);
       all_ok=false;
       break;
     }
     else {
       ATH_MSG_INFO("Share of " << evtnum << " succeeded");
     }
       }
       else if(m_evtSelector) {
     m_chronoStatSvc->chronoStart("AthenaMP_seek");
         if (m_evtSeek) {
           sc=m_evtSeek->seek(evtnum);
         }
         else {
           sc=m_evtSelSeek->seek(*m_evtContext, evtnum);
         }
     if(sc.isFailure()){
       ATH_MSG_ERROR("Unable to seek to " << evtnum);
       all_ok=false;
       break;
     }
     else {
       ATH_MSG_INFO("Seek to " << evtnum << " succeeded");
     }
     m_chronoStatSvc->chronoStop("AthenaMP_seek");
       }
       m_chronoStatSvc->chronoStart("AthenaMP_nextEvent");
       sc = m_evtProcessor->nextEvent(nEvt);
       nEventsProcessed += chunkSize;
       if(sc.isFailure()){
     if(chunkSize==1) {
       ATH_MSG_ERROR("Unable to process event " << evtnum);
     }
     else {
       ATH_MSG_ERROR("Unable to process the chunk (" << evtnum << "," << evtnum+chunkSize-1 << ")");
     }
     all_ok=false;
     break;
       }
       m_chronoStatSvc->chronoStop("AthenaMP_nextEvent");
       if(m_mpRunStop->stopScheduled()) {
         ATH_MSG_INFO("Scheduled stop");
         break;
       }
     }
     fs.close();
   }
   System::ProcessTime time_delta = System::getProcessTime() - time_start;
   TimeValType elapsedTime = time_delta.elapsedTime<System::Sec>();
  
   if(all_ok) {
     if(m_evtProcessor->executeRun(0).isFailure()) {
       ATH_MSG_ERROR("Could not finalize the Run");
       all_ok=false;
     }
     else if(!m_useSharedReader && m_evtSelector) {
       StatusCode sc;
       if (m_evtSeek) {
         sc = m_evtSeek->seek(evtnumAndChunk+m_nSkipEvents);
       }
       else {
         sc = m_evtSelSeek->seek(*m_evtContext, evtnumAndChunk+m_nSkipEvents);
       }
       if(sc.isFailure()) {
     ATH_MSG_WARNING("Seek past maxevt to " << evtnumAndChunk+m_nSkipEvents << " returned failure.");
       }
     }
   }
  
   std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
  
   // Return value: "ERRCODE|Func_Flag|NEvt|EvtLoopTime"
   int outsize = 2*sizeof(int)+sizeof(AthenaMPToolBase::Func_Flag)+sizeof(elapsedTime);
   void* outdata = CxxUtils::xmalloc(outsize);
   *(int*)(outdata) = (all_ok?0:1); // Error code: for now use 0 success, 1 failure
   AthenaMPToolBase::Func_Flag func = AthenaMPToolBase::FUNC_EXEC;
   memcpy((char*)outdata+sizeof(int),&func,sizeof(func));
   memcpy((char*)outdata+sizeof(int)+sizeof(func),&nEventsProcessed,sizeof(int));
   memcpy((char*)outdata+2*sizeof(int)+sizeof(func),&elapsedTime,sizeof(elapsedTime));
   outwork->data = outdata;
   outwork->size = outsize;
   // ...
   // (possible) TODO: extend outwork with some error message, which will be eventually
   // reported in the master proces
   // ...
   return outwork;
 }

◆ fin_func()

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

overridevirtual

Implements AthenaMPToolBase.

Definition at line 642 of file SharedEvtQueueConsumer.cxx.

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

◆ finalize()

StatusCode SharedEvtQueueConsumer::finalize ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 96 of file SharedEvtQueueConsumer.cxx.

 {
   if(getpid()==m_masterPid) {
     ATH_MSG_INFO("finalize() in the master process");
     // Merge saved event orders into one in the master run directory
  
     // 1. Check if master run directory already contains a file with saved orders
     // If so, then rename it with random suffix
     std::filesystem::path ordersFile(m_eventOrdersFile.value());
     if(std::filesystem::exists(ordersFile)) {
       srand((unsigned)time(0));
       std::ostringstream randname;
       randname << rand();
       std::string ordersFileBak = m_eventOrdersFile+std::string("-bak-")+randname.str();
       ATH_MSG_WARNING("File " << m_eventOrdersFile << " already exists in the master run directory!");
       ATH_MSG_WARNING("Saving a backup with new name " << ordersFileBak);
  
       std::filesystem::path ordersFileBakpath(ordersFileBak);
       std::filesystem::rename(ordersFile,ordersFileBakpath);
     }
  
     // 2. Merge workers event orders into the master file
     std::fstream fs(m_eventOrdersFile,std::fstream::out);
     for(int i=0; i<m_nprocs; ++i) {
       std::ostringstream workerIndex;
       workerIndex << i;
       std::filesystem::path worker_rundir(m_subprocTopDir);
       worker_rundir /= std::filesystem::path(m_subprocDirPrefix+workerIndex.str());
       std::string ordersFileWorker(worker_rundir.string()+std::string("/")+m_eventOrdersFile);
       ATH_MSG_INFO("Processing " << ordersFileWorker << " ...");
       std::fstream fs_worker(ordersFileWorker.c_str(),std::fstream::in);
       std::string line;
       while(fs_worker.good()) {
     std::getline(fs_worker,line);
     fs << line << std::endl;
       }
       fs_worker.close();
     }
     fs.close();
   } // if(getpid()==m_masterPid)
  
   if (m_evtContext) {
     ATH_CHECK( m_evtSelector->releaseContext (m_evtContext) );
     m_evtContext = nullptr;
   }
  
   return StatusCode::SUCCESS;
 }

◆ fmterror()

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

protectedinherited

Definition at line 333 of file AthenaMPToolBase.cxx.

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

◆ generateOutputReport()

AthenaMP::AllWorkerOutputs_ptr AthenaMPToolBase::generateOutputReport ( )

overridevirtualinherited

Reimplemented in EvtRangeProcessor, EvtRangeScatterer, and SharedWriterTool.

Definition at line 119 of file AthenaMPToolBase.cxx.

 {
   AthenaMP::AllWorkerOutputs_ptr jobOutputs(new AthenaMP::AllWorkerOutputs());
  
   if(m_fileMgrLog.empty()) {
     ATH_MSG_WARNING(name() << " cannot make output report because FileMgr has not been configured to write log file!");
   }
   else {
     // Collect output files made by the workers
     std::string line;
     for(int i=0;i<m_nprocs;++i) {
       // Get the name of FileMgr log
       std::ostringstream workindex;
       workindex<<i;
       std::filesystem::path logFilePath(m_subprocTopDir);
       logFilePath /= std::filesystem::path(m_subprocDirPrefix+workindex.str());
       std::filesystem::path logFile(logFilePath);
       logFile /= std::filesystem::path(m_fileMgrLog);
       if(!(std::filesystem::exists(logFile)&&std::filesystem::is_regular_file(logFile))) {
         ATH_MSG_WARNING(logFile.string() << " either does not exist or is not a regular file. Skipping");
         continue;
       }
  
       ATH_MSG_DEBUG("FileMgr log file (" << i << ") " << logFile);
  
       std::ifstream inpStream(logFile.string().c_str());
       std::set<std::string> reportedFiles; // Don't report the same file twice
       while(!inpStream.eof()) {
     std::getline(inpStream,line);
         if(line.find("WRITE")!=std::string::npos) {
           // Parse the entry
           size_t startpos(0);
       std::vector<std::string> entries;
           while(startpos<line.size()) {
             while(line[startpos]==' ')
               startpos++;
  
             size_t endpos = line.find(' ',startpos);
             if(endpos==std::string::npos) endpos = line.size();
             entries.push_back(line.substr(startpos,endpos-startpos));
             startpos=endpos+1;
           }
  
           // enties[0] is filename
       std::filesystem::path filenamePath(entries[0]);
       std::filesystem::path basename = filenamePath.filename();
           if(reportedFiles.find(basename.string())==reportedFiles.end())
             reportedFiles.insert(basename.string());
           else
             continue;
       std::filesystem::path absolutename = basename.is_absolute() ? basename : std::filesystem::absolute(std::filesystem::path(logFilePath)/=basename);
       AthenaMP::AllWorkerOutputsIterator it1 = jobOutputs->find(basename.string());
           if(it1==jobOutputs->end()) {
             (*jobOutputs)[basename.string()] = AthenaMP::SingleWorkerOutputs();
             (*jobOutputs)[basename.string()].reserve(m_nprocs);
           }
  
       AthenaMP::WorkerOutput newOutput;
           newOutput.filename = absolutename.string();
           newOutput.technology = entries[1];
           newOutput.description = entries[2];
           newOutput.access_mode = entries[3];
           newOutput.shared = (line.find("SHARED")!=std::string::npos);
  
           (*jobOutputs)[basename.string()].push_back(newOutput);
         }
       }
     }
   }
   return jobOutputs;
 }

◆ handleSavedPfc()

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

protectedinherited

Definition at line 396 of file AthenaMPToolBase.cxx.

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

◆ initialize()

StatusCode SharedEvtQueueConsumer::initialize ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 48 of file SharedEvtQueueConsumer.cxx.

 {
   ATH_MSG_DEBUG("In initialize");
  
   ATH_CHECK(AthenaMPToolBase::initialize());
  
   // For pile-up jobs use event loop manager for seeking
   // otherwise use event selector
   if(m_isPileup) {
     m_evtSeek = SmartIF<IEventSeek>(m_evtProcessor.get());
     if(!m_evtSeek) {
       ATH_MSG_ERROR("Unable to dyn-cast PileUpEventLoopMgr to IEventSeek");
       return StatusCode::FAILURE;
     }
   }
   else if(m_evtSelector) {
     m_evtSelSeek = serviceLocator()->service(m_evtSelName);
     ATH_CHECK(m_evtSelSeek.isValid());
   }
  
   if(m_evtSelector) {
     ATH_CHECK( m_evtSelector->createContext (m_evtContext) );
  
     m_evtShare = serviceLocator()->service(m_evtSelName);
     if(!m_evtShare) {
       if(m_useSharedReader) {
         ATH_MSG_ERROR("Error retrieving IEventShare");
         return StatusCode::FAILURE;
       }
       ATH_MSG_INFO("Could not retrieve IEventShare");
     }
  
     //FIXME: AthenaPool dependent for now
  
     if(m_useSharedWriter) {
       m_dataShare = SmartIF<IDataShare>(serviceLocator()->service("AthenaPoolSharedIOCnvSvc"));
       if(!m_dataShare) {
         ATH_MSG_ERROR("Error retrieving AthenaPoolSharedIOCnvSvc");
         return StatusCode::FAILURE;
       }
     }
   }
  
   ATH_CHECK(m_chronoStatSvc.retrieve());
   
   return StatusCode::SUCCESS;
 }

◆ killChildren()

void AthenaMPToolBase::killChildren ( )

overridevirtualinherited

Definition at line 201 of file AthenaMPToolBase.cxx.

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

◆ operator()

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

pure virtualinherited

◆ redirectLog()

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

protectedinherited

Definition at line 269 of file AthenaMPToolBase.cxx.

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

◆ reopenFd()

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

privateinherited

Definition at line 419 of file AthenaMPToolBase.cxx.

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

◆ reopenFds()

int AthenaMPToolBase::reopenFds ( )

protectedinherited

Definition at line 340 of file AthenaMPToolBase.cxx.

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

◆ reportSubprocessStatuses()

void SharedEvtQueueConsumer::reportSubprocessStatuses ( )

overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 223 of file SharedEvtQueueConsumer.cxx.

 {
   ATH_MSG_INFO("Statuses of event processors");
   const std::vector<AthenaInterprocess::ProcessStatus>& statuses = m_processGroup->getStatuses();
   for(size_t i=0; i<statuses.size(); ++i) {
     // Get the number of events processed by this worker
     auto it = m_eventStat.find(statuses[i].pid);
     msg(MSG::INFO) << "*** Process PID=" << statuses[i].pid 
            << ". Status " << ((statuses[i].exitcode)?"FAILURE":"SUCCESS") 
            << ". Number of events processed: ";
     if(it==m_eventStat.end())
       msg(MSG::INFO) << "N/A" << endmsg;
     else
       msg(MSG::INFO) << it->second.first
              << ", Event Loop Time: " << it->second.second << "sec."
              << endmsg;
   }
 }

◆ setMaxEvt()

virtual void AthenaMPToolBase::setMaxEvt ( int maxEvt )

inlineoverridevirtualinherited

Definition at line 44 of file AthenaMPToolBase.h.

44 {m_maxEvt=maxEvt;}

◆ setMPRunStop()

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

inlineoverridevirtualinherited

Definition at line 45 of file AthenaMPToolBase.h.

45 {m_mpRunStop=runStop;}

◆ setRandString()

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

overridevirtualinherited

Definition at line 196 of file AthenaMPToolBase.cxx.

 {
   m_randStr = randStr;
 }

◆ subProcessLogs()

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

overridevirtual

Definition at line 242 of file SharedEvtQueueConsumer.cxx.

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

◆ updateIoReg()

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

protectedinherited

Definition at line 322 of file AthenaMPToolBase.cxx.

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

◆ useFdsRegistry()

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

overridevirtualinherited

Definition at line 191 of file AthenaMPToolBase.cxx.

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

◆ waitForSignal()

void AthenaMPToolBase::waitForSignal ( )

protectedinherited

Definition at line 403 of file AthenaMPToolBase.cxx.

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

Member Data Documentation

◆ m_appMgr

ServiceHandle<IAppMgrUI> AthenaMPToolBase::m_appMgr

protectedinherited

Definition at line 95 of file AthenaMPToolBase.h.

◆ m_chronoStatSvc

ServiceHandle<IChronoStatSvc> SharedEvtQueueConsumer::m_chronoStatSvc

private

Definition at line 66 of file SharedEvtQueueConsumer.h.

◆ m_dataShare

SmartIF<IDataShare> SharedEvtQueueConsumer::m_dataShare

private

Definition at line 71 of file SharedEvtQueueConsumer.h.

◆ m_debug

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

private

Definition at line 58 of file SharedEvtQueueConsumer.h.

◆ m_eventOrders

std::vector<int> SharedEvtQueueConsumer::m_eventOrders

private

Definition at line 81 of file SharedEvtQueueConsumer.h.

◆ m_eventOrdersFile

Gaudi::Property<std::string> SharedEvtQueueConsumer::m_eventOrdersFile {this, "EventOrdersFile", "athenamp_eventorders.txt"}

private

Definition at line 61 of file SharedEvtQueueConsumer.h.

◆ m_eventStat

std::map<pid_t,std::pair<int,TimeValType> > SharedEvtQueueConsumer::m_eventStat

private

Definition at line 77 of file SharedEvtQueueConsumer.h.

◆ m_evtContext

IEvtSelector::Context* SharedEvtQueueConsumer::m_evtContext {nullptr}

private

Definition at line 69 of file SharedEvtQueueConsumer.h.

◆ m_evtProcessor

ServiceHandle<IEventProcessor> AthenaMPToolBase::m_evtProcessor

protectedinherited

Definition at line 94 of file AthenaMPToolBase.h.

◆ m_evtSeek

SmartIF<IEventSeek> SharedEvtQueueConsumer::m_evtSeek

private

Definition at line 67 of file SharedEvtQueueConsumer.h.

◆ m_evtSelector

SmartIF<IEvtSelector> AthenaMPToolBase::m_evtSelector

protectedinherited

Definition at line 98 of file AthenaMPToolBase.h.

◆ m_evtSelName

std::string AthenaMPToolBase::m_evtSelName

protectedinherited

Name of the event selector.

Definition at line 89 of file AthenaMPToolBase.h.

◆ m_evtSelSeek

SmartIF<IEvtSelectorSeek> SharedEvtQueueConsumer::m_evtSelSeek

private

Definition at line 68 of file SharedEvtQueueConsumer.h.

◆ m_evtShare

SmartIF<IEventShare> SharedEvtQueueConsumer::m_evtShare

private

Definition at line 70 of file SharedEvtQueueConsumer.h.

◆ m_fdsRegistry

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

protectedinherited

Definition at line 100 of file AthenaMPToolBase.h.

◆ m_fileMgr

ServiceHandle<IFileMgr> AthenaMPToolBase::m_fileMgr

protectedinherited

Definition at line 96 of file AthenaMPToolBase.h.

◆ m_fileMgrLog

std::string AthenaMPToolBase::m_fileMgrLog

protectedinherited

Definition at line 99 of file AthenaMPToolBase.h.

◆ m_finQueue

std::queue<pid_t> SharedEvtQueueConsumer::m_finQueue

private

Definition at line 78 of file SharedEvtQueueConsumer.h.

◆ m_ioMgr

ServiceHandle<IIoComponentMgr> AthenaMPToolBase::m_ioMgr

protectedinherited

Definition at line 97 of file AthenaMPToolBase.h.

◆ m_isPileup

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

protectedinherited

Definition at line 103 of file AthenaMPToolBase.h.

◆ m_isRoundRobin

Gaudi::Property<bool> SharedEvtQueueConsumer::m_isRoundRobin {this, "IsRoundRobin", false, "Are we running in the 'reproducible mode'?"}

private

Definition at line 57 of file SharedEvtQueueConsumer.h.

◆ m_masterPid

pid_t SharedEvtQueueConsumer::m_masterPid

private

Definition at line 82 of file SharedEvtQueueConsumer.h.

◆ m_maxEvt

int AthenaMPToolBase::m_maxEvt {-1}

protectedinherited

Maximum number of events assigned to the job.

Definition at line 86 of file AthenaMPToolBase.h.

◆ m_mpRunStop

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

protectedinherited

Definition at line 92 of file AthenaMPToolBase.h.

◆ m_nEventsBeforeFork

Gaudi::Property<int> SharedEvtQueueConsumer::m_nEventsBeforeFork {this, "EventsBeforeFork", 0}

private

Definition at line 60 of file SharedEvtQueueConsumer.h.

◆ m_nprocs

int AthenaMPToolBase::m_nprocs {-1}

protectedinherited

Number of workers spawned by the master process.

Definition at line 85 of file AthenaMPToolBase.h.

◆ m_nSkipEvents

int SharedEvtQueueConsumer::m_nSkipEvents {0}

private

Definition at line 64 of file SharedEvtQueueConsumer.h.

◆ m_processGroup

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

protectedinherited

Definition at line 91 of file AthenaMPToolBase.h.

◆ m_randStr

std::string AthenaMPToolBase::m_randStr

protectedinherited

Definition at line 101 of file AthenaMPToolBase.h.

◆ m_rankId

int SharedEvtQueueConsumer::m_rankId {-1}

private

Definition at line 63 of file SharedEvtQueueConsumer.h.

◆ m_readEventOrders

Gaudi::Property<bool> SharedEvtQueueConsumer::m_readEventOrders {this, "ReadEventOrders", false}

private

Definition at line 59 of file SharedEvtQueueConsumer.h.

◆ m_sharedEventQueue

AthenaInterprocess::SharedQueue* SharedEvtQueueConsumer::m_sharedEventQueue {nullptr}

private

Definition at line 73 of file SharedEvtQueueConsumer.h.

◆ m_sharedRankQueue

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

private

Definition at line 74 of file SharedEvtQueueConsumer.h.

◆ m_subprocDirPrefix

std::string AthenaMPToolBase::m_subprocDirPrefix

protectedinherited

For ex. "worker__".

Definition at line 88 of file AthenaMPToolBase.h.

◆ m_subprocTopDir

std::string AthenaMPToolBase::m_subprocTopDir

protectedinherited

Top run directory for subprocesses.

Definition at line 87 of file AthenaMPToolBase.h.

◆ m_useSharedReader

Gaudi::Property<bool> SharedEvtQueueConsumer::m_useSharedReader {this, "UseSharedReader", false, "Work in pair with a SharedReader"}

private

Definition at line 55 of file SharedEvtQueueConsumer.h.

◆ m_useSharedWriter

Gaudi::Property<bool> SharedEvtQueueConsumer::m_useSharedWriter {this, "UseSharedWriter", false, "Work in pair with a SharedWriter"}

private

Definition at line 56 of file SharedEvtQueueConsumer.h.

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

Public Member Functions

Protected Types

Protected Member Functions

Protected Attributes

Private Types

Private Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ TimeValType

Member Enumeration Documentation

◆ ESRange_Status

◆ Func_Flag

Constructor & Destructor Documentation

◆ SharedEvtQueueConsumer()

◆ ~SharedEvtQueueConsumer()

Member Function Documentation

◆ ATLAS_NOT_THREAD_SAFE() [1/5]

◆ ATLAS_NOT_THREAD_SAFE() [2/5]

◆ ATLAS_NOT_THREAD_SAFE() [3/5]

◆ ATLAS_NOT_THREAD_SAFE() [4/5]

◆ ATLAS_NOT_THREAD_SAFE() [5/5]

◆ bootstrap_func() [1/2]

◆ bootstrap_func() [2/2]

◆ evtSelector()

◆ exec_func()

◆ fin_func()

◆ finalize()

◆ fmterror()

◆ generateOutputReport()

◆ handleSavedPfc()

◆ initialize()

◆ killChildren()

◆ operator()

◆ redirectLog()

◆ reopenFd()

◆ reopenFds()

◆ reportSubprocessStatuses()

◆ setMaxEvt()

◆ setMPRunStop()

◆ setRandString()

◆ subProcessLogs()

◆ updateIoReg()

◆ useFdsRegistry()

◆ waitForSignal()

Member Data Documentation

◆ m_appMgr

◆ m_chronoStatSvc

◆ m_dataShare

◆ m_debug

◆ m_eventOrders

◆ m_eventOrdersFile

◆ m_eventStat

◆ m_evtContext

◆ m_evtProcessor

◆ m_evtSeek

◆ m_evtSelector

◆ m_evtSelName

◆ m_evtSelSeek

◆ m_evtShare

◆ m_fdsRegistry

◆ m_fileMgr

◆ m_fileMgrLog

◆ m_finQueue

◆ m_ioMgr

◆ m_isPileup

◆ m_isRoundRobin

◆ m_masterPid

◆ m_maxEvt

◆ m_mpRunStop

◆ m_nEventsBeforeFork

◆ m_nprocs

◆ m_nSkipEvents

◆ m_processGroup

◆ m_randStr

◆ m_rankId

◆ m_readEventOrders

◆ m_sharedEventQueue

◆ m_sharedRankQueue

◆ m_subprocDirPrefix