EventSelectorByteStream Class Reference

concrete implementation of IEvtSelector for ByteStream More...

#include <EventSelectorByteStream.h>

Inheritance diagram for EventSelectorByteStream:

Collaboration diagram for EventSelectorByteStream:

Public Member Functions
	EventSelectorByteStream (const std::string &name, ISvcLocator *svcloc)
	Standard Constructor. More...
virtual	~EventSelectorByteStream ()
	Standard Destructor. More...
virtual StatusCode	initialize () override
	Implementation of Service base class methods. More...
virtual StatusCode	start () override
virtual StatusCode	stop () override
virtual StatusCode	finalize () override
virtual StatusCode	createContext (Context *&it) const override
	create context More...
virtual StatusCode	next (Context &it) const override
virtual StatusCode	next (Context &it, int jump) const override
virtual StatusCode	previous (Context &it) const override
virtual StatusCode	previous (Context &it, int jump) const override
virtual StatusCode	last (Context &it) const override
virtual StatusCode	rewind (Context &it) const override
virtual StatusCode	createAddress (const Context &it, IOpaqueAddress *&iop) const override
virtual StatusCode	releaseContext (Context *&it) const override
virtual StatusCode	resetCriteria (const std::string &criteria, Context &context) const override
	Set a selection criteria. More...
virtual StatusCode	seek (Context &, int evtnum) const override
	Seek to a given event number. More...
virtual int	curEvent (const Context &it) const override
	Return the current event number. More...
virtual int	size (Context &it) const override
	Always returns -1. More...
virtual StatusCode	makeServer (int num) override
	Make this a server. More...
virtual StatusCode	makeClient (int num) override
	Make this a client. More...
virtual StatusCode	share (int evtnum) override
	Request to share a given event number. More...
virtual StatusCode	readEvent (int maxevt) override
	Read the next maxevt events. More...
virtual StatusCode	io_reinit () override
	Callback method to reinitialize the internal state of the component for I/O purposes (e.g. upon fork(2)) More...

Protected Types
typedef std::recursive_mutex	mutex_t
typedef std::lock_guard< mutex_t >	lock_t

Protected Member Functions
virtual StatusCode	nextHandleFileTransition (IEvtSelector::Context &ctxt) const override
	Handle file transition at the next iteration. More...
virtual StatusCode	nextWithSkip (IEvtSelector::Context &ctxt) const override
	Go to next event and skip if necessary. More...
virtual StatusCode	recordAttributeList () const override
	Record AttributeList in StoreGate. More...
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. More...
virtual bool	disconnectIfFinished (const SG::SourceID &fid) const override

Private Member Functions
StatusCode	nextImpl (Context &it, lock_t &lock) const
StatusCode	nextImpl (Context &it, int jump, lock_t &lock) const
StatusCode	previousImpl (Context &it, lock_t &lock) const
StatusCode	previousImpl (Context &it, int jump, lock_t &lock) const
StatusCode	nextHandleFileTransitionImpl (IEvtSelector::Context &ctxt, lock_t &lock) const
StatusCode	nextWithSkipImpl (IEvtSelector::Context &ctxt, lock_t &lock) const
StatusCode	recordAttributeListImpl (lock_t &lock) const
StatusCode	fillAttributeListImpl (coral::AttributeList *attrList, const std::string &suffix, bool copySource, lock_t &lock) const
StatusCode	reinit (lock_t &lock)
	Reinitialize the service when a fork() occurred/was-issued. More...
StatusCode	openNewRun (lock_t &lock) const
void	nextFile (lock_t &lock) const
int	findEvent (int evtNum, lock_t &lock) const
	Search for event with number evtNum. More...
StoreGateSvc *	eventStore () const
void	inputCollectionsHandler (Gaudi::Details::PropertyBase &)

Private Attributes
mutex_t	m_mutex
int m_fileCount	ATLAS_THREAD_SAFE = 0
	number of files to process. More...
std::vector< int > m_numEvt	ATLAS_THREAD_SAFE
std::vector< int > m_firstEvt	ATLAS_THREAD_SAFE
std::vector< std::string >::const_iterator m_inputCollectionsIterator	ATLAS_THREAD_SAFE
std::vector< long > m_skipEventSequence	ATLAS_THREAD_SAFE
long m_NumEvents	ATLAS_THREAD_SAFE = 0
ToolHandle< IAthenaIPCTool > m_eventStreamingTool	ATLAS_THREAD_SAFE {this, "SharedMemoryTool", "", ""}
Gaudi::Property< bool >	m_isSecondary {this, "IsSecondary", false, ""}
	IsSecondary, know if this is an instance of secondary event selector. More...
Gaudi::Property< std::string >	m_eventSourceName {this, "ByteStreamInputSvc", "", ""}
Gaudi::Property< bool >	m_procBadEvent {this, "ProcessBadEvent", false, ""}
	process bad events, which fail check_tree(). More...
Gaudi::Property< int >	m_maxBadEvts {this, "MaxBadEvents", -1, ""}
	number of bad events allowed before quitting. More...
EventContextByteStream *	m_beginIter {}
EventContextByteStream *	m_endIter {}
SmartIF< IByteStreamInputSvc >	m_eventSource
Gaudi::Property< std::vector< std::string > >	m_inputCollectionsProp {this, "Input", {}, ""}
ServiceHandle< IIncidentSvc >	m_incidentSvc {this, "IncidentSvc", "IncidentSvc", ""}
Gaudi::Property< long >	m_skipEvents {this, "SkipEvents", 0, ""}
Gaudi::Property< std::vector< long > >	m_skipEventSequenceProp {this, "SkipEventSequence", {}, ""}
bool	m_firstFileFired {}
bool	m_beginFileFired {}
ToolHandleArray< IAthenaSelectorTool >	m_helperTools {this}
	HelperTools, vector of names of AlgTools that are executed by the EventSelector. More...
ToolHandle< IAthenaSelectorTool >	m_counterTool {this, "CounterTool", "", ""}
Gaudi::Property< bool >	m_overrideRunNumber {this, "OverrideRunNumber", false, ""}
	The following are included for compatibility with McEventSelector and are not really used. More...
Gaudi::Property< bool >	m_filebased {this, "FileBased", true, ""}
Gaudi::CheckedProperty< uint32_t >	m_runNo {this, "RunNumber", 0, ""}
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, ""}

Detailed Description

concrete implementation of IEvtSelector for ByteStream

An event selector service that steps through Event Filter packed raw events by use of an event source object and an iterator object called EventIteratorByteStream. The type of event source is specified at run time. It

can be a single file, a set of files or the network.

Definition at line 42 of file EventSelectorByteStream.h.

Member Typedef Documentation

lock_t

typedef std::lock_guard<mutex_t> EventSelectorByteStream::lock_t

protected

Definition at line 127 of file EventSelectorByteStream.h.

mutex_t

