SharedEvtQueueProvider Class Reference

#include <SharedEvtQueueProvider.h>

Inheritance diagram for SharedEvtQueueProvider:

Collaboration diagram for SharedEvtQueueProvider:

Public Member Functions
	SharedEvtQueueProvider (const std::string &type, const std::string &name, const IInterface *parent)
virtual	~SharedEvtQueueProvider () 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 void	subProcessLogs (std::vector< std::string > &) override
virtual AthenaMP::AllWorkerOutputs_ptr	generateOutputReport () override
virtual void	handle (const Incident &inc) 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

Private Member Functions
	SharedEvtQueueProvider ()
	SharedEvtQueueProvider (const SharedEvtQueueProvider &)
SharedEvtQueueProvider &	operator= (const SharedEvtQueueProvider &)
void	addEventsToQueue ()

Private Attributes
Gaudi::Property< bool >	m_useSharedReader {this, "UseSharedReader", false, "Use shared reader"}
Gaudi::Property< int >	m_nEventsBeforeFork {this, "EventsBeforeFork", 0, "Number of events before forking"}
Gaudi::Property< int >	m_nChunkSize {this, "ChunkSize", 1}
int	m_nprocesses {-1}
	We use this data member for adding negative numbers at the end of the event queue.
int	m_nChunkStart {0}
	The beginning of the current chunk.
int	m_nPositionInChunk {0}
	Position within the current chunk.
int	m_nEvtRequested {-1}
	Max event received from AppMgr.
int	m_nEvtCounted {0}
	The number of events this tool has counted itself in the input files.
AthenaInterprocess::SharedQueue *	m_sharedEventQueue {nullptr}
SmartIF< IEventShare >	m_evtShare

Detailed Description

Definition at line 14 of file SharedEvtQueueProvider.h.

Constructor & Destructor Documentation

SharedEvtQueueProvider() [1/3]

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

Definition at line 28 of file SharedEvtQueueProvider.cxx.

  : base_class(type,name,parent)
{
  m_subprocDirPrefix = "evt_counter";
}

~SharedEvtQueueProvider()

SharedEvtQueueProvider::~SharedEvtQueueProvider ( )

overridevirtual

Definition at line 36 of file SharedEvtQueueProvider.cxx.

{
}

SharedEvtQueueProvider() [2/3]

SharedEvtQueueProvider::SharedEvtQueueProvider ( )

private

SharedEvtQueueProvider() [3/3]

SharedEvtQueueProvider::SharedEvtQueueProvider ( const SharedEvtQueueProvider & )

private

Member Function Documentation

addEventsToQueue()

void SharedEvtQueueProvider::addEventsToQueue ( )

private

Definition at line 357 of file SharedEvtQueueProvider.cxx.

{
  ATH_MSG_DEBUG("in addEventsToQueue");
  long newValueForQueue = ((long)(m_nPositionInChunk-m_nChunkStart)<<(sizeof(int)*8))|m_nChunkStart;
  while(!m_sharedEventQueue->try_send_basic<long>(newValueForQueue)) {
    usleep(100);
  }
  ATH_MSG_INFO("Sent to the queue 0x" << std::hex << newValueForQueue << std::dec
           << " which corresponds to Chunks start " << m_nChunkStart
           << " and chunk size " << m_nPositionInChunk-m_nChunkStart);
}

ATLAS_NOT_THREAD_SAFE() [1/2]

virtual StatusCode exec SharedEvtQueueProvider::ATLAS_NOT_THREAD_SAFE ( )

overridevirtual

ATLAS_NOT_THREAD_SAFE() [2/2]

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

overridevirtual

bootstrap_func()

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

overridevirtual

Definition at line 110 of file SharedEvtQueueProvider.cxx.

