ByteStreamEventStorageInputSvc Class Reference

This class is the ByteStreamInputSvc for reading events written by EventStorage. More...

#include <ByteStreamEventStorageInputSvc.h>

Inheritance diagram for ByteStreamEventStorageInputSvc:

Collaboration diagram for ByteStreamEventStorageInputSvc:

Classes
struct	EventCache

Public Member Functions
	ByteStreamEventStorageInputSvc (const std::string &name, ISvcLocator *pSvcLocator)
	Constructors:
virtual	~ByteStreamEventStorageInputSvc ()
	Destructor.
virtual StatusCode	initialize () override
	Required of all Gaudi Services.
virtual StatusCode	stop () override
virtual StatusCode	finalize () override
virtual const RawEvent *	currentEvent () const override
	Implementation of the IByteStreamInputSvc interface methods.
virtual const RawEvent *	nextEvent () override
	++, new
virtual const RawEvent *	previousEvent () override
	–, old
virtual void	setEvent (void *data, unsigned int eventStatus) override
virtual unsigned int	currentEventStatus () const override
	Return the current event status.
virtual void	validateEvent () override
virtual long	positionInBlock () override
virtual std::pair< long, std::string >	getBlockIterator (const std::string &fileName) override
virtual void	closeBlockIterator (bool clearMetadata=true) override
virtual bool	setSequentialRead ()
virtual bool	ready () override
virtual StatusCode	generateDataHeader () override

Private Types
enum	Advance { PREVIOUS = -1 , NEXT = 1 }

Private Member Functions
StatusCode	loadMetadata ()
void	buildFragment (EventCache *cache, uint32_t eventSize, bool validate) const
bool	readerReady ()
bool	ROBFragmentCheck (const RawEvent *) const
unsigned	validateEvent (const RawEvent *const rawEvent) const
void	setEvent (const EventContext &context, void *data, unsigned int eventStatus)
const RawEvent *	getEvent (Advance step)
std::unique_ptr< DataHeaderElement >	makeBSProvenance () const
template<typename T>
StatusCode	deleteEntry (const std::string &key)

Private Attributes
std::mutex	m_readerMutex
SG::SlotSpecificObj< EventCache >	m_eventsCache
std::unique_ptr< EventStorage::DataReader >	m_reader
	DataReader from EventStorage.
std::vector< long long int >	m_evtOffsets
	offset for event i in that file
unsigned int	m_evtInFile
long long int	m_evtFileOffset
	last read in event offset within a file, can be -1
std::string	m_fileGUID
	current file GUID
ServiceHandle< StoreGateSvc >	m_storeGate {this, "EventStore", "StoreGateSvc"}
	Pointer to StoreGate.
ServiceHandle< StoreGateSvc >	m_inputMetadata {this, "MetaDataStore", "StoreGateSvc/InputMetaDataStore"}
ServiceHandle< IROBDataProviderSvc >	m_robProvider
Gaudi::Property< bool >	m_sequential {this, "EnableSequential", false, "enable sequential reading"}
Gaudi::Property< bool >	m_dump {this, "DumpFlag", false, "Dump fragments"}
Gaudi::Property< float >	m_wait {this, "WaitSecs", 0.0f, "Seconds to wait if input is in wait state"}
Gaudi::Property< bool >	m_valEvent {this, "ValidateEvent", false, "switch on check_tree when reading events"}
Gaudi::Property< std::string >	m_eventInfoKey {this, "EventInfoKey", "EventInfo", "Key of EventInfo in metadata store"}

Detailed Description

This class is the ByteStreamInputSvc for reading events written by EventStorage.

Definition at line 35 of file ByteStreamEventStorageInputSvc.h.

Member Enumeration Documentation

Advance

enum ByteStreamEventStorageInputSvc::Advance

private

Enumerator
PREVIOUS
NEXT

Definition at line 113 of file ByteStreamEventStorageInputSvc.h.

{ PREVIOUS = -1, NEXT = 1 };

Constructor & Destructor Documentation

