SchedulerMonSvc Class Reference

Service monitoring the Scheduler status and producing relevant online histograms. More...

#include <SchedulerMonSvc.h>

Inheritance diagram for SchedulerMonSvc:

Collaboration diagram for SchedulerMonSvc:

Public Types
using	ClockType = decltype(IScheduler::OccupancySnapshot::time)::clock

Public Member Functions
	SchedulerMonSvc (const std::string &name, ISvcLocator *svcLoc)
virtual	~SchedulerMonSvc () override=default
virtual StatusCode	initialize () override
virtual StatusCode	startMonitoring () override
	Start querying and monitoring Scheduler status. More...
virtual StatusCode	stopMonitoring () override
	Stop querying and monitoring Scheduler status. More...

Private Attributes
Gaudi::Property< std::string >	m_schedulerName
Gaudi::Property< unsigned int >	m_monIntervalMillisec
ToolHandle< GenericMonitoringTool >	m_monTool
SmartIF< IScheduler >	m_scheduler {nullptr}
std::atomic_bool	m_running {false}
ClockType::time_point	m_startTime {}
std::atomic_size_t	m_lastSnapCounter {0}

Detailed Description

Service monitoring the Scheduler status and producing relevant online histograms.

Definition at line 26 of file SchedulerMonSvc.h.

Member Typedef Documentation

ClockType

using SchedulerMonSvc::ClockType = decltype(IScheduler::OccupancySnapshot::time)::clock

Definition at line 30 of file SchedulerMonSvc.h.

Constructor & Destructor Documentation

SchedulerMonSvc()

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

Definition at line 43 of file SchedulerMonSvc.cxx.

: base_class(name, svcLoc) {}

~SchedulerMonSvc()

virtual SchedulerMonSvc::~SchedulerMonSvc ( )

overridevirtualdefault

Member Function Documentation

initialize()

StatusCode SchedulerMonSvc::initialize ( )

overridevirtual

Definition at line 47 of file SchedulerMonSvc.cxx.

                                       {
  if (!m_monTool.empty()) ATH_CHECK(m_monTool.retrieve());
  return StatusCode::SUCCESS;
}

startMonitoring()

StatusCode SchedulerMonSvc::startMonitoring ( )

overridevirtual

Start querying and monitoring Scheduler status.

