EventSelectorAthenaPool Class Reference

This class is the EventSelector for event data. More...

#include <EventSelectorAthenaPool.h>

Inheritance diagram for EventSelectorAthenaPool:

Public Member Functions
	EventSelectorAthenaPool (const std::string &name, ISvcLocator *pSvcLocator)
	Standard Service Constructor.
virtual	~EventSelectorAthenaPool ()
	Destructor.
virtual StatusCode	initialize () override
	Required of all Gaudi Services.
virtual StatusCode	start () override
virtual StatusCode	stop () override
virtual StatusCode	finalize () override
virtual StatusCode	createContext (IEvtSelector::Context *&ctxt) const override
	create context
virtual StatusCode	next (IEvtSelector::Context &ctxt) const override
virtual StatusCode	next (IEvtSelector::Context &ctxt, int jump) const override
virtual StatusCode	previous (IEvtSelector::Context &ctxt) const override
virtual StatusCode	previous (IEvtSelector::Context &ctxt, int jump) const override
virtual StatusCode	last (IEvtSelector::Context &ctxt) const override
virtual StatusCode	rewind (IEvtSelector::Context &ctxt) const override
virtual StatusCode	createAddress (const IEvtSelector::Context &ctxt, IOpaqueAddress *&iop) const override
virtual StatusCode	releaseContext (IEvtSelector::Context *&ctxt) const override
virtual StatusCode	resetCriteria (const std::string &criteria, IEvtSelector::Context &ctxt) const override
	Set a selection criteria.
virtual StatusCode	seek (Context &ctxt, int evtnum) const override
	Seek to a given event number.
virtual int	curEvent (const Context &ctxt) const override
	Return the current event number.
virtual int	size (Context &ctxt) const override
	Return the size of the collection.
virtual StatusCode	makeServer (int num) override
	Make this a server.
virtual StatusCode	makeClient (int num) override
	Make this a client.
virtual StatusCode	share (int evtnum) override
	Request to share a given event number.
virtual StatusCode	readEvent (int maxevt) override
	Read the next maxevt events.
virtual StatusCode	io_reinit () override
	Callback method to reinitialize the internal state of the component for I/O purposes (e.g. upon fork(2))
virtual StatusCode	io_finalize () override
	Callback method to finalize the internal state of the component for I/O purposes (e.g. before fork(2))
virtual void	handle (const Incident &incident) override
	Incident service handle listening for BeginProcessing and EndProcessing.

Protected Member Functions
virtual StatusCode	nextHandleFileTransition (IEvtSelector::Context &ctxt) const override
	Handle file transition at the next iteration.
virtual StatusCode	nextWithSkip (IEvtSelector::Context &ctxt) const override
	Go to next event and skip if necessary.
virtual StatusCode	recordAttributeList () const override
	Record AttributeList in StoreGate.
virtual StatusCode	fillAttributeList (coral::AttributeList *attrList, const std::string &suffix, bool copySource) const override
	Fill AttributeList with specific items from the selector and a suffix.
virtual bool	disconnectIfFinished (const SG::SourceID &fid) const override

Private Types
typedef std::mutex	CallMutex

Private Member Functions
StoreGateSvc *	eventStore () const
	Return pointer to active event SG.
StatusCode	reinit () const
	Reinitialize the service when a fork() occurred/was-issued.
std::unique_ptr< PoolCollectionConverter >	getCollectionCnv (bool throwIncidents=false) const
	Return pointer to new PoolCollectionConverter.
int	findEvent (int evtNum) const
	Search for event with number evtNum.
void	fireEndFileIncidents (bool isLastFile) const
	Fires the EndInputFile incident (if there is an open file) at end of selector.
void	inputCollectionsHandler (Gaudi::Details::PropertyBase &)

Private Attributes
EventContextAthenaPool *	m_endIter {}
std::unique_ptr< PoolCollectionConverter > m_poolCollectionConverter	ATLAS_THREAD_SAFE {}
pool::ICollectionCursor *m_headerIterator	ATLAS_THREAD_SAFE {}
Guid m_guid	ATLAS_THREAD_SAFE {}
std::map< SG::SourceID, int > m_activeEventsPerSource	ATLAS_THREAD_SAFE
ServiceHandle< IAthenaPoolCnvSvc >	m_athenaPoolCnvSvc {this, "ConversionService", "AthenaPoolCnvSvc", ""}
ServiceHandle< IIncidentSvc >	m_incidentSvc {this, "IncidentSvc", "IncidentSvc", ""}
Gaudi::Property< bool >	m_isSecondary {this, "IsSecondary", false, ""}
	IsSecondary, know if this is an instance of secondary event selector.
Gaudi::Property< bool >	m_processMetadata {this, "ProcessMetadata", true, ""}
	ProcessMetadata, switch on firing of FileIncidents which will trigger processing of metadata: default = true.
Gaudi::Property< std::string >	m_collectionType {this, "CollectionType", "ImplicitCollection", ""}
	CollectionType, type of the collection: default = "ImplicitCollection".
std::string	m_attrListKey {"Input"}
	AttributeList SG key.
Gaudi::Property< std::vector< std::string > >	m_inputCollectionsProp {this, "InputCollections", {}, ""}
	InputCollections, vector with names of the input collections.
std::vector< std::string >::const_iterator m_inputCollectionsIterator	ATLAS_THREAD_SAFE
bool m_inputCollectionsChanged	ATLAS_THREAD_SAFE
	flag to notify the EvSel that the inputs were changed and reinit() needs to be called ASAP
Gaudi::Property< bool >	m_keepInputFilesOpen {this, "KeepInputFilesOpen", false, ""}
	KeepInputFilesOpen, boolean flag to keep files open after PoolCollection reaches end: default = false.
ToolHandleArray< IAthenaSelectorTool > m_helperTools	ATLAS_THREAD_SAFE {this, "HelperTools", {}, ""}
	HelperTools, vector of names of AlgTools that are executed by the EventSelector.
ToolHandle< IAthenaSelectorTool >	m_counterTool {this, "CounterTool", "", ""}
ToolHandle< IAthenaIPCTool >	m_eventStreamingTool {this, "SharedMemoryTool", "", ""}
Gaudi::Property< int >	m_makeStreamingToolClient {this, "MakeStreamingToolClient",0}
	Make this instance a Streaming Client during first iteration automatically.
Gaudi::CheckedProperty< uint32_t >	m_runNo {this, "RunNumber", 0, ""}
	The following are included for compatibility with McEventSelector and are not really used.
Gaudi::CheckedProperty< uint32_t >	m_oldRunNo {this, "OldRunNumber", 0, ""}
Gaudi::Property< bool >	m_overrideRunNumber {this, "OverrideRunNumber", false, ""}
Gaudi::Property< bool >	m_overrideRunNumberFromInput {this, "OverrideRunNumberFromInput", false, ""}
Gaudi::CheckedProperty< uint64_t >	m_firstEventNo {this, "FirstEvent", 1, ""}
Gaudi::CheckedProperty< uint64_t >	m_eventsPerRun {this, "EventsPerRun", 1000000, ""}
Gaudi::CheckedProperty< uint32_t >	m_firstLBNo {this, "FirstLB", 0, ""}
Gaudi::CheckedProperty< uint32_t >	m_eventsPerLB {this, "EventsPerLB", 1000, ""}
Gaudi::CheckedProperty< uint32_t >	m_initTimeStamp {this, "InitialTimeStamp", 0, ""}
Gaudi::Property< uint32_t >	m_timeStampInterval {this, "TimeStampInterval", 0, ""}
std::atomic_long	m_curCollection {}
std::vector< int > m_numEvt	ATLAS_THREAD_SAFE
std::vector< int > m_firstEvt	ATLAS_THREAD_SAFE
Gaudi::Property< int >	m_skipEvents {this, "SkipEvents", 0, ""}
	SkipEvents, numbers of events to skip: default = 0.
Gaudi::Property< std::vector< long > >	m_skipEventSequenceProp {this, "SkipEventSequence", {}, ""}
Gaudi::Property< std::string >	m_skipEventRangesProp {this, "SkipEventRanges", {}, ""}
	Skip Events - comma separated list of event to skip, ranges with '-': <start> - <end>
std::vector< std::pair< long, long > > m_skipEventRanges	ATLAS_THREAD_SAFE
std::atomic_int	m_evtCount {}
std::atomic_bool	m_firedIncident {}
CallMutex	m_callLock
SG::SlotSpecificObj< SG::SourceID >	m_sourceID

Friends
class	DoubleEventSelectorAthenaPool
	make the DoubleEventSelectorAthenaPool a friend so it can access the internal EventSelectorAthenaPool methods and members

Detailed Description

This class is the EventSelector for event data.

Definition at line 48 of file EventSelectorAthenaPool.h.

Member Typedef Documentation

CallMutex

typedef std::mutex EventSelectorAthenaPool::CallMutex

private

Definition at line 238 of file EventSelectorAthenaPool.h.

Constructor & Destructor Documentation

EventSelectorAthenaPool()

EventSelectorAthenaPool::EventSelectorAthenaPool	(	const std::string &	name,
		ISvcLocator *	pSvcLocator )

Standard Service Constructor.