ByteStreamEventStorageInputSvc()

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

Constructors:

Definition at line 38 of file ByteStreamEventStorageInputSvc.cxx.

  : base_class(name, pSvcLocator)
  , m_readerMutex()
  , m_eventsCache()
  , m_reader()
  , m_evtOffsets()
  , m_evtInFile(0)
  , m_evtFileOffset(0)
  , m_fileGUID("")
  , m_robProvider("ROBDataProviderSvc", name)
{
  assert(pSvcLocator != nullptr);
}

~ByteStreamEventStorageInputSvc()

ByteStreamEventStorageInputSvc::~ByteStreamEventStorageInputSvc ( )

virtualdefault

Destructor.

Member Function Documentation

buildFragment()

void ByteStreamEventStorageInputSvc::buildFragment ( EventCache * cache, uint32_t eventSize, bool validate ) const

private

Definition at line 304 of file ByteStreamEventStorageInputSvc.cxx.

                                                                {
  using OFFLINE_FRAGMENTS_NAMESPACE::DataType;
  using OFFLINE_FRAGMENTS_NAMESPACE::PointerType;
  DataType* fragment = reinterpret_cast<DataType*>(cache->data);
 
  if (validate) {
    // check fragment type
    const DataType headWord = fragment[0];
    ATH_MSG_DEBUG("First word of the fragment " << MSG::hex << headWord << MSG::dec);
    // format version
    const DataType formatVersion = eformat::helper::Version(fragment[3]).major_version();
    ATH_MSG_DEBUG("Format version" << MSG::hex << formatVersion << MSG::dec);
    // error message
    if((formatVersion != eformat::MAJOR_DEFAULT_VERSION) &&
       (formatVersion != eformat::MAJOR_V24_VERSION) &&
       (formatVersion != eformat::MAJOR_V30_VERSION) &&
       (formatVersion != eformat::MAJOR_V40_VERSION) &&
       (formatVersion != eformat::MAJOR_V31_VERSION) ) {
      ATH_MSG_ERROR("unsupported Format Version : "
          << MSG::hex << formatVersion << MSG::dec);
    }
 
    if(eformat::FULL_EVENT == headWord || 0xcc1234cc == headWord) { // ROS = 0xcc1234cc
      try {
        // convert old fragment
        if(formatVersion != eformat::MAJOR_DEFAULT_VERSION) {
          // 100 for increase of data-size due to header conversion
          const uint32_t newEventSize = eventSize + 1000;
          DataType* newFragment  = new DataType[newEventSize];
          eformat::old::convert(fragment, newFragment, newEventSize);
 
          // delete old fragment
          delete [] fragment; fragment = nullptr;
 
          // set new pointer
          fragment = newFragment;
          cache->data = reinterpret_cast< char* >(fragment);
        }
      } catch (const eformat::Issue& ex) {
        // bad event
        ATH_MSG_WARNING(ex.what());
        ATH_MSG_ERROR("Skipping bad event");
        throw ByteStreamExceptions::badFragment();
      }
    } else {
      // Unknown fragment
      ATH_MSG_FATAL("Unknown Header work in input fragment " << MSG::hex << headWord);
      throw ByteStreamExceptions::badFragment();
    }
  }
  if (!fragment){
    ATH_MSG_ERROR("fragment is nullptr!");
    throw ByteStreamExceptions::badFragment();
  }
  // This is a FullEventFragment
  // make a new FEFrag in memory from it
  cache->eventStatus = 0;
  if(fragment[5] > 0) {
    cache->eventStatus += eformat::helper::Status(fragment[6]).specific();
    cache->eventStatus += (eformat::helper::Status(fragment[6]).generic() & 0x000000ff) << 16;
  }
 
  // This is a FullEventFragment
  // make a new RawEvent in memory from it
  cache->rawEvent = std::make_unique<RawEvent>(fragment);
  ATH_MSG_DEBUG("Made an FullEventFragment from ES " << fragment);
}

closeBlockIterator()

