A ByteStreamInputSvc implementation for online use, reading events from hltinterface::DataCollector. More...

#include <TrigByteStreamInputSvc.h>

Inheritance diagram for TrigByteStreamInputSvc:

Collaboration diagram for TrigByteStreamInputSvc:

Classes
struct	EventCache

Public Member Functions
	TrigByteStreamInputSvc (const std::string &name, ISvcLocator *svcLoc)
	Standard constructor. More...

virtual	~TrigByteStreamInputSvc ()
	Standard destructor. More...

virtual StatusCode	queryInterface (const InterfaceID &riid, void **ppvInterface) override

virtual StatusCode	initialize () override

virtual StatusCode	finalize () override

virtual const RawEvent *	nextEvent () override
	virtual method for advance to the next event More...

virtual const RawEvent *	previousEvent () override

virtual const RawEvent *	currentEvent () const override
	virtual method for accessing the current event More...

virtual void	setEvent (void *, unsigned int)

virtual unsigned int	currentEventStatus () const
	virtual method for accessing the current event status More...

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

virtual void	closeBlockIterator (bool)

virtual bool	ready ()

virtual StatusCode	generateDataHeader ()

virtual long	positionInBlock ()

virtual void	validateEvent ()

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 Attributes
ServiceHandle< IROBDataProviderSvc >	m_robDataProviderSvc {this, "ROBDataProvider", "ROBDataProviderSvc"}

ServiceHandle< StoreGateSvc >	m_evtStore {this, "EventStore", "StoreGateSvc"}

ToolHandle< GenericMonitoringTool >	m_monTool {this, "MonTool", "" , "Monitoring tool"}

Gaudi::Property< int >	m_checkCTPFragmentModuleID

SG::SlotSpecificObj< EventCache >	m_eventsCache
	Cache of RawEvent pointer per event slot. More...

uint16_t	m_maxLB {0}
	Maximum lumi block number seen so far, used for monitoring. More...

Detailed Description

A ByteStreamInputSvc implementation for online use, reading events from hltinterface::DataCollector.

The layout and implementation are based on ByteStreamEventStorageInputSvc

Definition at line 23 of file TrigByteStreamInputSvc.h.

Constructor & Destructor Documentation

◆ TrigByteStreamInputSvc()

TrigByteStreamInputSvc::TrigByteStreamInputSvc	(	const std::string &	name,
		ISvcLocator *	svcLoc
	)

Standard constructor.

Definition at line 26 of file TrigByteStreamInputSvc.cxx.

27 : ByteStreamInputSvc(name, svcLoc) {}

◆ ~TrigByteStreamInputSvc()

TrigByteStreamInputSvc::~TrigByteStreamInputSvc ( )

virtual

Standard destructor.

Definition at line 32 of file TrigByteStreamInputSvc.cxx.

32 {}

Member Function Documentation

◆ closeBlockIterator()

void ByteStreamInputSvc::closeBlockIterator ( bool )

inlinevirtualinherited

Reimplemented in ByteStreamEventStorageInputSvc.

Definition at line 61 of file ByteStreamInputSvc.h.

61 {}

◆ currentEvent()

const RawEvent * TrigByteStreamInputSvc::currentEvent ( ) const

overridevirtual

virtual method for accessing the current event

Implements ByteStreamInputSvc.

Definition at line 207 of file TrigByteStreamInputSvc.cxx.

                                                            {
   ATH_MSG_FATAL("The method " << __FUNCTION__ << " is not implemented for online running");
   return nullptr;
 }

◆ currentEventStatus()

unsigned int ByteStreamInputSvc::currentEventStatus ( ) const

inlinevirtualinherited

virtual method for accessing the current event status

Reimplemented in ByteStreamEventStorageInputSvc.

Definition at line 55 of file ByteStreamInputSvc.h.

                                                                  {
   return(0);
 }

◆ finalize()

StatusCode TrigByteStreamInputSvc::finalize ( )

overridevirtual

Definition at line 67 of file TrigByteStreamInputSvc.cxx.

                                             {
   ATH_MSG_VERBOSE("start of " << __FUNCTION__);
   if (m_robDataProviderSvc.release().isFailure()) {
     ATH_MSG_WARNING("Cannot release rob data provider");
   }
   ATH_MSG_VERBOSE("end of " << __FUNCTION__);
   return StatusCode::SUCCESS;
 }

