TrigCostSvc Class Reference

AthenaMT service to collect trigger cost data from all threads and summarise it at the end of the event. More...

#include <TrigCostSvc.h>

Inheritance diagram for TrigCostSvc:

Collaboration diagram for TrigCostSvc:

Classes
class	ThreadHashCompare
	Static hash and equal members as required by tbb::concurrent_hash_map. More...

Public Member Functions
	TrigCostSvc (const std::string &name, ISvcLocator *pSvcLocator)
	Standard ATLAS Service constructor.
virtual	~TrigCostSvc ()
	Destructor.
virtual StatusCode	initialize () override
	Initialise, create enough storage to store m_eventSlots.
virtual StatusCode	finalize () override
	Finalize, act on m_saveHashes.
virtual StatusCode	startEvent (const EventContext &context, const bool enableMonitoring=true) override
	Implementation of ITrigCostSvc::startEvent.
virtual StatusCode	processAlg (const EventContext &context, const std::string &caller, const AuditType type) override
	Implementation of ITrigCostSvc::processAlg.
virtual StatusCode	endEvent (const EventContext &context, SG::WriteHandle< xAOD::TrigCompositeContainer > &costOutputHandle, SG::WriteHandle< xAOD::TrigCompositeContainer > &rosOutputHandle) override
	Implementation of ITrigCostSvc::endEvent.
virtual bool	isMonitoredEvent (const EventContext &context, const bool includeMultiSlot=true) const override
virtual StatusCode	monitorROS (const EventContext &context, robmonitor::ROBDataMonitorStruct payload) override
	Implementation of ITrigCostSvc::monitorROS.
virtual StatusCode	generateTimeoutReport (const EventContext &context, std::string &report) override
virtual StatusCode	discardEvent (const EventContext &context) override
	Discard a cost monitored event.

Private Member Functions
StatusCode	monitor (const EventContext &context, const AlgorithmIdentifier &ai, const TrigTimeStamp &now, const AuditType type)
	Internal call to save monitoring data for a given AlgorithmIdentifier.
StatusCode	checkSlot (const EventContext &context) const
	Sanity check that the job is respecting the number of slots which were declared at config time.
int32_t	getROIID (const EventContext &context)
	@breif Internal function to return a RoI from an extended event context context

Private Attributes
size_t	m_eventSlots
	Number of concurrent processing slots.
std::unique_ptr< std::atomic< bool >[] >	m_eventMonitored
	Used to cache if the event in a given slot is being monitored.
std::unique_ptr< std::shared_mutex[] >	m_slotMutex
	Used to control and protect whole-table operations.
std::mutex	m_globalMutex
	Used to protect all-slot modifications.
TrigCostDataStore< AlgorithmPayload >	m_algStartInfo
	Thread-safe store of algorithm start payload.
TrigCostDataStore< TrigTimeStamp >	m_algStopTime
	Thread-safe store of algorithm stop times.
TrigCostDataStore< std::vector< robmonitor::ROBDataMonitorStruct > >	m_rosData
	Thread-safe store of ROS data.
tbb::concurrent_hash_map< std::thread::id, AlgorithmIdentifier, ThreadHashCompare >	m_threadToAlgMap
	Keeps track of what is running right now in each thread.
std::unordered_map< uint32_t, uint32_t >	m_threadToCounterMap
	Map thread's hash ID to a counting numeral.
size_t	m_threadCounter
	Count how many unique thread ID we have seen.
Gaudi::Property< bool >	m_monitorAllEvents {this, "MonitorAllEvents", false, "Monitor every HLT event, e.g. for offline validation."}
Gaudi::Property< bool >	m_enableMultiSlot {this, "EnableMultiSlot", false, "Monitored events in the MasterSlot collect data from events running in other slots."}
Gaudi::Property< bool >	m_saveHashes {this, "SaveHashes", false, "Store a copy of the hash dictionary for easier debugging"}
Gaudi::Property< size_t >	m_masterSlot {this, "MasterSlot", 0, "The slot responsible for saving MultiSlot data"}
Gaudi::Property< std::string >	m_costSupervisorAlgName {this, "CostSupervisorAlgName", "TrigCostSupervisorAlg", "The name of cost monitoring supervising algorithm, starting at the begining of the event"}
Gaudi::Property< std::string >	m_costFinalizeAlgName {this, "CostFinalizeAlgName", "TrigCostFinalizeAlg", "The name of cost monitoring finalize algorithm, starting at the end of the event"}

Detailed Description

AthenaMT service to collect trigger cost data from all threads and summarise it at the end of the event.

The main hooks into this service are: HLTSeeding - To clear the internal storage and flag the event for monitoring. TrigCostAuditor - To inform the service when algorithms start and stop executing HLTROBDataProviderSvc - To inform the service about requests for data ROBs HLTSummaryAlg - To inform the service when the HLT has finished, and to receive the persistent payload