void ByteStreamEventStorageInputSvc::closeBlockIterator ( bool clearMetadata = true )

overridevirtual

Definition at line 469 of file ByteStreamEventStorageInputSvc.cxx.

{
  if (clearMetadata) {
    ATH_MSG_WARNING("Clearing input metadata store");
    StatusCode status = m_inputMetadata->clearStore();
    if (!status.isSuccess()) {
      ATH_MSG_WARNING("Unable to clear Input MetaData Proxies");
    }
  }
 
  if (!readerReady()) {
    ATH_MSG_INFO("No more events in this run, high water mark for this file = "
        << m_evtOffsets.size()-1);
  }
 
  m_reader.reset();
}

currentEvent()

const RawEvent * ByteStreamEventStorageInputSvc::currentEvent ( ) const

overridevirtual

Implementation of the IByteStreamInputSvc interface methods.

Definition at line 636 of file ByteStreamEventStorageInputSvc.cxx.

{
  const EventContext& context{Gaudi::Hive::currentContext()};
  return m_eventsCache.get(context)->rawEvent.get();
}

currentEventStatus()

unsigned int ByteStreamEventStorageInputSvc::currentEventStatus ( ) const

overridevirtual

Return the current event status.

Definition at line 645 of file ByteStreamEventStorageInputSvc.cxx.

{
  const EventContext& context{Gaudi::Hive::currentContext()};
  return m_eventsCache.get(context)->eventStatus;
}

deleteEntry()

template<typename T>

StatusCode ByteStreamEventStorageInputSvc::deleteEntry ( const std::string & key )

inlineprivate

Definition at line 118 of file ByteStreamEventStorageInputSvc.h.

  {
    if (m_storeGate->contains<T>(key)) {
      const T* tmp = m_storeGate->tryConstRetrieve<T>(key);
      if (tmp != nullptr) ATH_CHECK(m_storeGate->remove<T>(tmp));
    }
    return StatusCode::SUCCESS;
  }

finalize()

StatusCode ByteStreamEventStorageInputSvc::finalize ( )

overridevirtual

Definition at line 83 of file ByteStreamEventStorageInputSvc.cxx.

                                         {
  return StatusCode::SUCCESS;
}

generateDataHeader()

StatusCode ByteStreamEventStorageInputSvc::generateDataHeader ( )

overridevirtual

Definition at line 376 of file ByteStreamEventStorageInputSvc.cxx.

{
  std::lock_guard<std::mutex> lock(m_readerMutex);
 
  // get file GUID
  m_fileGUID = m_reader->GUID();
 
  // reader returns -1 when end of the file is reached
  if(m_evtFileOffset != -1) {
    ATH_MSG_DEBUG("ByteStream File GUID:              " << m_fileGUID);
    ATH_MSG_DEBUG("ByteStream Event Position in File: " << m_evtFileOffset);
 
    // To accommodate for skipEvents option in EventSelector
    // While skipping BS event Selector does not return SUCCESS code,
    // just advances silently through events. So SG content is not refreshed
    // Lets do refresh of the event header here
    std::string key = "ByteStreamDataHeader";
    ATH_CHECK(deleteEntry<DataHeader>(key));
 
    // Created data header element with BS provenance information
    std::unique_ptr<DataHeaderElement> dataHeaderElement = makeBSProvenance();
    // Create data header itself
    std::unique_ptr<DataHeader> dataHeader = std::make_unique<DataHeader>();
    // Declare header primary
    dataHeader->setStatus(DataHeader::Input);
    // Set processTag
    dataHeader->setProcessTag(dataHeaderElement->getKey());
    //add the data header element self reference to the object vector
    dataHeader->insert(*std::move(dataHeaderElement));
 
    // Clean up EventInfo from the previous event
    key = m_eventInfoKey.value();
    ATH_CHECK(deleteEntry<xAOD::EventInfo>(key));
    // Now add ref to xAOD::EventInfo
    std::unique_ptr<IOpaqueAddress> iopx = std::make_unique<ByteStreamAddress>(
        ClassID_traits<xAOD::EventInfo>::ID(), key, "");
    ATH_CHECK(m_storeGate->recordAddress(key, std::move(iopx)));
    const SG::DataProxy* ptmpx = m_storeGate->transientProxy(
        ClassID_traits<xAOD::EventInfo>::ID(), key);
    if (ptmpx != nullptr) {
      DataHeaderElement element(ptmpx, 0, key);
      dataHeader->insert(element);
    }
 
    // Clean up auxiliary EventInfo from the previous event
    key = m_eventInfoKey.value() + "Aux.";
    ATH_CHECK(deleteEntry<xAOD::EventAuxInfo>(key));
    // Now add ref to xAOD::EventAuxInfo
    std::unique_ptr<IOpaqueAddress> iopaux = std::make_unique<ByteStreamAddress>(
        ClassID_traits<xAOD::EventAuxInfo>::ID(), key, "");
    ATH_CHECK(m_storeGate->recordAddress(key, std::move(iopaux)));
    const SG::DataProxy* ptmpaux = m_storeGate->transientProxy(
        ClassID_traits<xAOD::EventAuxInfo>::ID(), key);
    if (ptmpaux !=0) {
      DataHeaderElement element(ptmpaux, 0, key);
      dataHeader->insert(element);
    }
 
    // Record new data header.Boolean flags will allow it's deletion in case
    // of skipped events.
    ATH_CHECK(m_storeGate->record<DataHeader>(dataHeader.release(),
          "ByteStreamDataHeader", true, false, true));
  }
  return StatusCode::SUCCESS;
}