typedef std::recursive_mutex EventSelectorByteStream::mutex_t

protected

Definition at line 126 of file EventSelectorByteStream.h.

Constructor & Destructor Documentation

EventSelectorByteStream()

EventSelectorByteStream::EventSelectorByteStream	(	const std::string &	name,
		ISvcLocator *	svcloc
	)

Standard Constructor.

Definition at line 41 of file EventSelectorByteStream.cxx.

    : base_class(name, svcloc) {
  declareProperty("HelperTools", m_helperTools);
 
  // RunNumber, OldRunNumber and OverrideRunNumberFromInput are used
  // to override the run number coming in on the input stream
  m_runNo.verifier().setLower(0);
  // The following properties are only for compatibility with
  // McEventSelector and are not really used anywhere
  // TODO(berghaus): validate if those are even used
  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(
      &EventSelectorByteStream::inputCollectionsHandler, this);
}

~EventSelectorByteStream()

EventSelectorByteStream::~EventSelectorByteStream ( )

virtual

Standard Destructor.

Definition at line 72 of file EventSelectorByteStream.cxx.

                                                  {
}

Member Function Documentation

createAddress()

StatusCode EventSelectorByteStream::createAddress ( const Context &  it, IOpaqueAddress *&  iop  ) const

overridevirtual

Parameters

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

Definition at line 1033 of file EventSelectorByteStream.cxx.

                                            {
   SG::DataProxy* proxy = eventStore()->proxy(ClassID_traits<DataHeader>::ID(),"ByteStreamDataHeader");
   if (proxy !=0) {
     iop = proxy->address();
     return(StatusCode::SUCCESS);
   } else {
     iop = 0;
     return(StatusCode::FAILURE);
   }
}

createContext()

StatusCode EventSelectorByteStream::createContext ( IEvtSelector::Context *&  it ) const

overridevirtual

create context

Definition at line 338 of file EventSelectorByteStream.cxx.

                                                                                {
   it = new EventContextByteStream(this);
   return(StatusCode::SUCCESS);
}

curEvent()

int EventSelectorByteStream::curEvent ( const Context &  it ) const

overridevirtual

Return the current event number.

Parameters

it	[IN/OUT] current event context.

Definition at line 938 of file EventSelectorByteStream.cxx.

                                                             {
   // event counter in IEvtSelectorSeek interface
   lock_t lock (m_mutex);
   return int(m_NumEvents);
}

disconnectIfFinished()

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

overrideprotectedvirtual

Definition at line 1094 of file EventSelectorByteStream.cxx.

{
   return true;
}

eventStore()

StoreGateSvc * EventSelectorByteStream::eventStore ( ) const

private

Definition at line 78 of file EventSelectorByteStream.cxx.

                                          {
   return StoreGateSvc::currentStoreGate();
}

fillAttributeList()

StatusCode EventSelectorByteStream::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 803 of file EventSelectorByteStream.cxx.

{
  lock_t lock (m_mutex);
  return fillAttributeListImpl (attrList, suffix, copySource, lock);
}

fillAttributeListImpl()

StatusCode EventSelectorByteStream::fillAttributeListImpl ( coral::AttributeList *  attrList, const std::string &  suffix, bool  copySource, lock_t &  lock  ) const

private

Definition at line 808 of file EventSelectorByteStream.cxx.

{
   attrList->extend("RunNumber"   + suffix, "unsigned int");
   attrList->extend("EventNumber" + suffix, "unsigned long long");
   attrList->extend("LumiBlockN"  + suffix, "unsigned int");
   attrList->extend("BunchId"     + suffix, "unsigned int");
   attrList->extend("EventTime"        + suffix, "unsigned int");
   attrList->extend("EventTimeNanoSec" + suffix, "unsigned int");
 
   // fill attribute list
   const RawEvent* event = m_eventSource->currentEvent();
 
   (*attrList)["RunNumber" + suffix].data<unsigned int>() = event->run_no();
   if (event->version() < 0x03010000) {
      (*attrList)["EventNumber" + suffix].data<unsigned long long>() = event->lvl1_id();
   } else {
      (*attrList)["EventNumber" + suffix].data<unsigned long long>() = event->global_id();
   }
   (*attrList)["LumiBlockN" + suffix].data<unsigned int>() = event->lumi_block();
   (*attrList)["BunchId" + suffix].data<unsigned int>() = event->bc_id();
 
   unsigned int bc_time_sec = event->bc_time_seconds();
   unsigned int bc_time_ns  = event->bc_time_nanoseconds();
   // bc_time_ns should be lt 1e9.
   if (bc_time_ns > 1000000000) {
      // round it off to 1e9
      ATH_MSG_WARNING(" bc_time nanosecond number larger than 1e9, it is " << bc_time_ns << ", reset it to 1 sec");
      bc_time_ns = 1000000000;
   }
   (*attrList)["EventTime" + suffix].data<unsigned int>() = bc_time_sec;
   (*attrList)["EventTimeNanoSec" + suffix].data<unsigned int>() = bc_time_ns;
 
   const OFFLINE_FRAGMENTS_NAMESPACE::DataType* buffer;
 
   event->status(buffer);
   attrList->extend("TriggerStatus" + suffix, "unsigned int");
   (*attrList)["TriggerStatus" + suffix].data<unsigned int>() = *buffer;
 
   attrList->extend("ExtendedL1ID"  + suffix, "unsigned int");
   attrList->extend("L1TriggerType" + suffix, "unsigned int");
   (*attrList)["ExtendedL1ID" + suffix].data<unsigned int>() = event->lvl1_id();
   (*attrList)["L1TriggerType" + suffix].data<unsigned int>() = event->lvl1_trigger_type();
 
   // Grab L1 words
   event->lvl1_trigger_info(buffer);
   for (uint32_t iT1 = 0; iT1 < event->nlvl1_trigger_info(); ++iT1) {
      std::stringstream name;
      name << "L1TriggerInfo_" << iT1;
      attrList->extend(name.str() + suffix, "unsigned int");
      (*attrList)[name.str() + suffix].data<unsigned int>() = *buffer;
      ++buffer;
   }
 
   // Grab L2 words
   event->lvl2_trigger_info(buffer);
   for (uint32_t iT1 = 0; iT1 < event->nlvl2_trigger_info(); ++iT1) {
      if (*buffer != 0) {
         std::stringstream name;
         name << "L2TriggerInfo_" << iT1;
         attrList->extend(name.str() + suffix, "unsigned int");
         (*attrList)[name.str() + suffix].data<unsigned int>() = *buffer;
      }
      ++buffer;
   }
 
   // Grab EF words
   event->event_filter_info(buffer);
   for (uint32_t iT1 = 0; iT1 < event->nevent_filter_info(); ++iT1) {
      if (*buffer != 0) {
         std::stringstream name;
         name << "EFTriggerInfo_" << iT1;
         attrList->extend(name.str() + suffix, "unsigned int");
         (*attrList)[name.str() + suffix].data<unsigned int>() = *buffer;
      }
      ++buffer;
   }
 
   // Grab stream tags
   event->stream_tag(buffer);
   std::vector<eformat::helper::StreamTag> onl_streamTags;
   eformat::helper::decode(event->nstream_tag(), buffer, onl_streamTags);
   for (std::vector<eformat::helper::StreamTag>::const_iterator itS = onl_streamTags.begin(),
      itSE = onl_streamTags.end(); itS != itSE; ++itS) {
      attrList->extend(itS->name + suffix, "string");
      (*attrList)[itS->name + suffix].data<std::string>() = itS->type;
   }
 
   return StatusCode::SUCCESS;
}

