AvalancheSchedulerSvc Class Reference

#include </builds/gaudi/Gaudi/GaudiHive/src/AvalancheSchedulerSvc.h>

Inheritance diagram for AvalancheSchedulerSvc:

Collaboration diagram for AvalancheSchedulerSvc:

Classes
struct	TaskSpec
	Struct to hold entries in the alg queues. More...
struct	AlgQueueSort
	Comparison operator to sort the queues. More...

Public Member Functions
StatusCode	initialize () override
	Initialise.
StatusCode	finalize () override
	Finalise.
StatusCode	pushNewEvent (EventContext *eventContext) override
	Make an event available to the scheduler.
StatusCode	pushNewEvents (std::vector< EventContext * > &eventContexts) override
StatusCode	popFinishedEvent (EventContext *&eventContext) override
	Blocks until an event is available.
StatusCode	tryPopFinishedEvent (EventContext *&eventContext) override
	Try to fetch an event from the scheduler.
unsigned int	freeSlots () override
	Get free slots number.
void	dumpState () override
	Dump scheduler state for all slots.
virtual StatusCode	scheduleEventView (const EventContext *sourceContext, const std::string &nodeName, std::unique_ptr< EventContext > viewContext) override
	Method to inform the scheduler about event views.
virtual void	recordOccupancy (int samplePeriod, std::function< void(OccupancySnapshot)> callback) override
	Sample occupancy at fixed interval (ms) Negative value to deactivate, 0 to snapshot every change Each sample, apply the callback function to the result.
bool	next (TaskSpec &ts, bool asynchronous)
Public Member Functions inherited from extends< Service, IScheduler >
void const *	i_cast (const InterfaceID &tid) const override
	Implementation of IInterface::i_cast.
StatusCode	queryInterface (const InterfaceID &ti, void **pp) override
	Implementation of IInterface::queryInterface.
std::vector< std::string >	getInterfaceNames () const override
	Implementation of IInterface::getInterfaceNames.
Public Member Functions inherited from Service
const std::string &	name () const override
	Retrieve name of the service.
StatusCode	configure () override
StatusCode	initialize () override
StatusCode	start () override
StatusCode	stop () override
StatusCode	finalize () override
StatusCode	terminate () override
Gaudi::StateMachine::State	FSMState () const override
Gaudi::StateMachine::State	targetFSMState () const override
StatusCode	reinitialize () override
StatusCode	restart () override
StatusCode	sysInitialize () override
	Initialize Service.
StatusCode	sysStart () override
	Initialize Service.
StatusCode	sysStop () override
	Initialize Service.
StatusCode	sysFinalize () override
	Finalize Service.
StatusCode	sysReinitialize () override
	Re-initialize the Service.
StatusCode	sysRestart () override
	Re-initialize the Service.
	Service (std::string name, ISvcLocator *svcloc)
	Standard Constructor.
SmartIF< ISvcLocator > &	serviceLocator () const override
	Retrieve pointer to service locator.
template<typename IFace = IService>
SmartIF< IFace >	service (const std::string &name, bool createIf=true) const
template<class T>
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, ToolHandle< T > &hndl, const std::string &doc="none")
template<class T>
StatusCode	declareTool (ToolHandle< T > &handle, bool createIf=true)
template<class T>
StatusCode	declareTool (ToolHandle< T > &handle, const std::string &toolTypeAndName, bool createIf=true)
	Declare used tool.
template<class T>
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, ToolHandleArray< T > &hndlArr, const std::string &doc="none")
template<class T>
void	addToolsArray (ToolHandleArray< T > &hndlArr)
const std::vector< IAlgTool * > &	tools () const
SmartIF< IAuditorSvc > &	auditorSvc () const
	The standard auditor service.May not be invoked before sysInitialize() has been invoked.
Public Member Functions inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
StatusCode	setProperty (const Gaudi::Details::PropertyBase &p)
	Set the property from a property.
StatusCode	setProperty (const std::string &name, const char *v)
	Special case for string literals.
StatusCode	setProperty (const std::string &name, const std::string &v)
	Special case for std::string.
StatusCode	setProperty (const std::string &name, const TYPE &value)
	set the property form the value
	PropertyHolder ()=default
Gaudi::Details::PropertyBase &	declareProperty (Gaudi::Details::PropertyBase &prop)
	Declare a property.
Gaudi::Details::PropertyBase *	declareRemoteProperty (const std::string &name, IProperty *rsvc, const std::string &rname="")
	Declare a remote property.
StatusCode	setPropertyRepr (const std::string &n, const std::string &r) override
	set the property from name and value string representation
StatusCode	getProperty (Gaudi::Details::PropertyBase *p) const override
	get the property
const std::vector< Gaudi::Details::PropertyBase * > &	getProperties () const override
	get all properties
bool	hasProperty (std::string_view name) const override
	Return true if we have a property with the given name.
Gaudi::Details::PropertyBase *	property (std::string_view name) const
	\fixme property and bindPropertiesTo should be protected
void	bindPropertiesTo (Gaudi::Interfaces::IOptionsSvc &optsSvc)
PropertyHolder &	operator= (const PropertyHolder &)=delete
Public Member Functions inherited from CommonMessagingBase
virtual	~CommonMessagingBase ()=default
	Virtual destructor.
const SmartIF< IMessageSvc > &	msgSvc () const
	The standard message service.
MsgStream &	msgStream () const
	Return an uninitialized MsgStream.
MsgStream &	msgStream (const MSG::Level level) const
	Predefined configurable message stream for the efficient printouts.
MsgStream &	always () const
	shortcut for the method msgStream(MSG::ALWAYS)
MsgStream &	fatal () const
	shortcut for the method msgStream(MSG::FATAL)
MsgStream &	err () const
	shortcut for the method msgStream(MSG::ERROR)
MsgStream &	error () const
	shortcut for the method msgStream(MSG::ERROR)
MsgStream &	warning () const
	shortcut for the method msgStream(MSG::WARNING)
MsgStream &	info () const
	shortcut for the method msgStream(MSG::INFO)
MsgStream &	debug () const
	shortcut for the method msgStream(MSG::DEBUG)
MsgStream &	verbose () const
	shortcut for the method msgStream(MSG::VERBOSE)
MsgStream &	msg () const
	shortcut for the method msgStream(MSG::INFO)
MSG::Level	msgLevel () const
	get the cached level (originally extracted from the embedded MsgStream)
bool	msgLevel (MSG::Level lvl) const
	get the output level from the embedded MsgStream

Private Types
enum	ActivationState { INACTIVE = 0 , ACTIVE = 1 , FAILURE = 2 }
using	AState = AlgsExecutionStates::State
using	action = std::function<StatusCode()>

Private Member Functions
StatusCode	dumpDataDepsGraphFile (const std::map< std::string, DataObjIDColl > &inDeps, const std::map< std::string, DataObjIDColl > &outDeps) const
void	activate ()
	Activate scheduler.
StatusCode	deactivate ()
	Deactivate scheduler.
unsigned int	algname2index (const std::string &algoname)
	Convert a name to an integer.
const std::string &	index2algname (unsigned int index)
	Convert an integer to a name.
StatusCode	iterate ()
	Loop on all slots to schedule DATAREADY algorithms and sign off ready events.
StatusCode	revise (unsigned int iAlgo, EventContext *contextPtr, AState state, bool iterate=false)
StatusCode	schedule (TaskSpec &&)
StatusCode	signoff (const TaskSpec &)
	The call to this method is triggered only from within the AlgTask.
bool	isStalled (const EventSlot &) const
	Check if scheduling in a particular slot is in a stall.
void	eventFailed (EventContext *eventContext)
	Method to execute if an event failed.
void	dumpSchedulerState (int iSlot)
	Dump the state of the scheduler.