getBlockIterator()

std::pair< long, std::string > ByteStreamEventStorageInputSvc::getBlockIterator ( const std::string & fileName )

overridevirtual

Definition at line 508 of file ByteStreamEventStorageInputSvc.cxx.

{
  // open the file
  if(m_reader != 0) closeBlockIterator();
 
  m_reader = std::unique_ptr<EventStorage::DataReader>(pickDataReader(fileName));
 
  if(m_reader == nullptr) {
    ATH_MSG_ERROR("Failed to open file " << fileName);
    closeBlockIterator();
    return std::make_pair(-1,"END");
  }
 
  // Initialize offset vector
  m_evtOffsets.resize(m_reader->eventsInFile(), -1);
  m_evtOffsets.clear();
 
  // Get ByteStream Metadata into Input MetaData Store
  // (include begin Input File Incident)
  if (loadMetadata().isSuccess()) {
    ATH_MSG_DEBUG("Recorded ByteStreamMetadata in InputMetaDataStore");
  } else {
    ATH_MSG_ERROR("Unable to record ByteStreamMetadata in InputMetaDataStore");
    return std::make_pair(-1, "FAIL");
  }
 
  m_evtInFile = 0;
 
  // enable sequentialReading if multiple files
  if(m_sequential) {
    bool test = setSequentialRead();
    if (!test) return std::make_pair(-1,"SEQ");
  }
 
  ATH_MSG_INFO("Picked valid file: " << m_reader->fileName());
  // initialize offsets and counters
  m_evtOffsets.push_back(static_cast<long long>(m_reader->getPosition()));
  return std::make_pair(m_reader->eventsInFile(), m_reader->GUID());
}

getEvent()

const RawEvent * ByteStreamEventStorageInputSvc::getEvent ( Advance step )

private

initialize()

StatusCode ByteStreamEventStorageInputSvc::initialize ( )

overridevirtual

Required of all Gaudi Services.

Definition at line 60 of file ByteStreamEventStorageInputSvc.cxx.

{
  ATH_MSG_INFO("Initializing");
 
  ATH_CHECK(m_inputMetadata.retrieve());
  ATH_CHECK(m_storeGate.retrieve());
  ATH_CHECK(m_robProvider.retrieve());
 
  return StatusCode::SUCCESS;
}

loadMetadata()

StatusCode ByteStreamEventStorageInputSvc::loadMetadata ( )

private

Definition at line 99 of file ByteStreamEventStorageInputSvc.cxx.