Definition at line 36 of file TrigCostSvc.h.

Constructor & Destructor Documentation

TrigCostSvc()

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

Standard ATLAS Service constructor.

Parameters

[in]	name	The service's name
[in]	svcloc	A pointer to a service location service

Definition at line 14 of file TrigCostSvc.cxx.

                                                                        :
base_class(name, pSvcLocator), // base_class = AthService
m_eventSlots(),
m_eventMonitored(),
m_slotMutex(),
m_globalMutex(),
m_algStartInfo(),
m_algStopTime(),
m_threadToAlgMap(),
m_threadToCounterMap(),
m_threadCounter(0)
{
  ATH_MSG_DEBUG("TrigCostSvc regular constructor");
}

~TrigCostSvc()

TrigCostSvc::~TrigCostSvc ( )

virtual

Destructor.

Currently nothing to delete.

Definition at line 31 of file TrigCostSvc.cxx.

                          {
  // delete[] m_eventMonitored;
  ATH_MSG_DEBUG("TrigCostSvc destructor()");
}

Member Function Documentation

checkSlot()

StatusCode TrigCostSvc::checkSlot ( const EventContext & context ) const

private

Sanity check that the job is respecting the number of slots which were declared at config time.

Parameters

[in]

context

The event context

Returns

Success if the m_eventMonitored array is range, Failure if access request would overflow

Definition at line 506 of file TrigCostSvc.cxx.

                                                                   {
  if (context.slot() >= m_eventSlots) {
    ATH_MSG_FATAL("Job is using event slot #" << context.slot() << ", but we only reserved space for: " << m_eventSlots);
    return StatusCode::FAILURE;
  }
  return StatusCode::SUCCESS;
}

discardEvent()

StatusCode TrigCostSvc::discardEvent ( const EventContext & context )

overridevirtual

Discard a cost monitored event.

Parameters

[in]

context

The event context

Definition at line 484 of file TrigCostSvc.cxx.

                                                                {
  
  if (m_monitorAllEvents) {
    ATH_MSG_DEBUG("All events are monitored - event will not be discarded");
    return StatusCode::SUCCESS;
  }
 
  ATH_MSG_DEBUG("Cost Event will be discarded");
  ATH_CHECK(checkSlot(context));
  {
    std::unique_lock lockUnique( m_slotMutex[ context.slot() ] );
 
    // Reset eventMonitored flags
    m_eventMonitored[ context.slot() ] = false;
 
    // tables are cleared at the start of the event
  }
  return StatusCode::SUCCESS;
}

endEvent()

StatusCode TrigCostSvc::endEvent ( const EventContext & context, SG::WriteHandle< xAOD::TrigCompositeContainer > & costOutputHandle, SG::WriteHandle< xAOD::TrigCompositeContainer > & rosOutputHandle )

overridevirtual

Implementation of ITrigCostSvc::endEvent.

Parameters

[in]	context	The event context
[out]	costOutputHandle	Write handle to fill with execution summary if the event was monitored
[out]	rosOutputHandle	Write handle to fill with ROS requests summary if the event was monitored