◆ generateDataHeader()

StatusCode ByteStreamInputSvc::generateDataHeader ( )

inlinevirtualinherited

Reimplemented in ByteStreamEventStorageInputSvc.

Definition at line 63 of file ByteStreamInputSvc.h.

63 {return StatusCode::SUCCESS;}

◆ getBlockIterator()

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

inlinevirtualinherited

Reimplemented in ByteStreamEventStorageInputSvc.

Definition at line 60 of file ByteStreamInputSvc.h.

60 {return std::make_pair(-1,"GUID");}

◆ initialize()

StatusCode TrigByteStreamInputSvc::initialize ( )

overridevirtual

Definition at line 53 of file TrigByteStreamInputSvc.cxx.

                                               {
   ATH_MSG_VERBOSE("start of " << __FUNCTION__);
  
   ATH_CHECK(m_robDataProviderSvc.retrieve());
   ATH_CHECK(m_evtStore.retrieve());
   if (!m_monTool.empty()) ATH_CHECK(m_monTool.retrieve());
  
   ATH_MSG_VERBOSE("end of " << __FUNCTION__);
   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);
 }

◆ 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 * TrigByteStreamInputSvc::nextEvent ( )

overridevirtual

virtual method for advance to the next event

Implements ByteStreamInputSvc.