Definition at line 56 of file EventSelectorAthenaPool.cxx.

                                                                                                :
    base_class(name, pSvcLocator)
{
   // TODO: validate if those are even used
   m_runNo.verifier().setLower(0);
   m_oldRunNo.verifier().setLower(0);
   m_eventsPerRun.verifier().setLower(0);
   m_firstEventNo.verifier().setLower(1);
   m_firstLBNo.verifier().setLower(0);
   m_eventsPerLB.verifier().setLower(0);
   m_initTimeStamp.verifier().setLower(0);
 
   m_inputCollectionsProp.declareUpdateHandler(&EventSelectorAthenaPool::inputCollectionsHandler, this);
   m_inputCollectionsChanged = false;
}

~EventSelectorAthenaPool()

EventSelectorAthenaPool::~EventSelectorAthenaPool ( )

virtual

Destructor.

Definition at line 78 of file EventSelectorAthenaPool.cxx.

                                                  {
}

Member Function Documentation

createAddress()

StatusCode EventSelectorAthenaPool::createAddress ( const IEvtSelector::Context & ctxt, IOpaqueAddress *& iop ) const

overridevirtual

Parameters

ctxt	[IN] current event context.
iop	[OUT] pointer to IOpaqueAddress reference of selection context.

Definition at line 688 of file EventSelectorAthenaPool.cxx.

                                    {
   std::string tokenStr;
   SG::ReadHandle<AthenaAttributeList> attrList(m_attrListKey, eventStore()->name());
   if (attrList.isValid()) {
      try {
         tokenStr = (*attrList)["eventRef"].data<std::string>();
         ATH_MSG_DEBUG("found AthenaAttribute, name = eventRef = " << tokenStr);
      } catch (std::exception &e) {
         ATH_MSG_ERROR(e.what());
         return StatusCode::FAILURE;
      }
   } else {
      ATH_MSG_WARNING("Cannot find AthenaAttribute, key = " << m_attrListKey);
      tokenStr = m_headerIterator->eventRef().toString();
   }
   auto token = std::make_unique<Token>();
   token->fromString(tokenStr);
   iop = new TokenAddress(pool::POOL_StorageType.type(), ClassID_traits<DataHeader>::ID(), "", "EventSelector", IPoolSvc::kInputStream, std::move(token));
   return StatusCode::SUCCESS;
}

createContext()

StatusCode EventSelectorAthenaPool::createContext ( IEvtSelector::Context *& ctxt ) const

overridevirtual

create context

Definition at line 386 of file EventSelectorAthenaPool.cxx.

                                                                                  {
   ctxt = new EventContextAthenaPool(this);
   return StatusCode::SUCCESS;
}

curEvent()

int EventSelectorAthenaPool::curEvent ( const Context & ctxt ) const

overridevirtual

Return the current event number.

Parameters

ctxt	[IN/OUT] current event context.

Definition at line 781 of file EventSelectorAthenaPool.cxx.

                                                             {
  return(m_evtCount);
}

disconnectIfFinished()

bool EventSelectorAthenaPool::disconnectIfFinished ( const SG::SourceID & fid ) const

overrideprotectedvirtual

Definition at line 1143 of file EventSelectorAthenaPool.cxx.

{
   if( m_eventStreamingTool.empty() && m_activeEventsPerSource.find(fid) != m_activeEventsPerSource.end() 
           && m_activeEventsPerSource[fid] <= 0 && m_guid != fid ) {
      // Explicitly disconnect file corresponding to old FID to release memory
      if( !m_keepInputFilesOpen.value() ) {
         // Assume that the end of collection file indicates the end of payload file.
         if (m_processMetadata.value()) {
            FileIncident endInputFileIncident(name(), "EndInputFile", "FID:" + fid, fid);
            m_incidentSvc->fireIncident(endInputFileIncident);
         }
         ATH_MSG_INFO("Disconnecting input sourceID: " << fid );
         m_athenaPoolCnvSvc->getPoolSvc()->disconnectDb("FID:" + fid, IPoolSvc::kInputStream).ignore();
         m_activeEventsPerSource.erase( fid );
         return true;
      }
   }
   return false;
}

eventStore()

StoreGateSvc * EventSelectorAthenaPool::eventStore ( ) const

private

Return pointer to active event SG.

Definition at line 81 of file EventSelectorAthenaPool.cxx.

                                                        {
   return StoreGateSvc::currentStoreGate();
}

fillAttributeList()

StatusCode EventSelectorAthenaPool::fillAttributeList ( coral::AttributeList * attrList, const std::string & suffix, bool copySource ) const

overrideprotectedvirtual

Fill AttributeList with specific items from the selector and a suffix.

Definition at line 1009 of file EventSelectorAthenaPool.cxx.

{
   const auto& row = m_headerIterator->currentRow();
   attrList->extend( row.tokenName() + suffix, "string" );
   (*attrList)[ row.tokenName() + suffix ].data<std::string>() = row.token().toString();
   ATH_MSG_DEBUG("record AthenaAttribute, name = " << row.tokenName() + suffix << " = " << row.token().toString() << ".");
 
   std::string eventRef = "eventRef";
   if (m_isSecondary.value()) {
      eventRef.append(suffix);
   }
   attrList->extend(eventRef, "string");
   (*attrList)[eventRef].data<std::string>() = m_headerIterator->eventRef().toString();
   ATH_MSG_DEBUG("record AthenaAttribute, name = " + eventRef + " = " <<  m_headerIterator->eventRef().toString() << ".");
 
   if (copySource) {
      const coral::AttributeList& sourceAttrList = m_headerIterator->currentRow().attributeList();
      for (const auto &attr : sourceAttrList) {
         attrList->extend(attr.specification().name() + suffix, attr.specification().type());
         (*attrList)[attr.specification().name() + suffix] = attr;
      }
   }
 
   return StatusCode::SUCCESS;
}

finalize()

StatusCode EventSelectorAthenaPool::finalize ( )

overridevirtual

Definition at line 365 of file EventSelectorAthenaPool.cxx.

                                             {
   if (m_eventStreamingTool.empty() || !m_eventStreamingTool->isClient()) {
      if (!m_counterTool.empty() && !m_counterTool->preFinalize().isSuccess()) {
         ATH_MSG_WARNING("Failed to preFinalize() CounterTool");
      }
      for (auto& tool : m_helperTools) {
         if (!tool->preFinalize().isSuccess()) {
            ATH_MSG_WARNING("Failed to preFinalize() " << tool->name());
         }
      }
   }
   delete m_endIter;   m_endIter   = nullptr;
   m_headerIterator = nullptr;
   if (m_poolCollectionConverter) {
     m_poolCollectionConverter.reset();
   }
   // Finalize the Service base class.
   return ::AthService::finalize();
}

findEvent()

int EventSelectorAthenaPool::findEvent ( int evtNum ) const

private

Search for event with number evtNum.

Definition at line 789 of file EventSelectorAthenaPool.cxx.

                                                       {
   for (std::size_t i = 0, imax = m_numEvt.size(); i < imax; i++) {
      if (m_numEvt[i] == -1) {
         PoolCollectionConverter pcc(m_collectionType.value(),
             m_inputCollectionsProp.value()[i],
             IPoolSvc::kInputStream,
             m_athenaPoolCnvSvc->getPoolSvc());
         if (!pcc.initialize().isSuccess()) {
            break;
         }
         int collection_size = 0;
         if (pcc.isValid()) {
            pool::ICollectionCursor* hi = &pcc.selectAll();
            collection_size = hi->size();
         }
         if (i > 0) {
            m_firstEvt[i] = m_firstEvt[i - 1] + m_numEvt[i - 1];
         } else {
            m_firstEvt[i] = 0;
         }
         m_numEvt[i] = collection_size;
      }
      if (evtNum >= m_firstEvt[i] && evtNum < m_firstEvt[i] + m_numEvt[i]) {
         return(i);
      }
   }
   return(-1);
}

fireEndFileIncidents()

void EventSelectorAthenaPool::fireEndFileIncidents ( bool isLastFile ) const

private

Fires the EndInputFile incident (if there is an open file) at end of selector.

Definition at line 348 of file EventSelectorAthenaPool.cxx.

                                                                        {
   if (m_processMetadata.value()) {
      if (m_evtCount >= 0) {
         // Assume that the end of collection file indicates the end of payload file.
         if (m_guid != Guid::null()) {
            // Fire EndInputFile incident
            FileIncident endInputFileIncident(name(), "EndInputFile", "FID:" + m_guid.toString(), m_guid.toString());
            m_incidentSvc->fireIncident(endInputFileIncident);
         }
      }
      if (isLastFile && m_firedIncident) {
         m_firedIncident = false;
      }
   }
}

getCollectionCnv()

std::unique_ptr< PoolCollectionConverter > EventSelectorAthenaPool::getCollectionCnv ( bool throwIncidents = false ) const

private

Return pointer to new PoolCollectionConverter.