Definition at line 206 of file TrigCostSvc.cxx.

                                                                                                                                                                                     { 
  ATH_CHECK(checkSlot(context));
  if (m_eventMonitored[ context.slot() ] == false) {
    // This event was not monitored - nothing to do.
    ATH_MSG_DEBUG("Not a monitored event.");
    return StatusCode::SUCCESS;
  }
 
  // As we will miss the AuditType::After of the TrigCostFinalizeAlg (which is calling this TrigCostSvc::endEvent), let's add it now.
  // This will be our canonical final timestamps for measuring this event. Similar was done for HLTSeeding at the start
  ATH_CHECK(processAlg(context, m_costFinalizeAlgName, AuditType::After));
 
  // Reset eventMonitored flags
  m_eventMonitored[ context.slot() ] = false;
 
  // Now that this atomic is set to FALSE, additional algs in this instance which trigger this service will 
  // not be able to call TrigCostSvc::monitor
 
  // ... but processAlg might already be running in other threads... 
  // Wait to obtain an exclusive lock.
  std::unique_lock lockUnique( m_slotMutex[ context.slot() ] );
  
  // we can now perform whole-map inspection of this event's TrigCostDataStores without the danger that it will be changed further
 
  // Let's start by getting the global STOP time we just wrote
  uint64_t eventStopTime = 0;
  {
    const AlgorithmIdentifier myAi = AlgorithmIdentifierMaker::make(context, m_costFinalizeAlgName, msg());
    ATH_CHECK( myAi.isValid() );
    tbb::concurrent_hash_map<AlgorithmIdentifier, TrigTimeStamp, AlgorithmIdentifierHashCompare>::const_accessor stopTimeAcessor;
    if (m_algStopTime.retrieve(myAi, stopTimeAcessor, msg()).isFailure()) {
      ATH_MSG_ERROR("No end time for '" << myAi.m_caller << "', '" << myAi.m_store << "'"); // Error as we JUST entered this info!
    } else { // retrieve was a success
      eventStopTime = stopTimeAcessor->second.microsecondsSinceEpoch();
    }
  }
 
  // And the global START time for the event
  uint64_t eventStartTime = 0;
  {
    const AlgorithmIdentifier hltSeedingAi = AlgorithmIdentifierMaker::make(context, m_costSupervisorAlgName, msg());
    ATH_CHECK( hltSeedingAi.isValid() );
    tbb::concurrent_hash_map<AlgorithmIdentifier, AlgorithmPayload, AlgorithmIdentifierHashCompare>::const_accessor startAcessor;
    if (m_algStartInfo.retrieve(hltSeedingAi, startAcessor, msg()).isFailure()) {
      ATH_MSG_ERROR("No alg info for '" << hltSeedingAi.m_caller << "', '" << hltSeedingAi.m_store << "'"); // Error as we know this info must be present
    } else { // retrieve was a success
      eventStartTime = startAcessor->second.m_algStartTime.microsecondsSinceEpoch();
    }
  }
 
  // Read payloads. Write to persistent format
  tbb::concurrent_hash_map< AlgorithmIdentifier, AlgorithmPayload, AlgorithmIdentifierHashCompare>::const_iterator beginIt;
  tbb::concurrent_hash_map< AlgorithmIdentifier, AlgorithmPayload, AlgorithmIdentifierHashCompare>::const_iterator endIt;
  tbb::concurrent_hash_map< AlgorithmIdentifier, AlgorithmPayload, AlgorithmIdentifierHashCompare>::const_iterator it;
  ATH_CHECK(m_algStartInfo.getIterators(context, msg(), beginIt, endIt));
 
  ATH_MSG_DEBUG("Monitored event with " << std::distance(beginIt, endIt) << " AlgorithmPayload objects.");
 
  std::map<size_t, size_t> aiToHandleIndex;
  for (it = beginIt; it != endIt; ++it) {
    const AlgorithmIdentifier& ai = it->first;
    const AlgorithmPayload& ap = it->second;
    uint64_t startTime = ap.m_algStartTime.microsecondsSinceEpoch();
 
    // Can we find the end time for this alg? If not, it is probably still running. Hence we use "now" as the default time.
    uint64_t stopTime = eventStopTime;
    {
      tbb::concurrent_hash_map<AlgorithmIdentifier, TrigTimeStamp, AlgorithmIdentifierHashCompare>::const_accessor stopTimeAcessor;
      if (m_algStopTime.retrieve(ai, stopTimeAcessor, msg()).isFailure()) {
        ATH_MSG_DEBUG("No end time for '" << ai.m_caller << "', '" << ai.m_store << "'");
      } else { // retrieve was a success
        stopTime = stopTimeAcessor->second.microsecondsSinceEpoch();
      }
      // stopTimeAcessor goes out of scope - lock released
    }
 
    // It is possible (when in the master-slot) to catch just the END of an Alg's exec from another slot, and then the START of the same
    // alg executing in the next event in that same other-slot.
    // This gives us an end time which is before the start time. Disregard these entries.
    if (startTime > stopTime) {
      ATH_MSG_VERBOSE("Disregard start-time:" << startTime << " > stop-time:" << stopTime 
        << " for " << TrigConf::HLTUtils::hash2string( ai.callerHash(msg()), "ALG") << " in slot " << ap.m_slot << ", this is slot " << context.slot());
      continue;
    }
 
    // Lock the start and stop times to be no later than eventStopTime.
    // E.g. it's possible for an alg in another slot to start or stop running after 'processAlg(context, m_costFinalizeAlgName, AuditType::After))'
    // but before 'lockUnique( m_slotMutex[ context.slot() ] )', creating a timestamp after the nominal end point for this event.
    // If the alg starts afterwards, we disregard it in lieu of setting to have zero walltime.
    // If the alg stops afterwards, we truncate its stop time to be no later than eventStopTime
    if (startTime > eventStopTime) {
      ATH_MSG_VERBOSE("Disregard " << TrigConf::HLTUtils::hash2string( ai.callerHash(msg()), "ALG") << " as it started after endEvent() was finished being called" );
      continue;
    }
    if (stopTime > eventStopTime) {
      ATH_MSG_VERBOSE(TrigConf::HLTUtils::hash2string( ai.callerHash(msg()), "ALG") << " stopped after endEvent() was called, but before the cost container was locked," 
        << " truncating its ending time stamp from " << stopTime << " to " << eventStopTime);
      stopTime = eventStopTime;
    }
 
    // Do the same, locking the start and stop times to be no earlier than eventStartTime
    // If the alg stops before eventStartTime, we disregard it in lieu of setting it to have zero walltime
    // If the alg starts before eventStartTime, we truncate its start time to be no later than eventStopTime
    if (stopTime < eventStartTime) {
      ATH_MSG_VERBOSE("Disregard " << TrigConf::HLTUtils::hash2string( ai.callerHash(msg()), "ALG") << " as it stopped before startEvent() was finished being called" );
      continue;
    }
    if (startTime < eventStartTime) {
      ATH_MSG_VERBOSE(TrigConf::HLTUtils::hash2string( ai.callerHash(msg()), "ALG") << " started just after the cost container was unlocked, but before the HLTSeeding record was written." 
        << " truncating its starting time stamp from " << startTime << " to " << eventStartTime);
      startTime = eventStartTime;
    }
 
    // Make a new TrigComposite to persist monitoring payload for this alg
    xAOD::TrigComposite* tc = new xAOD::TrigComposite();
    costOutputHandle->push_back( tc ); 
    // tc is now owned by storegate and, and has an aux store provided by the TrigCompositeCollection
 
    const uint32_t threadID = static_cast<uint32_t>( std::hash< std::thread::id >()(ap.m_algThreadID) );
    uint32_t threadEnumerator = 0; 
    {
      // We can have multiple slots get here at the same time
      std::lock_guard<std::mutex> lock(m_globalMutex);
      const std::unordered_map<uint32_t, uint32_t>::const_iterator mapIt = m_threadToCounterMap.find(threadID);
      if (mapIt == m_threadToCounterMap.end()) {
        threadEnumerator = m_threadCounter;
        m_threadToCounterMap.insert( std::make_pair(threadID, m_threadCounter++) );
      } else {
        threadEnumerator = mapIt->second;
      }
    }
 
    bool result = true;
    result &= tc->setDetail("alg", ai.callerHash(msg()));
    result &= tc->setDetail("store", ai.storeHash(msg()));
    result &= tc->setDetail("view", ai.m_viewID);
    result &= tc->setDetail("thread", threadEnumerator);
    result &= tc->setDetail("thash", threadID);
    result &= tc->setDetail("slot", ap.m_slot);
    result &= tc->setDetail("roi", ap.m_algROIID);
    result &= tc->setDetail("start", startTime);
    result &= tc->setDetail("stop", stopTime);
    if (!result) ATH_MSG_WARNING("Failed to append one or more details to trigger cost TC");
 
    aiToHandleIndex[ai.m_hash] = costOutputHandle->size() - 1;
  }
 
  typedef tbb::concurrent_hash_map< AlgorithmIdentifier, std::vector<robmonitor::ROBDataMonitorStruct>, AlgorithmIdentifierHashCompare>::const_iterator ROBConstIt;
  ROBConstIt beginRob;
  ROBConstIt endRob;
  
  ATH_CHECK(m_rosData.getIterators(context, msg(), beginRob, endRob));
  
  for (ROBConstIt it = beginRob; it != endRob; ++it) {
    size_t aiHash = it->first.m_hash;
 
    if (aiToHandleIndex.count(aiHash) == 0) {
      ATH_MSG_WARNING("Algorithm with hash " << aiHash << " not found!");
    }
 
    // Save ROB data via TrigComposite
    for (const robmonitor::ROBDataMonitorStruct& robData : it->second) {
      xAOD::TrigComposite* tc = new xAOD::TrigComposite();
      rosOutputHandle->push_back(tc); 
 
      // Retrieve ROB requests data into primitives vectors
      std::vector<uint32_t> robs_id;
      std::vector<uint32_t> robs_size;
      std::vector<unsigned> robs_history;
      std::vector<unsigned short> robs_status;
 
      robs_id.reserve(robData.requested_ROBs.size());
      robs_size.reserve(robData.requested_ROBs.size());
      robs_history.reserve(robData.requested_ROBs.size());
      robs_status.reserve(robData.requested_ROBs.size());
 
      for (const auto& rob : robData.requested_ROBs) {
        robs_id.push_back(rob.second.rob_id);
        robs_size.push_back(rob.second.rob_size);
        robs_history.push_back(rob.second.rob_history);
        robs_status.push_back(rob.second.isStatusOk());
      }
 
      bool result = true;
      result &= tc->setDetail("alg_idx", aiToHandleIndex[aiHash]);
      result &= tc->setDetail("lvl1ID", robData.lvl1ID);
      result &= tc->setDetail<std::vector<uint32_t>>("robs_id", robs_id);
      result &= tc->setDetail<std::vector<uint32_t>>("robs_size", robs_size);
      result &= tc->setDetail<std::vector<unsigned>>("robs_history", robs_history);
      result &= tc->setDetail<std::vector<unsigned short>>("robs_status", robs_status);
      result &= tc->setDetail("start", robData.start_time);
      result &= tc->setDetail("stop", robData.end_time);
 
      if (!result) ATH_MSG_WARNING("Failed to append one or more details to trigger cost ROS TC");
    }
  }
 
  if (msg().level() <= MSG::VERBOSE) {
    ATH_MSG_VERBOSE("--- Trig Cost Event Summary ---");
    for ( const xAOD::TrigComposite* tc : *costOutputHandle ) {
      ATH_MSG_VERBOSE("Algorithm:'" << TrigConf::HLTUtils::hash2string( tc->getDetail<TrigConf::HLTHash>("alg"), "ALG") << "'");
      ATH_MSG_VERBOSE("  Store:'" << TrigConf::HLTUtils::hash2string( tc->getDetail<TrigConf::HLTHash>("store"), "STORE") << "'");
      ATH_MSG_VERBOSE("  View ID:" << tc->getDetail<int16_t>("view"));
      ATH_MSG_VERBOSE("  Thread #:" << tc->getDetail<uint32_t>("thread") );
      ATH_MSG_VERBOSE("  Thread ID Hash:" << tc->getDetail<uint32_t>("thash") );
      ATH_MSG_VERBOSE("  Slot:" << tc->getDetail<uint32_t>("slot") );
      ATH_MSG_VERBOSE("  RoI ID Hash:" << tc->getDetail<int32_t>("roi") );
      ATH_MSG_VERBOSE("  Start Time:" << tc->getDetail<uint64_t>("start") << " mu s");
      ATH_MSG_VERBOSE("  Stop Time:" << tc->getDetail<uint64_t>("stop") << " mu s");
    }
  }
 
  return StatusCode::SUCCESS;
}

