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: More...
virtual	~ByteStreamEventStorageInputSvc ()
	Destructor. More...
virtual StatusCode	initialize () override
	Required of all Gaudi Services. More...
virtual StatusCode	stop () override
virtual StatusCode	finalize () override
virtual StatusCode	queryInterface (const InterfaceID &riid, void **ppvInterface) override
	Required of all Gaudi services: see Gaudi documentation for details. More...
virtual const RawEvent *	currentEvent () const override
	Implementation of the ByteStreamInputSvc interface methods. More...
virtual const RawEvent *	nextEvent () override
	++, new More...
virtual const RawEvent *	previousEvent () override
	–, old More...
virtual void	setEvent (void *data, unsigned int eventStatus) override
virtual unsigned int	currentEventStatus () const override
	Return the current event status. More...
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
MsgStream &	msg () const
MsgStream &	msg (const MSG::Level lvl) const
bool	msgLvl (const MSG::Level lvl) const

Static Public Member Functions
static const InterfaceID &	interfaceID ()
	Retrieve interface ID. More...

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. More...
std::vector< long long int >	m_evtOffsets
	offset for event i in that file More...
unsigned int	m_evtInFile
long long int	m_evtFileOffset
	last read in event offset within a file, can be -1 More...
std::string	m_fileGUID
	current file GUID More...
ServiceHandle< StoreGateSvc >	m_storeGate
	Pointer to StoreGate. More...
ServiceHandle< StoreGateSvc >	m_inputMetadata
	StoreGateSvc. More...
ServiceHandle< IROBDataProviderSvc >	m_robProvider
Gaudi::Property< bool >	m_sequential
	enable sequential reading. More...
Gaudi::Property< bool >	m_dump
Gaudi::Property< float >	m_wait
Gaudi::Property< bool >	m_valEvent
Gaudi::Property< std::string >	m_eventInfoKey

Detailed Description

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

Definition at line 34 of file ByteStreamEventStorageInputSvc.h.

Member Enumeration Documentation

Advance

enum ByteStreamEventStorageInputSvc::Advance

private

Enumerator
PREVIOUS
NEXT

Definition at line 116 of file ByteStreamEventStorageInputSvc.h.

{ PREVIOUS = -1, NEXT = 1 };

Constructor & Destructor Documentation

ByteStreamEventStorageInputSvc()

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

Constructors:

Definition at line 37 of file ByteStreamEventStorageInputSvc.cxx.

  : ByteStreamInputSvc(name, pSvcLocator)
  , m_readerMutex()
  , m_eventsCache()
  , m_reader()
  , m_evtOffsets()
  , m_evtInFile(0)
  , m_evtFileOffset(0)
  , m_fileGUID("")
  , m_storeGate    ("StoreGateSvc", name)
  , m_inputMetadata("StoreGateSvc/InputMetaDataStore", name)
  , m_robProvider  ("ROBDataProviderSvc", name)
  , m_sequential   (this, "EnableSequential",   false, "")
  , m_dump         (this, "DumpFlag",           false, "Dump fragments")
  , m_wait         (this, "WaitSecs",              0., "Seconds to wait if input is in wait state")
  , m_valEvent     (this, "ValidateEvent",       true, "switch on check_tree when reading events")
  , m_eventInfoKey (this, "EventInfoKey", "EventInfo", "Key of EventInfo in metadata store")
{
  assert(pSvcLocator != nullptr);
 
  declareProperty("EventStore",    m_storeGate);
  declareProperty("MetaDataStore", m_inputMetadata);
}

~ByteStreamEventStorageInputSvc()

ByteStreamEventStorageInputSvc::~ByteStreamEventStorageInputSvc ( )

virtual

Destructor.

Definition at line 64 of file ByteStreamEventStorageInputSvc.cxx.

{}

Member Function Documentation

buildFragment()

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

private

Definition at line 322 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();
    }
  }
 
  // 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

Reimplemented from ByteStreamInputSvc.

Definition at line 482 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 ByteStreamInputSvc interface methods.

Implements ByteStreamInputSvc.