{
  // default goes into ByteStreamMetadata
  auto bsmdc = std::make_unique<ByteStreamMetadataContainer>();
  bsmdc->push_back(std::make_unique<ByteStreamMetadata>(*m_reader));
  ATH_MSG_DEBUG("ByteStreamMetadata:\n" << *(bsmdc->front()));
 
  return m_inputMetadata->record(std::move(bsmdc), "ByteStreamMetadata");
}

makeBSProvenance()

std::unique_ptr< DataHeaderElement > ByteStreamEventStorageInputSvc::makeBSProvenance ( ) const

private

Definition at line 654 of file ByteStreamEventStorageInputSvc.cxx.

{
    Token token;
    token.setDb(m_fileGUID);
    token.setTechnology(0x00001000);
    token.setOid(Token::OID_t(0LL, m_evtFileOffset));
 
    return std::make_unique<DataHeaderElement>(ClassID_traits<DataHeader>::ID(),
        "StreamRAW", std::move(token));
}

nextEvent()

const RawEvent * ByteStreamEventStorageInputSvc::nextEvent ( )

overridevirtual

++, new

Definition at line 179 of file ByteStreamEventStorageInputSvc.cxx.

                                          {
 
  std::lock_guard<std::mutex> lock( m_readerMutex );
  const EventContext& context{ Gaudi::Hive::currentContext() };
  EventCache* cache = m_eventsCache.get(context);
 
  // initialize before building RawEvent
  cache->releaseEvent();
 
  // Load data buffer from file
  unsigned int eventSize;
  if (readerReady()) {
    DRError ecode;
    // Check if have moved back from high water mark
    m_evtInFile++; // increment iterator
    if (m_evtInFile+1 > m_evtOffsets.size()) {
      //get current event position (cast to long long until native tdaq implementation)
      ATH_MSG_DEBUG("nextEvent _above_ high water mark");
      m_evtFileOffset = static_cast<long long>(m_reader->getPosition());
      m_evtOffsets.push_back(m_evtFileOffset);
      ecode = m_reader->getData(eventSize, &(cache->data));
    } else {
      // Load from previous offset
      ATH_MSG_DEBUG("nextEvent below high water mark");
      m_evtFileOffset = m_evtOffsets.at(m_evtInFile - 1);
      ecode = m_reader->getData(eventSize, &(cache->data), m_evtFileOffset);
    }
 
    if (DRWAIT == ecode && m_wait > 0) {
      do {
        // wait for n seconds
        ATH_MSG_DEBUG("Waiting for input:  " << m_wait << " seconds");
        int result = usleep(static_cast<int>(m_wait * 1e6));
        if (result != 0) {
          ATH_MSG_ERROR("System Error while running sleep");
          return nullptr;
        }
      } while(m_reader->getData(eventSize, &(cache->data)) == DRWAIT);
    } else if (DROK != ecode) {
      ATH_MSG_ERROR("Error reading next event");
      throw ByteStreamExceptions::readError();
    }
    ATH_MSG_DEBUG("Event Size " << eventSize);
 
  } else {
    ATH_MSG_ERROR("DataReader not ready. Need to getBlockIterator first");
    return nullptr;
  }
 
  // Use buffer to build FullEventFragment
  try {
    buildFragment(cache, eventSize, true);
  }
  catch (...) {
    // rethrow any exceptions
    throw;
  }
 
  if (cache->rawEvent == nullptr) {
    ATH_MSG_ERROR("Failure to build fragment");
    return nullptr;
  }
 
  // Set it for the data provider
  m_robProvider->setNextEvent(context, cache->rawEvent.get());
  m_robProvider->setEventStatus(context, cache->eventStatus);
 
  //++m_totalEventCounter;
 
  // dump
  if (m_dump) {
    DumpFrags::dump(cache->rawEvent.get());
  }
  ATH_MSG_DEBUG("switched to next event in slot " << context);
  return cache->rawEvent.get();
}

positionInBlock()