finalize()

StatusCode TrigCostSvc::finalize ( )

overridevirtual

Finalize, act on m_saveHashes.

Definition at line 65 of file TrigCostSvc.cxx.

                                 {
  ATH_MSG_DEBUG("TrigCostSvc finalize()");
  if (m_saveHashes) {
    TrigConf::HLTUtils::hashes2file();
    ATH_MSG_INFO("Calling hashes2file, saving dump of job's HLT hashing dictionary to disk.");
  }
  return StatusCode::SUCCESS;
}

generateTimeoutReport()

StatusCode TrigCostSvc::generateTimeoutReport ( const EventContext & context, std::string & report )

overridevirtual

Returns

Generate timeout report with the most time consuming algorithms

Parameters

[in]	context	The event context
[out]	report	Created report with algorithms and times (in ms)

Definition at line 423 of file TrigCostSvc.cxx.

                                                                                            {
 
  ATH_CHECK(checkSlot(context));
  if (!m_eventMonitored[context.slot()]) {
    ATH_MSG_DEBUG("Not a monitored event.");
    report = "";
    return StatusCode::SUCCESS;
  }
 
  std::unique_lock lockUnique(m_slotMutex[context.slot()]);
 
  tbb::concurrent_hash_map< AlgorithmIdentifier, AlgorithmPayload, AlgorithmIdentifierHashCompare>::const_iterator beginIt;
  tbb::concurrent_hash_map< AlgorithmIdentifier, AlgorithmPayload, AlgorithmIdentifierHashCompare>::const_iterator endIt;
  tbb::concurrent_hash_map< AlgorithmIdentifier, AlgorithmPayload, AlgorithmIdentifierHashCompare>::const_iterator it;
  ATH_CHECK(m_algStartInfo.getIterators(context, msg(), beginIt, endIt));
 
  // Create map that sorts in descending order
  std::map<uint64_t, std::string, std::greater<uint64_t>> timeToAlgMap;
 
  for (it = beginIt; it != endIt; ++it) {
    const AlgorithmIdentifier& ai = it->first;
    const AlgorithmPayload& ap = it->second;
 
    // Don't look at any records from other slots
    if (ai.m_realSlot != context.slot()) continue;
 
    uint64_t startTime = ap.m_algStartTime.microsecondsSinceEpoch();
    uint64_t stopTime = 0;
    {
      tbb::concurrent_hash_map<AlgorithmIdentifier, TrigTimeStamp, AlgorithmIdentifierHashCompare>::const_accessor stopTimeAcessor;
      if (m_algStopTime.retrieve(ai, stopTimeAcessor, msg()).isFailure()) {
        ATH_MSG_DEBUG("No end time for '" << ai.m_caller << "', '" << ai.m_store << "'");
      } else { // retrieve was a success
        stopTime = stopTimeAcessor->second.microsecondsSinceEpoch();
      }
      // stopTimeAcessor goes out of scope - lock released
    }
 
    if (stopTime == 0) continue; 
 
    timeToAlgMap[stopTime-startTime] = ai.m_caller;
  }
 
  // Save top 5 times to the report
  report = "Timeout detected with the following algorithms consuming the most time: ";
  int algCounter = 0;
  for(const std::pair<const uint64_t, std::string>& p : timeToAlgMap){
    // Save time in miliseconds instead of microseconds
    report += p.second + " (" + std::to_string(std::lround(p.first/1e3)) + " ms)";
    ++algCounter;
    if (algCounter >= 5){
      break;
    }
    report += ", ";
  }
 
  return StatusCode::SUCCESS;
}