Definition at line 79 of file TrigByteStreamInputSvc.cxx.

                                                   {
   ATH_MSG_VERBOSE("start of " << __FUNCTION__);
  
   // Get the EventContext via event store because the interface doesn't allow passing it explicitly as an argument
   // and we don't want to use ThreadLocalContext. Don't use ReadHandle here because it calls ThreadLocalContext if
   // not given a context (which we want to retrieve).
   const EventContext* eventContext = nullptr;
   if (m_evtStore->retrieve(eventContext).isFailure()) {
     ATH_MSG_ERROR("Failed to retrieve EventContext from the event store, new event cannot be read");
     return nullptr;
   }
  
   ATH_MSG_DEBUG("Reading new event for event context " << *eventContext);
  
   // Find the cache corresponding to the current slot
   EventCache* cache = m_eventsCache.get(*eventContext);
  
   // Free the memory allocated to the previous event processed in the current slot
   cache->releaseEvent();
  
   using DCStatus = hltinterface::DataCollector::Status;
   DCStatus status = DCStatus::NO_EVENT;
   auto monLBN = Monitored::Scalar<uint16_t>("getNext_LBN", m_maxLB);
   auto monNoEvent = Monitored::Scalar<bool>("getNext_noEvent", false);
   try {
     auto t_getNext = Monitored::Timer<std::chrono::duration<float, std::milli>>("TIME_getNext");
     status = hltinterface::DataCollector::instance()->getNext(cache->rawData);
     auto mon = Monitored::Group(m_monTool, t_getNext);
   }
   catch (const std::exception& ex) {
     ATH_MSG_ERROR("Failed to read new event, caught an unexpected exception: " << ex.what()
                   << ". Throwing hltonl::Exception::EventSourceCorrupted" );
     throw hltonl::Exception::EventSourceCorrupted();
   }
   catch (...) {
     ATH_MSG_ERROR("Failed to read new event, caught an unexpected exception. "
                   << "Throwing hltonl::Exception::EventSourceCorrupted" );
     throw hltonl::Exception::EventSourceCorrupted();
   }
  
   if (status == DCStatus::STOP) {
     ATH_MSG_DEBUG("DataCollector::getNext returned STOP - no more events available");
     throw hltonl::Exception::NoMoreEvents();
   }
   else if (status == DCStatus::NO_EVENT) {
     ATH_MSG_DEBUG("DataCollector::getNext returned NO_EVENT - no events available temporarily");
     monNoEvent = true;
     auto mon = Monitored::Group(m_monTool, monLBN, monNoEvent);
     throw hltonl::Exception::NoEventsTemporarily();
   }
   else if (status != DCStatus::OK) {
     ATH_MSG_ERROR("Unhandled return Status " << static_cast<int>(status) << " from DataCollector::getNext");
     return nullptr;
   }
   ATH_MSG_VERBOSE("DataCollector::getNext returned Status::OK");
  
   // Create a cached FullEventFragment object from the cached raw data
   cache->fullEventFragment.reset(new RawEvent(cache->rawData.get()));
  
   // Update LB number for monitoring
   if (m_maxLB < cache->fullEventFragment->lumi_block()) {
     m_maxLB = cache->fullEventFragment->lumi_block();
     monLBN = m_maxLB;
   }
  
   // Monitor the input
   auto numROBs = Monitored::Scalar<int>("L1Result_NumROBs",
                                         cache->fullEventFragment->nchildren());
   auto fragSize = Monitored::Scalar<float>("L1Result_FullEvFragSize",
                                            cache->fullEventFragment->fragment_size_word()*wordsToKiloBytes);
   std::vector<eformat::read::ROBFragment> robVec;
   cache->fullEventFragment->robs(robVec);
   std::vector<std::string> subdetNameVec;
   for (const eformat::read::ROBFragment& rob : robVec) {
     eformat::helper::SourceIdentifier sid(rob.rob_source_id());
     subdetNameVec.push_back(sid.human_detector());
   }
   auto subdets = Monitored::Collection<std::vector<std::string>>("L1Result_SubDets", subdetNameVec);
   auto mon = Monitored::Group(m_monTool, numROBs, fragSize, subdets, monLBN, monNoEvent);
  
   // Give the FullEventFragment pointer to ROBDataProviderSvc
   m_robDataProviderSvc->setNextEvent(*eventContext, cache->fullEventFragment.get());
   ATH_MSG_VERBOSE("end of " << __FUNCTION__);
  
   // Check the CTP fragment (request from readout if not part of the cache), ATR-25217
   if (m_checkCTPFragmentModuleID.value() >= 0) {
     const eformat::helper::SourceIdentifier sid{eformat::SubDetector::TDAQ_CTP,
                                                 static_cast<uint16_t>(m_checkCTPFragmentModuleID.value())};
     std::vector<const OFFLINE_FRAGMENTS_NAMESPACE::ROBFragment*> vrobf;
     m_robDataProviderSvc->getROBData(*eventContext, {sid.code()}, vrobf, name());
     if (vrobf.empty()) {
       ATH_MSG_INFO("The CTP ROB fragment 0x" << std::hex << sid.code() << std::dec << " is missing. "
                     << "Throwing hltonl::Exception::MissingCTPFragment");
       throw hltonl::Exception::MissingCTPFragment();
     }
     uint32_t robStatus = vrobf.at(0)->nstatus()>0 ? *(vrobf.at(0)->status()) : 0;
     if (robStatus!=0) {
       std::string hexStatus(8, char{0});
       std::to_chars(hexStatus.data(), hexStatus.data()+hexStatus.size(), robStatus, 16);
       ATH_MSG_INFO("The CTP ROB fragment 0x" << std::hex << sid.code() << std::dec << " has non-zero status word: 0x"
                     << hexStatus << ". Throwing hltonl::Exception::BadCTPFragment");
       throw hltonl::Exception::BadCTPFragment("Non-zero ROB status 0x"+hexStatus);
     }
     try {
       vrobf.at(0)->check();
     }
     catch (const std::exception& ex) {
       ATH_MSG_INFO("The CTP ROB fragment 0x" << std::hex << sid.code() << std::dec << " is corrupted: "
                     << ex.what() << ". Throwing hltonl::Exception::BadCTPFragment");
       throw hltonl::Exception::BadCTPFragment(ex.what());
     }
   }
  
   // Return the FullEventFragment pointer (do not transfer ownership)
   return cache->fullEventFragment.get();
 }

◆ positionInBlock()

long ByteStreamInputSvc::positionInBlock ( )

inlinevirtualinherited

Reimplemented in ByteStreamEventStorageInputSvc.

Definition at line 64 of file ByteStreamInputSvc.h.

64 {return -1;}

◆ previousEvent()

const RawEvent * TrigByteStreamInputSvc::previousEvent ( )

overridevirtual