Definition at line 951 of file EventSelectorAthenaPool.cxx.

                                                                   {
   while (m_inputCollectionsIterator != m_inputCollectionsProp.value().end()) {
      if (m_curCollection != 0) {
         m_numEvt[m_curCollection] = m_evtCount - m_firstEvt[m_curCollection];
         m_curCollection++;
         m_firstEvt[m_curCollection] = m_evtCount;
      }
      ATH_MSG_DEBUG("Try item: \"" << *m_inputCollectionsIterator << "\" from the collection list.");
      auto pCollCnv = std::make_unique<PoolCollectionConverter>(m_collectionType.value(),
          *m_inputCollectionsIterator,
          IPoolSvc::kInputStream,
          m_athenaPoolCnvSvc->getPoolSvc());
      StatusCode status = pCollCnv->initialize();
      if (!status.isSuccess()) {
         // Close previous collection.
         pCollCnv.reset();
         if (!status.isRecoverable()) {
            ATH_MSG_ERROR("Unable to initialize PoolCollectionConverter.");
            throw GaudiException("Unable to read: " + *m_inputCollectionsIterator, name(), StatusCode::FAILURE);
         } else {
            ATH_MSG_ERROR("Unable to open: " << *m_inputCollectionsIterator);
            throw GaudiException("Unable to open: " + *m_inputCollectionsIterator, name(), StatusCode::FAILURE);
         }
      } else {
         if (!pCollCnv->isValid().isSuccess()) {
            pCollCnv.reset();
            ATH_MSG_DEBUG("No events found in: " << *m_inputCollectionsIterator << " skipped!!!");
            if (throwIncidents && m_processMetadata.value()) {
               FileIncident beginInputFileIncident(name(), "BeginInputFile", *m_inputCollectionsIterator);
               m_incidentSvc->fireIncident(beginInputFileIncident);
               FileIncident endInputFileIncident(name(), "EndInputFile", "eventless " + *m_inputCollectionsIterator);
               m_incidentSvc->fireIncident(endInputFileIncident);
            }
            m_athenaPoolCnvSvc->getPoolSvc()->disconnectDb(*m_inputCollectionsIterator).ignore();
            ++m_inputCollectionsIterator;
         } else {
            return(pCollCnv);
         }
      }
   }
   return(nullptr);
}

handle()

void EventSelectorAthenaPool::handle ( const Incident & incident )

overridevirtual

Incident service handle listening for BeginProcessing and EndProcessing.

Reimplemented in DoubleEventSelectorAthenaPool.

Definition at line 1102 of file EventSelectorAthenaPool.cxx.

{
   SG::SourceID fid;
   if (inc.type() == IncidentType::BeginProcessing) {
     if ( Atlas::hasExtendedEventContext(inc.context()) ) {
       fid = Atlas::getExtendedEventContext(inc.context()).proxy()->sourceID();
     }
     *m_sourceID.get(inc.context()) = fid;
   }
   else {
     fid = *m_sourceID.get(inc.context());
   }
 
   if( fid.empty() ) {
      ATH_MSG_WARNING("could not read event source ID from incident event context");
      return;
   }
   if( m_activeEventsPerSource.find( fid ) == m_activeEventsPerSource.end()) {
      ATH_MSG_DEBUG("Incident handler ignoring unknown input FID: " << fid );
      return;
   }
   ATH_MSG_DEBUG("**  MN Incident handler " << inc.type() << " Event source ID=" << fid );
   if( inc.type() == IncidentType::BeginProcessing ) {
      // increment the events-per-file counter for FID
      m_activeEventsPerSource[fid]++;
   } else if( inc.type() == IncidentType::EndProcessing ) {
      m_activeEventsPerSource[fid]--;
      disconnectIfFinished( fid );
      *m_sourceID.get(inc.context()) = "";
   }
   if( msgLvl(MSG::DEBUG) ) {
      for( auto& source: m_activeEventsPerSource )
         msg(MSG::DEBUG) << "SourceID: " << source.first << " active events: " << source.second << endmsg;
   }
}

initialize()

StatusCode EventSelectorAthenaPool::initialize ( )

overridevirtual

Required of all Gaudi Services.

Reimplemented in DoubleEventSelectorAthenaPool.

Definition at line 85 of file EventSelectorAthenaPool.cxx.

                                               {
 
   m_autoRetrieveTools = false;
   m_checkToolDeps = false;
  
   if (m_isSecondary.value()) {
      ATH_MSG_DEBUG("Initializing secondary event selector " << name());
   } else {
      ATH_MSG_DEBUG("Initializing " << name());
   }
 
   ATH_CHECK(::AthService::initialize());
   // Check for input collection
   if (m_inputCollectionsProp.value().empty()) {
      ATH_MSG_FATAL("Use the property: EventSelector.InputCollections = "
              << "[ \"<collectionName>\" ] (list of collections)");
      return StatusCode::FAILURE;
   }
   boost::char_separator<char> sep_coma(","), sep_hyph("-");
   boost::tokenizer  ranges(m_skipEventRangesProp.value(), sep_coma);
   for( const std::string& r: ranges ) {
      boost::tokenizer  fromto(r, sep_hyph);
      auto from_iter = fromto.begin();
      long from = std::stol(*from_iter);
      long to = from;
      if( ++from_iter != fromto.end() ) {
         to = std::stol(*from_iter);
      }
      m_skipEventRanges.emplace_back(from, to);
   }
 
   for( auto v : m_skipEventSequenceProp.value() ) {
      m_skipEventRanges.emplace_back(v, v);
   }
   std::sort(m_skipEventRanges.begin(), m_skipEventRanges.end());
   if( msgLvl(MSG::DEBUG) ) {
      std::string skip_ranges_str;
      for( const auto& [first, second] : m_skipEventRanges ) {
         if( !skip_ranges_str.empty() ) skip_ranges_str += ", ";
         skip_ranges_str += std::to_string(first);
         if( first != second) skip_ranges_str += std::format("-{}", second);
      }
      if( !skip_ranges_str.empty() )
         ATH_MSG_DEBUG("Events to skip: " << skip_ranges_str);
   }
   // Get IncidentSvc
   ATH_CHECK(m_incidentSvc.retrieve());
   // Listen to the Event Processing incidents
   if (m_eventStreamingTool.empty()) {
      m_incidentSvc->addListener(this, IncidentType::BeginProcessing, 0);
      m_incidentSvc->addListener(this, IncidentType::EndProcessing, 0);
   }
 
   // Get AthenaPoolCnvSvc
   ATH_CHECK(m_athenaPoolCnvSvc.retrieve());
   // Get CounterTool (if configured)
   if (!m_counterTool.empty()) {
      ATH_CHECK(m_counterTool.retrieve());
   }
   // Get HelperTools
   ATH_CHECK(m_helperTools.retrieve());
   // Get SharedMemoryTool (if configured)
   if (!m_eventStreamingTool.empty() && !m_eventStreamingTool.retrieve().isSuccess()) {
      ATH_MSG_FATAL("Cannot get " << m_eventStreamingTool.typeAndName() << "");
      return StatusCode::FAILURE;
   } else if (m_makeStreamingToolClient.value() == -1) {
      std::string dummyStr;
      ATH_CHECK(m_eventStreamingTool->makeClient(m_makeStreamingToolClient.value(), dummyStr));
   }
 
   // Ensure the xAODCnvSvc is listed in the EventPersistencySvc
   ServiceHandle<IProperty> epSvc("EventPersistencySvc", name());
   std::vector<std::string> propVal;
   ATH_CHECK(Gaudi::Parsers::parse(propVal , epSvc->getProperty("CnvServices").toString()));
   bool foundCnvSvc = false;
   for (const auto& property : propVal) {
      if (property == m_athenaPoolCnvSvc.type()) { foundCnvSvc = true; }
   }
   if (!foundCnvSvc) {
      propVal.push_back(m_athenaPoolCnvSvc.type());
      if (!epSvc->setProperty("CnvServices", Gaudi::Utils::toString(propVal)).isSuccess()) {
         ATH_MSG_FATAL("Cannot set EventPersistencySvc Property for CnvServices");
         return StatusCode::FAILURE;
      }
   }
 
   // Register this service for 'I/O' events
   ServiceHandle<IIoComponentMgr> iomgr("IoComponentMgr", name());
   ATH_CHECK(iomgr.retrieve());
   ATH_CHECK(iomgr->io_register(this));
   // Register input file's names with the I/O manager
   const std::vector<std::string>& incol = m_inputCollectionsProp.value();
   bool allGood = true;
   for (const auto& inputCollection : incol) {
      if (!iomgr->io_register(this, IIoComponentMgr::IoMode::READ, inputCollection, inputCollection).isSuccess()) {
         ATH_MSG_FATAL("could not register [" << inputCollection << "] for output !");
         allGood = false;
      } else {
         ATH_MSG_VERBOSE("io_register[" << this->name() << "](" << inputCollection << ") [ok]");
      }
   }
   if (!allGood) {
      return StatusCode::FAILURE;
   }
 
   // Connect to PersistencySvc
   if (!m_athenaPoolCnvSvc->getPoolSvc()->connect(pool::ITransaction::READ, IPoolSvc::kInputStream).isSuccess()) {
      ATH_MSG_FATAL("Cannot connect to POOL PersistencySvc.");
      return StatusCode::FAILURE;
   }
   // Jump to reinit() to execute common init/reinit actions
   m_guid = Guid::null();
   return reinit();
}

inputCollectionsHandler()

void EventSelectorAthenaPool::inputCollectionsHandler ( Gaudi::Details::PropertyBase & )

private