{
  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
  outwork->data = CxxUtils::xmalloc(sizeof(int));
  *(int*)(outwork->data) = 1; // Error code: for now use 0 success, 1 failure
  outwork->size = sizeof(int);
 
  // ...
  // (possible) TODO: extend outwork with some error message, which will be eventually
  // reported in the master proces
  // ...
 
  // ________________________ mkdir ________________________
  std::filesystem::path counter_rundir(m_subprocTopDir);
  counter_rundir /= std::filesystem::path(m_subprocDirPrefix);
 
  if(mkdir(counter_rundir.string().c_str(),S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH)==-1) {
    ATH_MSG_ERROR( "Unable to make event counter run directory: " << counter_rundir.string() << ". " << fmterror(errno) );
    return outwork;
  }
 
  // ________________________ Redirect logs ________________________
  if(redirectLog(counter_rundir.string()))
    return outwork;
 
  ATH_MSG_INFO( "Logs redirected in the AthenaMP event event counter PID=" << getpid() );
 
  // ________________________ Update Io Registry ____________________________
  if(updateIoReg(counter_rundir.string()))
    return outwork;
 
  ATH_MSG_INFO( "Io registry updated in the AthenaMP event event counter PID=" << getpid() );
 
  // _______________________ Handle saved PFC (if any) ______________________
  std::filesystem::path abs_counter_rundir = std::filesystem::absolute(counter_rundir);
  if(handleSavedPfc(abs_counter_rundir))
    return outwork;
 
  // ________________________ reopen descriptors ____________________________
  if(reopenFds())
    return outwork;
 
  ATH_MSG_INFO( "File descriptors re-opened in the AthenaMP event event counter PID=" << getpid() );
 
  // _______________________ event counting ________________________________
  // Use incident service for registering a EndInputFile handler 
  SmartIF<IIncidentSvc> incsvc(serviceLocator()->service("IncidentSvc"));
  if(!incsvc) {
    ATH_MSG_ERROR( "Error retrieving IncidentSvc" );
    return outwork;
  }
  
  incsvc->addListener(this,"EndInputFile");
  ATH_MSG_DEBUG( "Added self as listener to EndInputFile" );
 
  // _______________________ event sharing ________________________________
  // Use EventSelector as SharedReader (if configured) and enable output streaming
  if (m_useSharedReader) {
    m_evtShare = SmartIF<IEventShare>(m_evtSelector);
    if(!m_evtShare) {
      ATH_MSG_ERROR( "Failed to dyncast event selector to IEventShare" );
      return outwork;
    } else {
      if(!m_evtShare->makeServer(m_nprocs+1).isSuccess()) {
        ATH_MSG_ERROR("Failed to make the event selector a share server");
        return outwork;
      } else {
        ATH_MSG_DEBUG("Successfully made the event selector a share server");
      }
    }
  }
 
  // ________________________ I/O reinit ________________________
  if(!m_ioMgr->io_reinitialize().isSuccess()) {
    ATH_MSG_ERROR( "Failed to reinitialize I/O" );
    return outwork;
  } else {
    ATH_MSG_DEBUG( "Successfully reinitialized I/O" );
  }
 
  // ________________________ Event selector restart ________________________
  if(m_evtSelector) {
    SmartIF<IService> evtSelSvc(m_evtSelector);
    if(!evtSelSvc) {
      ATH_MSG_ERROR( "Failed to dyncast event selector to IService" );
      return outwork;
    }
    if(!evtSelSvc->start().isSuccess()) {
      ATH_MSG_ERROR( "Failed to restart the event selector" );
      return outwork;
    } else {
      ATH_MSG_DEBUG( "Successfully restarted the event selector" );
    }
  }
  // ________________________ chdir ________________________
  if(chdir(counter_rundir.string().c_str())==-1) {
    ATH_MSG_ERROR( "Failed to chdir to " << counter_rundir.string() );
    return outwork;
  }
 
  // Declare success and return
  *(int*)(outwork->data) = 0; // Error code: for now use 0 success, 1 failure
  return outwork;
}

exec_func()

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

overridevirtual

Definition at line 215 of file SharedEvtQueueProvider.cxx.