finalize()

StatusCode EventSelectorByteStream::finalize ( )

overridevirtual

Definition at line 249 of file EventSelectorByteStream.cxx.

                                             {
   if (!m_counterTool.empty()) {
      if (!m_counterTool->preFinalize().isSuccess()) {
         ATH_MSG_WARNING("Failed to preFinalize() CounterTool");
      }
   }
   for (ToolHandle<IAthenaSelectorTool>& tool : m_helperTools) {
      if (!tool->preFinalize().isSuccess()) {
         ATH_MSG_WARNING("Failed to preFinalize() " << tool->name());
      }
   }
   delete m_beginIter; m_beginIter = 0;
   delete m_endIter; m_endIter = 0;
   // Release AthenaSharedMemoryTool
   if (!m_eventStreamingTool.empty() && !m_eventStreamingTool.release().isSuccess()) {
      ATH_MSG_WARNING("Cannot release AthenaSharedMemoryTool");
   }
   // Release CounterTool
   if (!m_counterTool.empty()) {
      if (!m_counterTool.release().isSuccess()) {
         ATH_MSG_WARNING("Cannot release CounterTool.");
      }
   }
   // Release HelperTools
   if (!m_helperTools.release().isSuccess()) {
      ATH_MSG_WARNING("Cannot release " << m_helperTools);
   }
   if (m_eventSource) m_eventSource->release();
   // Finalize the Service base class.
   return(AthService::finalize());
}

findEvent()

int EventSelectorByteStream::findEvent ( int  evtNum, lock_t &  lock  ) const

private

Search for event with number evtNum.

Definition at line 900 of file EventSelectorByteStream.cxx.

                                                                  {
   // Loop over file event counts
   //ATH_MSG_INFO("try to find evnum = " << evtNum << " in " << m_numEvt.size() << " files");
   for (size_t i = 0; i < m_inputCollectionsProp.value().size(); i++) {
      if (m_inputCollectionsProp.value().size() != m_numEvt.size()) {
         ATH_MSG_ERROR("vector size incompatibility");
         break;
      }
      // if file not opened yet, check it
      if (m_numEvt[i] == -1) {
         std::string fileName = m_inputCollectionsProp.value()[i];
         auto nevguid = m_eventSource->getBlockIterator(fileName);
         long nev = nevguid.first;
         // if failure on file open, exit
         if (nev==-1) {
            break;
         }
         // set initial event counter for that file
         if (i > 0) {
            m_firstEvt[i] = m_firstEvt[i - 1] + m_numEvt[i - 1];
         } else {
            m_firstEvt[i] = 0;
         }
         // log number of events in that file
         m_numEvt[i] = nev;
      }
      // if sought event is in this file, then return the index of that file
      if (evtNum >= m_firstEvt[i] && evtNum < m_firstEvt[i] + m_numEvt[i]) {
         ATH_MSG_INFO("found " << evtNum << " in file " << i);
         return(i);
      }
   }
   ATH_MSG_INFO("did not find ev " << evtNum);
   // return file not found marker
   return(-1);
}

initialize()

StatusCode EventSelectorByteStream::initialize ( )

overridevirtual

Implementation of Service base class methods.

Definition at line 84 of file EventSelectorByteStream.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());
   }
 
   if (!::AthService::initialize().isSuccess()) {
      ATH_MSG_FATAL("Cannot initialize AthService base class.");
      return(StatusCode::FAILURE);
   }
 
   // Check for input setting
   if (m_filebased && m_inputCollectionsProp.value().empty()) {
     ATH_MSG_FATAL("Unable to retrieve valid input list");
     return(StatusCode::FAILURE);
   }
   m_skipEventSequence = m_skipEventSequenceProp.value();
   std::sort(m_skipEventSequence.begin(), m_skipEventSequence.end());
 
   // Check ByteStreamCnvSvc
   m_eventSource = serviceLocator()->service(m_eventSourceName.value());
   if (!m_eventSource) {
      ATH_MSG_FATAL("Cannot get ByteStreamInputSvc");
      return(StatusCode::FAILURE);
   }
 
   // Get CounterTool (if configured)
   if (!m_counterTool.empty()) {
      if (!m_counterTool.retrieve().isSuccess()) {
         ATH_MSG_FATAL("Cannot get CounterTool.");
         return(StatusCode::FAILURE);
      }
   }
   // Get HelperTools
   if (!m_helperTools.empty()) {
      if (!m_helperTools.retrieve().isSuccess()) {
         ATH_MSG_FATAL("Cannot get " << m_helperTools);
         return(StatusCode::FAILURE);
      }
   }
   // Get SharedMemoryTool (if configured)
   if (!m_eventStreamingTool.empty() && !m_eventStreamingTool.retrieve().isSuccess()) {
      ATH_MSG_FATAL("Cannot get AthenaSharedMemoryTool");
      return(StatusCode::FAILURE);
   }
 
   // Register this service for 'I/O' events
   ServiceHandle<IIoComponentMgr> iomgr("IoComponentMgr", name());
   if (!iomgr.retrieve().isSuccess()) {
      ATH_MSG_FATAL("Cannot retrieve IoComponentMgr.");
      return(StatusCode::FAILURE);
   }
   if (!iomgr->io_register(this).isSuccess()) {
      ATH_MSG_FATAL("Cannot register myself with the IoComponentMgr.");
      return(StatusCode::FAILURE);
   }
 
   // Register the input files with the iomgr
   bool allGood = true;
   const std::vector<std::string>& incol = m_inputCollectionsProp.value();
   for (std::size_t icol = 0, imax = incol.size(); icol != imax; ++icol) {
      if (!iomgr->io_register(this, IIoComponentMgr::IoMode::READ, incol[icol]).isSuccess()) {
         ATH_MSG_FATAL("could not register [" << incol[icol] << "] for output !");
         allGood = false;
      } else {
         ATH_MSG_VERBOSE("io_register[" << this->name() << "](" << incol[icol] << ") [ok]");
      }
   }
   if (!allGood) {
      return(StatusCode::FAILURE);
   }
 
   // Make sure MetaDataSvc is initialized before the first file is opened
   ServiceHandle<IAthMetaDataSvc> metaDataSvc("MetaDataSvc", name());
   ATH_CHECK(metaDataSvc.retrieve());
 
   // Must happen before trying to open a file
   lock_t lock (m_mutex);
   StatusCode risc = this->reinit(lock);
 
   return risc;
}