Private Attributes
std::chrono::duration< int64_t, std::milli >	m_snapshotInterval = std::chrono::duration<int64_t, std::milli>::min()
std::chrono::system_clock::time_point	m_lastSnapshot = std::chrono::system_clock::now()
std::function< void(OccupancySnapshot)>	m_snapshotCallback
Gaudi::Property< int >	m_threadPoolSize
Gaudi::Property< int >	m_maxParallelismExtra
Gaudi::Property< std::string >	m_whiteboardSvcName { this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name" }
Gaudi::Property< bool >	m_simulateExecution
Gaudi::Property< std::string >	m_optimizationMode
Gaudi::Property< bool >	m_dumpIntraEventDynamics
Gaudi::Property< int >	m_numOffloadThreads
Gaudi::Property< std::size_t >	m_fiberStackSize
Gaudi::Property< bool >	m_checkDeps
Gaudi::Property< bool >	m_checkOutput
Gaudi::Property< std::vector< std::string > >	m_checkOutputIgnoreList
Gaudi::Property< std::string >	m_useDataLoader
Gaudi::Property< bool >	m_enableCondSvc { this, "EnableConditions", false, "Enable ConditionsSvc" }
Gaudi::Property< bool >	m_showDataDeps
Gaudi::Property< bool >	m_showDataFlow
Gaudi::Property< bool >	m_showControlFlow
Gaudi::Property< bool >	m_verboseSubSlots { this, "VerboseSubSlots", false, "Dump algorithm states for all sub-slots" }
Gaudi::Property< std::string >	m_dataDepsGraphFile
Gaudi::Property< std::string >	m_dataDepsGraphAlgoPattern
Gaudi::Property< std::string >	m_dataDepsGraphObjectPattern
std::atomic< ActivationState >	m_isActive { INACTIVE }
	Flag to track if the scheduler is active or not.
std::thread	m_thread
	The thread in which the activate function runs.
std::unordered_map< std::string, unsigned int >	m_algname_index_map
	Map to bookkeep the information necessary to the name2index conversion.
std::vector< std::string >	m_algname_vect
	Vector to bookkeep the information necessary to the index2name conversion.
SmartIF< IPrecedenceSvc >	m_precSvc
	A shortcut to the Precedence Service.
SmartIF< IHiveWhiteBoard >	m_whiteboard
	A shortcut to the whiteboard.
std::vector< EventSlot >	m_eventSlots
	Vector of events slots.
std::atomic_int	m_freeSlots { 0 }
	Atomic to account for asyncronous updates by the scheduler wrt the rest.
tbb::concurrent_bounded_queue< EventContext * >	m_finishedEvents
	Queue of finished events.
SmartIF< IAlgExecStateSvc >	m_algExecStateSvc
	Algorithm execution state manager.
SmartIF< ICondSvc >	m_condSvc
	A shortcut to service for Conditions handling.
unsigned int	m_algosInFlight = 0
	Number of algorithms presently in flight.
SmartIF< IAlgResourcePool >	m_algResourcePool
	Cache for the algorithm resource pool.
tbb::concurrent_bounded_queue< action >	m_actionsQueue
	Queue where closures are stored and picked for execution.
tbb::concurrent_priority_queue< TaskSpec, AlgQueueSort >	m_scheduledQueue
	Queues for scheduled algorithms.
tbb::concurrent_priority_queue< TaskSpec, AlgQueueSort >	m_scheduledAsynchronousQueue
std::queue< TaskSpec >	m_retryQueue
std::atomic< bool >	m_needsUpdate { true }
SmartIF< IThreadPoolSvc >	m_threadPoolSvc
tbb::task_arena *	m_arena { nullptr }
std::unique_ptr< FiberManager >	m_fiberManager { nullptr }
size_t	m_maxEventsInFlight { 0 }

Friends
class	AlgTask

Additional Inherited Members
Public Types inherited from extends< Service, IScheduler >
using	base_class
	Typedef to this class.
using	extend_interfaces_base
	Typedef to the base of this class.
Public Types inherited from Service
using	Factory = Gaudi::PluginService::Factory<IService*( const std::string&, ISvcLocator* )>
Public Types inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
using	PropertyHolderImpl
	Typedef used to refer to this class from derived classes, as in.
Public Types inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
using	base_class
Public Types inherited from extend_interfaces< Interfaces... >
using	ext_iids
	take union of the ext_iids of all Interfaces...
Protected Member Functions inherited from Service
std::vector< IAlgTool * > &	tools ()
	~Service () override
int	outputLevel () const
	get the Service's output level
Protected Member Functions inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
MSG::Level	setUpMessaging () const override final
	Set up local caches.
MSG::Level	resetMessaging ()
	Reinitialize internal states.
void	updateMsgStreamOutputLevel (int level)
	Update the output level of the cached MsgStream.
Protected Attributes inherited from Service
Gaudi::StateMachine::State	m_state = Gaudi::StateMachine::OFFLINE
	Service state.
Gaudi::StateMachine::State	m_targetState = Gaudi::StateMachine::OFFLINE
	Service state.
Gaudi::Property< int >	m_outputLevel { this, "OutputLevel", MSG::NIL, "output level" }
	flag indicating whether ToolHandle tools have been added to m_tools
Gaudi::Property< bool >	m_auditorInitialize { this, "AuditInitialize", false, "trigger auditor on initialize()" }
Gaudi::Property< bool >	m_auditorStart { this, "AuditStart", false, "trigger auditor on start()" }
Gaudi::Property< bool >	m_auditorStop { this, "AuditStop", false, "trigger auditor on stop()" }
Gaudi::Property< bool >	m_auditorFinalize { this, "AuditFinalize", false, "trigger auditor on finalize()" }
Gaudi::Property< bool >	m_auditorReinitialize { this, "AuditReinitialize", false, "trigger auditor on reinitialize()" }
Gaudi::Property< bool >	m_auditorRestart { this, "AuditRestart", false, "trigger auditor on restart()" }
Gaudi::Property< bool >	m_autoRetrieveTools
Gaudi::Property< bool >	m_checkToolDeps
SmartIF< IAuditorSvc >	m_pAuditorSvc
	Auditor Service.

Detailed Description

Introduction

The scheduler is named after its ability to generically maximize the average intra-event task occupancy by inducing avalanche-like concurrency disclosure waves in conditions of arbitrary intra-event task precedence constraints (see section 3.2 of http://cern.ch/go/7Jn7).

Task precedence management

The scheduler is driven by graph-based task precedence management. When compared to approach used in the ForwardSchedulerSvc, the following advantages can be emphasized:

(1) Faster decision making (thus lower concurrency disclosure downtime); (2) Capacity for proactive task scheduling decision making.

Point (2) allowed to implement a number of generic, non-intrusive intra-event throughput maximization scheduling strategies.

Scheduling principles

o Task scheduling prerequisites

A task is scheduled ASA all following conditions are met:

• if a control flow (CF) graph traversal reaches the task;
• when all data flow (DF) dependencies of the task are satisfied;
• when the DF-ready task pool parsing mechanism (*) considers it, and:
  • a free (or re-entrant) algorithm instance to run within the task is available;
  • there is a free computational resource to run the task.

o (*) Avalanche induction strategies

The scheduler is able to maximize the intra-event throughput by applying several search strategies within the pool, prioritizing tasks according to the following types of precedence rules graph asymmetries:

(A) Local task-to-data asymmetry; (B) Local task-to-task asymmetry; (C) Global task-to-task asymmetry.

o Other mechanisms of throughput maximization

The scheduler supports cooperative scheduling with fibers, allowing a task to suspend its execution while waiting for an off-CPU operation to complete. The worker thread is then free to execute other work and resumes the suspended task once the operation completes.

The mechanism can be applied to the following types of tasks:

• I/O-bound tasks;
• tasks with computation offloading (accelerators, GPGPUs, clouds);
• synchronization-bound tasks.

Credits

Historically, the AvalancheSchedulerSvc branched off the ForwardSchedulerSvc and in many ways built its success on ideas and code of the latter.

Author

Illya Shapoval

Version

1.0

Definition at line 114 of file AvalancheSchedulerSvc.h.

Member Typedef Documentation

action

using AvalancheSchedulerSvc::action = std::function<StatusCode()>

private

Definition at line 161 of file AvalancheSchedulerSvc.h.

AState

using AvalancheSchedulerSvc::AState = AlgsExecutionStates::State

private

Definition at line 160 of file AvalancheSchedulerSvc.h.

Member Enumeration Documentation

ActivationState

enum AvalancheSchedulerSvc::ActivationState

private

Enumerator
INACTIVE
ACTIVE
FAILURE

Definition at line 163 of file AvalancheSchedulerSvc.h.

{ INACTIVE = 0, ACTIVE = 1, FAILURE = 2 };

Member Function Documentation

activate()

void AvalancheSchedulerSvc::activate ( )

private

Activate scheduler.

Activate the scheduler.

From this moment on the queue of actions is checked. The checking will stop when the m_isActive flag is false and the queue is not empty. This will guarantee that all actions are executed and a stall is not created. The TBB pool must be initialised in the thread from where the tasks are launched (http://threadingbuildingblocks.org/docs/doxygen/a00342.html) The scheduler is initialised here since this method runs in a separate thread and spawns the tasks (through the execution of the lambdas)

Definition at line 467 of file AvalancheSchedulerSvc.cpp.

                                     {
 
  ON_DEBUG debug() << "AvalancheSchedulerSvc::activate()" << endmsg;
 
  if ( m_threadPoolSvc->initPool( m_threadPoolSize, m_maxParallelismExtra ).isFailure() ) {
    error() << "problems initializing ThreadPoolSvc" << endmsg;
    m_isActive = FAILURE;
    return;
  }
 
  // Wait for actions pushed into the queue by finishing tasks.
  action     thisAction;
  StatusCode sc( StatusCode::SUCCESS );
 
  m_isActive = ACTIVE;
 
  // Continue to wait if the scheduler is running or there is something to do
  ON_DEBUG debug() << "Start checking the actionsQueue" << endmsg;
  while ( m_isActive == ACTIVE || m_actionsQueue.size() != 0 ) {
    m_actionsQueue.pop( thisAction );
    sc = thisAction();
    ON_VERBOSE {
      if ( sc.isFailure() )
        verbose() << "Action did not succeed (which is not bad per se)." << endmsg;
      else
        verbose() << "Action succeeded." << endmsg;
    }
    else sc.ignore();
 
    // If all queued actions have been processed, update the slot states
    if ( m_needsUpdate.load() && m_actionsQueue.empty() ) {
      sc = iterate();
      ON_VERBOSE {
        if ( sc.isFailure() )
          verbose() << "Iteration did not succeed (which is not bad per se)." << endmsg;
        else
          verbose() << "Iteration succeeded." << endmsg;
      }
      else sc.ignore();
    }
  }
 
  ON_DEBUG debug() << "Terminating thread-pool resources" << endmsg;
  if ( m_threadPoolSvc->terminatePool().isFailure() ) {
    error() << "Problems terminating thread pool" << endmsg;
    m_isActive = FAILURE;
  }
}

algname2index()

unsigned int AvalancheSchedulerSvc::algname2index ( const std::string & algoname )

inlineprivate

Convert a name to an integer.

Definition at line 250 of file AvalancheSchedulerSvc.h.

{ return m_algname_index_map[algoname]; }

deactivate()

StatusCode AvalancheSchedulerSvc::deactivate ( )

private

Deactivate scheduler.

Deactivates the scheduler.

Two actions are pushed into the queue: 1) Drain the scheduler until all events are finished. 2) Flip the status flag m_isActive to false This second action is the last one to be executed by the scheduler.

Definition at line 524 of file AvalancheSchedulerSvc.cpp.

                                             {
 
  if ( m_isActive == ACTIVE ) {
 
    // Set the number of slots available to an error code
    m_freeSlots.store( 0 );
 
    // Empty queue
    action thisAction;
    while ( m_actionsQueue.try_pop( thisAction ) ) {};
 
    // This would be the last action
    m_actionsQueue.push( [this]() -> StatusCode {
      ON_VERBOSE verbose() << "Deactivating scheduler" << endmsg;
      m_isActive = INACTIVE;
      return StatusCode::SUCCESS;
    } );
  }
 
  return StatusCode::SUCCESS;
}

dumpDataDepsGraphFile()

StatusCode AvalancheSchedulerSvc::dumpDataDepsGraphFile ( const std::map< std::string, DataObjIDColl > & inDeps, const std::map< std::string, DataObjIDColl > & outDeps ) const

private

Definition at line 1202 of file AvalancheSchedulerSvc.cpp.

                                                                                                                 {
  // Both maps should have the same algorithm entries
  assert( inDeps.size() == outDeps.size() );
 
  Gaudi::Hive::Graph g{ m_dataDepsGraphFile.value() };
  info() << "Dumping data dependencies graph to file: " << g.fileName() << endmsg;
 
  // define algs and objects
  std::set<std::size_t> definedObjects;
 
  // Regex for selection of algs and objects
  std::regex algNameRegex( m_dataDepsGraphAlgoPattern.value() );
  std::regex objNameRegex( m_dataDepsGraphObjectPattern.value() );
 
  // inDeps and outDeps should have the same entries
  std::size_t algoIndex = 0ul;
  for ( const auto& [algName, ideps] : inDeps ) {
    if ( not std::regex_search( algName, algNameRegex ) ) continue;
    std::string algIndex = "Alg_" + std::to_string( algoIndex );
    g.addNode( algIndex, algName );
 
    // inputs
    for ( const auto& dep : ideps ) {
      if ( not std::regex_search( dep.fullKey(), objNameRegex ) ) continue;
 
      const auto [itr, inserted] = definedObjects.insert( dep.hash() );
      std::string objIndex       = "obj_" + std::to_string( dep.hash() );
      if ( inserted ) g.addNode( objIndex, dep.key() );
 
      g.addEdge( objIndex, algIndex );
    } // loop on ideps
 
    const auto& odeps = outDeps.at( algName );
    for ( const auto& dep : odeps ) {
      if ( not std::regex_search( dep.fullKey(), objNameRegex ) ) continue;
 
      const auto [itr, inserted] = definedObjects.insert( dep.hash() );
      std::string objIndex       = "obj_" + std::to_string( dep.hash() );
      if ( inserted ) g.addNode( objIndex, dep.key() );
 
      g.addEdge( algIndex, objIndex );
    } // loop on odeps
 
    ++algoIndex;
  } // loop on inDeps
 
  return StatusCode::SUCCESS;
}

dumpSchedulerState()

void AvalancheSchedulerSvc::dumpSchedulerState ( int iSlot )

private

Dump the state of the scheduler.

Used for debugging purposes, the state of the scheduler is dumped on screen in order to be inspected.

Definition at line 912 of file AvalancheSchedulerSvc.cpp.

                                                          {
 
  // To have just one big message
  std::ostringstream outputMS;
 
  outputMS << "Dumping scheduler state\n"
           << "=========================================================================================\n"
           << "++++++++++++++++++++++++++++++++++++ SCHEDULER STATE ++++++++++++++++++++++++++++++++++++\n"
           << "=========================================================================================\n\n";
 
  //===========================================================================
 
  outputMS << "------------------ Last schedule: Task/Event/Slot/Thread/State Mapping "
           << "------------------\n\n";
 
  // Figure if TimelineSvc is available (used below to detect threads IDs)
  auto timelineSvc = serviceLocator()->service<ITimelineSvc>( "TimelineSvc", false );
  if ( !timelineSvc.isValid() || !timelineSvc->isEnabled() ) {
    outputMS << "WARNING Enable TimelineSvc in record mode (RecordTimeline = True) to trace the mapping\n";
  } else {
 
    // Figure optimal printout layout
    size_t indt( 0 );
    for ( auto& slot : m_eventSlots ) {
 
      const auto& schedAlgs = slot.algsStates.algsInState( AState::SCHEDULED );
      for ( uint algIndex : schedAlgs ) {
        if ( index2algname( algIndex ).length() > indt ) indt = index2algname( algIndex ).length();
      }
    }
 
    // Figure the last running schedule across all slots
    for ( auto& slot : m_eventSlots ) {
 
      const auto& schedAlgs = slot.algsStates.algsInState( AState::SCHEDULED );
      for ( uint algIndex : schedAlgs ) {
 
        const std::string& algoName{ index2algname( algIndex ) };
 
        outputMS << "  task: " << std::setw( indt ) << algoName << " evt/slot: " << slot.eventContext->evt() << "/"
                 << slot.eventContext->slot();
 
        // Try to get POSIX threads IDs the currently running tasks are scheduled to
        if ( timelineSvc.isValid() ) {
          TimelineEvent te{};
          te.algorithm = algoName;
          te.slot      = slot.eventContext->slot();
          te.event     = slot.eventContext->evt();
 
          if ( timelineSvc->getLastMatching( te ) )
            outputMS << " thread.id: 0x" << std::hex << te.thread << std::dec;
          else
            outputMS << " thread.id: [unknown]"; // this means a task has just
                                                 // been signed off as SCHEDULED,
                                                 // but has not been assigned to a thread yet
                                                 // (i.e., not running yet)
        }
        outputMS << " state: [" << m_algExecStateSvc->algExecState( algoName, *( slot.eventContext ) ) << "]\n";
      }
    }
  }
 
  //===========================================================================
 
  outputMS << "\n---------------------------- Task/CF/FSM Mapping "
           << ( 0 > iSlot ? "[all slots] --" : "[target slot] " ) << "--------------------------\n\n";
 
  int  slotCount   = -1;
  bool wasAlgError = ( iSlot >= 0 ) ? m_eventSlots[iSlot].algsStates.containsAny( { AState::ERROR } ) ||
                                          subSlotAlgsInStates( m_eventSlots[iSlot], { AState::ERROR } )
                                    : false;
 
  for ( auto& slot : m_eventSlots ) {
    ++slotCount;
    if ( slot.complete ) continue;
 
    outputMS << "[ slot: "
             << ( slot.eventContext->valid() ? std::to_string( slot.eventContext->slot() ) : "[ctx invalid]" )
             << ", event: "
             << ( slot.eventContext->valid() ? std::to_string( slot.eventContext->evt() ) : "[ctx invalid]" );
 
    if ( slot.eventContext->eventID().isValid() ) { outputMS << ", eventID: " << slot.eventContext->eventID(); }
    outputMS << " ]:\n\n";
 
    if ( 0 > iSlot || iSlot == slotCount ) {
 
      // If an alg has thrown an error then it's not a failure of the CF/DF graph
      if ( wasAlgError ) {
        outputMS << "ERROR alg(s):";
        int         errorCount = 0;
        const auto& errorAlgs  = slot.algsStates.algsInState( AState::ERROR );
        for ( uint algIndex : errorAlgs ) {
          outputMS << " " << index2algname( algIndex );
          ++errorCount;
        }
        if ( errorCount == 0 ) outputMS << " in subslot(s)";
        outputMS << "\n\n";
      } else {
        // Snapshot of the Control Flow and FSM states
        outputMS << m_precSvc->printState( slot ) << "\n";
      }
 
      // Mention sub slots (this is expensive if the number of sub-slots is high)
      if ( m_verboseSubSlots && !slot.allSubSlots.empty() ) {
        outputMS << "\nNumber of sub-slots: " << slot.allSubSlots.size() << "\n\n";
        auto slotID = slot.eventContext->valid() ? std::to_string( slot.eventContext->slot() ) : "[ctx invalid]";
        for ( auto& ss : slot.allSubSlots ) {
          outputMS << "[ slot: " << slotID << ", sub-slot: "
                   << ( ss.eventContext->valid() ? std::to_string( ss.eventContext->subSlot() ) : "[ctx invalid]" )
                   << ", entry: " << ss.entryPoint << ", event: "
                   << ( ss.eventContext->valid() ? std::to_string( ss.eventContext->evt() ) : "[ctx invalid]" )
                   << " ]:\n\n";
          if ( wasAlgError ) {
            outputMS << "ERROR alg(s):";
            const auto& errorAlgs = ss.algsStates.algsInState( AState::ERROR );
            for ( uint algIndex : errorAlgs ) { outputMS << " " << index2algname( algIndex ); }
            outputMS << "\n\n";
          } else {
            // Snapshot of the Control Flow and FSM states in sub slot
            outputMS << m_precSvc->printState( ss ) << "\n";
          }
        }
      }
    }
  }
 
  //===========================================================================
 
  if ( 0 <= iSlot && !wasAlgError ) {
    outputMS << "\n------------------------------ Algorithm Execution States -----------------------------\n\n";
    m_algExecStateSvc->dump( outputMS, *( m_eventSlots[iSlot].eventContext ) );
  }
 
  outputMS << "\n=========================================================================================\n"
           << "++++++++++++++++++++++++++++++++++++++ END OF DUMP ++++++++++++++++++++++++++++++++++++++\n"
           << "=========================================================================================\n\n";
 
  info() << outputMS.str() << endmsg;
}

dumpState()

void AvalancheSchedulerSvc::dumpState ( )

override

Dump scheduler state for all slots.

Definition at line 628 of file AvalancheSchedulerSvc.cpp.

                                      {
  // The state of the event slots is only modified by the scheduler thread (through the actions in
  // m_actionsQueue), so, unless we are already on it or it is not running, let it produce the dump
  // instead of reading the slots concurrently from the calling thread (typically the EventWatchdog).
  if ( m_isActive != ACTIVE || std::this_thread::get_id() == m_thread.get_id() ) {
    dumpSchedulerState( -1 );
    return;
  }
 
  auto done   = std::make_shared<std::promise<void>>();
  auto future = done->get_future();
  m_actionsQueue.push( [this, done]() -> StatusCode {
    dumpSchedulerState( -1 );
    done->set_value();
    return StatusCode::SUCCESS;
  } );
 
  // Do not block the caller for ever if the scheduler thread is stuck: the dump stays queued
  // and is produced as soon as the scheduler thread gets to it.
  if ( future.wait_for( s_dumpStateTimeout ) != std::future_status::ready ) {
    warning() << "The scheduler thread did not produce the state dump within " << s_dumpStateTimeout.count()
              << "s (it may be blocked): the dump will appear when it proceeds" << endmsg;
  }
}

eventFailed()

void AvalancheSchedulerSvc::eventFailed ( EventContext * eventContext )

private

Method to execute if an event failed.

It can be possible that an event fails.

In this case this method is called. It dumps the state of the scheduler and marks the event as finished.

Definition at line 891 of file AvalancheSchedulerSvc.cpp.

                                                                    {
  const uint slotIdx = eventContext->slot();
 
  error() << "Event " << eventContext->evt() << " on slot " << slotIdx << " failed" << endmsg;
 
  dumpSchedulerState( msgLevel( MSG::VERBOSE ) ? -1 : slotIdx );
 
  // dump temporal and topological precedence analysis (if enabled in the PrecedenceSvc)
  m_precSvc->dumpPrecedenceRules( m_eventSlots[slotIdx] );
 
  // Push into the finished events queue the failed context
  m_eventSlots[slotIdx].complete = true;
  m_finishedEvents.push( m_eventSlots[slotIdx].eventContext.release() );
}

finalize()

StatusCode AvalancheSchedulerSvc::finalize ( )

override

Finalise.

Here the scheduler is deactivated and the thread joined.

Definition at line 433 of file AvalancheSchedulerSvc.cpp.

                                           {
 
  StatusCode sc( Service::finalize() );
  if ( sc.isFailure() ) warning() << "Base class could not be finalized" << endmsg;
 
  sc = deactivate();
  if ( sc.isFailure() ) warning() << "Scheduler could not be deactivated" << endmsg;
 
  debug() << "Deleting FiberManager" << endmsg;
  m_fiberManager.reset();
 
  info() << "Joining Scheduler thread" << endmsg;
  m_thread.join();
 
  // Final error check after thread pool termination
  if ( m_isActive == FAILURE ) {
    error() << "problems in scheduler thread" << endmsg;
    return StatusCode::FAILURE;
  }
 
  return sc;
}

freeSlots()

unsigned int AvalancheSchedulerSvc::freeSlots ( )

override

Get free slots number.

Definition at line 624 of file AvalancheSchedulerSvc.cpp.

{ return std::max( m_freeSlots.load(), 0 ); }

index2algname()

const std::string & AvalancheSchedulerSvc::index2algname ( unsigned int index )

inlineprivate

Convert an integer to a name.

Definition at line 256 of file AvalancheSchedulerSvc.h.

{ return m_algname_vect[index]; }

initialize()

StatusCode AvalancheSchedulerSvc::initialize ( )

override

Initialise.

Here, among some "bureaucracy" operations, the scheduler is activated, executing the activate() function in a new thread.

In addition the algorithms list is acquired from the algResourcePool.

Definition at line 83 of file AvalancheSchedulerSvc.cpp.

                                             {
 
  // Initialise mother class (read properties, ...)
  StatusCode sc( Service::initialize() );
  if ( sc.isFailure() ) warning() << "Base class could not be initialized" << endmsg;
 
  // Get hold of the TBBSvc. This should initialize the thread pool
  m_threadPoolSvc = serviceLocator()->service( "ThreadPoolSvc" );
  if ( !m_threadPoolSvc.isValid() ) {
    fatal() << "Error retrieving ThreadPoolSvc" << endmsg;
    return StatusCode::FAILURE;
  }
  auto castTPS = dynamic_cast<ThreadPoolSvc*>( m_threadPoolSvc.get() );
  if ( !castTPS ) {
    fatal() << "Cannot cast ThreadPoolSvc" << endmsg;
    return StatusCode::FAILURE;
  }
  m_arena = castTPS->getArena();
  if ( !m_arena ) {
    fatal() << "Cannot find valid TBB task_arena" << endmsg;
    return StatusCode::FAILURE;
  }
 
  // Activate the scheduler in another thread.
  info() << "Activating scheduler in a separate thread" << endmsg;
  std::binary_semaphore fiber_manager_initalized{ 0 };
  m_thread = std::thread( [this, &fiber_manager_initalized]() {
    // Initialize FiberManager
    this->m_fiberManager =
        std::make_unique<FiberManager>( this->m_numOffloadThreads.value(), this->m_fiberStackSize.value() );
    fiber_manager_initalized.release();
    this->activate();
  } );
  // Wait for initialization to complete
  fiber_manager_initalized.acquire();
 
  while ( m_isActive != ACTIVE ) {
    if ( m_isActive == FAILURE ) {
      fatal() << "Terminating initialization" << endmsg;
      return StatusCode::FAILURE;
    } else {
      ON_DEBUG debug() << "Waiting for AvalancheSchedulerSvc to activate" << endmsg;
      sleep( 1 );
    }
  }
 
  if ( m_enableCondSvc ) {
    // Get hold of the CondSvc
    m_condSvc = serviceLocator()->service( "CondSvc" );
    if ( !m_condSvc.isValid() ) {
      warning() << "No CondSvc found, or not enabled. "
                << "Will not manage CondAlgorithms" << endmsg;
      m_enableCondSvc = false;
    }
  }
 
  // Get the algo resource pool
  m_algResourcePool = serviceLocator()->service( "AlgResourcePool" );
  if ( !m_algResourcePool.isValid() ) {
    fatal() << "Error retrieving AlgoResourcePool" << endmsg;
    return StatusCode::FAILURE;
  }
 
  m_algExecStateSvc = serviceLocator()->service( "AlgExecStateSvc" );
  if ( !m_algExecStateSvc.isValid() ) {
    fatal() << "Error retrieving AlgExecStateSvc" << endmsg;
    return StatusCode::FAILURE;
  }
 
  // Get Whiteboard
  m_whiteboard = serviceLocator()->service( m_whiteboardSvcName );
  if ( !m_whiteboard.isValid() ) {
    fatal() << "Error retrieving EventDataSvc interface IHiveWhiteBoard." << endmsg;
    return StatusCode::FAILURE;
  }
 
  // Set the MaxEventsInFlight parameters from the number of WB stores
  m_maxEventsInFlight = m_whiteboard->getNumberOfStores();
 
  // Set the number of free slots
  m_freeSlots = m_maxEventsInFlight;
 
  // Get the list of algorithms
  const std::list<IAlgorithm*>& algos      = m_algResourcePool->getFlatAlgList();
  const unsigned int            algsNumber = algos.size();
  if ( algsNumber != 0 ) {
    info() << "Found " << algsNumber << " algorithms" << endmsg;
  } else {
    error() << "No algorithms found" << endmsg;
    return StatusCode::FAILURE;
  }
 
  /* Dependencies
   1) Look for handles in algo, if none
   2) Assume none are required
  */
 
  DataObjIDColl globalInp, globalOutp;
 
  auto hasAsynchronousAlgorithms = false;
 
  // figure out all outputs and check for asynchronous algorithms
  std::map<std::string, DataObjIDColl> algosOutputDependenciesMap;
  for ( IAlgorithm* ialgoPtr : algos ) {
    Gaudi::Algorithm* algoPtr = dynamic_cast<Gaudi::Algorithm*>( ialgoPtr );
    if ( !algoPtr ) {
      fatal() << "Could not convert IAlgorithm into Gaudi::Algorithm: this will result in a crash." << endmsg;
      return StatusCode::FAILURE;
    }
 
    DataObjIDColl algoOutputs;
    for ( auto id : algoPtr->outputDataObjs() ) {
      globalOutp.insert( id );
      algoOutputs.insert( id );
    }
    algosOutputDependenciesMap[algoPtr->name()] = algoOutputs;
 
    hasAsynchronousAlgorithms = hasAsynchronousAlgorithms || algoPtr->isAsynchronous();
  }
 
  if ( m_numOffloadThreads <= 0 && hasAsynchronousAlgorithms ) {
    fatal() << "Found asynchronous algorithms, but NumOffloadThreads is " << m_numOffloadThreads
            << "; no fiber pool threads will be available to execute them." << endmsg;
    return StatusCode::FAILURE;
  }
 
  std::ostringstream ostdd;
  ostdd << "Data Dependencies for Algorithms:";
 
  std::map<std::string, DataObjIDColl> algosInputDependenciesMap;
  for ( IAlgorithm* ialgoPtr : algos ) {
    Gaudi::Algorithm* algoPtr = dynamic_cast<Gaudi::Algorithm*>( ialgoPtr );
    if ( nullptr == algoPtr ) {
      fatal() << "Could not convert IAlgorithm into Gaudi::Algorithm for " << ialgoPtr->name()
              << ": this will result in a crash." << endmsg;
      return StatusCode::FAILURE;
    }
 
    DataObjIDColl i1, i2;
    DHHVisitor    avis( i1, i2 );
    algoPtr->acceptDHVisitor( &avis );
 
    ostdd << "\n  " << algoPtr->name();
 
    auto write_owners = [&avis, &ostdd]( const DataObjID& id ) {
      auto owners = avis.owners_names_of( id );
      if ( !owners.empty() ) { GaudiUtils::operator<<( ostdd << ' ', owners ); }
    };
 
    DataObjIDColl algoDependencies;
    if ( !algoPtr->inputDataObjs().empty() || !algoPtr->outputDataObjs().empty() ) {
      for ( const DataObjID* idp : sortedDataObjIDColl( algoPtr->inputDataObjs() ) ) {
        DataObjID id = *idp;
        ostdd << "\n    o INPUT  " << id;
        write_owners( id );
        algoDependencies.insert( id );
        globalInp.insert( id );
      }
      for ( const DataObjID* id : sortedDataObjIDColl( algoPtr->outputDataObjs() ) ) {
        ostdd << "\n    o OUTPUT " << *id;
        write_owners( *id );
        if ( id->key().find( ":" ) != std::string::npos ) {
          error() << " in Alg " << algoPtr->name() << " alternatives are NOT allowed for outputs! id: " << *id
                  << endmsg;
          m_showDataDeps = true;
        }
      }
    } else {
      ostdd << "\n      none";
    }
    algosInputDependenciesMap[algoPtr->name()] = algoDependencies;
  }
 
  if ( m_showDataDeps ) { info() << ostdd.str() << endmsg; }
 
  // If requested, dump a graph of the data dependencies in a .dot, .md or .graphml file
  if ( not m_dataDepsGraphFile.empty() ) {
    if ( dumpDataDepsGraphFile( algosInputDependenciesMap, algosOutputDependenciesMap ).isFailure() ) {
      return StatusCode::FAILURE;
    }
  }
 
  // Check if we have unmet global input dependencies, and, optionally, heal them
  // WARNING: this step must be done BEFORE the Precedence Service is initialized
  DataObjIDColl unmetDepInp, unusedOutp;
  if ( m_checkDeps || m_checkOutput ) {
    std::set<std::string> requiredInputKeys;
    for ( auto o : globalInp ) {
      // track aliases
      // (assuming there should be no items with different class and same key corresponding to different objects)
      requiredInputKeys.insert( o.key() );
      if ( globalOutp.find( o ) == globalOutp.end() ) unmetDepInp.insert( o );
    }
    if ( m_checkOutput ) {
      for ( auto o : globalOutp ) {
        if ( globalInp.find( o ) == globalInp.end() && requiredInputKeys.find( o.key() ) == requiredInputKeys.end() ) {
          // check ignores
          bool ignored{};
          for ( const std::string& algoName : m_checkOutputIgnoreList ) {
            auto it = algosOutputDependenciesMap.find( algoName );
            if ( it != algosOutputDependenciesMap.end() ) {
              if ( it->second.find( o ) != it->second.end() ) {
                ignored = true;
                break;
              }
            }
          }
          if ( !ignored ) { unusedOutp.insert( o ); }
        }
      }
    }
  }
 
  if ( m_checkDeps ) {
    if ( unmetDepInp.size() > 0 ) {
 
      auto printUnmet = [&]( auto msg ) {
        for ( const DataObjID* o : sortedDataObjIDColl( unmetDepInp ) ) {
          msg << "   o " << *o << "    required by Algorithm: " << endmsg;
 
          for ( const auto& p : algosInputDependenciesMap )
            if ( p.second.find( *o ) != p.second.end() ) msg << "       * " << p.first << endmsg;
        }
      };
 
      if ( !m_useDataLoader.empty() ) {
 
        // Find the DataLoader Alg
        IAlgorithm* dataLoaderAlg( nullptr );
        for ( IAlgorithm* algo : algos )
          if ( m_useDataLoader == algo->name() ) {
            dataLoaderAlg = algo;
            break;
          }
 
        if ( dataLoaderAlg == nullptr ) {
          fatal() << "No DataLoader Algorithm \"" << m_useDataLoader.value()
                  << "\" found, and unmet INPUT dependencies "
                  << "detected:" << endmsg;
          printUnmet( fatal() );
          return StatusCode::FAILURE;
        }
 
        info() << "Will attribute the following unmet INPUT dependencies to \"" << dataLoaderAlg->type() << "/"
               << dataLoaderAlg->name() << "\" Algorithm" << endmsg;
        printUnmet( info() );
 
        // Set the property Load of DataLoader Alg
        Gaudi::Algorithm* dataAlg = dynamic_cast<Gaudi::Algorithm*>( dataLoaderAlg );
        if ( !dataAlg ) {
          fatal() << "Unable to dcast DataLoader \"" << m_useDataLoader.value() << "\" IAlg to Gaudi::Algorithm"
                  << endmsg;
          return StatusCode::FAILURE;
        }
 
        for ( auto& id : unmetDepInp ) {
          ON_DEBUG debug() << "adding OUTPUT dep \"" << id << "\" to " << dataLoaderAlg->type() << "/"
                           << dataLoaderAlg->name() << endmsg;
          dataAlg->addDependency( id, Gaudi::DataHandle::Writer );
        }
 
      } else {
        fatal() << "Auto DataLoading not requested, "
                << "and the following unmet INPUT dependencies were found:" << endmsg;
        printUnmet( fatal() );
        return StatusCode::FAILURE;
      }
 
    } else {
      info() << "No unmet INPUT data dependencies were found" << endmsg;
    }
  }
 
  if ( m_checkOutput ) {
    if ( unusedOutp.size() > 0 ) {
 
      auto printUnusedOutp = [&]( auto msg ) {
        for ( const DataObjID* o : sortedDataObjIDColl( unusedOutp ) ) {
          msg << "   o " << *o << "    produced by Algorithm: " << endmsg;
 
          for ( const auto& p : algosOutputDependenciesMap )
            if ( p.second.find( *o ) != p.second.end() ) msg << "       * " << p.first << endmsg;
        }
      };
 
      fatal() << "The following unused OUTPUT items were found:" << endmsg;
      printUnusedOutp( fatal() );
      return StatusCode::FAILURE;
    } else {
      info() << "No unused OUTPUT items were found" << endmsg;
    }
  }
 
  // Get the precedence service
  m_precSvc = serviceLocator()->service( "PrecedenceSvc" );
  if ( !m_precSvc.isValid() ) {
    fatal() << "Error retrieving PrecedenceSvc" << endmsg;
    return StatusCode::FAILURE;
  }
  const PrecedenceSvc* precSvc = dynamic_cast<const PrecedenceSvc*>( m_precSvc.get() );
  if ( !precSvc ) {
    fatal() << "Unable to dcast PrecedenceSvc" << endmsg;
    return StatusCode::FAILURE;
  }
 
  // Fill the containers to convert algo names to index
  m_algname_vect.resize( algsNumber );
  for ( IAlgorithm* algo : algos ) {
    const std::string& name    = algo->name();
    auto               index   = precSvc->getRules()->getAlgorithmNode( name )->getAlgoIndex();
    m_algname_index_map[name]  = index;
    m_algname_vect.at( index ) = name;
  }
 
  // Shortcut for the message service
  SmartIF<IMessageSvc> messageSvc( serviceLocator() );
  if ( !messageSvc.isValid() ) error() << "Error retrieving MessageSvc interface IMessageSvc." << endmsg;
 
  m_eventSlots.reserve( m_maxEventsInFlight );
  for ( size_t i = 0; i < m_maxEventsInFlight; ++i ) {
    m_eventSlots.emplace_back( algsNumber, precSvc->getRules()->getControlFlowNodeCounter(), messageSvc );
    m_eventSlots.back().complete = true;
  }
 
  // Clearly inform about the level of concurrency
  info() << "Concurrency level information:" << endmsg;
  info() << " o Number of events in flight: " << m_maxEventsInFlight << endmsg;
  info() << " o TBB thread pool size: " << m_threadPoolSize << endmsg;
  info() << " o Fiber thread pool size: " << m_numOffloadThreads << endmsg;
 
  // Inform about task scheduling prescriptions
  info() << "Task scheduling settings:" << endmsg;
  info() << " o Avalanche generation mode: "
         << ( m_optimizationMode.empty() ? "disabled" : m_optimizationMode.toString() ) << endmsg;
  info() << " o Scheduling of condition tasks: " << ( m_enableCondSvc ? "enabled" : "disabled" ) << endmsg;
 
  if ( m_showControlFlow ) m_precSvc->dumpControlFlow();
 
  if ( m_showDataFlow ) m_precSvc->dumpDataFlow();
 
  // Simulate execution flow
  if ( m_simulateExecution ) sc = m_precSvc->simulate( m_eventSlots[0] );
 
  return sc;
}

isStalled()

bool AvalancheSchedulerSvc::isStalled ( const EventSlot & slot ) const

private

Check if scheduling in a particular slot is in a stall.

Check if we are in present of a stall condition for a particular slot.

This is the case when a slot has no actions queued in the actionsQueue, has no scheduled algorithms and has no algorithms with all of its dependencies satisfied.

Definition at line 873 of file AvalancheSchedulerSvc.cpp.

                                                                   {
 
  if ( !slot.algsStates.containsAny( { AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS } ) &&
       !subSlotAlgsInStates( slot, { AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS } ) ) {
 
    error() << "*** Stall detected, event context: " << slot.eventContext.get() << endmsg;
 
    return true;
  }
  return false;
}

iterate()

StatusCode AvalancheSchedulerSvc::iterate ( )

private

Loop on all slots to schedule DATAREADY algorithms and sign off ready events.

Loop on all slots to schedule DATAREADY algorithms, sign off ready ones or detect execution stalls.

To check if an event is finished the method verifies that the root control flow decision of the task precedence graph is resolved and there are no algorithms moving in-between INITIAL and EVTACCEPTED FSM states.

Definition at line 698 of file AvalancheSchedulerSvc.cpp.

                                          {
 
  StatusCode global_sc( StatusCode::SUCCESS );
 
  // Retry algorithms
  const size_t retries = m_retryQueue.size();
  for ( unsigned int retryIndex = 0; retryIndex < retries; ++retryIndex ) {
    TaskSpec retryTS = std::move( m_retryQueue.front() );
    m_retryQueue.pop();
    global_sc = schedule( std::move( retryTS ) );
  }
 
  // Loop over all slots
  OccupancySnapshot nextSnap;
  auto              now = std::chrono::system_clock::now();
  for ( EventSlot& thisSlot : m_eventSlots ) {
 
    // Ignore slots without a valid context (relevant when populating scheduler for first time)
    if ( !thisSlot.eventContext ) continue;
 
    int iSlot = thisSlot.eventContext->slot();
 
    // Cache the states of the algorithms to improve readability and performance
    AlgsExecutionStates& thisAlgsStates = thisSlot.algsStates;
 
    StatusCode partial_sc = StatusCode::FAILURE;
 
    // Make an occupancy snapshot
    if ( m_snapshotInterval != std::chrono::duration<int64_t, std::milli>::min() &&
         now - m_lastSnapshot >= m_snapshotInterval ) {
 
      // Initialise snapshot
      if ( nextSnap.states.empty() ) {
        nextSnap.time = now;
        nextSnap.states.resize( m_eventSlots.size() );
      }
 
      // Store alg states
      std::vector<int>& slotStateTotals = nextSnap.states[iSlot];
      slotStateTotals.resize( AState::MAXVALUE );
      for ( uint8_t state = 0; state < AState::MAXVALUE; ++state ) {
        slotStateTotals[state] = thisSlot.algsStates.sizeOfSubset( AState( state ) );
      }
 
      // Add subslot alg states
      for ( auto& subslot : thisSlot.allSubSlots ) {
        for ( uint8_t state = 0; state < AState::MAXVALUE; ++state ) {
          slotStateTotals[state] += subslot.algsStates.sizeOfSubset( AState( state ) );
        }
      }
    }
 
    // Perform DR->SCHEDULED
    const auto& drAlgs = thisAlgsStates.algsInState( AState::DATAREADY );
    for ( uint algIndex : drAlgs ) {
      const std::string& algName{ index2algname( algIndex ) };
      unsigned int       rank{ m_optimizationMode.empty() ? 0 : m_precSvc->getPriority( algName ) };
      bool               asynchronous{ m_precSvc->isAsynchronous( algName ) };
 
      partial_sc =
          schedule( TaskSpec( nullptr, algIndex, algName, rank, asynchronous, iSlot, thisSlot.eventContext.get() ) );
 
      ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
          << "Could not apply transition from " << AState::DATAREADY << " for algorithm " << algName
          << " on processing slot " << iSlot << endmsg;
    }
 
    // Check for algorithms ready in sub-slots
    for ( auto& subslot : thisSlot.allSubSlots ) {
      const auto& drAlgsSubSlot = subslot.algsStates.algsInState( AState::DATAREADY );
      for ( uint algIndex : drAlgsSubSlot ) {
        const std::string& algName{ index2algname( algIndex ) };
        unsigned int       rank{ m_optimizationMode.empty() ? 0 : m_precSvc->getPriority( algName ) };
        bool               asynchronous{ m_precSvc->isAsynchronous( algName ) };
        partial_sc =
            schedule( TaskSpec( nullptr, algIndex, algName, rank, asynchronous, iSlot, subslot.eventContext.get() ) );
      }
    }
 
    if ( m_dumpIntraEventDynamics ) {
      std::stringstream s;
      s << "START, " << thisAlgsStates.sizeOfSubset( AState::CONTROLREADY ) << ", "
        << thisAlgsStates.sizeOfSubset( AState::DATAREADY ) << ", " << thisAlgsStates.sizeOfSubset( AState::SCHEDULED )
        << ", " << std::chrono::high_resolution_clock::now().time_since_epoch().count() << "\n";
      auto          threads = ( m_threadPoolSize != -1 ) ? std::to_string( m_threadPoolSize )
                                                         : std::to_string( std::thread::hardware_concurrency() );
      std::ofstream myfile;
      myfile.open( "IntraEventFSMOccupancy_" + threads + "T.csv", std::ios::app );
      myfile << s.str();
      myfile.close();
    }
 
    // Not complete because this would mean that the slot is already free!
    if ( m_precSvc->CFRulesResolved( thisSlot ) &&
         !thisSlot.algsStates.containsAny(
             { AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS } ) &&
         !subSlotAlgsInStates( thisSlot,
                               { AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS } ) &&
         !thisSlot.complete ) {
 
      thisSlot.complete = true;
      // if the event did not fail, add it to the finished events
      // otherwise it is taken care of in the error handling
      if ( m_algExecStateSvc->eventStatus( *thisSlot.eventContext ) == EventStatus::Success ) {
        ON_DEBUG debug() << "Event " << thisSlot.eventContext->evt() << " finished (slot "
                         << thisSlot.eventContext->slot() << ")." << endmsg;
        m_finishedEvents.push( thisSlot.eventContext.release() );
      }
 
      // now let's return the fully evaluated result of the control flow
      ON_DEBUG debug() << m_precSvc->printState( thisSlot ) << endmsg;
 
      thisSlot.eventContext.reset( nullptr );
 
    } else if ( isStalled( thisSlot ) ) {
      m_algExecStateSvc->setEventStatus( EventStatus::AlgStall, *thisSlot.eventContext );
      eventFailed( thisSlot.eventContext.get() ); // can't release yet
    }
    partial_sc.ignore();
  } // end loop on slots
 
  // Process snapshot
  if ( !nextSnap.states.empty() ) {
    m_lastSnapshot = nextSnap.time;
    m_snapshotCallback( std::move( nextSnap ) );
  }
 
  ON_VERBOSE verbose() << "Iteration done." << endmsg;
  m_needsUpdate.store( false );
  return global_sc;
}

next()

bool AvalancheSchedulerSvc::next ( TaskSpec & ts, bool asynchronous )

inline

Definition at line 371 of file AvalancheSchedulerSvc.h.

                                               {
    if ( asynchronous ) { return m_scheduledAsynchronousQueue.try_pop( ts ); }
    return m_scheduledQueue.try_pop( ts );
  }

popFinishedEvent()

StatusCode AvalancheSchedulerSvc::popFinishedEvent ( EventContext *& eventContext )

override

Blocks until an event is available.

Get a finished event or block until one becomes available.

Definition at line 657 of file AvalancheSchedulerSvc.cpp.

                                                                                {
 
  // ON_DEBUG debug() << "popFinishedEvent: queue size: " << m_finishedEvents.size() << endmsg;
  if ( m_freeSlots.load() == (int)m_maxEventsInFlight || m_isActive == INACTIVE ) {
    // ON_DEBUG debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
    //      << " active: " << m_isActive << endmsg;
    return StatusCode::FAILURE;
  } else {
    // ON_DEBUG debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
    //      << " active: " << m_isActive << endmsg;
    m_finishedEvents.pop( eventContext );
    ++m_freeSlots;
    ON_DEBUG debug() << "Popped slot " << eventContext->slot() << " (event " << eventContext->evt() << ")" << endmsg;
    return StatusCode::SUCCESS;
  }
}

pushNewEvent()

StatusCode AvalancheSchedulerSvc::pushNewEvent ( EventContext * eventContext )

override

Make an event available to the scheduler.

Add event to the scheduler.

There are two cases possible: 1) No slot is free. A StatusCode::FAILURE is returned. 2) At least one slot is free. An action which resets the slot and kicks off its update is queued.