Definition at line 72 of file EventSelectorAthenaPool.cxx.

                                                                               {
   if (this->FSMState() != Gaudi::StateMachine::OFFLINE) {
      m_inputCollectionsChanged = true;
   }
}

io_finalize()

StatusCode EventSelectorAthenaPool::io_finalize ( )

overridevirtual

Callback method to finalize the internal state of the component for I/O purposes (e.g. before fork(2))

Definition at line 1087 of file EventSelectorAthenaPool.cxx.

                                                {
   ATH_MSG_INFO("I/O finalization...");
   if (m_poolCollectionConverter) {
      m_poolCollectionConverter->disconnectDb().ignore();
      m_poolCollectionConverter.reset();
   }
   return StatusCode::SUCCESS;
}

io_reinit()

StatusCode EventSelectorAthenaPool::io_reinit ( )

overridevirtual

Callback method to reinitialize the internal state of the component for I/O purposes (e.g. upon fork(2))

Definition at line 1035 of file EventSelectorAthenaPool.cxx.

                                              {
   ATH_MSG_INFO("I/O reinitialization...");
   if (m_poolCollectionConverter) {
      m_poolCollectionConverter->disconnectDb().ignore();
      m_poolCollectionConverter.reset();
   }
   m_headerIterator = nullptr;
   ServiceHandle<IIoComponentMgr> iomgr("IoComponentMgr", name());
   if (!iomgr.retrieve().isSuccess()) {
      ATH_MSG_FATAL("Could not retrieve IoComponentMgr !");
      return StatusCode::FAILURE;
   }
   if (!iomgr->io_hasitem(this)) {
      ATH_MSG_FATAL("IoComponentMgr does not know about myself !");
      return StatusCode::FAILURE;
   }
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      m_guid = Guid::null();
      return(this->reinit());
   }
   std::vector<std::string> inputCollections = m_inputCollectionsProp.value();
   std::set<std::size_t> updatedIndexes;
   for (std::size_t i = 0, imax = m_inputCollectionsProp.value().size(); i < imax; i++) {
      if (updatedIndexes.find(i) != updatedIndexes.end()) continue;
      std::string savedName = inputCollections[i];
      std::string &fname = inputCollections[i];
      if (!iomgr->io_contains(this, fname)) {
         ATH_MSG_ERROR("IoComponentMgr does not know about [" << fname << "] !");
         return StatusCode::FAILURE;
      }
      if (!iomgr->io_retrieve(this, fname).isSuccess()) {
         ATH_MSG_FATAL("Could not retrieve new value for [" << fname << "] !");
         return StatusCode::FAILURE;
      }
      if (savedName != fname) {
         ATH_MSG_DEBUG("Mapping value for [" << savedName << "] to [" << fname << "]");
         m_athenaPoolCnvSvc->getPoolSvc()->renamePfn(savedName, fname);
      }
      updatedIndexes.insert(i);
      for (std::size_t j = i + 1; j < imax; j++) {
         if (inputCollections[j] == savedName) {
            inputCollections[j] = fname;
            updatedIndexes.insert(j);
         }
      }
   }
   // all good... copy over.
   m_inputCollectionsProp = inputCollections;
   m_guid = Guid::null();
   return reinit();
}

last()

StatusCode EventSelectorAthenaPool::last ( IEvtSelector::Context & ctxt ) const

overridevirtual

Parameters

ctxt	[IN/OUT] current event context is interated to last event.

Definition at line 674 of file EventSelectorAthenaPool.cxx.

                                                                        {
   if (ctxt.identifier() == m_endIter->identifier()) {
      ATH_MSG_DEBUG("last(): Last event in InputStream.");
      return StatusCode::SUCCESS;
   }
   return StatusCode::FAILURE;
}

makeClient()

StatusCode EventSelectorAthenaPool::makeClient ( int num )

overridevirtual

Make this a client.

Definition at line 844 of file EventSelectorAthenaPool.cxx.

                                                      {
   IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
   if (ds == nullptr) {
      ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
      return StatusCode::FAILURE;
   }
   if (ds->makeClient(num + 1).isFailure()) {
      ATH_MSG_ERROR("Failed to switch AthenaPoolCnvSvc to DataStreaming client");
      return StatusCode::FAILURE;
   }
   if (m_eventStreamingTool.empty()) {
      return StatusCode::SUCCESS;
   }
   ATH_MSG_DEBUG("makeClient: " << m_eventStreamingTool << " = " << num);
   std::string dummyStr;
   return(m_eventStreamingTool->makeClient(0, dummyStr));
}

makeServer()

StatusCode EventSelectorAthenaPool::makeServer ( int num )

overridevirtual

Make this a server.

Definition at line 819 of file EventSelectorAthenaPool.cxx.

                                                      {
   IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
   if (ds == nullptr) {
      ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
      return StatusCode::FAILURE;
   }
   if (num < 0) {
      if (ds->makeServer(num - 1).isFailure()) {
         ATH_MSG_ERROR("Failed to switch AthenaPoolCnvSvc to output DataStreaming server");
      }
      return StatusCode::SUCCESS;
   }
   if (ds->makeServer(num + 1).isFailure()) {
      ATH_MSG_ERROR("Failed to switch AthenaPoolCnvSvc to input DataStreaming server");
      return StatusCode::FAILURE;
   }
   if (m_eventStreamingTool.empty()) {
      return StatusCode::SUCCESS;
   }
   m_processMetadata = false;
   ATH_MSG_DEBUG("makeServer: " << m_eventStreamingTool << " = " << num);
   return(m_eventStreamingTool->makeServer(1, ""));
}

next() [1/2]

StatusCode EventSelectorAthenaPool::next ( IEvtSelector::Context & ctxt ) const

overridevirtual

Parameters

ctxt	[IN/OUT] current event context is interated to next event.

Reimplemented in DoubleEventSelectorAthenaPool.

Definition at line 391 of file EventSelectorAthenaPool.cxx.

                                                                        {
   std::lock_guard<CallMutex> lockGuard(m_callLock);
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      if (m_makeStreamingToolClient.value() == -1) {
         StatusCode sc = m_eventStreamingTool->lockEvent(m_evtCount);
         while (sc.isRecoverable()) {
            usleep(1000);
            sc = m_eventStreamingTool->lockEvent(m_evtCount);
         }
      }
      // Increase event count
      ++m_evtCount;
      void* tokenStr = nullptr;
      unsigned int status = 0;
      StatusCode sc = m_eventStreamingTool->getLockedEvent(&tokenStr, status);
      if (sc.isRecoverable()) {
         delete [] (char*)tokenStr; tokenStr = nullptr;
         // Return end iterator
         ctxt = *m_endIter;
         // This is not a real failure but a Gaudi way of handling "end of job"
         return StatusCode::FAILURE;
      }
      if (sc.isFailure()) {
         ATH_MSG_FATAL("Cannot get NextEvent from AthenaSharedMemoryTool");
         delete [] (char*)tokenStr; tokenStr = nullptr;
         return StatusCode::FAILURE;
      }
      if (!eventStore()->clearStore().isSuccess()) {
         ATH_MSG_WARNING("Cannot clear Store");
      }
      std::unique_ptr<AthenaAttributeList> athAttrList(new AthenaAttributeList());
      athAttrList->extend("eventRef", "string");
      (*athAttrList)["eventRef"].data<std::string>() = std::string((char*)tokenStr);
      SG::WriteHandle<AthenaAttributeList> wh(m_attrListKey, eventStore()->name());
      if (!wh.record(std::move(athAttrList)).isSuccess()) {
         delete [] (char*)tokenStr; tokenStr = nullptr;
         ATH_MSG_ERROR("Cannot record AttributeList to StoreGate " << StoreID::storeName(eventStore()->storeID()));
         return StatusCode::FAILURE;
      }
      Token token;
      token.fromString(std::string((char*)tokenStr));
      delete [] (char*)tokenStr; tokenStr = nullptr;
      Guid guid = token.dbID();
      if (guid != m_guid && m_processMetadata.value()) {
         if (m_evtCount >= 0 && m_guid != Guid::null()) {
            // Fire EndInputFile incident
            FileIncident endInputFileIncident(name(), "EndInputFile", "FID:" + m_guid.toString(), m_guid.toString());
            m_incidentSvc->fireIncident(endInputFileIncident);
         }
         m_guid = guid;
         FileIncident beginInputFileIncident(name(), "BeginInputFile", "FID:" + m_guid.toString(), m_guid.toString());
         m_incidentSvc->fireIncident(beginInputFileIncident);
      }
      return StatusCode::SUCCESS;
   }
   for (const auto& tool : m_helperTools) {
      if (!tool->preNext().isSuccess()) {
         ATH_MSG_WARNING("Failed to preNext() " << tool->name());
      }
   }
   for (;;) {
      // Handle possible file transition
      StatusCode sc = nextHandleFileTransition(ctxt);
      if (sc.isRecoverable()) {
        continue; // handles empty files
      }
      if (sc.isFailure()) {
         return StatusCode::FAILURE;
      }
      // Increase event count
      ++m_evtCount;
      if (!m_counterTool.empty() && !m_counterTool->preNext().isSuccess()) {
         ATH_MSG_WARNING("Failed to preNext() CounterTool.");
      }
      if( m_evtCount > m_skipEvents
          && (m_skipEventRanges.empty() || m_evtCount < m_skipEventRanges.front().first))
      {
         if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isServer()) {
            IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
            if (ds == nullptr) {
               ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
               return StatusCode::FAILURE;
            }
            std::string token = m_headerIterator->eventRef().toString();
            StatusCode sc;
            while ( (sc = putEvent_ST(*m_eventStreamingTool,
                                      m_evtCount - 1, token.c_str(),
                                      token.length() + 1, 0)).isRecoverable() ) {
               while (ds->readData().isSuccess()) {
                  ATH_MSG_VERBOSE("Called last readData, while putting next event in next()");
               }
               // Nothing to do right now, trigger alternative (e.g. caching) here? Currently just fast loop.
            }
            if (!sc.isSuccess()) {
               ATH_MSG_ERROR("Cannot put Event " << m_evtCount - 1 << " to AthenaSharedMemoryTool");
               return StatusCode::FAILURE;
            }
         } else {
            if (!m_isSecondary.value()) {
               if (!eventStore()->clearStore().isSuccess()) {
                  ATH_MSG_WARNING("Cannot clear Store");
               }
               if (!recordAttributeList().isSuccess()) {
                  ATH_MSG_ERROR("Failed to record AttributeList.");
                  return StatusCode::FAILURE;
               }
            }
         }
         StatusCode status = StatusCode::SUCCESS;
         for (const auto& tool : m_helperTools) {
            StatusCode toolStatus = tool->postNext();
            if (toolStatus.isRecoverable()) {
               ATH_MSG_INFO("Request skipping event from: " << tool->name());
               if (status.isSuccess()) {
                  status = StatusCode::RECOVERABLE;
               }
            } else if (toolStatus.isFailure()) {
               ATH_MSG_WARNING("Failed to postNext() " << tool->name());
               status = StatusCode::FAILURE;
            }
         }
         if (status.isRecoverable()) {
            ATH_MSG_INFO("skipping event " << m_evtCount);
         } else if (status.isFailure()) {
            ATH_MSG_WARNING("Failed to postNext() HelperTool.");
         } else {
            if (!m_counterTool.empty() && !m_counterTool->postNext().isSuccess()) {
               ATH_MSG_WARNING("Failed to postNext() CounterTool.");
            }
            break;
         }
      } else {
         while( !m_skipEventRanges.empty() && m_evtCount >= m_skipEventRanges.front().second ) {
            m_skipEventRanges.erase(m_skipEventRanges.begin());
         }
         ATH_MSG_INFO("skipping event " << m_evtCount);
      }
   }
   return StatusCode::SUCCESS;
}