Definition at line 649 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.

Reimplemented from ByteStreamInputSvc.

Definition at line 658 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 121 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 96 of file ByteStreamEventStorageInputSvc.cxx.

                                         {
 
  ATH_CHECK(m_storeGate.release());
  ATH_CHECK(m_robProvider.release());
  ATH_CHECK(m_inputMetadata.release());
 
  return(ByteStreamInputSvc::finalize());
}

generateDataHeader()

StatusCode ByteStreamEventStorageInputSvc::generateDataHeader ( )

overridevirtual

Reimplemented from ByteStreamInputSvc.

Definition at line 391 of file ByteStreamEventStorageInputSvc.cxx.

{
  // 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);
    //add the data header element self reference to the object vector
    dataHeader->insert(*std::move(dataHeaderElement));
    // Set processTag
    dataHeader->setProcessTag(dataHeaderElement->getKey());
 
    // 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, iopx.release()));
    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, iopaux.release()));
    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

Reimplemented from ByteStreamInputSvc.

Definition at line 521 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 70 of file ByteStreamEventStorageInputSvc.cxx.

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

interfaceID()

const InterfaceID & ByteStreamInputSvc::interfaceID ( )

inlinestaticinherited

Retrieve interface ID.

Declaration of the interface ID ( interface id, major version, minor version)

Definition at line 49 of file ByteStreamInputSvc.h.

                                                          {
  static const InterfaceID IID_ByteStreamInputSvc("ByteStreamInputSvc", 1, 0);
  return(IID_ByteStreamInputSvc);
}

loadMetadata()

StatusCode ByteStreamEventStorageInputSvc::loadMetadata ( )

private

Definition at line 117 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 682 of file ByteStreamEventStorageInputSvc.cxx.

{
    std::unique_ptr<Token> token = std::make_unique<Token>();
    token->setDb(m_fileGUID);
    token->setTechnology(0x00001000);
    token->setOid(Token::OID_t(0LL, m_evtFileOffset));
 
    // note: passing ownership of token to DataHeaderElement
    return std::make_unique<DataHeaderElement>(ClassID_traits<DataHeader>::ID(),
        "StreamRAW", token.release());
}

msg() [1/2]

MsgStream& AthCommonMsg< Service >::msg ( ) const

inlineinherited

Definition at line 24 of file AthCommonMsg.h.

                                {
    return this->msgStream();
  }

msg() [2/2]

MsgStream& AthCommonMsg< Service >::msg ( const MSG::Level  lvl ) const

inlineinherited

Definition at line 27 of file AthCommonMsg.h.

                                                  {
    return this->msgStream(lvl);
  }

msgLvl()

bool AthCommonMsg< Service >::msgLvl ( const MSG::Level  lvl ) const

inlineinherited

Definition at line 30 of file AthCommonMsg.h.

                                               {
    return this->msgLevel(lvl);
  }

nextEvent()

const RawEvent * ByteStreamEventStorageInputSvc::nextEvent ( )

overridevirtual

++, new

Implements ByteStreamInputSvc.

Definition at line 197 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 0;
        }
      } 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 0;
  }
 
  // 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

Reimplemented from ByteStreamInputSvc.

Definition at line 109 of file ByteStreamEventStorageInputSvc.cxx.

{
  return m_evtOffsets.size();
}

previousEvent()

const RawEvent * ByteStreamEventStorageInputSvc::previousEvent ( )

overridevirtual

–, old

Implements ByteStreamInputSvc.

Definition at line 130 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 0;
  }
 
  // 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());
}

queryInterface()

StatusCode ByteStreamEventStorageInputSvc::queryInterface ( const InterfaceID &  riid, void **  ppvInterface  )

overridevirtual

Required of all Gaudi services: see Gaudi documentation for details.

Definition at line 667 of file ByteStreamEventStorageInputSvc.cxx.

{
  if(ByteStreamInputSvc::interfaceID().versionMatch(riid))
    *ppvInterface = dynamic_cast<ByteStreamInputSvc*>(this);
  else // Interface is not directly available: try out a base class
    return(::AthService::queryInterface(riid, ppvInterface));
 
  addRef();
  return(StatusCode::SUCCESS);
}