inputCollectionsHandler()

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

private

Definition at line 63 of file EventSelectorByteStream.cxx.

                                                                               {
  lock_t lock (m_mutex);
  if (this->FSMState() != Gaudi::StateMachine::OFFLINE) {
    this->reinit(lock).ignore();
  }
}

io_reinit()

StatusCode EventSelectorByteStream::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 1052 of file EventSelectorByteStream.cxx.

                                              {
   lock_t lock (m_mutex);
   ATH_MSG_INFO("I/O reinitialization...");
      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);
   }
   std::vector<std::string> inputCollections = m_inputCollectionsProp.value();
   for (std::size_t i = 0, imax = inputCollections.size(); i != imax; ++i) {
      ATH_MSG_INFO("I/O reinitialization, file = "  << 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);
      }
   }
   // all good... copy over.
   m_beginFileFired = false;
 
   // Set m_inputCollectionsProp.  But we _dont_ want to run the update
   // handler --- that calls reinit(), which will deadlock since
   // we're holding the lock.  Instead, we'll call reinit() ourselves.
   auto old_cb = m_inputCollectionsProp.updateCallBack();
   m_inputCollectionsProp.declareUpdateHandler(
      [] (Gaudi::Details::PropertyBase&) {}
   );
   m_inputCollectionsProp = inputCollections;
   m_inputCollectionsProp.declareUpdateHandler (std::move(old_cb));
 
   return(this->reinit(lock));
}

last()

StatusCode EventSelectorByteStream::last ( IEvtSelector::Context &  it ) const

overridevirtual

Parameters

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

Definition at line 693 of file EventSelectorByteStream.cxx.

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

makeClient()

StatusCode EventSelectorByteStream::makeClient ( int  num )

overridevirtual

Make this a client.

Definition at line 959 of file EventSelectorByteStream.cxx.

                                                    {
   lock_t lock (m_mutex);
   if (m_eventStreamingTool.empty()) {
      return(StatusCode::FAILURE);
   }
   std::string dummyStr;
   return(m_eventStreamingTool->makeClient(0, dummyStr));
}

makeServer()

StatusCode EventSelectorByteStream::makeServer ( int  num )

overridevirtual

Make this a server.

Definition at line 950 of file EventSelectorByteStream.cxx.

                                                    {
   lock_t lock (m_mutex);
   if (m_eventStreamingTool.empty()) {
      return(StatusCode::FAILURE);
   }
   return(m_eventStreamingTool->makeServer(1, ""));
}

next() [1/2]

StatusCode EventSelectorByteStream::next ( IEvtSelector::Context &  it ) const

overridevirtual

Parameters

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

Definition at line 343 of file EventSelectorByteStream.cxx.

                                                                      {
   lock_t lock (m_mutex);
   return nextImpl (it, lock);
}

next() [2/2]

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

overridevirtual

Parameters

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

Definition at line 481 of file EventSelectorByteStream.cxx.

                                                                                  {
  lock_t lock (m_mutex);
  return nextImpl (ctxt, jump, lock);
}

nextFile()

void EventSelectorByteStream::nextFile ( lock_t &  lock ) const

private

Definition at line 281 of file EventSelectorByteStream.cxx.

                                                      {
   FileIncident endInputFileIncident(name(), "EndInputFile", "BSF:" + *m_inputCollectionsIterator);
   m_incidentSvc->fireIncident(endInputFileIncident);
   ++m_inputCollectionsIterator;
   ++m_fileCount;
}

nextHandleFileTransition()

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

overrideprotectedvirtual

Handle file transition at the next iteration.

Definition at line 512 of file EventSelectorByteStream.cxx.

{
  lock_t lock (m_mutex);
  return nextHandleFileTransitionImpl (ctxt, lock);
}

nextHandleFileTransitionImpl()

StatusCode EventSelectorByteStream::nextHandleFileTransitionImpl ( IEvtSelector::Context &  ctxt, lock_t &  lock  ) const

private

Definition at line 517 of file EventSelectorByteStream.cxx.

{
   const RawEvent* pre{};
   bool badEvent{};
   // if event source not ready from init, try next file
   if (m_filebased && !m_eventSource->ready()) {
      // next file
      this->nextFile(lock);
      if (this->openNewRun(lock).isFailure()) {
         ATH_MSG_DEBUG("Event source found no more valid files left in input list");
         m_NumEvents = -1;
         return StatusCode::FAILURE;
      }
   }
   try {
      pre = m_eventSource->nextEvent();
   }
   catch (const ByteStreamExceptions::readError&) {
      ATH_MSG_FATAL("Caught ByteStreamExceptions::readError");
      return StatusCode::FAILURE;
   }
   catch (const ByteStreamExceptions::badFragment&) {
      ATH_MSG_ERROR("badFragment encountered");
      badEvent = true;
   }
   catch (const ByteStreamExceptions::badFragmentData&) {
      ATH_MSG_ERROR("badFragment data encountered");
      badEvent = true;
   }
   // Check whether a RawEvent has actually been provided
   if (pre == nullptr) {
      ctxt = *m_endIter;
      return StatusCode::FAILURE;
   }
 
   // If not secondary just return the status code based on if the event is bas
   if (!m_isSecondary.value()) {
      // check bad event flag and handle as configured
      return badEvent ? StatusCode::RECOVERABLE : StatusCode::SUCCESS;
   }
 
   // Build a DH for use by other components
   StatusCode rec_sg = m_eventSource->generateDataHeader();
   if (rec_sg != StatusCode::SUCCESS) {
      ATH_MSG_ERROR("Fail to record BS DataHeader in StoreGate. Skipping events?! " << rec_sg);
   }
 
   return StatusCode::SUCCESS;
}

nextImpl() [1/2]

StatusCode EventSelectorByteStream::nextImpl ( IEvtSelector::Context &  ctxt, int  jump, lock_t &  lock  ) const

private

Definition at line 487 of file EventSelectorByteStream.cxx.

{
   if (jump > 0) {
      if ( m_NumEvents+jump != m_skipEvents.value()) {
         // Save initial event count
         unsigned int cntr = m_NumEvents;
         // In case NumEvents increments multiple times in a single next call
         while (m_NumEvents+1 <= cntr + jump) {
            if (!nextImpl(ctxt, lock).isSuccess()) {
               return(StatusCode::FAILURE);
            }
         }
      }
      else ATH_MSG_DEBUG("Jump covered by skip event " << m_skipEvents.value());
      return(StatusCode::SUCCESS);
   }
   else {
      ATH_MSG_WARNING("Called jump next with non-multiple jump");
   }
   return(StatusCode::SUCCESS);
}