next() [2/2]

StatusCode EventSelectorAthenaPool::next ( IEvtSelector::Context & ctxt, int jump ) const

overridevirtual

Parameters

ctxt	[IN/OUT] current event context is interated to next event.
jump	[IN] number of events to jump (currently not supported).

Reimplemented in DoubleEventSelectorAthenaPool.

Definition at line 532 of file EventSelectorAthenaPool.cxx.

                                                                                  {
   if (jump > 0) {
      for (int i = 0; i < jump; i++) {
         ATH_CHECK(next(ctxt));
      }
      return StatusCode::SUCCESS;
   }
   return StatusCode::FAILURE;
}

nextHandleFileTransition()

StatusCode EventSelectorAthenaPool::nextHandleFileTransition ( IEvtSelector::Context & ctxt ) const

overrideprotectedvirtual

Handle file transition at the next iteration.

Definition at line 542 of file EventSelectorAthenaPool.cxx.

{
   if( m_inputCollectionsChanged ) {
      StatusCode rc = reinit();
      if( rc != StatusCode::SUCCESS ) return rc;
   }
   else {   // advance to the next (not needed after reinit)
      // Check if we're at the end of file
      if (m_headerIterator == nullptr || m_headerIterator->next() == 0) {
         m_headerIterator = nullptr;
         // Close previous collection.
         m_poolCollectionConverter.reset();
 
         // zero the current DB ID (m_guid) before disconnect() to indicate it is no longer in use
         const SG::SourceID old_guid = m_guid.toString();
         m_guid = Guid::null();
         disconnectIfFinished( old_guid );
 
         // check if somebody updated Inputs in the EOF incident (like VP1 does)
         if( m_inputCollectionsChanged ) {
            StatusCode rc = reinit();
            if( rc != StatusCode::SUCCESS ) return rc;
         } else {
            // Open next file from inputCollections list.
            ++m_inputCollectionsIterator;
            // Create PoolCollectionConverter for input file
            m_poolCollectionConverter = getCollectionCnv(true);
            if (!m_poolCollectionConverter) {
               // Return end iterator
               ctxt = *m_endIter;
               // This is not a real failure but a Gaudi way of handling "end of job"
               return StatusCode::FAILURE;
            }
            // Get DataHeader iterator
            m_headerIterator = &m_poolCollectionConverter->selectAll();
 
            // Return RECOVERABLE to mark we should still continue
            return StatusCode::RECOVERABLE;
         }
      }
   }
   const Token& headRef = m_headerIterator->eventRef();
   const Guid guid = headRef.dbID();
   const int tech = headRef.technology();
   ATH_MSG_VERBOSE("next(): DataHeder Token=" << headRef.toString() );
 
   if (guid != m_guid) {
      // we are starting reading from a new DB. Check if the old one needs to be retired
      if (m_guid != Guid::null()) {
         // zero the current DB ID (m_guid) before trying disconnect() to indicate it is no longer in use
         const SG::SourceID old_guid = m_guid.toString();
         m_guid = Guid::null();
         disconnectIfFinished( old_guid );
      }
      m_guid = guid;
      m_activeEventsPerSource[guid.toString()] = 0;
      // Fire BeginInputFile incident if current InputCollection is a payload file;
      // otherwise, ascertain whether the pointed-to file is reachable before firing any incidents and/or proceeding
      if (m_collectionType.value() == "ImplicitCollection") {
         // For now, we can only deal with input metadata from POOL files, but we know we have a POOL file here
         if (!m_athenaPoolCnvSvc->setInputAttributes(*m_inputCollectionsIterator).isSuccess()) {
               ATH_MSG_ERROR("Failed to set input attributes.");
               return StatusCode::FAILURE;
         }
         if (m_processMetadata.value()) {
            FileIncident beginInputFileIncident(name(), "BeginInputFile", *m_inputCollectionsIterator, m_guid.toString());
            m_incidentSvc->fireIncident(beginInputFileIncident);
         }
      } else {
         // Check if File is BS
         if (tech != 0x00001000 && m_processMetadata.value()) {
            FileIncident beginInputFileIncident(name(), "BeginInputFile", "FID:" + m_guid.toString(), m_guid.toString());
            m_incidentSvc->fireIncident(beginInputFileIncident);
         }
      }
   }  // end if (guid != m_guid)
   return StatusCode::SUCCESS;
}

nextWithSkip()

StatusCode EventSelectorAthenaPool::nextWithSkip ( IEvtSelector::Context & ctxt ) const

overrideprotectedvirtual

Go to next event and skip if necessary.

Definition at line 621 of file EventSelectorAthenaPool.cxx.

                                                                                {
   ATH_MSG_DEBUG("EventSelectorAthenaPool::nextWithSkip");
 
   for (;;) {
      // Check if we're at the end of file
      StatusCode sc = nextHandleFileTransition(ctxt);
      if (sc.isRecoverable()) {
         continue; // handles empty files
      }
      if (sc.isFailure()) {
         return StatusCode::FAILURE;
      }
 
      // Increase event count
      ++m_evtCount;
 
      if (!m_counterTool.empty() && !m_counterTool->preNext().isSuccess()) {
         ATH_MSG_WARNING("Failed to preNext() CounterTool.");
      }
      if( m_evtCount > m_skipEvents
         && (m_skipEventRanges.empty() || m_evtCount < m_skipEventRanges.front().first))
      {
         return StatusCode::SUCCESS;
      } else {
         while( !m_skipEventRanges.empty() && m_evtCount >= m_skipEventRanges.front().second ) {
            m_skipEventRanges.erase(m_skipEventRanges.begin());
         }
         if (m_isSecondary.value()) {
            ATH_MSG_INFO("skipping secondary event " << m_evtCount);
         } else {
            ATH_MSG_INFO("skipping event " << m_evtCount);
         }
      }
   }
 
   return StatusCode::SUCCESS;
}

previous() [1/2]

StatusCode EventSelectorAthenaPool::previous ( IEvtSelector::Context & ctxt ) const

overridevirtual

Parameters

ctxt	[IN/OUT] current event context is interated to previous event.

Definition at line 659 of file EventSelectorAthenaPool.cxx.

                                                                         {
   ATH_MSG_ERROR("previous() not implemented");
   return StatusCode::FAILURE;
}

previous() [2/2]

StatusCode EventSelectorAthenaPool::previous ( IEvtSelector::Context & ctxt, int jump ) const

overridevirtual

Parameters

ctxt	[IN/OUT] current event context is interated to previous event.
jump	[IN] number of events to jump (currently not supported).

Definition at line 664 of file EventSelectorAthenaPool.cxx.

                                                                                      {
   if (jump > 0) {
      for (int i = 0; i < jump; i++) {
         ATH_CHECK(previous(ctxt));
      }
      return StatusCode::SUCCESS;
   }
   return StatusCode::FAILURE;
}