{
  ATH_MSG_INFO("Exec function in the AthenaMP Event Counter PID=" << getpid());
 
  bool all_ok(true);
 
  int skipEvents(0);
  IEvtSelector::Context* evtContext(nullptr);
 
  // Get SkipEvents property of the event selector
  if(m_evtSelector) {
    SmartIF<IProperty> propertyServer(m_evtSelector);
    if(propertyServer==0) {
      ATH_MSG_ERROR( "Unable to cast event selector to IProperty" );
      all_ok=false;
    }
    else {
      std::string propertyName("SkipEvents");
      IntegerProperty skipEventsProp(std::move(propertyName),skipEvents);
      if(propertyServer->getProperty(&skipEventsProp).isFailure()) {
    ATH_MSG_INFO( "Event Selector does not have SkipEvents property" );
      }
      else {
    skipEvents = skipEventsProp.value();
      }
    }
 
    if(all_ok) {
      StatusCode sc = m_evtSelector->createContext(evtContext);
      if(sc.isFailure()) {
    ATH_MSG_ERROR("Failed to create the event selector context");
    all_ok=false;
      }
      else {
    // advance to nEventsBeforeFork
    for(int i(0); i<m_nEventsBeforeFork;++i) {
      if(!m_evtSelector->next(*evtContext).isSuccess()) {
        ATH_MSG_ERROR("Unexpected error: EventsBeforeFork>EventsInInputFiles");
        all_ok=false;
        break;
      }
    }
      }
    }
  }
 
  if(all_ok) {
    if(m_nEvtRequested!=0) { // Take into account corner case with evtMax=0
      m_nChunkStart = skipEvents+m_nEventsBeforeFork;
      m_nEvtCounted = m_nEventsBeforeFork;
      m_nPositionInChunk = m_nChunkStart;      
      ATH_MSG_VERBOSE("Starting to go through events. Chunk start = " << m_nChunkStart+1);
      
      // Loop through all remaining events 
      while(!m_evtSelector || m_evtSelector->next(*evtContext).isSuccess()) {
    m_nPositionInChunk++;
    m_nEvtCounted++;
    ATH_MSG_VERBOSE("Events Counted " << m_nEvtCounted << ", Position in Chunk " << m_nPositionInChunk);
    if(((m_nPositionInChunk-m_nChunkStart)==m_nChunkSize)
       || (m_nEvtCounted==m_nEvtRequested)) {
      addEventsToQueue();
      m_nChunkStart = m_nPositionInChunk;
      if(m_nEvtCounted==m_nEvtRequested) break;
    }
      }
    }
 
    // We are done. Add -m_nEvtCounted  m_nprocesses-times to the queue
    long newValueForQueue = (long)(-m_nEvtCounted);
    for(int i=0;i<m_nprocesses;++i) {
      while(!m_sharedEventQueue->try_send_basic<long>(newValueForQueue)) {
    usleep(1000);
      }
    }
 
    ATH_MSG_INFO("Done counting events and populating shared queue. Total number of events to be processed: " << std::max(m_nEvtCounted - m_nEventsBeforeFork,0) 
         << ", Event Chunk size in the queue is " << m_nChunkSize);
 
    if(m_useSharedReader && m_evtShare) {
      if(m_evtShare->readEvent(0).isFailure()) {
        ATH_MSG_ERROR("Failed to read " << m_nEvtRequested << " events");
        all_ok=false;
      } else {
        ATH_MSG_DEBUG("readEvent succeeded");
      }
    }
 
    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);
  outwork->data = CxxUtils::xmalloc(sizeof(int));
  *(int*)(outwork->data) = (all_ok?0:1); // Error code: for now use 0 success, 1 failure
  outwork->size = sizeof(int);
 
  // ...
  // (possible) TODO: extend outwork with some error message, which will be eventually
  // reported in the master proces
  // ...
  return outwork;
}

fin_func()

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

overridevirtual

Definition at line 327 of file SharedEvtQueueProvider.cxx.