nextImpl() [2/2]

StatusCode EventSelectorByteStream::nextImpl ( IEvtSelector::Context &  it, lock_t &  lock  ) const

private

Definition at line 347 of file EventSelectorByteStream.cxx.

{
    static std::atomic<int> n_bad_events = 0;   // cross loop counter of bad events
   // Check if this is an athenaMP client process
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isClient()) {
      void* source = 0;
      unsigned int status = 0;
      if (!m_eventStreamingTool->getLockedEvent(&source, status).isSuccess()) {
         ATH_MSG_FATAL("Cannot get NextEvent from AthenaSharedMemoryTool");
         return(StatusCode::FAILURE);
      }
      m_eventSource->setEvent(static_cast<char*>(source), status);
      return(StatusCode::SUCCESS);
   }
   // Call all selector tool preNext before starting loop
   for (const ToolHandle<IAthenaSelectorTool>& tool : m_helperTools) {
      if (!tool->preNext().isSuccess()) {
         ATH_MSG_WARNING("Failed to preNext() " << tool->name());
      }
   }
   if (!m_counterTool.empty()) {
      if (!m_counterTool->preNext().isSuccess()) {
         ATH_MSG_WARNING("Failed to preNext() CounterTool.");
      }
   }
   // Find an event to return
   for (;;) {
      bool badEvent{};
      StatusCode sc = nextHandleFileTransitionImpl(it, lock);
      if (sc.isRecoverable()) {
         badEvent = true;
      } else if (sc.isFailure()) {
         return StatusCode::FAILURE;
      }
 
      // increment that an event was found
      ++m_NumEvents;
 
      // check bad event flag and handle as configured
      if (badEvent) {
         int nbad = ++n_bad_events;
         ATH_MSG_WARNING("Bad event encountered, current count at " << nbad);
         bool toomany = (m_maxBadEvts >= 0 && nbad > m_maxBadEvts);
         if (!m_procBadEvent || toomany) {
            ATH_MSG_ERROR("Cannot continue processing: bad event handling disabled or limit exceeded (" << nbad << " events)");
            it = *m_endIter;
            return StatusCode::FAILURE;
         }
         ATH_MSG_WARNING("Continue with bad event");
      }
 
      // Check whether properties or tools reject this event
      if ( m_NumEvents > m_skipEvents.value() &&
           (m_skipEventSequence.empty() || m_NumEvents != m_skipEventSequence.front()) ) {
 
         // Build a DH for use by other components
         StatusCode rec_sg = m_eventSource->generateDataHeader();
         if (rec_sg != StatusCode::SUCCESS) {
            ATH_MSG_ERROR("Fail to record BS DataHeader in StoreGate. Skipping events?! " << rec_sg);
         }
 
         // Build event info attribute list
         if (recordAttributeListImpl(lock).isFailure()) ATH_MSG_WARNING("Unable to build event info att list");
 
         StatusCode status(StatusCode::SUCCESS);
         for (const ToolHandle<IAthenaSelectorTool>& tool : m_helperTools) {
            StatusCode toolStatus = tool->postNext();
            if (toolStatus.isRecoverable()) {
               ATH_MSG_INFO("Request skipping event from: " << tool->name());
               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_NumEvents);
         } else if (status.isFailure()) {
            ATH_MSG_WARNING("Failed to postNext() HelperTool.");
         } else {
               if (!m_counterTool.empty()) {
                  if (!m_counterTool->postNext().isSuccess()) {
                     ATH_MSG_WARNING("Failed to postNext() CounterTool.");
                  }
               }
               // Validate the event
               try {
                  m_eventSource->validateEvent();
               }
               catch (const ByteStreamExceptions::badFragmentData&) {
                  int nbad = ++n_bad_events;
                  ATH_MSG_WARNING("Bad fragment data encountered, current count at " << nbad);
 
                  bool toomany = (m_maxBadEvts >= 0 && nbad > m_maxBadEvts);
                  if (!m_procBadEvent || toomany) {
                     ATH_MSG_ERROR("Cannot continue processing: bad event handling disabled or limit exceeded (" << nbad << " events)");
                     it = *m_endIter;
                     return StatusCode::FAILURE;
                  }
                  ATH_MSG_WARNING("Continue with bad event");
               }
               break;
         }
      } else {
         if (!m_skipEventSequence.empty() && m_NumEvents == m_skipEventSequence.front()) {
            m_skipEventSequence.erase(m_skipEventSequence.begin());
         }
         if ( m_NumEvents % 1'000 == 0 ) {
            ATH_MSG_INFO("Skipping event " << m_NumEvents - 1);
         } else {
            ATH_MSG_DEBUG("Skipping event " << m_NumEvents - 1);
         }
      }
   } // for loop
 
   if (!m_eventStreamingTool.empty() && m_eventStreamingTool->isServer()) { // For SharedReader Server, put event into SHM
      const RawEvent* pre = m_eventSource->currentEvent();
      StatusCode sc;
      while ( (sc = putEvent_ST(*m_eventStreamingTool,
                                m_NumEvents - 1, pre->start(),
                                pre->fragment_size_word() * sizeof(uint32_t),
                                m_eventSource->currentEventStatus())).isRecoverable() ) {
         usleep(1000);
      }
      if (!sc.isSuccess()) {
         ATH_MSG_ERROR("Cannot put Event " << m_NumEvents - 1 << " to AthenaSharedMemoryTool");
         return(StatusCode::FAILURE);
      }
   }
   return(StatusCode::SUCCESS);
}

nextWithSkip()

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

overrideprotectedvirtual

Go to next event and skip if necessary.

Definition at line 568 of file EventSelectorByteStream.cxx.

{
   lock_t lock (m_mutex);
   return nextWithSkipImpl (ctxt, lock);
}

nextWithSkipImpl()

StatusCode EventSelectorByteStream::nextWithSkipImpl ( IEvtSelector::Context &  ctxt, lock_t &  lock  ) const

private

Definition at line 573 of file EventSelectorByteStream.cxx.

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

openNewRun()

StatusCode EventSelectorByteStream::openNewRun ( lock_t &  lock ) const

private