long ByteStreamEventStorageInputSvc::positionInBlock ( )

overridevirtual

Definition at line 91 of file ByteStreamEventStorageInputSvc.cxx.

{
  return m_evtOffsets.size();
}

previousEvent()

const RawEvent * ByteStreamEventStorageInputSvc::previousEvent ( )

overridevirtual

–, old

Definition at line 112 of file ByteStreamEventStorageInputSvc.cxx.

{
  std::lock_guard<std::mutex> lock(m_readerMutex);
  const EventContext& context{Gaudi::Hive::currentContext()};
  EventCache* cache = m_eventsCache.get(context);
  // initialize before building RawEvent
  cache->releaseEvent();
 
  // Load data buffer from file
  unsigned int eventSize;
  if (readerReady()) {
    //get current event position (cast to long long until native tdaq implementation)
    m_evtInFile--;
    m_evtFileOffset = m_evtOffsets.at(m_evtInFile);
    DRError ecode = m_reader->getData(eventSize, &(cache->data), m_evtOffsets.at(m_evtInFile - 1));
 
    if (DRWAIT == ecode && m_wait > 0) {
      do {
        // wait for n seconds
        ATH_MSG_DEBUG("Waiting for input:  " << m_wait << " seconds");
        int result = usleep(static_cast<int>(m_wait * 1e6));
        if (result != 0) {
          ATH_MSG_ERROR("System Error while running sleep");
          return nullptr;
        }
      } while(m_reader->getData(eventSize, &(cache->data), m_evtFileOffset) == DRWAIT);
    } else if (DROK != ecode) {
      ATH_MSG_ERROR("Error reading previous event");
      throw ByteStreamExceptions::readError();
    }
    ATH_MSG_DEBUG("Event Size " << eventSize);
  }
  else {
    ATH_MSG_ERROR("DataReader not ready. Need to getBlockIterator first");
    return nullptr;
  }
 
  // Use buffer to build FullEventFragment
  try {
    buildFragment(cache, eventSize, true);
  }
  catch (...) {
    // rethrow any exceptions
    throw;
  }
 
  if (cache->rawEvent == nullptr) {
    ATH_MSG_ERROR("Failure to build fragment");
    return nullptr;
  }
 
  // Set it for the data provider
  m_robProvider->setNextEvent(context, cache->rawEvent.get());
  m_robProvider->setEventStatus(context, cache->eventStatus);
 
  // dump
  if (m_dump) {
    DumpFrags::dump(cache->rawEvent.get());
  }
  ATH_MSG_DEBUG( "switched to previous event in slot " << context);
  return cache->rawEvent.get();
}

readerReady()

bool ByteStreamEventStorageInputSvc::readerReady ( )

private

Definition at line 551 of file ByteStreamEventStorageInputSvc.cxx.

{
  bool eofFlag(false);
 
  if (m_reader) eofFlag = m_reader->endOfFile();
  else {
    ATH_MSG_INFO("eformat reader object not initialized");
    return false;
  }
 
  bool moreEvent = m_reader->good();
 
  return (!eofFlag) && moreEvent;
}

ready()

bool ByteStreamEventStorageInputSvc::ready ( )

overridevirtual

Definition at line 500 of file ByteStreamEventStorageInputSvc.cxx.

{
  return readerReady();
}

ROBFragmentCheck()

bool ByteStreamEventStorageInputSvc::ROBFragmentCheck ( const RawEvent * re ) const

private

Definition at line 569 of file ByteStreamEventStorageInputSvc.cxx.

{
  bool     allOK  = true;
  uint32_t total  = re->nchildren();
  uint32_t lastId = 0;
  std::vector<eformat::FragmentProblem> problems;
 
  for (size_t i = 0; i<total; ++i) {
    OFFLINE_FRAGMENTS_NAMESPACE::PointerType fp;
    re->child(fp, i);
 
    OFFLINE_FRAGMENTS_NAMESPACE::ROBFragment f(fp);
    lastId = f.source_id();
 
    problems.clear();
    f.problems(problems);
    if(!problems.empty()) {
      allOK = false;
      for(const auto& problem : problems) {
        ATH_MSG_WARNING("Failed to create ROBFragment id = " << lastId << ", "
            << eformat::helper::SourceIdentifier(lastId).human()  << " : "
            << eformat::helper::FragmentProblemDictionary.string(problem));
      }
    }
  }
 
  return allOK;
}