Implements ByteStreamInputSvc.

Definition at line 199 of file TrigByteStreamInputSvc.cxx.

                                                       {
   ATH_MSG_FATAL("The method " << __FUNCTION__ << " is not implemented for online running");
   return nullptr;
 }

◆ queryInterface()

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

overridevirtual

Definition at line 37 of file TrigByteStreamInputSvc.cxx.

                                                                                               {
   ATH_MSG_VERBOSE("start of " << __FUNCTION__);
  
   if(ByteStreamInputSvc::interfaceID().versionMatch(riid))
     *ppvInterface = static_cast<ByteStreamInputSvc*>(this);
   else
     return AthService::queryInterface(riid, ppvInterface);
  
   addRef();
   ATH_MSG_VERBOSE("end of " << __FUNCTION__);
   return StatusCode::SUCCESS;
 }

◆ ready()

bool ByteStreamInputSvc::ready ( )

inlinevirtualinherited

Reimplemented in ByteStreamEventStorageInputSvc.

Definition at line 62 of file ByteStreamInputSvc.h.

62 {return false;}

◆ setEvent()

virtual void ByteStreamInputSvc::setEvent	(	void *	,
		unsigned int
	)

inlinevirtualinherited

Reimplemented in ByteStreamEventStorageInputSvc.

Definition at line 36 of file ByteStreamInputSvc.h.

36 {}

◆ validateEvent()

void ByteStreamInputSvc::validateEvent ( )

inlinevirtualinherited

Reimplemented in ByteStreamEventStorageInputSvc.

Definition at line 65 of file ByteStreamInputSvc.h.

65 {}

Member Data Documentation

◆ m_checkCTPFragmentModuleID

Gaudi::Property<int> TrigByteStreamInputSvc::m_checkCTPFragmentModuleID

private

Initial value:

{this, "CheckCTPFragmentModuleID", -1,
    "After reading a new event, assert we can retrieve the CTP fragment with Module ID given by this property, "
    "and that has no errors. A value <0 disables the check."}

Definition at line 49 of file TrigByteStreamInputSvc.h.

◆ m_eventsCache

SG::SlotSpecificObj<EventCache> TrigByteStreamInputSvc::m_eventsCache

private

Cache of RawEvent pointer per event slot.

Definition at line 61 of file TrigByteStreamInputSvc.h.

◆ m_evtStore

ServiceHandle<StoreGateSvc> TrigByteStreamInputSvc::m_evtStore {this, "EventStore", "StoreGateSvc"}

private

Definition at line 45 of file TrigByteStreamInputSvc.h.

◆ m_maxLB

uint16_t TrigByteStreamInputSvc::m_maxLB {0}

private

Maximum lumi block number seen so far, used for monitoring.

Definition at line 62 of file TrigByteStreamInputSvc.h.

◆ m_monTool

ToolHandle<GenericMonitoringTool> TrigByteStreamInputSvc::m_monTool {this, "MonTool", "" , "Monitoring tool"}

private

Definition at line 46 of file TrigByteStreamInputSvc.h.

◆ m_robDataProviderSvc

ServiceHandle<IROBDataProviderSvc> TrigByteStreamInputSvc::m_robDataProviderSvc {this, "ROBDataProvider", "ROBDataProviderSvc"}

private

Definition at line 44 of file TrigByteStreamInputSvc.h.

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

Classes

Public Member Functions

Static Public Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ TrigByteStreamInputSvc()

◆ ~TrigByteStreamInputSvc()

Member Function Documentation

◆ closeBlockIterator()

◆ currentEvent()

◆ currentEventStatus()

◆ finalize()

◆ generateDataHeader()

◆ getBlockIterator()

◆ initialize()

◆ interfaceID()

◆ msg() [1/2]

◆ msg() [2/2]

◆ msgLvl()

◆ nextEvent()

◆ positionInBlock()

◆ previousEvent()

◆ queryInterface()

◆ ready()

◆ setEvent()

◆ validateEvent()

Member Data Documentation

◆ m_checkCTPFragmentModuleID

◆ m_eventsCache

◆ m_evtStore

◆ m_maxLB

◆ m_monTool

◆ m_robDataProviderSvc