Definition at line 288 of file EventSelectorByteStream.cxx.

                                                                 {
   // Should be protected upstream, but this is further protection
   if (!m_filebased) {
      ATH_MSG_ERROR("cannot open new run for non-filebased inputs");
      return(StatusCode::FAILURE);
   }
   // Check for end of file list
   if (m_inputCollectionsIterator == m_inputCollectionsProp.value().end()) {
      ATH_MSG_INFO("End of input file list reached");
      return(StatusCode::FAILURE);
   }
   std::string blockname = *m_inputCollectionsIterator;
   // try to open a file, if failure go to next FIXME: PVG: silent failure?
   //long nev = m_eventSource->getBlockIterator(blockname);
   auto nevguid = m_eventSource->getBlockIterator(blockname);
   long nev = nevguid.first;
   if (nev == -1) {
      ATH_MSG_FATAL("Unable to access file " << *m_inputCollectionsIterator << ", stopping here");
      throw ByteStreamExceptions::fileAccessError();
   }
   // Fire the incident
   if (!m_beginFileFired) {
     FileIncident beginInputFileIncident(name(), "BeginInputFile", "BSF:" + *m_inputCollectionsIterator,nevguid.second);
     m_incidentSvc->fireIncident(beginInputFileIncident);
     //m_beginFileFired = true;   // Should go here, but can't because IEvtSelector next is const
   }
 
   // check if file is empty
   if (nev == 0) {
      ATH_MSG_WARNING("no events in file " << blockname << " try next");
      if (m_eventSource->ready()) m_eventSource->closeBlockIterator(true);
      this->nextFile(lock);
      return openNewRun(lock);
   // check if skipping all events in that file (minus events already skipped)
   } else if (m_skipEvents.value() - m_NumEvents > nev) {
      ATH_MSG_WARNING("skipping more events " << m_skipEvents.value() - m_NumEvents << "(" << nev <<") than in file " << *m_inputCollectionsIterator << ", try next");
      m_NumEvents += nev;
      m_numEvt[m_fileCount] = nev;
      if (m_eventSource->ready()) m_eventSource->closeBlockIterator(true);
      this->nextFile(lock);
      return openNewRun(lock);
   }
 
   ATH_MSG_DEBUG("Opened block/file " << blockname);
   m_firstEvt[m_fileCount] = m_NumEvents;
   m_numEvt[m_fileCount] = nev;
 
   return(StatusCode::SUCCESS);
}

previous() [1/2]

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

overridevirtual

Parameters

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

Definition at line 611 of file EventSelectorByteStream.cxx.

{
  lock_t lock (m_mutex);
  return previousImpl (ctxt, lock);
}

previous() [2/2]

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

overridevirtual

Parameters

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

Definition at line 672 of file EventSelectorByteStream.cxx.

                                                                                      {
  lock_t lock (m_mutex);
  return previousImpl (ctxt, jump, lock);
}

previousImpl() [1/2]

StatusCode EventSelectorByteStream::previousImpl ( IEvtSelector::Context &  ctxt, int  jump, lock_t &  lock  ) const

private

Definition at line 678 of file EventSelectorByteStream.cxx.

{
   if (jump > 0) {
      for (int i = 0; i < jump; i++) {
         if (!previousImpl(ctxt, lock).isSuccess()) {
            return(StatusCode::FAILURE);
         }
      }
      return(StatusCode::SUCCESS);
   }
   return(StatusCode::FAILURE);
}

previousImpl() [2/2]

StatusCode EventSelectorByteStream::previousImpl ( IEvtSelector::Context &  it, lock_t &  lock  ) const

private

Definition at line 616 of file EventSelectorByteStream.cxx.

                                                                  {
    ATH_MSG_DEBUG(" ... previous");
    const RawEvent* pre = 0;
    bool badEvent(false);
    // if event source not ready from init, try next file
    if (m_eventSource->ready()) {
       try {
          pre = m_eventSource->previousEvent();
       }
       catch (const ByteStreamExceptions::readError&) {
          ATH_MSG_FATAL("Caught ByteStreamExceptions::readError");
          return StatusCode::FAILURE;
       }
       catch (const ByteStreamExceptions::badFragment&) {
          ATH_MSG_ERROR("badFragment encountered");
          badEvent = true;
       }
       catch (const ByteStreamExceptions::badFragmentData&) {
          ATH_MSG_ERROR("badFragment data encountered");
          badEvent = true;
       }
       // Check whether a RawEvent has actually been provided
       if (pre == 0) {
          ATH_MSG_ERROR("No event built");
     //it = *m_endIter;
     return(StatusCode::FAILURE);
       }
    }
    else {
       ATH_MSG_FATAL("Attempt to read previous data on invalid reader");
       return(StatusCode::FAILURE);
    }
    // increment that an event was found
    //++m_NumEvents;
 
    // check bad event flag and handle as configured
    if (badEvent) {
       ATH_MSG_ERROR("Called previous for bad event");
       if (!m_procBadEvent) {
          // End of file
          //it = *m_endIter;
          return(StatusCode::FAILURE);
       }
       ATH_MSG_WARNING("Continue with bad event");
    }
 
    // Build a DH for use by other components
    StatusCode rec_sg = m_eventSource->generateDataHeader();
      if (rec_sg != StatusCode::SUCCESS) {
         ATH_MSG_ERROR("Fail to record BS DataHeader in StoreGate. Skipping events?! " << rec_sg);
    }
 
    return StatusCode::SUCCESS;
}

readEvent()

StatusCode EventSelectorByteStream::readEvent ( int  maxevt )

overridevirtual

Read the next maxevt events.

Parameters

evtnum

[IN] The number of events to read.

Definition at line 986 of file EventSelectorByteStream.cxx.

                                                        {
   lock_t lock (m_mutex);
   if (m_eventStreamingTool.empty()) {
      ATH_MSG_ERROR("No AthenaSharedMemoryTool configured for readEvent()");
      return(StatusCode::FAILURE);
   }
   ATH_MSG_VERBOSE("Called read Event " << maxevt);
   for (int i = 0; i < maxevt || maxevt == -1; ++i) {
      const RawEvent* pre = 0;
      if (this->nextImpl(*m_beginIter, lock).isSuccess()) {
         pre = m_eventSource->currentEvent();
      } else {
         if (m_NumEvents == -1) {
            ATH_MSG_VERBOSE("Called read Event and read last event from input: " << i);
            break;
         }
         ATH_MSG_ERROR("Unable to retrieve next event for " << i << "/" << maxevt);
         return(StatusCode::FAILURE);
      }
      if (m_eventStreamingTool->isServer()) {
         StatusCode sc;
         while ( (sc = putEvent_ST(*m_eventStreamingTool,
                                   m_NumEvents - 1,
                                   pre->start(),
                                   pre->fragment_size_word() * sizeof(uint32_t),
                                   m_eventSource->currentEventStatus())).isRecoverable() ) {
            usleep(1000);
         }
         if (!sc.isSuccess()) {
            ATH_MSG_ERROR("Cannot put Event " << m_NumEvents - 1 << " to AthenaSharedMemoryTool");
            return(StatusCode::FAILURE);
         }
      }
   }
   // End of file, wait for last event to be taken
   StatusCode sc;
   while ( (sc = putEvent_ST(*m_eventStreamingTool, 0, 0, 0, 0)).isRecoverable() ) {
      usleep(1000);
   }
   if (!sc.isSuccess()) {
      ATH_MSG_ERROR("Cannot put last Event marker to AthenaSharedMemoryTool");
      return(StatusCode::FAILURE);
   }
   return(StatusCode::SUCCESS);
}

recordAttributeList()

