TrigByteStreamInputSvc Class Reference

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	initialize () override
virtual StatusCode	finalize () override
virtual const RawEvent *	nextEvent () override
virtual const RawEvent *	previousEvent () override
virtual const RawEvent *	currentEvent () const override

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 24 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.

: base_class(name, svcLoc) {}

~TrigByteStreamInputSvc()

TrigByteStreamInputSvc::~TrigByteStreamInputSvc ( )

virtual

Standard destructor.

Definition at line 32 of file TrigByteStreamInputSvc.cxx.

{}

Member Function Documentation

currentEvent()

const RawEvent * TrigByteStreamInputSvc::currentEvent ( ) const

overridevirtual

Definition at line 191 of file TrigByteStreamInputSvc.cxx.

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

finalize()

StatusCode TrigByteStreamInputSvc::finalize ( )

overridevirtual

Definition at line 51 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;
}

initialize()

StatusCode TrigByteStreamInputSvc::initialize ( )

overridevirtual

Definition at line 37 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;
}

nextEvent()

const RawEvent * TrigByteStreamInputSvc::nextEvent ( )

overridevirtual

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

previousEvent()

const RawEvent * TrigByteStreamInputSvc::previousEvent ( )

overridevirtual

Definition at line 183 of file TrigByteStreamInputSvc.cxx.

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

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 47 of file TrigByteStreamInputSvc.h.

m_eventsCache

SG::SlotSpecificObj<EventCache> TrigByteStreamInputSvc::m_eventsCache

private

Cache of RawEvent pointer per event slot.

Definition at line 59 of file TrigByteStreamInputSvc.h.

m_evtStore

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

private

Definition at line 43 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 60 of file TrigByteStreamInputSvc.h.

m_monTool

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

private

Definition at line 44 of file TrigByteStreamInputSvc.h.

m_robDataProviderSvc

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

private

Definition at line 42 of file TrigByteStreamInputSvc.h.

---

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

• TrigByteStreamInputSvc.h
• TrigByteStreamInputSvc.cxx