Definition at line 555 of file AvalancheSchedulerSvc.cpp.

                                                                           {
 
  if ( !eventContext ) {
    fatal() << "Event context is nullptr" << endmsg;
    return StatusCode::FAILURE;
  }
 
  if ( m_freeSlots.load() == 0 ) {
    ON_DEBUG debug() << "A free processing slot could not be found." << endmsg;
    return StatusCode::FAILURE;
  }
 
  // no problem as push new event is only called from one thread (event loop manager)
  --m_freeSlots;
 
  auto action = [this, eventContext]() -> StatusCode {
    // Event processing slot forced to be the same as the wb slot
    const unsigned int thisSlotNum = eventContext->slot();
    EventSlot&         thisSlot    = m_eventSlots[thisSlotNum];
    if ( !thisSlot.complete ) {
      fatal() << "The slot " << thisSlotNum << " is supposed to be a finished event but it's not" << endmsg;
      return StatusCode::FAILURE;
    }
 
    ON_DEBUG debug() << "Executing event " << eventContext->evt() << " on slot " << thisSlotNum << endmsg;
    thisSlot.reset( eventContext );
 
    // Result status code:
    StatusCode result = StatusCode::SUCCESS;
 
    // promote to CR and DR the initial set of algorithms
    Cause cs = { Cause::source::Root, "RootDecisionHub" };
    if ( m_precSvc->iterate( thisSlot, cs ).isFailure() ) {
      error() << "Failed to call IPrecedenceSvc::iterate for slot " << thisSlotNum << endmsg;
      result = StatusCode::FAILURE;
    }
 
    if ( this->iterate().isFailure() ) {
      error() << "Failed to call AvalancheSchedulerSvc::updateStates for slot " << thisSlotNum << endmsg;
      result = StatusCode::FAILURE;
    }
 
    return result;
  }; // end of lambda
 
  // Kick off scheduling
  ON_VERBOSE {
    verbose() << "Pushing the action to update the scheduler for slot " << eventContext->slot() << endmsg;
    verbose() << "Free slots available " << m_freeSlots.load() << endmsg;
  }
 
  m_actionsQueue.push( action );
 
  return StatusCode::SUCCESS;
}