readEvent()

StatusCode EventSelectorAthenaPool::readEvent ( int maxevt )

overridevirtual

Read the next maxevt events.

Parameters

evtnum

[IN] The number of events to read.

Definition at line 896 of file EventSelectorAthenaPool.cxx.

                                                        {
   IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
   if (ds == nullptr) {
      ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
      return StatusCode::FAILURE;
   }
   if (m_eventStreamingTool.empty()) {
      ATH_MSG_ERROR("No AthenaSharedMemoryTool configured for readEvent()");
      return StatusCode::FAILURE;
   }
   ATH_MSG_VERBOSE("Called read Event " << maxevt);
   IEvtSelector::Context* ctxt = new EventContextAthenaPool(this);
   for (int i = 0; i < maxevt || maxevt == -1; ++i) {
      if (!next(*ctxt).isSuccess()) {
         if (m_evtCount == -1) {
            ATH_MSG_VERBOSE("Called read Event and read last event from input: " << i);
            break;
         }
         ATH_MSG_ERROR("Cannot read Event " << m_evtCount - 1 << " into AthenaSharedMemoryTool");
         delete ctxt; ctxt = nullptr;
         return StatusCode::FAILURE;
      } else {
         ATH_MSG_VERBOSE("Called next, read Event " << m_evtCount - 1);
      }
   }
   delete ctxt; ctxt = nullptr;
   // End of file, wait for last event to be taken
   StatusCode sc;
   while ( (sc = putEvent_ST(*m_eventStreamingTool, 0, 0, 0, 0)).isRecoverable() ) {
      while (ds->readData().isSuccess()) {
         ATH_MSG_VERBOSE("Called last readData, while marking last event in readEvent()");
      }
      usleep(1000);
   }
   if (!sc.isSuccess()) {
      ATH_MSG_ERROR("Cannot put last Event marker to AthenaSharedMemoryTool");
      return StatusCode::FAILURE;
   } else {
      sc = ds->readData();
      while (sc.isSuccess() || sc.isRecoverable()) {
         sc = ds->readData();
      }
      ATH_MSG_DEBUG("Failed last readData -> Clients are stopped, after marking last event in readEvent()");
   }
   return StatusCode::SUCCESS;
}

recordAttributeList()

StatusCode EventSelectorAthenaPool::recordAttributeList ( ) const

overrideprotectedvirtual

Record AttributeList in StoreGate.

Reimplemented in DoubleEventSelectorAthenaPool.

Definition at line 994 of file EventSelectorAthenaPool.cxx.

                                                              {
   // Get access to AttributeList
   ATH_MSG_DEBUG("Get AttributeList from the collection");
   // MN: accessing only attribute list, ignoring token list
   const coral::AttributeList& attrList = m_headerIterator->currentRow().attributeList();
   ATH_MSG_DEBUG("AttributeList size " << attrList.size());
   std::unique_ptr<AthenaAttributeList> athAttrList(new AthenaAttributeList(attrList));
   // Fill the new attribute list
   ATH_CHECK(fillAttributeList(athAttrList.get(), "", false));
   // Write the AttributeList
   SG::WriteHandle<AthenaAttributeList> wh(m_attrListKey, eventStore()->name());
   ATH_CHECK(wh.record(std::move(athAttrList)));
   return StatusCode::SUCCESS;
}

reinit()

StatusCode EventSelectorAthenaPool::reinit ( ) const

private

Reinitialize the service when a fork() occurred/was-issued.

Definition at line 200 of file EventSelectorAthenaPool.cxx.

                                                 {
   ATH_MSG_DEBUG("reinitialization...");
 
   // reset markers
   m_numEvt.resize(m_inputCollectionsProp.value().size(), -1);
   m_firstEvt.resize(m_inputCollectionsProp.value().size(), -1);
 
   // Initialize InputCollectionsIterator
   m_inputCollectionsIterator = m_inputCollectionsProp.value().begin();
   m_curCollection = 0;
   if (!m_firstEvt.empty()) {
      m_firstEvt[0] = 0;
   }
   m_inputCollectionsChanged = false;
   m_evtCount = 0;
   m_headerIterator = 0;
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      ATH_MSG_INFO("Done reinitialization for shared reader client");
      return StatusCode::SUCCESS;
   }
   bool retError = false;
   for (auto& tool : m_helperTools) {
      if (!tool->postInitialize().isSuccess()) {
         ATH_MSG_FATAL("Failed to postInitialize() " << tool->name());
         retError = true;
      }
   }
   if (retError) {
      ATH_MSG_FATAL("Failed to postInitialize() helperTools");
      return StatusCode::FAILURE;
   }
 
   // Create an m_poolCollectionConverter to read the objects in
   m_poolCollectionConverter = getCollectionCnv();
   if (!m_poolCollectionConverter) {
      ATH_MSG_INFO("No Events found in any Input Collections");
      if (m_processMetadata.value()) {
     m_inputCollectionsIterator = m_inputCollectionsProp.value().end();
     if (!m_inputCollectionsProp.value().empty()) --m_inputCollectionsIterator;
    //NOTE (wb may 2016): this will make the FirstInputFile incident correspond to last file in the collection ... if want it to be first file then move iterator to begin and then move above two lines below this incident firing
         if (m_collectionType.value() == "ImplicitCollection" && !m_firedIncident && !m_inputCollectionsProp.value().empty()) {
            FileIncident firstInputFileIncident(name(), "FirstInputFile", *m_inputCollectionsIterator);
            m_incidentSvc->fireIncident(firstInputFileIncident);
            m_firedIncident = true;
         }
      }
      return StatusCode::SUCCESS;
   }
   // Get DataHeader iterator
   try {
      m_headerIterator = &m_poolCollectionConverter->selectAll();
   } catch (std::exception &e) {
      ATH_MSG_FATAL("Cannot open implicit collection - check data/software version.");
      ATH_MSG_ERROR(e.what());
      return StatusCode::FAILURE;
   }
   while (m_headerIterator == nullptr || m_headerIterator->next() == 0) { // no selected events
      if (m_poolCollectionConverter) {
         m_poolCollectionConverter->disconnectDb().ignore();
         m_poolCollectionConverter.reset();
      }
      ++m_inputCollectionsIterator;
      m_poolCollectionConverter = getCollectionCnv();
      if (m_poolCollectionConverter) {
         m_headerIterator = &m_poolCollectionConverter->selectAll();
      } else {
         break;
      }
   }
   if (!m_poolCollectionConverter || m_headerIterator == nullptr) { // no event selected in any collection
      m_inputCollectionsIterator = m_inputCollectionsProp.value().begin();
      m_curCollection = 0;
      m_poolCollectionConverter = getCollectionCnv();
      if (!m_poolCollectionConverter) {
         return StatusCode::SUCCESS;
      }
      m_headerIterator = &m_poolCollectionConverter->selectAll();
      while (m_headerIterator == nullptr || m_headerIterator->next() == 0) { // empty collection
         if (m_poolCollectionConverter) {
            m_poolCollectionConverter->disconnectDb().ignore();
            m_poolCollectionConverter.reset();
         }
         ++m_inputCollectionsIterator;
         m_poolCollectionConverter = getCollectionCnv();
         if (m_poolCollectionConverter) {
            m_headerIterator = &m_poolCollectionConverter->selectAll();
         } else {
            break;
         }
      }
   }
   if (!m_poolCollectionConverter || m_headerIterator == nullptr) {
      return StatusCode::SUCCESS;
   }
   const Token& headRef = m_headerIterator->eventRef();
   const std::string fid = headRef.dbID().toString();
   const int tech = headRef.technology();
   ATH_MSG_VERBOSE("reinit(): First DataHeder Token=" << headRef.toString() );
 
   // Check if File is BS, for which Incident is thrown by SingleEventInputSvc
   if (tech != 0x00001000 && m_processMetadata.value() && !m_firedIncident) {
      FileIncident firstInputFileIncident(name(), "FirstInputFile", "FID:" + fid, fid);
      m_incidentSvc->fireIncident(firstInputFileIncident);
      m_firedIncident = true;
   }
   return StatusCode::SUCCESS;
}

releaseContext()

StatusCode EventSelectorAthenaPool::releaseContext ( IEvtSelector::Context *& ctxt ) const

overridevirtual

Parameters

ctxt	[IN] current event context is released.

Definition at line 710 of file EventSelectorAthenaPool.cxx.

                                                                                {
   return StatusCode::SUCCESS;
}

resetCriteria()

StatusCode EventSelectorAthenaPool::resetCriteria ( const std::string & criteria, IEvtSelector::Context & ctxt ) const

overridevirtual

Set a selection criteria.

Parameters

criteria	filter predicate (SQL-style WHERE clause)
ctxt	[IN] current event context.

Definition at line 714 of file EventSelectorAthenaPool.cxx.

                                      {
   return StatusCode::SUCCESS;
}

rewind()

StatusCode EventSelectorAthenaPool::rewind ( IEvtSelector::Context & ctxt ) const

overridevirtual

Parameters

ctxt	[IN/OUT] current event context is rewound to first event.

Definition at line 682 of file EventSelectorAthenaPool.cxx.

                                                                          {
   ATH_CHECK(reinit());
   ctxt = EventContextAthenaPool(this);
   return StatusCode::SUCCESS;
}