{
  // Dummy
  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
  outwork->data = CxxUtils::xmalloc(sizeof(int));
  *(int*)(outwork->data) = 0; // Error code: for now use 0 success, 1 failure
  outwork->size = sizeof(int);
  return outwork;
}

generateOutputReport()

AthenaMP::AllWorkerOutputs_ptr SharedEvtQueueProvider::generateOutputReport ( )

overridevirtual

Definition at line 104 of file SharedEvtQueueProvider.cxx.

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

handle()

void SharedEvtQueueProvider::handle ( const Incident & inc )

overridevirtual

Definition at line 337 of file SharedEvtQueueProvider.cxx.

{
  ATH_MSG_DEBUG( "Handling incident" );
 
  const FileIncident* fileInc  = dynamic_cast<const FileIncident*>(&inc);
  if(fileInc == 0) {
    ATH_MSG_WARNING( "Failed to dyn-cast the incident" );
    return;
  } 
 
  if(fileInc->type()=="EndInputFile") {
    ATH_MSG_DEBUG( "End Input File reached!" );
 
    if(m_nChunkStart!=m_nPositionInChunk) {
      addEventsToQueue();
      m_nChunkStart = m_nPositionInChunk;
    }
  }
}

operator=()

SharedEvtQueueProvider & SharedEvtQueueProvider::operator= ( const SharedEvtQueueProvider & )

private

subProcessLogs()

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

overridevirtual

Definition at line 96 of file SharedEvtQueueProvider.cxx.

{
  filenames.clear();
  std::filesystem::path counter_rundir(m_subprocTopDir);
  counter_rundir /= std::filesystem::path(m_subprocDirPrefix);
  filenames.push_back(counter_rundir.string()+std::string("/AthenaMP.log"));
}

Member Data Documentation

m_evtShare

SmartIF<IEventShare> SharedEvtQueueProvider::m_evtShare

private

Definition at line 60 of file SharedEvtQueueProvider.h.

m_nChunkSize

Gaudi::Property<int> SharedEvtQueueProvider::m_nChunkSize {this, "ChunkSize", 1}

private

Definition at line 45 of file SharedEvtQueueProvider.h.

{this, "ChunkSize", 1};

m_nChunkStart

int SharedEvtQueueProvider::m_nChunkStart {0}

private

The beginning of the current chunk.

Definition at line 53 of file SharedEvtQueueProvider.h.

{0};      

m_nEventsBeforeFork

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

private

Definition at line 44 of file SharedEvtQueueProvider.h.

{this, "EventsBeforeFork", 0, "Number of events before forking"};

m_nEvtCounted

int SharedEvtQueueProvider::m_nEvtCounted {0}

private

The number of events this tool has counted itself in the input files.

Definition at line 57 of file SharedEvtQueueProvider.h.

{0};       

m_nEvtRequested

int SharedEvtQueueProvider::m_nEvtRequested {-1}

private

Max event received from AppMgr.

Definition at line 56 of file SharedEvtQueueProvider.h.

{-1};    

m_nPositionInChunk

int SharedEvtQueueProvider::m_nPositionInChunk {0}

private

Position within the current chunk.

Definition at line 54 of file SharedEvtQueueProvider.h.

{0}; 

m_nprocesses

int SharedEvtQueueProvider::m_nprocesses {-1}

private

We use this data member for adding negative numbers at the end of the event queue.

We cannot use m_nprocs for this purpose in order to avoid generating Output File Reports by Shared Queue Providers.

Definition at line 52 of file SharedEvtQueueProvider.h.

{-1};

m_sharedEventQueue

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

private

Definition at line 59 of file SharedEvtQueueProvider.h.

{nullptr};

m_useSharedReader

Gaudi::Property<bool> SharedEvtQueueProvider::m_useSharedReader {this, "UseSharedReader", false, "Use shared reader"}

private

Definition at line 43 of file SharedEvtQueueProvider.h.

{this, "UseSharedReader", false, "Use shared reader"};

---

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

• SharedEvtQueueProvider.h
• SharedEvtQueueProvider.cxx