pushNewEvents()

StatusCode AvalancheSchedulerSvc::pushNewEvents ( std::vector< EventContext * > & eventContexts )

override

Definition at line 613 of file AvalancheSchedulerSvc.cpp.

                                                                                         {
  StatusCode sc;
  for ( auto context : eventContexts ) {
    sc = pushNewEvent( context );
    if ( sc != StatusCode::SUCCESS ) return sc;
  }
  return sc;
}

recordOccupancy()

void AvalancheSchedulerSvc::recordOccupancy ( int samplePeriod, std::function< void(OccupancySnapshot)> callback )

overridevirtual

Sample occupancy at fixed interval (ms) Negative value to deactivate, 0 to snapshot every change Each sample, apply the callback function to the result.

Definition at line 1187 of file AvalancheSchedulerSvc.cpp.

                                                                                                               {
 
  auto action = [this, samplePeriod, callback = std::move( callback )]() -> StatusCode {
    if ( samplePeriod < 0 ) {
      this->m_snapshotInterval = std::chrono::duration<int64_t, std::milli>::min();
    } else {
      this->m_snapshotInterval = std::chrono::duration<int64_t, std::milli>( samplePeriod );
      m_snapshotCallback       = std::move( callback );
    }
    return StatusCode::SUCCESS;
  };
 
  m_actionsQueue.push( std::move( action ) );
}