getROIID()

int32_t TrigCostSvc::getROIID ( const EventContext & context )

private

@breif Internal function to return a RoI from an extended event context context

Parameters

[in]

context

The event context

Returns

RoIId from the ATLAS extended event context. Or, AlgorithmIdentifier::s_noView = -1 for no RoIIdentifier

Definition at line 516 of file TrigCostSvc.cxx.

                                                         {
  if (Atlas::hasExtendedEventContext(context)) {
    const TrigRoiDescriptor* roi = Atlas::getExtendedEventContext(context).roiDescriptor();
    if (roi) return static_cast<int32_t>(roi->roiId());
  }
  return AlgorithmIdentifier::s_noView;
}

initialize()

StatusCode TrigCostSvc::initialize ( )

overridevirtual

Initialise, create enough storage to store m_eventSlots.

Definition at line 39 of file TrigCostSvc.cxx.

                                   {
  ATH_MSG_DEBUG("TrigCostSvc initialize()");
  m_eventSlots = SG::getNSlots();
  // TODO Remove this when the configuration is correctly propagated in config-then-run jobs
  if (!m_eventSlots) {
    ATH_MSG_WARNING("numConcurrentEvents() == 0. This is a misconfiguration, probably coming from running from pickle. "
      "Setting local m_eventSlots to a 'large' number until this is fixed to allow the job to proceed.");
    m_eventSlots = 100;
  }
  ATH_MSG_INFO("Initializing TrigCostSvc with " << m_eventSlots << " event slots");
 
  // We cannot have a vector here as atomics are not movable nor copyable. Unique heap arrays are supported by C++
  m_eventMonitored = std::make_unique< std::atomic<bool>[] >( m_eventSlots );
  m_slotMutex = std::make_unique< std::shared_mutex[] >( m_eventSlots );
 
  for (size_t i = 0; i < m_eventSlots; ++i) m_eventMonitored[i] = false;
 
  ATH_CHECK(m_algStartInfo.initialize(m_eventSlots));
  ATH_CHECK(m_algStopTime.initialize(m_eventSlots));
  ATH_CHECK(m_rosData.initialize(m_eventSlots));
 
  return StatusCode::SUCCESS;
}