StatusCode EventSelectorByteStream::recordAttributeList ( ) const

overrideprotectedvirtual

Record AttributeList in StoreGate.

Definition at line 768 of file EventSelectorByteStream.cxx.

{
  lock_t lock (m_mutex);
  return recordAttributeListImpl (lock);
}

recordAttributeListImpl()

StatusCode EventSelectorByteStream::recordAttributeListImpl ( lock_t &  lock ) const

private

Definition at line 773 of file EventSelectorByteStream.cxx.

{
   std::string listName("EventInfoAtts");
 
   if (eventStore()->contains<AthenaAttributeList>(listName)) {
      const AthenaAttributeList* oldAttrList = nullptr;
      if (!eventStore()->retrieve(oldAttrList, listName).isSuccess()) {
         ATH_MSG_ERROR("Cannot retrieve old AttributeList from StoreGate.");
         return(StatusCode::FAILURE);
      }
      if (!eventStore()->removeDataAndProxy(oldAttrList).isSuccess()) {
         ATH_MSG_ERROR("Cannot remove old AttributeList from StoreGate.");
         return(StatusCode::FAILURE);
      }
   }
 
   // build the new attr list
   auto attrList = std::make_unique<AthenaAttributeList>();
 
   // fill the attr list
   ATH_CHECK(fillAttributeListImpl(attrList.get(), "", false, lock));
 
   // put result in event store
   if (eventStore()->record(std::move(attrList), listName).isFailure()) {
      return StatusCode::FAILURE;
   }
 
   return StatusCode::SUCCESS;
}

reinit()

StatusCode EventSelectorByteStream::reinit ( lock_t &  lock )

private

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

Definition at line 172 of file EventSelectorByteStream.cxx.

                                                       {
   ATH_MSG_INFO("reinitialization...");
   // reset markers
   if (m_inputCollectionsProp.value().size()>0) {
      m_numEvt.resize(m_inputCollectionsProp.value().size(), -1);
      m_firstEvt.resize(m_inputCollectionsProp.value().size(), -1);
   }
   else {
      m_numEvt.resize(1);
      m_firstEvt.resize(1);
   }
 
   // Initialize InputCollectionsIterator
   m_inputCollectionsIterator = m_inputCollectionsProp.value().begin();
   m_NumEvents = 0;
   bool retError = false;
   if (!m_helperTools.empty()) {
      for (ToolHandle<IAthenaSelectorTool>& 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);
   }
 
   // If file based input then fire appropriate incidents
   if (m_filebased) {
      if (!m_firstFileFired) {
         FileIncident firstInputFileIncident(name(), "FirstInputFile", "BSF:" + *m_inputCollectionsIterator);
         m_incidentSvc->fireIncident(firstInputFileIncident);
         m_firstFileFired = true;
      }
 
      // try to open a file
      if (this->openNewRun(lock).isFailure()) {
         ATH_MSG_FATAL("Unable to open any file in initialize");
         return(StatusCode::FAILURE);
      }
      // should be in openNewRun, but see comment there
      m_beginFileFired = true;
   }
 
   return(StatusCode::SUCCESS);
}

releaseContext()

StatusCode EventSelectorByteStream::releaseContext ( IEvtSelector::Context *&  it ) const

overridevirtual

Parameters

it	[IN] current event context is released.

Definition at line 1047 of file EventSelectorByteStream.cxx.

                                                                     {
   return(StatusCode::SUCCESS);
}

resetCriteria()

StatusCode EventSelectorByteStream::resetCriteria ( const std::string &  criteria, IEvtSelector::Context &  context  ) const

overridevirtual

Set a selection criteria.

Parameters

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

Definition at line 707 of file EventSelectorByteStream.cxx.

                                                                                                  {
   return(StatusCode::SUCCESS);
}

rewind()

StatusCode EventSelectorByteStream::rewind ( IEvtSelector::Context &  it ) const

overridevirtual

Parameters

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

Definition at line 701 of file EventSelectorByteStream.cxx.

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

seek()

StatusCode EventSelectorByteStream::seek ( Context &  , int  evtnum  ) const

overridevirtual

Seek to a given event number.

Parameters

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

Definition at line 712 of file EventSelectorByteStream.cxx.

                                                                     {
   lock_t lock (m_mutex);
   // Check that input is seekable
   if (!m_filebased) {
      ATH_MSG_ERROR("Input not seekable, choose different input svc");
      return StatusCode::FAILURE;
   }
   // find the file index with that event
   long fileNum = findEvent(evtNum, lock);
   if (fileNum == -1 && evtNum >= m_firstEvt[m_fileCount] && evtNum < m_NumEvents) {
      fileNum = m_fileCount;
   }
   // if unable to locate file, exit
   if (fileNum == -1) {
      ATH_MSG_INFO("seek: Reached end of Input.");
      return(StatusCode::RECOVERABLE);
   }
   // check if it is the current file
   if (fileNum != m_fileCount) { // event in different file
      // Close input file if open
      if (m_eventSource->ready()) m_eventSource->closeBlockIterator(true);
      ATH_MSG_DEBUG("Seek to item: \"" << m_inputCollectionsProp.value()[fileNum] << "\" from the explicit file list.");
      std::string fileName = m_inputCollectionsProp.value()[fileNum];
      m_fileCount = fileNum;
      // Open the correct file
      auto nevguid = m_eventSource->getBlockIterator(fileName);
      long nev = nevguid.first;
      if (nev == -1) {
         ATH_MSG_FATAL("Unable to open file with seeked event " << evtNum << " file " << fileName);
         return StatusCode::FAILURE;
      }
      int delta = evtNum - m_firstEvt[m_fileCount];
      if (delta > 0) {
        EventContextByteStream* beginIter ATLAS_THREAD_SAFE = m_beginIter;
        if (nextImpl(*beginIter,delta, lock).isFailure()) return StatusCode::FAILURE;
      }
   }
   // event in current file
   {
      int delta = (evtNum - m_firstEvt[m_fileCount] + 1) - m_eventSource->positionInBlock();
      ATH_MSG_DEBUG("Seeking event " << evtNum << " in current file with delta " << delta);
      if ( delta == 0 ) { // current event
         // nothing to do
      }
      else if ( delta > 0 ) { // forward
         EventContextByteStream* beginIter ATLAS_THREAD_SAFE = m_beginIter;
         if ( this->nextImpl(*beginIter, delta, lock).isFailure() ) return StatusCode::FAILURE;
      }
      else if ( delta < 0 ) { // backward
         EventContextByteStream* beginIter ATLAS_THREAD_SAFE = m_beginIter;
         if ( this->previousImpl(*beginIter, -1*delta, lock).isFailure() ) return(StatusCode::FAILURE);
      }
   }
   return StatusCode::SUCCESS;
}

share()

StatusCode EventSelectorByteStream::share ( int  evtnum )

overridevirtual

Request to share a given event number.

Parameters