revise()

StatusCode AvalancheSchedulerSvc::revise ( unsigned int iAlgo, EventContext * contextPtr, AState state, bool iterate = false )

private

Definition at line 832 of file AvalancheSchedulerSvc.cpp.

                                                                                                                   {
  StatusCode sc;
  auto       slotIndex = contextPtr->slot();
  EventSlot& slot      = m_eventSlots[slotIndex];
  Cause      cs        = { Cause::source::Task, index2algname( iAlgo ) };
 
  if ( contextPtr->usesSubSlot() ) {
    // Sub-slot
    auto       subSlotIndex = contextPtr->subSlot();
    EventSlot& subSlot      = slot.allSubSlots[subSlotIndex];
 
    sc = subSlot.algsStates.set( iAlgo, state );
 
    if ( sc.isSuccess() ) {
      ON_VERBOSE verbose() << "Promoted " << index2algname( iAlgo ) << " to " << state << " [slot:" << slotIndex
                           << ", subslot:" << subSlotIndex << ", event:" << contextPtr->evt() << "]" << endmsg;
      // Revise states of algorithms downstream the precedence graph
      if ( iterate ) sc = m_precSvc->iterate( subSlot, cs );
    }
  } else {
    // Event level (standard behaviour)
    sc = slot.algsStates.set( iAlgo, state );
 
    if ( sc.isSuccess() ) {
      ON_VERBOSE verbose() << "Promoted " << index2algname( iAlgo ) << " to " << state << " [slot:" << slotIndex
                           << ", event:" << contextPtr->evt() << "]" << endmsg;
      // Revise states of algorithms downstream the precedence graph
      if ( iterate ) sc = m_precSvc->iterate( slot, cs );
    }
  }
  return sc;
}