isMonitoredEvent()

bool TrigCostSvc::isMonitoredEvent ( const EventContext & context, const bool includeMultiSlot = true ) const

overridevirtual

Returns

If the current context is flagged as being monitored.

Parameters

[in]

context

The event context

Definition at line 526 of file TrigCostSvc.cxx.

                                                                                                 {
  if (m_eventMonitored[ context.slot() ]) {
    return true;
  }
  if (includeMultiSlot && m_enableMultiSlot) {
    return m_eventMonitored[ m_masterSlot ];
  }
  return false;
}

monitor()

StatusCode TrigCostSvc::monitor ( const EventContext & context, const AlgorithmIdentifier & ai, const TrigTimeStamp & now, const AuditType type )

private

Internal call to save monitoring data for a given AlgorithmIdentifier.

Parameters

[in]	context	The event context
[in]	ai	The AlgorithmIdentifier key to store
[in]	now	The timestamp to store (amoung other values)
[in]	type	The type of the audit event to store

Returns

Success if the data are saved

Definition at line 140 of file TrigCostSvc.cxx.

                                                                                                                                          {
 
  if (type == AuditType::Before) {
 
    AlgorithmPayload ap {
      now,
      std::this_thread::get_id(),
      getROIID(context),
      static_cast<uint32_t>(context.slot())
    };
    ATH_CHECK( m_algStartInfo.insert(ai, ap, msg()) );
 
    // Cache the AlgorithmIdentifier which has just started executing on this thread
    if (ai.m_realSlot == ai.m_slotToSaveInto) {
      tbb::concurrent_hash_map<std::thread::id, AlgorithmIdentifier, ThreadHashCompare>::accessor acc;
      m_threadToAlgMap.insert(acc, ap.m_algThreadID);
      acc->second = ai;
    }
 
  } else if (type == AuditType::After) {
 
    ATH_CHECK( m_algStopTime.insert(ai, now, msg()) );
 
  } else {
 
    ATH_MSG_ERROR("Only expecting AuditType::Before or AuditType::After");
    return StatusCode::FAILURE;
 
  }
 
  return StatusCode::SUCCESS;
}

monitorROS()

StatusCode TrigCostSvc::monitorROS ( const EventContext & context, robmonitor::ROBDataMonitorStruct payload )