setEvent() [1/2]

void ByteStreamEventStorageInputSvc::setEvent ( const EventContext & context, void * data, unsigned int eventStatus )

private

Definition at line 610 of file ByteStreamEventStorageInputSvc.cxx.

{
  using OFFLINE_FRAGMENTS_NAMESPACE::DataType;
  EventCache* cache = m_eventsCache.get(context);
  cache->releaseEvent();
 
  DataType* fragment = reinterpret_cast<DataType*>(data);
  cache->rawEvent = std::make_unique<RawEvent>(fragment);
  cache->eventStatus = eventStatus;
 
  // Set it for the data provider
  m_robProvider->setNextEvent(context, cache->rawEvent.get());
  m_robProvider->setEventStatus(context, cache->eventStatus);
 
  // Build a DH for use by other components
  StatusCode rec_sg = generateDataHeader();
  if (rec_sg != StatusCode::SUCCESS) {
    ATH_MSG_ERROR("Fail to record BS DataHeader in StoreGate. Skipping events?! "
        << rec_sg);
  }
}

setEvent() [2/2]

void ByteStreamEventStorageInputSvc::setEvent ( void * data, unsigned int eventStatus )

overridevirtual

Definition at line 601 of file ByteStreamEventStorageInputSvc.cxx.

{
  const EventContext& context{Gaudi::Hive::currentContext()};
  return setEvent(context, data, eventStatus);
}

setSequentialRead()

bool ByteStreamEventStorageInputSvc::setSequentialRead ( )

virtual

Definition at line 490 of file ByteStreamEventStorageInputSvc.cxx.

{
  // enable SequenceReading
  m_reader->enableSequenceReading();
  return true;
}

stop()

StatusCode ByteStreamEventStorageInputSvc::stop ( )

overridevirtual

Definition at line 74 of file ByteStreamEventStorageInputSvc.cxx.

{
  // close moved to EventSelector for explicit coupling with incident
  return StatusCode::SUCCESS;
}

validateEvent() [1/2]

void ByteStreamEventStorageInputSvc::validateEvent ( )

overridevirtual

Definition at line 259 of file ByteStreamEventStorageInputSvc.cxx.

{
  const EventContext& context{Gaudi::Hive::currentContext()};
  const RawEvent* const event = m_eventsCache.get(context)->rawEvent.get();
  m_eventsCache.get(context)->eventStatus = validateEvent(event);
}

validateEvent() [2/2]

unsigned ByteStreamEventStorageInputSvc::validateEvent ( const RawEvent *const rawEvent ) const

private

Definition at line 269 of file ByteStreamEventStorageInputSvc.cxx.

{
  unsigned int status = 0;
  if (m_valEvent) {
    // check validity
    std::vector<eformat::FragmentProblem> problems;
    rawEvent->problems(problems);
 
    if(!problems.empty()) {
      status += 0x01000000;
 
      // bad event
      ATH_MSG_WARNING("Failed to create FullEventFragment");
      for(const auto& problem : problems)
        ATH_MSG_WARNING(eformat::helper::FragmentProblemDictionary.string(problem));
 
      throw ByteStreamExceptions::badFragmentData();
    }
 
    if(!ROBFragmentCheck(rawEvent)) {
      status += 0x02000000;
 
      // bad event
      ATH_MSG_ERROR("Skipping bad event");
      throw ByteStreamExceptions::badFragmentData();
    }
  } else {
    ATH_MSG_DEBUG("Processing event without validating.");
  }
  return status;
}

Member Data Documentation

m_dump

Gaudi::Property<bool> ByteStreamEventStorageInputSvc::m_dump {this, "DumpFlag", false, "Dump fragments"}