Definition at line 53 of file SchedulerMonSvc.cxx.

                                            {
  // Get a handle to the scheduler
  if (!m_scheduler.isValid()) {
    m_scheduler = serviceLocator()->service<IScheduler>(m_schedulerName, false);
    if (!m_scheduler.isValid()) {
      ATH_MSG_ERROR("Failed to retrieve the Scheduler service with name " << m_schedulerName);
      return StatusCode::FAILURE;
    }
  }
 
  // Get the number of threads and slots
  const int numThreads = std::stoi( SmartIF<IProperty>(m_scheduler)->getProperty("ThreadPoolSize").toString() );
  const int numSlots = std::stoi( serviceLocator()->service("EventDataSvc").as<IProperty>()->getProperty("NSlots").toString() );
 
  // Flag the monitoring as running (prevents going past this point twice)
  if (bool expected = false; not m_running.compare_exchange_strong(expected, true)) {
    ATH_MSG_ERROR("startMonitoring called but it is already running");
    return StatusCode::FAILURE;
  }
 
  // Construct the callback and pass it to the scheduler monitoring API
  // Note: capture by value as this lambda is called from outside the scope of this method
  auto monCallback = [this, numThreads, numSlots](IScheduler::OccupancySnapshot snap) -> void {
    auto monTime = Monitored::Timer("TIME_monCallback");
    // Calculate and update snap counters
    const ClockType::duration wallTime = snap.time - m_startTime;
    const size_t thisSnapCounter = std::chrono::duration_cast<std::chrono::milliseconds>(wallTime).count() / m_monIntervalMillisec.value();
    const size_t lastSnapCounter = m_lastSnapCounter.exchange(thisSnapCounter);
    const int periodsSinceLastSnap = thisSnapCounter - lastSnapCounter;
 
    // If new snap comes before next sampling point, discard it
    if (periodsSinceLastSnap <= 0) {
      ATH_MSG_DEBUG("Discarding snap because periodsSinceLastSnap=" << periodsSinceLastSnap << " is not positive");
      return;
    }
 
    // Monitor total state counts across all slots
    std::vector<int> stateTotalCounts(static_cast<size_t>(AlgState::MAXVALUE), 0);
    for (size_t slot=0; slot < snap.states.size(); ++slot) {
      for (size_t state=0; state < snap.states[slot].size(); ++state) {
        stateTotalCounts[state] += snap.states[slot][state];
      }
    }
    auto mon_stateNumber = Monitored::Collection("AlgStates", s_algStateNumbers);
    auto mon_stateTotalCounts = Monitored::Collection("StateTotalCounts", stateTotalCounts);
 
    // Monitor alg state counts absolute numbers and ratios to N threads and N active states
    std::vector<Monitored::Scalar<int>> mon_stateCounts;
    std::vector<Monitored::Scalar<double>> mon_stateCountsOverThreads;
    std::vector<Monitored::Scalar<double>> mon_stateCountsOverSlots;
    std::vector<Monitored::Scalar<double>> mon_stateCountsOverActive;
    mon_stateCounts.reserve(static_cast<size_t>(AlgState::MAXVALUE));
    mon_stateCountsOverThreads.reserve(static_cast<size_t>(AlgState::MAXVALUE));
    mon_stateCountsOverSlots.reserve(static_cast<size_t>(AlgState::MAXVALUE));
    mon_stateCountsOverActive.reserve(static_cast<size_t>(AlgState::MAXVALUE));
    int activeCount = 0;
    for (size_t i : s_activeAlgStateNumbers) {
      activeCount += stateTotalCounts[i];
    }
    for (size_t i : s_algStateNumbers) {
      mon_stateCounts.emplace_back(s_algStateNames[i].data(), stateTotalCounts[i]);
      mon_stateCountsOverThreads.emplace_back(s_algStateNames[i].data()+"_Over_Threads"s, divAsDouble(stateTotalCounts[i], numThreads));
      mon_stateCountsOverSlots.emplace_back(s_algStateNames[i].data()+"_Over_Slots"s, divAsDouble(stateTotalCounts[i], numSlots));
      double toActive = (activeCount > 0) ? divAsDouble(stateTotalCounts[i], activeCount) : 0;
      mon_stateCountsOverActive.emplace_back(s_algStateNames[i].data()+"_Over_Active"s, toActive);
    }
 
    // Monitor number of free slots
    auto mon_freeSlots = Monitored::Scalar("FreeSlots", m_scheduler->freeSlots());
    auto mon_freeSlotsFrac = Monitored::Scalar("FreeSlotsFraction", divAsDouble(m_scheduler->freeSlots(), numSlots));
 
    // Reserve vector of references with size equal to the number of variables added into the vector in the loop below
    std::vector<std::reference_wrapper<Monitored::IMonitoredVariable>> allMonVars;
    allMonVars.reserve(6 +
                       mon_stateCounts.size() +
                       mon_stateCountsOverThreads.size() +
                       mon_stateCountsOverSlots.size() +
                       mon_stateCountsOverActive.size());
    // Fill monitoring histograms once for each sampling period passed since the last fill
    // If multiple sampling periods passed, it means the scheduler state didn't change during that time
    for (size_t snapNumber=lastSnapCounter+1; snapNumber<=thisSnapCounter; ++snapNumber) {
      auto mon_snapNumber = Monitored::Scalar("SnapNumber", snapNumber);
      auto mon_wallTimeSec = Monitored::Scalar("WallTimeSeconds", snapNumber*m_monIntervalMillisec.value()*1e-3);
      allMonVars.clear();
      allMonVars.insert(allMonVars.end(), mon_stateCounts.begin(), mon_stateCounts.end());
      allMonVars.insert(allMonVars.end(), mon_stateCountsOverThreads.begin(), mon_stateCountsOverThreads.end());
      allMonVars.insert(allMonVars.end(), mon_stateCountsOverSlots.begin(), mon_stateCountsOverSlots.end());
      allMonVars.insert(allMonVars.end(), mon_stateCountsOverActive.begin(), mon_stateCountsOverActive.end());
      allMonVars.insert(allMonVars.end(), {mon_stateNumber, mon_stateTotalCounts, mon_freeSlots, mon_freeSlotsFrac,
                                           mon_snapNumber, mon_wallTimeSec});
      Monitored::Group(m_monTool, allMonVars);
    }
    monTime.stop();
    Monitored::Group(m_monTool, monTime);
  };
 
  // Start monitoring
  m_startTime = ClockType::now();
  m_scheduler->recordOccupancy(m_monIntervalMillisec.value(), std::move(monCallback));
 
  ATH_MSG_INFO("Scheduler monitoring started");
 
  return StatusCode::SUCCESS;
}

stopMonitoring()

StatusCode SchedulerMonSvc::stopMonitoring ( )

overridevirtual

Stop querying and monitoring Scheduler status.

Definition at line 159 of file SchedulerMonSvc.cxx.

                                           {
  if (bool expected = true; not m_running.compare_exchange_strong(expected, false)) {
    ATH_MSG_WARNING("stopMonitoring called but it was not running");
    return StatusCode::SUCCESS;
  }
  m_scheduler->recordOccupancy(-1, {});
  ATH_MSG_INFO("Scheduler monitoring stopped");
  return StatusCode::SUCCESS;
}

Member Data Documentation

m_lastSnapCounter

std::atomic_size_t SchedulerMonSvc::m_lastSnapCounter {0}

private

Definition at line 57 of file SchedulerMonSvc.h.

m_monIntervalMillisec

Gaudi::Property<unsigned int> SchedulerMonSvc::m_monIntervalMillisec

private

Initial value:

{
    this, "MonIntervalMillisec", 100, "Monitoring snapshot interval in milliseconds"}

Definition at line 48 of file SchedulerMonSvc.h.

m_monTool

ToolHandle<GenericMonitoringTool> SchedulerMonSvc::m_monTool

private

Initial value:

{
    this, "MonTool", "", "Monitoring tool"}

Definition at line 50 of file SchedulerMonSvc.h.

m_running

std::atomic_bool SchedulerMonSvc::m_running {false}

private

Definition at line 55 of file SchedulerMonSvc.h.

m_scheduler

SmartIF<IScheduler> SchedulerMonSvc::m_scheduler {nullptr}

private

Definition at line 54 of file SchedulerMonSvc.h.

m_schedulerName

Gaudi::Property<std::string> SchedulerMonSvc::m_schedulerName

private

Initial value:

{
    this, "SchedulerName", "AvalancheSchedulerSvc", "Name of the scheduler"}

Definition at line 46 of file SchedulerMonSvc.h.

m_startTime

ClockType::time_point SchedulerMonSvc::m_startTime {}

private

Definition at line 56 of file SchedulerMonSvc.h.

---

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

• SchedulerMonSvc.h
• SchedulerMonSvc.cxx