overridevirtual

Implementation of ITrigCostSvc::monitorROS.

Parameters

[in]	context	The event context
[in]	payload	ROB data to be associated with ROS

Definition at line 176 of file TrigCostSvc.cxx.

                                                                                                     {
  ATH_CHECK(checkSlot(context));
  ATH_MSG_DEBUG( "Received ROB payload " << payload );
 
  // Associate payload with an algorithm
  AlgorithmIdentifier theAlg;
  {
    tbb::concurrent_hash_map<std::thread::id, AlgorithmIdentifier, ThreadHashCompare>::const_accessor acc;
    bool result = m_threadToAlgMap.find(acc, std::this_thread::get_id());
    if (!result){
      ATH_MSG_WARNING( "Cannot find algorithm on this thread (id=" << std::this_thread::get_id() << "). Request "<< payload <<" won't be monitored");
      return StatusCode::SUCCESS;
    }
 
    theAlg = acc->second;
  }
 
  // Record data in TrigCostDataStore
  ATH_MSG_DEBUG( "Adding ROBs from" << payload.requestor_name << " to " << theAlg.m_hash );
  {
    std::shared_lock lockShared( m_slotMutex[ context.slot() ] );
    ATH_CHECK( m_rosData.push_back(theAlg, std::move(payload), msg()) );
  }
 
  return StatusCode::SUCCESS;
}

processAlg()

StatusCode TrigCostSvc::processAlg ( const EventContext & context, const std::string & caller, const AuditType type )

overridevirtual

Implementation of ITrigCostSvc::processAlg.

Parameters

[in]	context	The event context
[in]	caller	Name of the algorithm to audit CPU usage for
[in]	type	If we are Before or After the algorithm's execution

Definition at line 105 of file TrigCostSvc.cxx.

                                                                                                             {
  ATH_CHECK(checkSlot(context));
 
  TrigTimeStamp now;
 
  // Do per-event within-slot monitoring
  if (m_eventMonitored[ context.slot() ]) {
    // Multiple simultaneous calls allowed here, adding their data to the concurrent map.
    std::shared_lock lockShared( m_slotMutex[ context.slot() ] );
 
    AlgorithmIdentifier ai = AlgorithmIdentifierMaker::make(context, caller, msg());
    ATH_CHECK( ai.isValid() );
 
    ATH_CHECK(monitor(context, ai, now, type));
 
    ATH_MSG_VERBOSE("Caller '" << caller << "', '" << ai.m_store << "', slot:" << context.slot() << " "
      << (type == AuditType::Before ? "BEGAN" : "ENDED") << " at " << now.microsecondsSinceEpoch());
  }
 
  // MultiSlot mode: do per-event monitoring of all slots, but saving the data within the master-slot
  if (m_enableMultiSlot && context.slot() != m_masterSlot && m_eventMonitored[ m_masterSlot ]) {
    std::shared_lock lockShared( m_slotMutex[ m_masterSlot ] );
 
    // Note: we override the storage location of these data from all other slots to be saved in the MasterSlot
    AlgorithmIdentifier ai = AlgorithmIdentifierMaker::make(context, caller, msg(), m_masterSlot);
    ATH_CHECK( ai.isValid() );
 
    ATH_CHECK(monitor(context, ai, now, type));
  }
 
  return StatusCode::SUCCESS;
}

startEvent()

StatusCode TrigCostSvc::startEvent ( const EventContext & context, const bool enableMonitoring = true )

overridevirtual

Implementation of ITrigCostSvc::startEvent.

Parameters

[in]	context	The event context
[in]	enableMonitoring	Sets if the event should be monitored or not. Not monitoring will save CPU

Returns

Success unless monitoring is enabled and the service's data stores can not be cleared for some reason

Definition at line 76 of file TrigCostSvc.cxx.

                                                                                           {
  const bool monitoredEvent = (enableMonitoring || m_monitorAllEvents);
  ATH_CHECK(checkSlot(context));
 
  m_eventMonitored[ context.slot() ] = false;
 
  {
    // "clear" is a whole table operation, we need it all to ourselves
    std::unique_lock lockUnique( m_slotMutex[ context.slot() ] );
    if (monitoredEvent) {
      // Empty transient thread-safe stores in preparation for recording this event's cost data
      ATH_CHECK(m_algStartInfo.clear(context, msg()));
      ATH_CHECK(m_algStopTime.clear(context, msg()));
      ATH_CHECK(m_rosData.clear(context, msg()));
    }
 
    // Enable collection of data in this slot for monitoredEvents
    m_eventMonitored[ context.slot() ] = monitoredEvent;
  }
 
  // As we missed the AuditType::Before of the TrigCostSupervisorAlg (which is calling this TrigCostSvc::startEvent), let's add it now.
  // This will be our canonical initial timestamps for measuring this event. Similar will be done for DecisionSummaryMakerAlg at the end
  ATH_CHECK(processAlg(context, m_costSupervisorAlgName, AuditType::Before));
 
  return StatusCode::SUCCESS;
}