private

Definition at line 99 of file ByteStreamEventStorageInputSvc.h.

{this, "DumpFlag", false, "Dump fragments"};

m_eventInfoKey

Gaudi::Property<std::string> ByteStreamEventStorageInputSvc::m_eventInfoKey {this, "EventInfoKey", "EventInfo", "Key of EventInfo in metadata store"}

private

Definition at line 102 of file ByteStreamEventStorageInputSvc.h.

{this, "EventInfoKey", "EventInfo", "Key of EventInfo in metadata store"};

m_eventsCache

SG::SlotSpecificObj<EventCache> ByteStreamEventStorageInputSvc::m_eventsCache

private

Definition at line 81 of file ByteStreamEventStorageInputSvc.h.

m_evtFileOffset

long long int ByteStreamEventStorageInputSvc::m_evtFileOffset

private

last read in event offset within a file, can be -1

Definition at line 87 of file ByteStreamEventStorageInputSvc.h.

m_evtInFile

unsigned int ByteStreamEventStorageInputSvc::m_evtInFile

private

Definition at line 86 of file ByteStreamEventStorageInputSvc.h.

m_evtOffsets

std::vector<long long int> ByteStreamEventStorageInputSvc::m_evtOffsets

private

offset for event i in that file

Definition at line 85 of file ByteStreamEventStorageInputSvc.h.

m_fileGUID

std::string ByteStreamEventStorageInputSvc::m_fileGUID

private

current file GUID

Definition at line 89 of file ByteStreamEventStorageInputSvc.h.

m_inputMetadata

ServiceHandle<StoreGateSvc> ByteStreamEventStorageInputSvc::m_inputMetadata {this, "MetaDataStore", "StoreGateSvc/InputMetaDataStore"}

private

Definition at line 96 of file ByteStreamEventStorageInputSvc.h.

{this, "MetaDataStore", "StoreGateSvc/InputMetaDataStore"};

m_reader

std::unique_ptr<EventStorage::DataReader> ByteStreamEventStorageInputSvc::m_reader

private

DataReader from EventStorage.

Definition at line 83 of file ByteStreamEventStorageInputSvc.h.

m_readerMutex

std::mutex ByteStreamEventStorageInputSvc::m_readerMutex

private

Definition at line 70 of file ByteStreamEventStorageInputSvc.h.

m_robProvider

ServiceHandle<IROBDataProviderSvc> ByteStreamEventStorageInputSvc::m_robProvider

private

Definition at line 97 of file ByteStreamEventStorageInputSvc.h.

m_sequential

Gaudi::Property<bool> ByteStreamEventStorageInputSvc::m_sequential {this, "EnableSequential", false, "enable sequential reading"}

private

Definition at line 98 of file ByteStreamEventStorageInputSvc.h.

{this, "EnableSequential", false, "enable sequential reading"};

m_storeGate

ServiceHandle<StoreGateSvc> ByteStreamEventStorageInputSvc::m_storeGate {this, "EventStore", "StoreGateSvc"}

private

Pointer to StoreGate.

Definition at line 95 of file ByteStreamEventStorageInputSvc.h.

{this, "EventStore", "StoreGateSvc"};

m_valEvent

Gaudi::Property<bool> ByteStreamEventStorageInputSvc::m_valEvent {this, "ValidateEvent", false, "switch on check_tree when reading events"}

private

Definition at line 101 of file ByteStreamEventStorageInputSvc.h.

{this, "ValidateEvent", false, "switch on check_tree when reading events"};

m_wait

Gaudi::Property<float> ByteStreamEventStorageInputSvc::m_wait {this, "WaitSecs", 0.0f, "Seconds to wait if input is in wait state"}

private

Definition at line 100 of file ByteStreamEventStorageInputSvc.h.

{this, "WaitSecs", 0.0f, "Seconds to wait if input is in wait state"};

---

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

• ByteStreamEventStorageInputSvc.h
• ByteStreamEventStorageInputSvc.cxx