readerReady()

bool ByteStreamEventStorageInputSvc::readerReady ( )

private

Definition at line 564 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

Reimplemented from ByteStreamInputSvc.

Definition at line 513 of file ByteStreamEventStorageInputSvc.cxx.

{
  return readerReady();
}

ROBFragmentCheck()

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

private

Definition at line 582 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 623 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

Reimplemented from ByteStreamInputSvc.

Definition at line 614 of file ByteStreamEventStorageInputSvc.cxx.

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

setSequentialRead()

bool ByteStreamEventStorageInputSvc::setSequentialRead ( )

virtual

Definition at line 503 of file ByteStreamEventStorageInputSvc.cxx.

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

stop()

StatusCode ByteStreamEventStorageInputSvc::stop ( )

overridevirtual

Definition at line 85 of file ByteStreamEventStorageInputSvc.cxx.

{
  ATH_MSG_DEBUG("Calling ByteStreamInputSvc::stop()");
  // close moved to EventSelector for explicit coupling with incident
  //if (m_reader != 0) closeBlockIterator(false);
  return(StatusCode::SUCCESS);
}

validateEvent() [1/2]

void ByteStreamEventStorageInputSvc::validateEvent ( )

overridevirtual

Reimplemented from ByteStreamInputSvc.

Definition at line 277 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 287 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

private

Definition at line 102 of file ByteStreamEventStorageInputSvc.h.

m_eventInfoKey

Gaudi::Property<std::string> ByteStreamEventStorageInputSvc::m_eventInfoKey

private

Definition at line 105 of file ByteStreamEventStorageInputSvc.h.

m_eventsCache

SG::SlotSpecificObj<EventCache> ByteStreamEventStorageInputSvc::m_eventsCache

private

Definition at line 84 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 90 of file ByteStreamEventStorageInputSvc.h.

m_evtInFile

unsigned int ByteStreamEventStorageInputSvc::m_evtInFile

private

Definition at line 89 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 88 of file ByteStreamEventStorageInputSvc.h.

m_fileGUID

std::string ByteStreamEventStorageInputSvc::m_fileGUID

private

current file GUID

Definition at line 92 of file ByteStreamEventStorageInputSvc.h.

m_inputMetadata

ServiceHandle<StoreGateSvc> ByteStreamEventStorageInputSvc::m_inputMetadata

private

StoreGateSvc.

Definition at line 99 of file ByteStreamEventStorageInputSvc.h.

m_reader

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

private

DataReader from EventStorage.

Definition at line 86 of file ByteStreamEventStorageInputSvc.h.

m_readerMutex

std::mutex ByteStreamEventStorageInputSvc::m_readerMutex

private

Definition at line 73 of file ByteStreamEventStorageInputSvc.h.

m_robProvider

ServiceHandle<IROBDataProviderSvc> ByteStreamEventStorageInputSvc::m_robProvider

private

Definition at line 100 of file ByteStreamEventStorageInputSvc.h.

m_sequential

Gaudi::Property<bool> ByteStreamEventStorageInputSvc::m_sequential

private

enable sequential reading.

Definition at line 101 of file ByteStreamEventStorageInputSvc.h.

m_storeGate

ServiceHandle<StoreGateSvc> ByteStreamEventStorageInputSvc::m_storeGate

private

Pointer to StoreGate.

StoreGateSvc

Definition at line 98 of file ByteStreamEventStorageInputSvc.h.

m_valEvent

Gaudi::Property<bool> ByteStreamEventStorageInputSvc::m_valEvent

private

Definition at line 104 of file ByteStreamEventStorageInputSvc.h.

m_wait

Gaudi::Property<float> ByteStreamEventStorageInputSvc::m_wait

private

Definition at line 103 of file ByteStreamEventStorageInputSvc.h.

---

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

• ByteStreamEventStorageInputSvc.h
• ByteStreamEventStorageInputSvc.cxx