seek()

StatusCode EventSelectorAthenaPool::seek ( Context & ctxt, int evtnum ) const

overridevirtual

Seek to a given event number.

Parameters

ctxt	[IN/OUT] current event context.
evtnum	[IN] The event number to which to seek.

Reimplemented in DoubleEventSelectorAthenaPool.

Definition at line 719 of file EventSelectorAthenaPool.cxx.

                                                                     {
 
   if( m_inputCollectionsChanged ) {
      StatusCode rc = reinit();
      if( rc != StatusCode::SUCCESS ) return rc;
   }
 
   long newColl = findEvent(evtNum);
   if (newColl == -1 && evtNum >= m_firstEvt[m_curCollection] && evtNum < m_evtCount - 1) {
      newColl = m_curCollection;
   }
   if (newColl == -1) {
      m_headerIterator = nullptr;
      ATH_MSG_INFO("seek: Reached end of Input.");
      fireEndFileIncidents(true);
      return StatusCode::RECOVERABLE;
   }
   if (newColl != m_curCollection) {
      if (!m_keepInputFilesOpen.value() && m_poolCollectionConverter) {
         m_poolCollectionConverter->disconnectDb().ignore();
      }
      m_poolCollectionConverter.reset();
      m_curCollection = newColl;
      try {
         ATH_MSG_DEBUG("Seek to item: \""
             <<  m_inputCollectionsProp.value()[m_curCollection]
             << "\" from the collection list.");
         // Reset input collection iterator to the right place
         m_inputCollectionsIterator = m_inputCollectionsProp.value().begin();
         m_inputCollectionsIterator += m_curCollection;
         m_poolCollectionConverter = std::make_unique<PoolCollectionConverter>(m_collectionType.value(),
             m_inputCollectionsProp.value()[m_curCollection],
             IPoolSvc::kInputStream,
             m_athenaPoolCnvSvc->getPoolSvc());
         if (!m_poolCollectionConverter->initialize().isSuccess()) {
            m_headerIterator = nullptr;
            ATH_MSG_ERROR("seek: Unable to initialize PoolCollectionConverter.");
            return StatusCode::FAILURE;
         }
         // Create DataHeader iterators
         m_headerIterator = &m_poolCollectionConverter->selectAll();
         EventContextAthenaPool* beginIter = new EventContextAthenaPool(this);
         m_evtCount = m_firstEvt[m_curCollection];
         next(*beginIter).ignore();
         ATH_MSG_DEBUG("Token " << m_headerIterator->eventRef().toString());
      } catch (std::exception &e) {
         m_headerIterator = nullptr;
         ATH_MSG_ERROR(e.what());
         return StatusCode::FAILURE;
      }
   }
 
   if (m_headerIterator->seek(evtNum - m_firstEvt[m_curCollection]) == 0) {
      m_headerIterator = nullptr;
      ATH_MSG_ERROR("Did not find event, evtNum = " << evtNum);
      return StatusCode::FAILURE;
   } else {
      m_evtCount = evtNum + 1;
   }
   return StatusCode::SUCCESS;
}

share()

StatusCode EventSelectorAthenaPool::share ( int evtnum )

overridevirtual

Request to share a given event number.

Parameters

evtnum

[IN] The event number to share.

Definition at line 863 of file EventSelectorAthenaPool.cxx.

                                                    {
   IDataShare* ds = dynamic_cast<IDataShare*>(m_athenaPoolCnvSvc.get());
   if (ds == nullptr) {
      ATH_MSG_ERROR("Cannot cast AthenaPoolCnvSvc to DataShare");
      return StatusCode::FAILURE;
   }
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      StatusCode sc = m_eventStreamingTool->lockEvent(evtnum);
      while (sc.isRecoverable()) {
         usleep(1000);
         sc = m_eventStreamingTool->lockEvent(evtnum);
      }
// Send stop client and wait for restart
      if (sc.isFailure()) {
         if (ds->makeClient(0).isFailure()) {
            return StatusCode::FAILURE;
         }
         sc = m_eventStreamingTool->lockEvent(evtnum);
         while (sc.isRecoverable() || sc.isFailure()) {
            usleep(1000);
            sc = m_eventStreamingTool->lockEvent(evtnum);
         }
//FIXME
         if (ds->makeClient(1).isFailure()) {
            return StatusCode::FAILURE;
         }
      }
      return(sc);
   }
   return StatusCode::FAILURE;
}

size()

int EventSelectorAthenaPool::size ( Context & ctxt ) const

overridevirtual

Return the size of the collection.

Parameters

ctxt	[IN/OUT] current event context.

Reimplemented in DoubleEventSelectorAthenaPool.

Definition at line 944 of file EventSelectorAthenaPool.cxx.

                                                  {
   // Fetch sizes of all collections.
   findEvent(-1);
   return std::accumulate(m_numEvt.begin(), m_numEvt.end(), 0);
}

start()

StatusCode EventSelectorAthenaPool::start ( )

overridevirtual

Definition at line 308 of file EventSelectorAthenaPool.cxx.

                                          {
   if (m_poolCollectionConverter) {
      // Reset iterators and apply new query
      m_poolCollectionConverter->disconnectDb().ignore();
      m_poolCollectionConverter.reset();
   }
   m_inputCollectionsIterator = m_inputCollectionsProp.value().begin();
   m_curCollection = 0;
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      return StatusCode::SUCCESS;
   }
   m_poolCollectionConverter = getCollectionCnv(true);
   if (!m_poolCollectionConverter) {
      ATH_MSG_INFO("No Events found in any Input Collections");
      m_inputCollectionsIterator = m_inputCollectionsProp.value().end();
      if (!m_inputCollectionsProp.value().empty()) {
         --m_inputCollectionsIterator; //leave iterator in state of last input file
      }
   } else {
      m_headerIterator = &m_poolCollectionConverter->selectAll();
   }
   m_evtCount = 0;
   delete m_endIter;
   m_endIter = nullptr;
   m_endIter = new EventContextAthenaPool(nullptr);
   return StatusCode::SUCCESS;
}

stop()

StatusCode EventSelectorAthenaPool::stop ( )

overridevirtual

Definition at line 336 of file EventSelectorAthenaPool.cxx.

                                         {
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      return StatusCode::SUCCESS;
   }
   IEvtSelector::Context* ctxt(nullptr);
   if (!releaseContext(ctxt).isSuccess()) {
      ATH_MSG_WARNING("Cannot release context");
   }
   return StatusCode::SUCCESS;
}

Friends And Related Symbol Documentation

DoubleEventSelectorAthenaPool

friend class DoubleEventSelectorAthenaPool

friend

make the DoubleEventSelectorAthenaPool a friend so it can access the internal EventSelectorAthenaPool methods and members

Definition at line 245 of file EventSelectorAthenaPool.h.

Member Data Documentation

ATLAS_THREAD_SAFE [1/10]

std::unique_ptr<PoolCollectionConverter> m_poolCollectionConverter EventSelectorAthenaPool::ATLAS_THREAD_SAFE {}

mutableprivate

Definition at line 171 of file EventSelectorAthenaPool.h.

{};

ATLAS_THREAD_SAFE [2/10]

pool::ICollectionCursor* m_headerIterator EventSelectorAthenaPool::ATLAS_THREAD_SAFE {}

mutableprivate

Definition at line 172 of file EventSelectorAthenaPool.h.

{};

ATLAS_THREAD_SAFE [3/10]

Guid m_guid EventSelectorAthenaPool::ATLAS_THREAD_SAFE {}

mutableprivate

Definition at line 173 of file EventSelectorAthenaPool.h.

{};

ATLAS_THREAD_SAFE [4/10]

std::map<SG::SourceID, int> m_activeEventsPerSource EventSelectorAthenaPool::ATLAS_THREAD_SAFE

mutableprivate

Definition at line 174 of file EventSelectorAthenaPool.h.

ATLAS_THREAD_SAFE [5/10]

std::vector<std::string>::const_iterator m_inputCollectionsIterator EventSelectorAthenaPool::ATLAS_THREAD_SAFE

mutableprivate

Definition at line 192 of file EventSelectorAthenaPool.h.

ATLAS_THREAD_SAFE [6/10]

bool m_inputCollectionsChanged EventSelectorAthenaPool::ATLAS_THREAD_SAFE

mutableprivate

flag to notify the EvSel that the inputs were changed and reinit() needs to be called ASAP

Definition at line 195 of file EventSelectorAthenaPool.h.

ATLAS_THREAD_SAFE [7/10]

ToolHandleArray<IAthenaSelectorTool> m_helperTools EventSelectorAthenaPool::ATLAS_THREAD_SAFE {this, "HelperTools", {}, ""}

mutableprivate

HelperTools, vector of names of AlgTools that are executed by the EventSelector.

Definition at line 202 of file EventSelectorAthenaPool.h.

{this, "HelperTools", {}, ""};

ATLAS_THREAD_SAFE [8/10]

std::vector<int> m_numEvt EventSelectorAthenaPool::ATLAS_THREAD_SAFE

mutableprivate

Definition at line 225 of file EventSelectorAthenaPool.h.

ATLAS_THREAD_SAFE [9/10]

std::vector<int> m_firstEvt EventSelectorAthenaPool::ATLAS_THREAD_SAFE