schedule()

StatusCode AvalancheSchedulerSvc::schedule ( TaskSpec && ts )

private

Definition at line 1054 of file AvalancheSchedulerSvc.cpp.

                                                          {
 
  // Check if a free Algorithm instance is available
  StatusCode getAlgSC( m_algResourcePool->acquireAlgorithm( ts.algName, ts.algPtr ) );
 
  // If an instance is available, proceed to scheduling
  StatusCode sc;
  if ( getAlgSC.isSuccess() ) {
 
    // Decide how to schedule the task and schedule it
    if ( -100 != m_threadPoolSize ) {
 
      // Cache values before moving the TaskSpec further
      unsigned int     algIndex{ ts.algIndex };
      std::string_view algName( ts.algName );
      unsigned int     algRank{ ts.algRank };
      bool             asynchronous{ ts.asynchronous };
      int              slotIndex{ ts.slotIndex };
      EventContext*    contextPtr{ ts.contextPtr };
 
      if ( asynchronous ) {
        // Add to asynchronous scheduled queue
        m_scheduledAsynchronousQueue.push( std::move( ts ) );
 
        // Schedule task
        m_fiberManager->schedule( AlgTask( this, serviceLocator(), m_algExecStateSvc, asynchronous ) );
      }
 
      if ( !asynchronous ) {
        // Add the algorithm to the scheduled queue
        m_scheduledQueue.push( std::move( ts ) );
 
        // Prepare a TBB task that will execute the Algorithm according to the above queued specs
        m_arena->enqueue( AlgTask( this, serviceLocator(), m_algExecStateSvc, asynchronous ) );
        ++m_algosInFlight;
      }
      sc = revise( algIndex, contextPtr, AState::SCHEDULED );
 
      ON_DEBUG debug() << "Scheduled " << algName << " [slot:" << slotIndex << ", event:" << contextPtr->evt()
                       << ", rank:" << algRank << ", asynchronous:" << ( asynchronous ? "yes" : "no" )
                       << "]. Scheduled algorithms: " << m_algosInFlight << endmsg;
 
    } else { // Avoid scheduling via TBB if the pool size is -100. Instead, run here in the scheduler's control thread
      // Beojan: I don't think this bit works. ts hasn't been pushed into any queue so AlgTask won't retrieve it
      ++m_algosInFlight;
      sc = revise( ts.algIndex, ts.contextPtr, AState::SCHEDULED );
      AlgTask( this, serviceLocator(), m_algExecStateSvc, ts.asynchronous )();
      --m_algosInFlight;
    }
  } else { // if no Algorithm instance available, retry later
 
    sc = revise( ts.algIndex, ts.contextPtr, AState::RESOURCELESS );
    // Add the algorithm to the retry queue
    m_retryQueue.push( std::move( ts ) );
  }
 
  ON_VERBOSE dumpSchedulerState( -1 );
 
  return sc;
}