evtnum

[IN] The event number to share.

Definition at line 969 of file EventSelectorByteStream.cxx.

                                                    {
   lock_t lock (m_mutex);
   if (m_eventStreamingTool.empty()) {
      return(StatusCode::FAILURE);
   }
   if (m_eventStreamingTool->isClient()) {
      StatusCode sc = m_eventStreamingTool->lockEvent(evtNum);
      while (sc.isRecoverable()) {
         usleep(1000);
         sc = m_eventStreamingTool->lockEvent(evtNum);
      }
      return(sc);
   }
   return(StatusCode::FAILURE);
}

size()

int EventSelectorByteStream::size ( Context &  it ) const

overridevirtual

Always returns -1.

Parameters

it	[IN/OUT] current event context.

Definition at line 945 of file EventSelectorByteStream.cxx.

                                                   {
  return -1;
}

start()

StatusCode EventSelectorByteStream::start ( )

overridevirtual

Definition at line 222 of file EventSelectorByteStream.cxx.

                                          {
   ATH_MSG_DEBUG("Calling EventSelectorByteStream::start()");
   lock_t lock (m_mutex);
   // Create the begin and end iterator's for this selector.
   m_beginIter =  new EventContextByteStream(this);
   // Increment to get the new event in.
   m_endIter   =  new EventContextByteStream(0);
 
   return(StatusCode::SUCCESS);
}

stop()

StatusCode EventSelectorByteStream::stop ( )

overridevirtual

Definition at line 234 of file EventSelectorByteStream.cxx.

                                         {
   ATH_MSG_DEBUG("Calling EventSelectorByteStream::stop()");
   // Handle open files
   if (m_filebased) {
      // Close the file
      if (m_eventSource->ready()) {
         m_eventSource->closeBlockIterator(false);
         FileIncident endInputFileIncident(name(), "EndInputFile", "stop");
         m_incidentSvc->fireIncident(endInputFileIncident);
      }
   }
   return(StatusCode::SUCCESS);
}

Member Data Documentation

ATLAS_THREAD_SAFE [1/7]

int m_fileCount EventSelectorByteStream::ATLAS_THREAD_SAFE = 0

mutableprivate

number of files to process.

Definition at line 171 of file EventSelectorByteStream.h.

ATLAS_THREAD_SAFE [2/7]

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

mutableprivate

Definition at line 172 of file EventSelectorByteStream.h.

ATLAS_THREAD_SAFE [3/7]

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

mutableprivate

Definition at line 173 of file EventSelectorByteStream.h.

ATLAS_THREAD_SAFE [4/7]

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

mutableprivate

Definition at line 174 of file EventSelectorByteStream.h.

ATLAS_THREAD_SAFE [5/7]

std::vector<long> m_skipEventSequence EventSelectorByteStream::ATLAS_THREAD_SAFE

mutableprivate

Definition at line 175 of file EventSelectorByteStream.h.

ATLAS_THREAD_SAFE [6/7]

long m_NumEvents EventSelectorByteStream::ATLAS_THREAD_SAFE = 0

mutableprivate

Definition at line 176 of file EventSelectorByteStream.h.

ATLAS_THREAD_SAFE [7/7]

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

mutableprivate

Definition at line 177 of file EventSelectorByteStream.h.

m_beginFileFired

bool EventSelectorByteStream::m_beginFileFired {}

private

Definition at line 197 of file EventSelectorByteStream.h.

m_beginIter

EventContextByteStream* EventSelectorByteStream::m_beginIter {}

private

Definition at line 186 of file EventSelectorByteStream.h.

m_counterTool

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

private

Definition at line 201 of file EventSelectorByteStream.h.

m_endIter

EventContextByteStream* EventSelectorByteStream::m_endIter {}

private

Definition at line 187 of file EventSelectorByteStream.h.

m_eventSource

SmartIF<IByteStreamInputSvc> EventSelectorByteStream::m_eventSource

private

Definition at line 188 of file EventSelectorByteStream.h.

m_eventSourceName

Gaudi::Property<std::string> EventSelectorByteStream::m_eventSourceName {this, "ByteStreamInputSvc", "", ""}

private

Definition at line 182 of file EventSelectorByteStream.h.

m_eventsPerLB

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

private

Definition at line 213 of file EventSelectorByteStream.h.

m_eventsPerRun

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

private

Definition at line 211 of file EventSelectorByteStream.h.

m_filebased

Gaudi::Property<bool> EventSelectorByteStream::m_filebased {this, "FileBased", true, ""}

private

Definition at line 207 of file EventSelectorByteStream.h.

m_firstEventNo

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

private

Definition at line 210 of file EventSelectorByteStream.h.

m_firstFileFired

bool EventSelectorByteStream::m_firstFileFired {}

private

Definition at line 196 of file EventSelectorByteStream.h.

m_firstLBNo

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

private

Definition at line 212 of file EventSelectorByteStream.h.

m_helperTools

ToolHandleArray<IAthenaSelectorTool> EventSelectorByteStream::m_helperTools {this}

private

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

Definition at line 200 of file EventSelectorByteStream.h.

m_incidentSvc

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

private

Definition at line 191 of file EventSelectorByteStream.h.

m_initTimeStamp

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

private

Definition at line 214 of file EventSelectorByteStream.h.

m_inputCollectionsProp

Gaudi::Property<std::vector<std::string> > EventSelectorByteStream::m_inputCollectionsProp {this, "Input", {}, ""}

private

Definition at line 189 of file EventSelectorByteStream.h.

m_isSecondary

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

private

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

Definition at line 180 of file EventSelectorByteStream.h.

m_maxBadEvts

Gaudi::Property<int> EventSelectorByteStream::m_maxBadEvts {this, "MaxBadEvents", -1, ""}

private

number of bad events allowed before quitting.

Definition at line 184 of file EventSelectorByteStream.h.

m_mutex

mutex_t EventSelectorByteStream::m_mutex

mutableprivate

Definition at line 170 of file EventSelectorByteStream.h.

m_overrideRunNumber

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

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 206 of file EventSelectorByteStream.h.

m_procBadEvent

Gaudi::Property<bool> EventSelectorByteStream::m_procBadEvent {this, "ProcessBadEvent", false, ""}

private

process bad events, which fail check_tree().

Definition at line 183 of file EventSelectorByteStream.h.

m_runNo

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

private

Definition at line 209 of file EventSelectorByteStream.h.

m_skipEvents

Gaudi::Property<long> EventSelectorByteStream::m_skipEvents {this, "SkipEvents", 0, ""}

private

Definition at line 193 of file EventSelectorByteStream.h.

m_skipEventSequenceProp

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

private

Definition at line 194 of file EventSelectorByteStream.h.

m_timeStampInterval

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

private

Definition at line 215 of file EventSelectorByteStream.h.

---

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

• EventSelectorByteStream.h
• EventSelectorByteStream.cxx