mutableprivate

Definition at line 226 of file EventSelectorAthenaPool.h.

ATLAS_THREAD_SAFE [10/10]

std::vector<std::pair<long,long> > m_skipEventRanges EventSelectorAthenaPool::ATLAS_THREAD_SAFE

mutableprivate

Definition at line 233 of file EventSelectorAthenaPool.h.

m_athenaPoolCnvSvc

ServiceHandle<IAthenaPoolCnvSvc> EventSelectorAthenaPool::m_athenaPoolCnvSvc {this, "ConversionService", "AthenaPoolCnvSvc", ""}

private

Definition at line 176 of file EventSelectorAthenaPool.h.

{this, "ConversionService", "AthenaPoolCnvSvc", ""};

m_attrListKey

std::string EventSelectorAthenaPool::m_attrListKey {"Input"}

private

AttributeList SG key.

Definition at line 188 of file EventSelectorAthenaPool.h.

{"Input"};

m_callLock

CallMutex EventSelectorAthenaPool::m_callLock

mutableprivate

Definition at line 239 of file EventSelectorAthenaPool.h.

m_collectionType

Gaudi::Property<std::string> EventSelectorAthenaPool::m_collectionType {this, "CollectionType", "ImplicitCollection", ""}

private

CollectionType, type of the collection: default = "ImplicitCollection".

Definition at line 185 of file EventSelectorAthenaPool.h.

{this, "CollectionType", "ImplicitCollection", ""};

m_counterTool

ToolHandle<IAthenaSelectorTool> EventSelectorAthenaPool::m_counterTool {this, "CounterTool", "", ""}

private

Definition at line 203 of file EventSelectorAthenaPool.h.

{this, "CounterTool", "", ""};

m_curCollection

std::atomic_long EventSelectorAthenaPool::m_curCollection {}

mutableprivate

Definition at line 224 of file EventSelectorAthenaPool.h.

{};

m_endIter

EventContextAthenaPool* EventSelectorAthenaPool::m_endIter {}

private

Definition at line 169 of file EventSelectorAthenaPool.h.

{};

m_eventsPerLB

Gaudi::CheckedProperty<uint32_t> EventSelectorAthenaPool::m_eventsPerLB {this, "EventsPerLB", 1000, ""}

private

Definition at line 220 of file EventSelectorAthenaPool.h.

{this, "EventsPerLB", 1000, ""};

m_eventsPerRun

Gaudi::CheckedProperty<uint64_t> EventSelectorAthenaPool::m_eventsPerRun {this, "EventsPerRun", 1000000, ""}

private

Definition at line 218 of file EventSelectorAthenaPool.h.

{this, "EventsPerRun", 1000000, ""};

m_eventStreamingTool

ToolHandle<IAthenaIPCTool> EventSelectorAthenaPool::m_eventStreamingTool {this, "SharedMemoryTool", "", ""}

private

Definition at line 204 of file EventSelectorAthenaPool.h.

{this, "SharedMemoryTool", "", ""};

m_evtCount

std::atomic_int EventSelectorAthenaPool::m_evtCount {}

mutableprivate

Definition at line 235 of file EventSelectorAthenaPool.h.

{}; // internal count of events

m_firedIncident

std::atomic_bool EventSelectorAthenaPool::m_firedIncident {}

mutableprivate

Definition at line 236 of file EventSelectorAthenaPool.h.

{};

m_firstEventNo

Gaudi::CheckedProperty<uint64_t> EventSelectorAthenaPool::m_firstEventNo {this, "FirstEvent", 1, ""}

private

Definition at line 216 of file EventSelectorAthenaPool.h.

{this, "FirstEvent", 1, ""};

m_firstLBNo

Gaudi::CheckedProperty<uint32_t> EventSelectorAthenaPool::m_firstLBNo {this, "FirstLB", 0, ""}

private

Definition at line 219 of file EventSelectorAthenaPool.h.

{this, "FirstLB", 0, ""};

m_incidentSvc

ServiceHandle<IIncidentSvc> EventSelectorAthenaPool::m_incidentSvc {this, "IncidentSvc", "IncidentSvc", ""}

private

Definition at line 177 of file EventSelectorAthenaPool.h.

{this, "IncidentSvc", "IncidentSvc", ""};

m_initTimeStamp

Gaudi::CheckedProperty<uint32_t> EventSelectorAthenaPool::m_initTimeStamp {this, "InitialTimeStamp", 0, ""}

private

Definition at line 221 of file EventSelectorAthenaPool.h.

{this, "InitialTimeStamp", 0, ""};

m_inputCollectionsProp

Gaudi::Property<std::vector<std::string> > EventSelectorAthenaPool::m_inputCollectionsProp {this, "InputCollections", {}, ""}

private

InputCollections, vector with names of the input collections.

Definition at line 191 of file EventSelectorAthenaPool.h.

{this, "InputCollections", {}, ""};

m_isSecondary

Gaudi::Property<bool> EventSelectorAthenaPool::m_isSecondary {this, "IsSecondary", false, ""}

private

IsSecondary, know if this is an instance of secondary event selector.

Definition at line 181 of file EventSelectorAthenaPool.h.

{this, "IsSecondary", false, ""};

m_keepInputFilesOpen

Gaudi::Property<bool> EventSelectorAthenaPool::m_keepInputFilesOpen {this, "KeepInputFilesOpen", false, ""}

private

KeepInputFilesOpen, boolean flag to keep files open after PoolCollection reaches end: default = false.

Needed for PilUp to run without PoolFileCatalog. Relies on POOL to close files when reaching DB_AGE_LIMIT.

Definition at line 199 of file EventSelectorAthenaPool.h.

{this, "KeepInputFilesOpen", false, ""};

m_makeStreamingToolClient

Gaudi::Property<int> EventSelectorAthenaPool::m_makeStreamingToolClient {this, "MakeStreamingToolClient",0}

private

Make this instance a Streaming Client during first iteration automatically.

Definition at line 206 of file EventSelectorAthenaPool.h.

{this, "MakeStreamingToolClient",0};

m_oldRunNo

Gaudi::CheckedProperty<uint32_t> EventSelectorAthenaPool::m_oldRunNo {this, "OldRunNumber", 0, ""}

private

Definition at line 212 of file EventSelectorAthenaPool.h.

{this, "OldRunNumber", 0, ""};

m_overrideRunNumber

Gaudi::Property<bool> EventSelectorAthenaPool::m_overrideRunNumber {this, "OverrideRunNumber", false, ""}

private

Definition at line 213 of file EventSelectorAthenaPool.h.

{this, "OverrideRunNumber", false, ""};

m_overrideRunNumberFromInput

Gaudi::Property<bool> EventSelectorAthenaPool::m_overrideRunNumberFromInput {this, "OverrideRunNumberFromInput", false, ""}

private

Definition at line 214 of file EventSelectorAthenaPool.h.

{this, "OverrideRunNumberFromInput", false, ""};

m_processMetadata

Gaudi::Property<bool> EventSelectorAthenaPool::m_processMetadata {this, "ProcessMetadata", true, ""}

private

ProcessMetadata, switch on firing of FileIncidents which will trigger processing of metadata: default = true.

Definition at line 183 of file EventSelectorAthenaPool.h.

{this, "ProcessMetadata", true, ""};

m_runNo

Gaudi::CheckedProperty<uint32_t> EventSelectorAthenaPool::m_runNo {this, "RunNumber", 0, ""}

private

The following are included for compatibility with McEventSelector and are not really used.

However runNo, oldRunNo and overrideRunNumberFromInput are used to reset run number for simulated events, needed to use condition

Definition at line 211 of file EventSelectorAthenaPool.h.

{this, "RunNumber", 0, ""};

m_skipEventRangesProp

Gaudi::Property<std::string> EventSelectorAthenaPool::m_skipEventRangesProp {this, "SkipEventRanges", {}, ""}

private

Skip Events - comma separated list of event to skip, ranges with '-': <start> - <end>

Definition at line 232 of file EventSelectorAthenaPool.h.

{this, "SkipEventRanges", {}, ""};

m_skipEvents

Gaudi::Property<int> EventSelectorAthenaPool::m_skipEvents {this, "SkipEvents", 0, ""}

private

SkipEvents, numbers of events to skip: default = 0.

Definition at line 229 of file EventSelectorAthenaPool.h.

{this, "SkipEvents", 0, ""};

m_skipEventSequenceProp

Gaudi::Property<std::vector<long> > EventSelectorAthenaPool::m_skipEventSequenceProp {this, "SkipEventSequence", {}, ""}

private

Definition at line 230 of file EventSelectorAthenaPool.h.

{this, "SkipEventSequence", {}, ""};

m_sourceID

SG::SlotSpecificObj<SG::SourceID> EventSelectorAthenaPool::m_sourceID

private

Definition at line 241 of file EventSelectorAthenaPool.h.

m_timeStampInterval

Gaudi::Property<uint32_t> EventSelectorAthenaPool::m_timeStampInterval {this, "TimeStampInterval", 0, ""}

private

Definition at line 222 of file EventSelectorAthenaPool.h.

{this, "TimeStampInterval", 0, ""};

---

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

• EventSelectorAthenaPool.h
• EventSelectorAthenaPool.cxx