scheduleEventView()

StatusCode AvalancheSchedulerSvc::scheduleEventView ( const EventContext * sourceContext, const std::string & nodeName, std::unique_ptr< EventContext > viewContext )

overridevirtual

Method to inform the scheduler about event views.

Definition at line 1147 of file AvalancheSchedulerSvc.cpp.

                                                                                               {
  //  Prevent view nesting
  if ( sourceContext->usesSubSlot() ) {
    fatal() << "Attempted to nest EventViews at node " << nodeName << ": this is not supported" << endmsg;
    return StatusCode::FAILURE;
  }
 
  ON_VERBOSE verbose() << "Queuing a view for [" << viewContext.get() << "]" << endmsg;
 
  // It's not possible to create an std::functional from a move-capturing lambda
  // So, we have to release the unique pointer
  auto action = [this, slotIndex = sourceContext->slot(), viewContextPtr = viewContext.release(),
                 &nodeName]() -> StatusCode {
    // Attach the sub-slot to the top-level slot
    EventSlot& topSlot = this->m_eventSlots[slotIndex];
 
    if ( viewContextPtr ) {
      // Re-create the unique pointer
      auto viewContext = std::unique_ptr<EventContext>( viewContextPtr );
      topSlot.addSubSlot( std::move( viewContext ), nodeName );
      return StatusCode::SUCCESS;
    } else {
      // Disable the view node if there are no views
      topSlot.disableSubSlots( nodeName );
      return StatusCode::SUCCESS;
    }
  };
 
  m_actionsQueue.push( std::move( action ) );
 
  return StatusCode::SUCCESS;
}

signoff()

StatusCode AvalancheSchedulerSvc::signoff ( const TaskSpec & ts )

private

The call to this method is triggered only from within the AlgTask.

Definition at line 1120 of file AvalancheSchedulerSvc.cpp.

                                                              {
 
  Gaudi::Hive::setCurrentContext( ts.contextPtr );
 
  --m_algosInFlight;
 
  const AlgExecStateRef algstate = m_algExecStateSvc->algExecState( ts.algPtr, *( ts.contextPtr ) );
  AState                state    = algstate.execStatus().isSuccess()
                                       ? ( algstate.filterPassed() ? AState::EVTACCEPTED : AState::EVTREJECTED )
                                       : AState::ERROR;
 
  // Update algorithm state and revise the downstream states
  auto sc = revise( ts.algIndex, ts.contextPtr, state, true );
 
  ON_DEBUG debug() << "Executed " << ts.algName << " [slot:" << ts.slotIndex << ", event:" << ts.contextPtr->evt()
                   << ", rank:" << ts.algRank << ", asynchronous:" << ( ts.asynchronous ? "yes" : "no" )
                   << "]. Scheduled algorithms: " << m_algosInFlight << endmsg;
 
  // Prompt a call to updateStates
  m_needsUpdate.store( true );
  return sc;
}

tryPopFinishedEvent()

StatusCode AvalancheSchedulerSvc::tryPopFinishedEvent ( EventContext *& eventContext )

override

Try to fetch an event from the scheduler.

Try to get a finished event, if not available just return a failure.

Definition at line 678 of file AvalancheSchedulerSvc.cpp.

                                                                                   {
 
  if ( m_finishedEvents.try_pop( eventContext ) ) {
    ON_DEBUG debug() << "Try Pop successful slot " << eventContext->slot() << "(event " << eventContext->evt() << ")"
                     << endmsg;
    ++m_freeSlots;
    return StatusCode::SUCCESS;
  }
  return StatusCode::FAILURE;
}

Friends And Related Symbol Documentation

AlgTask

friend class AlgTask

friend

Definition at line 116 of file AvalancheSchedulerSvc.h.

Member Data Documentation

m_actionsQueue

tbb::concurrent_bounded_queue<action> AvalancheSchedulerSvc::m_actionsQueue

private

Queue where closures are stored and picked for execution.

Definition at line 314 of file AvalancheSchedulerSvc.h.

m_algExecStateSvc

SmartIF<IAlgExecStateSvc> AvalancheSchedulerSvc::m_algExecStateSvc

private

Algorithm execution state manager.

Definition at line 277 of file AvalancheSchedulerSvc.h.

m_algname_index_map

std::unordered_map<std::string, unsigned int> AvalancheSchedulerSvc::m_algname_index_map

private

Map to bookkeep the information necessary to the name2index conversion.

Definition at line 253 of file AvalancheSchedulerSvc.h.

m_algname_vect

std::vector<std::string> AvalancheSchedulerSvc::m_algname_vect

private

Vector to bookkeep the information necessary to the index2name conversion.

Definition at line 259 of file AvalancheSchedulerSvc.h.

m_algosInFlight

unsigned int AvalancheSchedulerSvc::m_algosInFlight = 0

private

Number of algorithms presently in flight.

Definition at line 283 of file AvalancheSchedulerSvc.h.

m_algResourcePool

SmartIF<IAlgResourcePool> AvalancheSchedulerSvc::m_algResourcePool

private

Cache for the algorithm resource pool.

Definition at line 309 of file AvalancheSchedulerSvc.h.

m_arena

tbb::task_arena* AvalancheSchedulerSvc::m_arena { nullptr }

private

Definition at line 364 of file AvalancheSchedulerSvc.h.

{ nullptr };

m_checkDeps

Gaudi::Property<bool> AvalancheSchedulerSvc::m_checkDeps

private

Initial value:

{ this, "CheckDependencies", false,
                                     "Runtime check of Algorithm Input Data Dependencies" }

Definition at line 194 of file AvalancheSchedulerSvc.h.

                                                     { this, "CheckDependencies", false,
                                     "Runtime check of Algorithm Input Data Dependencies" };

m_checkOutput

Gaudi::Property<bool> AvalancheSchedulerSvc::m_checkOutput

private

Initial value:

{ this, "CheckOutputUsage", false,
                                       "Runtime check of Algorithm Output Data usage" }

Definition at line 196 of file AvalancheSchedulerSvc.h.

                                                       { this, "CheckOutputUsage", false,
                                       "Runtime check of Algorithm Output Data usage" };

m_checkOutputIgnoreList

Gaudi::Property<std::vector<std::string> > AvalancheSchedulerSvc::m_checkOutputIgnoreList

private

Initial value:

{
      this,
      "CheckOutputUsageIgnoreList",
      {},
      "Ignore outputs of the Algorithms of this name when doing the check",
      "OrderedSet<std::string>" }

Definition at line 198 of file AvalancheSchedulerSvc.h.

                                                             {
      this,
      "CheckOutputUsageIgnoreList",
      {},
      "Ignore outputs of the Algorithms of this name when doing the check",
      "OrderedSet<std::string>" };

m_condSvc

SmartIF<ICondSvc> AvalancheSchedulerSvc::m_condSvc

private

A shortcut to service for Conditions handling.

Definition at line 280 of file AvalancheSchedulerSvc.h.

m_dataDepsGraphAlgoPattern

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_dataDepsGraphAlgoPattern

private

Initial value:

{
      this, "DataDepsGraphAlgPattern", ".*",
      "Regex pattern for selecting desired Algorithms by name, whose data dependency has to be included in the data "
      "deps graph" }

Definition at line 226 of file AvalancheSchedulerSvc.h.

                                                     {
      this, "DataDepsGraphAlgPattern", ".*",
      "Regex pattern for selecting desired Algorithms by name, whose data dependency has to be included in the data "
      "deps graph" };

m_dataDepsGraphFile

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_dataDepsGraphFile

private

Initial value:

{
      this, "DataDepsGraphFile", "",
      "Name of the output file (.dot, .md or .graphml extensions allowed) containing the data dependency graph "
      "for some selected Algorithms" }

Definition at line 221 of file AvalancheSchedulerSvc.h.

                                              {
      this, "DataDepsGraphFile", "",
      "Name of the output file (.dot, .md or .graphml extensions allowed) containing the data dependency graph "
      "for some selected Algorithms" };

m_dataDepsGraphObjectPattern

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_dataDepsGraphObjectPattern