Member Data Documentation

m_algStartInfo

TrigCostDataStore<AlgorithmPayload> TrigCostSvc::m_algStartInfo

private

Thread-safe store of algorithm start payload.

Definition at line 151 of file TrigCostSvc.h.

m_algStopTime

TrigCostDataStore<TrigTimeStamp> TrigCostSvc::m_algStopTime

private

Thread-safe store of algorithm stop times.

Definition at line 152 of file TrigCostSvc.h.

m_costFinalizeAlgName

Gaudi::Property<std::string> TrigCostSvc::m_costFinalizeAlgName {this, "CostFinalizeAlgName", "TrigCostFinalizeAlg", "The name of cost monitoring finalize algorithm, starting at the end of the event"}

private

Definition at line 166 of file TrigCostSvc.h.

{this, "CostFinalizeAlgName", "TrigCostFinalizeAlg", "The name of cost monitoring finalize algorithm, starting at the end of the event"};

m_costSupervisorAlgName

Gaudi::Property<std::string> TrigCostSvc::m_costSupervisorAlgName {this, "CostSupervisorAlgName", "TrigCostSupervisorAlg", "The name of cost monitoring supervising algorithm, starting at the begining of the event"}

private

Definition at line 165 of file TrigCostSvc.h.

{this, "CostSupervisorAlgName", "TrigCostSupervisorAlg", "The name of cost monitoring supervising algorithm, starting at the begining of the event"};

m_enableMultiSlot

Gaudi::Property<bool> TrigCostSvc::m_enableMultiSlot {this, "EnableMultiSlot", false, "Monitored events in the MasterSlot collect data from events running in other slots."}

private

Definition at line 162 of file TrigCostSvc.h.

{this, "EnableMultiSlot", false, "Monitored events in the MasterSlot collect data from events running in other slots."};

m_eventMonitored

std::unique_ptr< std::atomic<bool>[] > TrigCostSvc::m_eventMonitored

private

Used to cache if the event in a given slot is being monitored.

Definition at line 148 of file TrigCostSvc.h.

m_eventSlots

size_t TrigCostSvc::m_eventSlots

private

Number of concurrent processing slots.

Cached from Gaudi

Definition at line 147 of file TrigCostSvc.h.

m_globalMutex

std::mutex TrigCostSvc::m_globalMutex

private

Used to protect all-slot modifications.

Definition at line 150 of file TrigCostSvc.h.

m_masterSlot

Gaudi::Property<size_t> TrigCostSvc::m_masterSlot {this, "MasterSlot", 0, "The slot responsible for saving MultiSlot data"}

private

Definition at line 164 of file TrigCostSvc.h.

{this, "MasterSlot", 0, "The slot responsible for saving MultiSlot data"};

m_monitorAllEvents

Gaudi::Property<bool> TrigCostSvc::m_monitorAllEvents {this, "MonitorAllEvents", false, "Monitor every HLT event, e.g. for offline validation."}

private

Definition at line 161 of file TrigCostSvc.h.

{this, "MonitorAllEvents", false, "Monitor every HLT event, e.g. for offline validation."};

m_rosData

TrigCostDataStore<std::vector<robmonitor::ROBDataMonitorStruct> > TrigCostSvc::m_rosData

private

Thread-safe store of ROS data.

Definition at line 153 of file TrigCostSvc.h.

m_saveHashes

Gaudi::Property<bool> TrigCostSvc::m_saveHashes {this, "SaveHashes", false, "Store a copy of the hash dictionary for easier debugging"}

private

Definition at line 163 of file TrigCostSvc.h.

{this, "SaveHashes", false, "Store a copy of the hash dictionary for easier debugging"};

m_slotMutex

std::unique_ptr< std::shared_mutex[] > TrigCostSvc::m_slotMutex

private

Used to control and protect whole-table operations.

Definition at line 149 of file TrigCostSvc.h.

m_threadCounter

size_t TrigCostSvc::m_threadCounter

private

Count how many unique thread ID we have seen.

Definition at line 158 of file TrigCostSvc.h.

m_threadToAlgMap

tbb::concurrent_hash_map<std::thread::id, AlgorithmIdentifier, ThreadHashCompare> TrigCostSvc::m_threadToAlgMap

private

Keeps track of what is running right now in each thread.

Definition at line 155 of file TrigCostSvc.h.

m_threadToCounterMap

std::unordered_map<uint32_t, uint32_t> TrigCostSvc::m_threadToCounterMap

private

Map thread's hash ID to a counting numeral.

Definition at line 157 of file TrigCostSvc.h.

---

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

• TrigCostSvc.h
• TrigCostSvc.cxx