private

Initial value:

{
      this, "DataDepsGraphObjectPattern", ".*",
      "Regex pattern for selecting desired input or output by their full key" }

Definition at line 231 of file AvalancheSchedulerSvc.h.

                                                       {
      this, "DataDepsGraphObjectPattern", ".*",
      "Regex pattern for selecting desired input or output by their full key" };

m_dumpIntraEventDynamics

Gaudi::Property<bool> AvalancheSchedulerSvc::m_dumpIntraEventDynamics

private

Initial value:

{ this, "DumpIntraEventDynamics", false,
                                                  "Dump intra-event concurrency dynamics to csv file" }

Definition at line 185 of file AvalancheSchedulerSvc.h.

                                                     { this, "DumpIntraEventDynamics", false,
                                                  "Dump intra-event concurrency dynamics to csv file" };

m_enableCondSvc

Gaudi::Property<bool> AvalancheSchedulerSvc::m_enableCondSvc { this, "EnableConditions", false, "Enable ConditionsSvc" }

private

Definition at line 208 of file AvalancheSchedulerSvc.h.

{ this, "EnableConditions", false, "Enable ConditionsSvc" };

m_eventSlots

std::vector<EventSlot> AvalancheSchedulerSvc::m_eventSlots

private

Vector of events slots.

Definition at line 268 of file AvalancheSchedulerSvc.h.

m_fiberManager

std::unique_ptr<FiberManager> AvalancheSchedulerSvc::m_fiberManager { nullptr }

private

Definition at line 365 of file AvalancheSchedulerSvc.h.

{ nullptr };

m_fiberStackSize

Gaudi::Property<std::size_t> AvalancheSchedulerSvc::m_fiberStackSize

private

Initial value:

{ this, "FiberStackSize", 16ULL * 1024 * 1024,
                                                 "Stack size for fibers (asynchronous algorithms)" }

Definition at line 192 of file AvalancheSchedulerSvc.h.

                                                        { this, "FiberStackSize", 16ULL * 1024 * 1024,
                                                 "Stack size for fibers (asynchronous algorithms)" };

m_finishedEvents

tbb::concurrent_bounded_queue<EventContext*> AvalancheSchedulerSvc::m_finishedEvents

private

Queue of finished events.

Definition at line 274 of file AvalancheSchedulerSvc.h.

m_freeSlots

std::atomic_int AvalancheSchedulerSvc::m_freeSlots { 0 }

private

Atomic to account for asyncronous updates by the scheduler wrt the rest.

Definition at line 271 of file AvalancheSchedulerSvc.h.

{ 0 };

m_isActive

std::atomic<ActivationState> AvalancheSchedulerSvc::m_isActive { INACTIVE }

private

Flag to track if the scheduler is active or not.

Definition at line 244 of file AvalancheSchedulerSvc.h.

{ INACTIVE };

m_lastSnapshot

std::chrono::system_clock::time_point AvalancheSchedulerSvc::m_lastSnapshot = std::chrono::system_clock::now()

private

Definition at line 167 of file AvalancheSchedulerSvc.h.

m_maxEventsInFlight

size_t AvalancheSchedulerSvc::m_maxEventsInFlight { 0 }

private

Definition at line 367 of file AvalancheSchedulerSvc.h.

{ 0 };

m_maxParallelismExtra

Gaudi::Property<int> AvalancheSchedulerSvc::m_maxParallelismExtra

private

Initial value:

{
      this, "maxParallelismExtra", 0,
      "Allows to add some extra threads to the maximum parallelism set in TBB"
      "The TBB max parallelism is set as: ThreadPoolSize + maxParallelismExtra + 1" }

Definition at line 175 of file AvalancheSchedulerSvc.h.

                                          {
      this, "maxParallelismExtra", 0,
      "Allows to add some extra threads to the maximum parallelism set in TBB"
      "The TBB max parallelism is set as: ThreadPoolSize + maxParallelismExtra + 1" };

m_needsUpdate

std::atomic<bool> AvalancheSchedulerSvc::m_needsUpdate { true }

private

Definition at line 358 of file AvalancheSchedulerSvc.h.

{ true };

m_numOffloadThreads

Gaudi::Property<int> AvalancheSchedulerSvc::m_numOffloadThreads

private

Initial value:

{
      this, "NumOffloadThreads", 0,
      "Number of threads to use for CPU portion of asynchronous algorithms. Asynchronous algorithms must be flagged "
              "and use Boost Fiber functionality to suspend while waiting for offloaded work." }

Definition at line 187 of file AvalancheSchedulerSvc.h.

                                                {
      this, "NumOffloadThreads", 0,
      "Number of threads to use for CPU portion of asynchronous algorithms. Asynchronous algorithms must be flagged "
              "and use Boost Fiber functionality to suspend while waiting for offloaded work." };

m_optimizationMode

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_optimizationMode

private

Initial value:

{ this, "Optimizer", "",
                                                   "The following modes are currently available: PCE, COD, DRE,  E" }

Definition at line 183 of file AvalancheSchedulerSvc.h.

                                             { this, "Optimizer", "",
                                                   "The following modes are currently available: PCE, COD, DRE,  E" };

m_precSvc

SmartIF<IPrecedenceSvc> AvalancheSchedulerSvc::m_precSvc

private

A shortcut to the Precedence Service.

Definition at line 262 of file AvalancheSchedulerSvc.h.

m_retryQueue

std::queue<TaskSpec> AvalancheSchedulerSvc::m_retryQueue

private

Definition at line 355 of file AvalancheSchedulerSvc.h.

m_scheduledAsynchronousQueue

tbb::concurrent_priority_queue<TaskSpec, AlgQueueSort> AvalancheSchedulerSvc::m_scheduledAsynchronousQueue

private

Definition at line 354 of file AvalancheSchedulerSvc.h.

m_scheduledQueue

tbb::concurrent_priority_queue<TaskSpec, AlgQueueSort> AvalancheSchedulerSvc::m_scheduledQueue

private

Queues for scheduled algorithms.

Definition at line 353 of file AvalancheSchedulerSvc.h.

m_showControlFlow

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showControlFlow

private

Initial value:

{ this, "ShowControlFlow", false,
                                           "Show the configuration of all Algorithms and Sequences" }

Definition at line 216 of file AvalancheSchedulerSvc.h.

                                       { this, "ShowControlFlow", false,
                                           "Show the configuration of all Algorithms and Sequences" };

m_showDataDeps

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showDataDeps

private

Initial value:

{ this, "ShowDataDependencies", true,
                                        "Show the INPUT and OUTPUT data dependencies of Algorithms" }

Definition at line 210 of file AvalancheSchedulerSvc.h.

                                    { this, "ShowDataDependencies", true,
                                        "Show the INPUT and OUTPUT data dependencies of Algorithms" };

m_showDataFlow

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showDataFlow

private

Initial value:

{ this, "ShowDataFlow", false,
                                        "Show the configuration of DataFlow between Algorithms" }

Definition at line 213 of file AvalancheSchedulerSvc.h.

                                    { this, "ShowDataFlow", false,
                                        "Show the configuration of DataFlow between Algorithms" };

m_simulateExecution

Gaudi::Property<bool> AvalancheSchedulerSvc::m_simulateExecution

private

Initial value:

{
      this, "SimulateExecution", false,
      "Flag to perform single-pass simulation of execution flow before the actual execution" }

Definition at line 180 of file AvalancheSchedulerSvc.h.

                                                {
      this, "SimulateExecution", false,
      "Flag to perform single-pass simulation of execution flow before the actual execution" };

m_snapshotCallback

std::function<void( OccupancySnapshot )> AvalancheSchedulerSvc::m_snapshotCallback

private

Definition at line 168 of file AvalancheSchedulerSvc.h.

m_snapshotInterval

std::chrono::duration<int64_t, std::milli> AvalancheSchedulerSvc::m_snapshotInterval = std::chrono::duration<int64_t, std::milli>::min()

private

Definition at line 166 of file AvalancheSchedulerSvc.h.

m_thread

std::thread AvalancheSchedulerSvc::m_thread

private

The thread in which the activate function runs.

Definition at line 247 of file AvalancheSchedulerSvc.h.

m_threadPoolSize

Gaudi::Property<int> AvalancheSchedulerSvc::m_threadPoolSize

private

Initial value:

{
      this, "ThreadPoolSize", -1,
      "Size of the global thread pool initialised by TBB; a value of -1 requests to use"
      "all available hardware threads; -100 requests to bypass TBB executing "
      "all algorithms in the scheduler's thread." }

Definition at line 170 of file AvalancheSchedulerSvc.h.

                                     {
      this, "ThreadPoolSize", -1,
      "Size of the global thread pool initialised by TBB; a value of -1 requests to use"
      "all available hardware threads; -100 requests to bypass TBB executing "
      "all algorithms in the scheduler's thread." };

m_threadPoolSvc

SmartIF<IThreadPoolSvc> AvalancheSchedulerSvc::m_threadPoolSvc

private

Definition at line 363 of file AvalancheSchedulerSvc.h.

m_useDataLoader

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_useDataLoader

private

Initial value:

{ this, "DataLoaderAlg", "",
                                                "Attribute unmet input dependencies to this DataLoader Algorithm" }

Definition at line 205 of file AvalancheSchedulerSvc.h.

                                          { this, "DataLoaderAlg", "",
                                                "Attribute unmet input dependencies to this DataLoader Algorithm" };

m_verboseSubSlots

Gaudi::Property<bool> AvalancheSchedulerSvc::m_verboseSubSlots { this, "VerboseSubSlots", false, "Dump algorithm states for all sub-slots" }

private

Definition at line 219 of file AvalancheSchedulerSvc.h.

{ this, "VerboseSubSlots", false, "Dump algorithm states for all sub-slots" };

m_whiteboard

SmartIF<IHiveWhiteBoard> AvalancheSchedulerSvc::m_whiteboard

private

A shortcut to the whiteboard.

Definition at line 265 of file AvalancheSchedulerSvc.h.

m_whiteboardSvcName

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_whiteboardSvcName { this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name" }

private

Definition at line 179 of file AvalancheSchedulerSvc.h.

{ this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name" };

---

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

• GaudiHive/src/AvalancheSchedulerSvc.h
• GaudiHive/src/AvalancheSchedulerSvc.cpp