ForwardSchedulerSvc Class Reference

The SchedulerSvc implements the IScheduler interface. More...

#include <GaudiKernel/ForwardSchedulerSvc.h>

Inheritance diagram for ForwardSchedulerSvc:

Collaboration diagram for ForwardSchedulerSvc:

Classes
class	SchedulerState

Public Member Functions
	~ForwardSchedulerSvc () override=default
	Destructor. More...
StatusCode	initialize () override
	Initialise. More...
StatusCode	finalize () override
	Finalise. More...
StatusCode	pushNewEvent (EventContext *eventContext) override
	Make an event available to the scheduler. More...
StatusCode	pushNewEvents (std::vector< EventContext * > &eventContexts) override
StatusCode	popFinishedEvent (EventContext *&eventContext) override
	Blocks until an event is availble. More...
StatusCode	tryPopFinishedEvent (EventContext *&eventContext) override
	Try to fetch an event from the scheduler. More...
unsigned int	freeSlots () override
	Get free slots number. More...
void	addAlg (Algorithm *, EventContext *, pthread_t)
bool	delAlg (Algorithm *)
void	dumpState () override
Public Member Functions inherited from extends< Service, IScheduler >
void *	i_cast (const InterfaceID &tid) const override
	Implementation of IInterface::i_cast. More...
StatusCode	queryInterface (const InterfaceID &ti, void **pp) override
	Implementation of IInterface::queryInterface. More...
std::vector< std::string >	getInterfaceNames () const override
	Implementation of IInterface::getInterfaceNames. More...
	~extends () override=default
	Virtual destructor. More...
Public Member Functions inherited from Service
const std::string &	name () const override
	Retrieve name of the service. More...
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. More...
StatusCode	sysStart () override
	Initialize Service. More...
StatusCode	sysStop () override
	Initialize Service. More...
StatusCode	sysFinalize () override
	Finalize Service. More...
StatusCode	sysReinitialize () override
	Re-initialize the Service. More...
StatusCode	sysRestart () override
	Re-initialize the Service. More...
	Service (std::string name, ISvcLocator *svcloc)
	Standard Constructor. More...
SmartIF< ISvcLocator > &	serviceLocator () const override
	Retrieve pointer to service locator. More...
StatusCode	setProperties ()
	Method for setting declared properties to the values specified for the job. More...
template<class T >
StatusCode	service (const std::string &name, const T *&psvc, bool createIf=true) const
	Access a service by name, creating it if it doesn't already exist. More...
template<class T >
StatusCode	service (const std::string &name, T *&psvc, bool createIf=true) const
template<typename IFace = IService>
SmartIF< IFace >	service (const std::string &name, bool createIf=true) const
template<class T >
StatusCode	service (const std::string &svcType, const std::string &svcName, T *&psvc) const
	Access a service by name and type, creating it if it doesn't already exist. More...
template<class T >
StatusCode	declarePrivateTool (ToolHandle< T > &handle, std::string toolTypeAndName="", bool createIf=true)
	Declare used Private tool. More...
template<class T >
StatusCode	declarePublicTool (ToolHandle< T > &handle, std::string toolTypeAndName="", bool createIf=true)
	Declare used Public tool. More...
SmartIF< IAuditorSvc > &	auditorSvc () const
	The standard auditor service.May not be invoked before sysInitialize() has been invoked. More...
Public Member Functions inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
	PropertyHolder ()=default
	~PropertyHolder () override=default
Gaudi::Details::PropertyBase &	declareProperty (Gaudi::Details::PropertyBase &prop)
	Declare a property. More...
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, TYPE &value, const std::string &doc="none")
	Helper to wrap a regular data member and use it as a regular property. More...
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, TYPE &value, const std::string &doc="none") const
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, Gaudi::Property< TYPE, VERIFIER, HANDLERS > &prop, const std::string &doc="none")
	Declare a PropertyBase instance setting name and documentation. More...
Gaudi::Details::PropertyBase *	declareRemoteProperty (const std::string &name, IProperty *rsvc, const std::string &rname="")
	Declare a remote property. More...
StatusCode	setProperty (const Gaudi::Details::PropertyBase &p) override
	set the property form another property More...
StatusCode	setProperty (const std::string &s) override
	set the property from the formatted string More...
StatusCode	setProperty (const std::string &n, const std::string &v) override
	set the property from name and the value More...
StatusCode	setProperty (const std::string &name, const TYPE &value)
	set the property form the value More...
StatusCode	getProperty (Gaudi::Details::PropertyBase *p) const override
	get the property More...
const Gaudi::Details::PropertyBase &	getProperty (const std::string &name) const override
	get the property by name More...
StatusCode	getProperty (const std::string &n, std::string &v) const override
	convert the property to the string More...
const std::vector< Gaudi::Details::PropertyBase * > &	getProperties () const override
	get all properties More...
bool	hasProperty (const std::string &name) const override
	Return true if we have a property with the given name. More...
	PropertyHolder (const PropertyHolder &)=delete
PropertyHolder &	operator= (const PropertyHolder &)=delete
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, GaudiHandleBase &ref, const std::string &doc="none")
	Specializations for various GaudiHandles. More...
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, GaudiHandleArrayBase &ref, const std::string &doc="none")
Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, DataObjectHandleBase &ref, const std::string &doc="none")
Public Member Functions inherited from CommonMessagingBase
virtual	~CommonMessagingBase ()=default
	Virtual destructor. More...
SmartIF< IMessageSvc > &	msgSvc () const
	The standard message service. More...
MsgStream &	msgStream () const
	Return an uninitialized MsgStream. More...
MsgStream &	msgStream (const MSG::Level level) const
	Predefined configurable message stream for the efficient printouts. More...
MsgStream &	always () const
	shortcut for the method msgStream(MSG::ALWAYS) More...
MsgStream &	fatal () const
	shortcut for the method msgStream(MSG::FATAL) More...
MsgStream &	err () const
	shortcut for the method msgStream(MSG::ERROR) More...
MsgStream &	error () const
	shortcut for the method msgStream(MSG::ERROR) More...
MsgStream &	warning () const
	shortcut for the method msgStream(MSG::WARNING) More...
MsgStream &	info () const
	shortcut for the method msgStream(MSG::INFO) More...
MsgStream &	debug () const
	shortcut for the method msgStream(MSG::DEBUG) More...
MsgStream &	verbose () const
	shortcut for the method msgStream(MSG::VERBOSE) More...
MsgStream &	msg () const
	shortcut for the method msgStream(MSG::INFO) More...
MSG::Level	msgLevel () const
	get the output level from the embedded MsgStream More...
MSG::Level	outputLevel () const __attribute__((deprecated))
	Backward compatibility function for getting the output level. More...
bool	msgLevel (MSG::Level lvl) const
	get the output level from the embedded MsgStream More...
Public Member Functions inherited from extend_interfaces< Interfaces... >
	~extend_interfaces () override=default
	Virtual destructor. More...

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

Private Member Functions
void	activate ()
	Activate scheduler. More...
StatusCode	deactivate ()
	Deactivate scheduler. More...
unsigned int	algname2index (const std::string &algoname)
	Convert a name to an integer. More...
const std::string &	index2algname (unsigned int index)
	Convert an integer to a name. More...
StatusCode	eventFailed (EventContext *eventContext)
	Method to check if an event failed and take appropriate actions. More...
StatusCode	updateStates (int si=-1, const std::string &algo_name=std::string())
	Loop on algorithm in the slots and promote them to successive states (-1 means all slots, while empty string means skipping an update of the Control Flow state) More...
StatusCode	promoteToControlReady (unsigned int iAlgo, int si)
	Algorithm promotion: Accepted by the control flow. More...
StatusCode	promoteToDataReady (unsigned int iAlgo, int si)
StatusCode	promoteToScheduled (unsigned int iAlgo, int si)
StatusCode	promoteToAsyncScheduled (unsigned int iAlgo, int si)
StatusCode	promoteToExecuted (unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
	The call to this method is triggered only from within the AlgoExecutionTask. More...
StatusCode	promoteToAsyncExecuted (unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
	The call to this method is triggered only from within the IOBoundAlgTask. More...
StatusCode	promoteToFinished (unsigned int iAlgo, int si)
StatusCode	isStalled (int si)
	Check if the scheduling is in a stall. More...
void	dumpSchedulerState (int iSlot)
	Dump the state of the scheduler. More...
StatusCode	m_drain ()
	Drain the actions present in the queue. More...
void	dumpState (std::ostringstream &)

Private Attributes
Gaudi::Property< int >	m_maxEventsInFlight
Gaudi::Property< int >	m_threadPoolSize
Gaudi::Property< std::string >	m_whiteboardSvcName {this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name"}
Gaudi::Property< std::string >	m_IOBoundAlgSchedulerSvcName {this, "IOBoundAlgSchedulerSvc", "IOBoundAlgSchedulerSvc"}
Gaudi::Property< unsigned int >	m_maxAlgosInFlight
Gaudi::Property< unsigned int >	m_maxIOBoundAlgosInFlight
Gaudi::Property< bool >	m_CFNext
Gaudi::Property< bool >	m_DFNext
Gaudi::Property< bool >	m_simulateExecution
Gaudi::Property< std::string >	m_optimizationMode
Gaudi::Property< bool >	m_dumpIntraEventDynamics
Gaudi::Property< bool >	m_useIOBoundAlgScheduler
Gaudi::Property< std::vector< std::vector< std::string > > >	m_algosDependencies
Gaudi::Property< bool >	m_checkDeps {this, "CheckDependencies", false, "[[deprecated]]"}
std::atomic< ActivationState >	m_isActive {INACTIVE}
	Flag to track if the scheduler is active or not. More...
std::thread	m_thread
	The thread in which the activate function runs. More...
std::unordered_map< std::string, unsigned int >	m_algname_index_map
	Map to bookkeep the information necessary to the name2index conversion. More...
std::vector< std::string >	m_algname_vect
	Vector to bookkeep the information necessary to the index2name conversion. More...
SmartIF< IHiveWhiteBoard >	m_whiteboard
	A shortcut to the whiteboard. More...
SmartIF< IAccelerator >	m_IOBoundAlgScheduler
	A shortcut to IO-bound algorithm scheduler. More...
std::vector< EventSlot >	m_eventSlots
	Vector of events slots. More...
std::atomic_int	m_freeSlots
	Atomic to account for asyncronous updates by the scheduler wrt the rest. More...
tbb::concurrent_bounded_queue< EventContext * >	m_finishedEvents
	Queue of finished events. More...
SmartIF< IAlgExecStateSvc >	m_algExecStateSvc
	Algorithm execution state manager. More...
unsigned int	m_algosInFlight = 0
	Number of algoritms presently in flight. More...
unsigned int	m_IOBoundAlgosInFlight = 0
	Number of algoritms presently in flight. More...
bool	m_updateNeeded = true
	Keep track of update actions scheduled. More...
SmartIF< IAlgResourcePool >	m_algResourcePool
	Cache for the algorithm resource pool. More...
tbb::concurrent_bounded_queue< action >	m_actionsQueue
	Queue where closures are stored and picked for execution. More...
concurrency::ExecutionFlowManager	m_efManager
	Member to take care of the control flow. More...
SmartIF< IThreadPoolSvc >	m_threadPoolSvc
bool	m_first = true

Static Private Attributes
static std::list< SchedulerState >	m_sState
static std::mutex	m_ssMut

Friends
class	AlgoExecutionTask
class	IOBoundAlgTask

Additional Inherited Members
Public Types inherited from extends< Service, IScheduler >
using	base_class = extends
	Typedef to this class. More...
using	extend_interfaces_base = extend_interfaces< Interfaces... >
	Typedef to the base of this class. More...
Public Types inherited from Service
typedef Gaudi::PluginService::Factory< IService *, const std::string &, ISvcLocator * >	Factory
Public Types inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
using	PropertyHolderImpl = PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
	Typedef used to refer to this class from derived classes, as in. More...
Public Types inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
using	base_class = CommonMessaging
Public Types inherited from extend_interfaces< Interfaces... >
using	ext_iids = typename Gaudi::interface_list_cat< typename Interfaces::ext_iids... >::type
	take union of the ext_iids of all Interfaces... More...
Protected Member Functions inherited from Service
	~Service () override
	Standard Destructor. More...
int	outputLevel () const
	get the Service's output level More...
Protected Member Functions inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
Gaudi::Details::PropertyBase *	property (const std::string &name) const
Protected Member Functions inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
void	updateMsgStreamOutputLevel (int level)
	Update the output level of the cached MsgStream. More...
Protected Attributes inherited from Service
Gaudi::StateMachine::State	m_state = Gaudi::StateMachine::OFFLINE
	Service state. More...
Gaudi::StateMachine::State	m_targetState = Gaudi::StateMachine::OFFLINE
	Service state. More...
Gaudi::Property< int >	m_outputLevel {this, "OutputLevel", MSG::NIL, "output level"}
Gaudi::Property< bool >	m_auditInit {this, "AuditServices", false, "[[deprecated]] unused"}
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()"}
SmartIF< IAuditorSvc >	m_pAuditorSvc
	Auditor Service. More...

Detailed Description

The SchedulerSvc implements the IScheduler interface.

It manages all the execution states of the algorithms and interacts with the TBB runtime for the algorithm tasks submission. A state machine takes care of the tracking of the execution state of the algorithms. This is a forward scheduler: algorithms are scheduled for execution as soon as their data dependencies are available in the whiteboard.

Algorithms management

The activate() method runs in a separate thread. It checks a TBB concurrent bounded queue of closures in a loop via the Pop method. This allows not to use a cpu entirely to check the presence of new actions to be taken. In other words, the asynchronous actions are serialised via the actions queue. Once a task terminates, a call to the promoteToExecuted method will be pushed into the actions queue. The promoteToExecuted method also triggers a call to the updateStates method, which brushes all algorithms, checking if their state can be changed. It's indeed possible that upon termination of an algorithm, the control flow and/or the data flow allow the submission of more algorithms.

Algorithms dependencies

There are two ways of declaring algorithms dependencies. One which is only temporarly available to ease developments consists in declaring them through AlgosDependencies property as a list of list. The order of these sublist must be the same one of the algorithms in the TopAlg list. The second one consists in declaring the data dependencies directly within the algorithms via data object handles.

Events management

The scheduler accepts events to be processed (in the form of eventContexts) and releases processed events. This flow is implemented through three methods:

• pushNewEvent: to make an event available to the scheduler.
• tryPopFinishedEvent: to retrieve an event from the scheduler
• popFinishedEvent: to retrieve an event from the scheduler (blocking)

Please refer to the full documentation of the methods for more details.

Author

Danilo Piparo

Benedikt Hegner

Version

1.1

Definition at line 84 of file ForwardSchedulerSvc.h.

Member Typedef Documentation

typedef std::function<StatusCode()> ForwardSchedulerSvc::action

private

Definition at line 235 of file ForwardSchedulerSvc.h.

Member Enumeration Documentation

enum ForwardSchedulerSvc::ActivationState

private

Enumerator
INACTIVE
ACTIVE
FAILURE

Definition at line 115 of file ForwardSchedulerSvc.h.

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

Constructor & Destructor Documentation

ForwardSchedulerSvc::~ForwardSchedulerSvc ( )

overridedefault

Destructor.

Member Function Documentation

void ForwardSchedulerSvc::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 317 of file ForwardSchedulerSvc.cpp.

{

  if (msgLevel(MSG::DEBUG))
    debug() << "ForwardSchedulerSvc::activate()" << endmsg;

  if ( m_threadPoolSvc->initPool( m_threadPoolSize ).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
  info() << "Start checking the actionsQueue" << endmsg;
  while ( m_isActive == ACTIVE or m_actionsQueue.size() != 0 ) {
    m_actionsQueue.pop( thisAction );
    sc = thisAction();
    if ( sc != StatusCode::SUCCESS )
      verbose() << "Action did not succeed (which is not bad per se)." << endmsg;
    else
      verbose() << "Action succeeded." << endmsg;
  }

  info() << "Terminating thread-pool resources" << endmsg;
  if ( m_threadPoolSvc->terminatePool().isFailure() ) {
    error() << "Problems terminating thread pool" << endmsg;
    m_isActive = FAILURE;
  }
}

void ForwardSchedulerSvc::addAlg	(	Algorithm *	a,
		EventContext *	e,
		pthread_t	t
	)

Definition at line 1187 of file ForwardSchedulerSvc.cpp.

{

  std::lock_guard<std::mutex> lock( m_ssMut );
  m_sState.push_back( SchedulerState( a, e, t ) );
}

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

inlineprivate

Convert a name to an integer.

Definition at line 386 of file ForwardSchedulerSvc.cpp.

{
  unsigned int index = m_algname_index_map[algoname];
  return index;
}

StatusCode ForwardSchedulerSvc::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 361 of file ForwardSchedulerSvc.cpp.

{

  if ( m_isActive == ACTIVE ) {
    // Drain the scheduler
    m_actionsQueue.push( std::bind( &ForwardSchedulerSvc::m_drain, this ) );
    // This would be the last action
    m_actionsQueue.push( [this]() -> StatusCode {
      m_isActive = INACTIVE;
      return StatusCode::SUCCESS;
    } );
  }

  return StatusCode::SUCCESS;
}

bool ForwardSchedulerSvc::delAlg

(

Algorithm *

a

)

Definition at line 1195 of file ForwardSchedulerSvc.cpp.

{

  std::lock_guard<std::mutex> lock( m_ssMut );

  for ( std::list<SchedulerState>::iterator itr = m_sState.begin(); itr != m_sState.end(); ++itr ) {
    if ( *itr == a ) {
      m_sState.erase( itr );
      return true;
    }
  }

  error() << "could not find Alg " << a->name() << " in Scheduler!" << endmsg;
  return false;
}

void ForwardSchedulerSvc::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.

The dependencies of each algo are printed and the missing ones specified.

Definition at line 829 of file ForwardSchedulerSvc.cpp.

{

  // To have just one big message
  std::ostringstream outputMessageStream;

  outputMessageStream << "============================== Execution Task State ============================="
                      << std::endl;
  dumpState( outputMessageStream );

  outputMessageStream << std::endl
                      << "============================== Scheduler State  ================================="
                      << std::endl;

  int slotCount = -1;
  for ( auto thisSlot : m_eventSlots ) {
    slotCount++;
    if ( thisSlot.complete ) continue;

    outputMessageStream << "-----------  slot: " << thisSlot.eventContext->slot()
                        << "  event: " << thisSlot.eventContext->evt() << " -----------" << std::endl;

    if ( 0 > iSlot or iSlot == slotCount ) {
      outputMessageStream << "Algorithms states:" << std::endl;

      const DataObjIDColl& wbSlotContent( thisSlot.dataFlowMgr.content() );
      for ( unsigned int algoIdx = 0; algoIdx < thisSlot.algsStates.size(); ++algoIdx ) {
        outputMessageStream << " o " << index2algname( algoIdx ) << " ["
                            << AlgsExecutionStates::stateNames[thisSlot.algsStates[algoIdx]] << "]  Data deps: ";
        DataObjIDColl deps( thisSlot.dataFlowMgr.dataDependencies( algoIdx ) );
        const int depsSize = deps.size();
        if ( depsSize == 0 ) outputMessageStream << " none";

        DataObjIDColl missing;
        for ( auto d : deps ) {
          outputMessageStream << d << " ";
          if ( wbSlotContent.find( d ) == wbSlotContent.end() ) {
            //      outputMessageStream << "[missing] ";
            missing.insert( d );
          }
        }

        if ( !missing.empty() ) {
          outputMessageStream << ". The following are missing: ";
          for ( auto d : missing ) {
            outputMessageStream << d << " ";
          }
        }

        outputMessageStream << std::endl;
      }

      // Snapshot of the WhiteBoard
      outputMessageStream << "\nWhiteboard contents: " << std::endl;
      for ( auto& product : wbSlotContent ) outputMessageStream << " o " << product << std::endl;

      // Snapshot of the ControlFlow
      outputMessageStream << "\nControl Flow:" << std::endl;
      std::stringstream cFlowStateStringStream;
      m_efManager.printEventState( cFlowStateStringStream, thisSlot.algsStates, thisSlot.controlFlowState, 0 );

      outputMessageStream << cFlowStateStringStream.str() << std::endl;
    }
  }

  outputMessageStream << "=================================== END ======================================" << std::endl;

  info() << "Dumping Scheduler State " << std::endl << outputMessageStream.str() << endmsg;
}

void ForwardSchedulerSvc::dumpState ( )

override

Definition at line 1223 of file ForwardSchedulerSvc.cpp.

{

  std::lock_guard<std::mutex> lock( m_ssMut );

  std::ostringstream ost;
  ost << "dumping Executing Threads: [" << m_sState.size() << "]" << std::endl;
  dumpState( ost );

  info() << ost.str() << endmsg;
}

void ForwardSchedulerSvc::dumpState ( std::ostringstream &  ost )

private

Definition at line 1212 of file ForwardSchedulerSvc.cpp.

{

  std::lock_guard<std::mutex> lock( m_ssMut );

  for ( auto it : m_sState ) {
    ost << "  " << it << std::endl;
  }
}

StatusCode ForwardSchedulerSvc::eventFailed ( EventContext *  eventContext )

private

Method to check if an event failed and take appropriate actions.

It can be possible that an event fails.

In this case this method is called. It dumps the state of the scheduler, drains the actions (without executing them) and events in the queues and returns a failure.

Definition at line 526 of file ForwardSchedulerSvc.cpp.

{

  // Set the number of slots available to an error code
  m_freeSlots.store( 0 );

  fatal() << "*** Event " << eventContext->evt() << " on slot "
          << eventContext->slot() << " failed! ***" << endmsg;

  std::ostringstream ost;
  m_algExecStateSvc->dump(ost, *eventContext);

  info() << "Dumping Alg Exec State for slot " << eventContext->slot()
         << ":\n" << ost.str() << endmsg;

  dumpSchedulerState(-1);

  // Empty queue and deactivate the service
  action thisAction;
  while ( m_actionsQueue.try_pop( thisAction ) ) {
  };
  deactivate();

  // Push into the finished events queue the failed context
  EventContext* thisEvtContext;
  while ( m_finishedEvents.try_pop( thisEvtContext ) ) {
    m_finishedEvents.push( thisEvtContext );
  };
  m_finishedEvents.push( eventContext );

  return StatusCode::FAILURE;
}

StatusCode ForwardSchedulerSvc::finalize ( )

override

Finalise.

Here the scheduler is deactivated and the thread joined.

Definition at line 284 of file ForwardSchedulerSvc.cpp.

{

  StatusCode sc( Service::finalize() );
  if ( !sc.isSuccess() ) warning() << "Base class could not be finalized" << endmsg;

  sc = deactivate();
  if ( !sc.isSuccess() ) warning() << "Scheduler could not be deactivated" << endmsg;

  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;
  }

  // m_efManager.getExecutionFlowGraph()->dumpExecutionPlan();

  return sc;
}

unsigned int ForwardSchedulerSvc::freeSlots ( )

override

Get free slots number.

Definition at line 462 of file ForwardSchedulerSvc.cpp.

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

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

inlineprivate

Convert an integer to a name.

Definition at line 382 of file ForwardSchedulerSvc.cpp.

{ return m_algname_vect[index]; }

StatusCode ForwardSchedulerSvc::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 53 of file ForwardSchedulerSvc.cpp.

{

  // Initialise mother class (read properties, ...)
  StatusCode sc( Service::initialize() );
  if ( !sc.isSuccess() ) 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;
  }

  // Activate the scheduler in another thread.
  info() << "Activating scheduler in a separate thread" << endmsg;
  m_thread = std::thread( std::bind( &ForwardSchedulerSvc::activate, this ) );

  while ( m_isActive != ACTIVE ) {
    if ( m_isActive == FAILURE ) {
      fatal() << "Terminating initialization" << endmsg;
      return StatusCode::FAILURE;
    } else {
      info() << "Waiting for ForwardSchedulerSvc to activate" << endmsg;
      sleep( 1 );
    }
  }

  // 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;
  }

  // Check the MaxEventsInFlight parameters and react
  // Deprecated for the moment
  size_t numberOfWBSlots = m_whiteboard->getNumberOfStores();
  if ( m_maxEventsInFlight != 0 ) {
    warning() << "Property MaxEventsInFlight was set. This works but it's deprecated. "
              << "Please migrate your code options files." << endmsg;

    if ( m_maxEventsInFlight != (int)numberOfWBSlots ) {
      warning() << "In addition, the number of events in flight (" << m_maxEventsInFlight
                << ") differs from the slots in the whiteboard (" << numberOfWBSlots
                << "). Setting the number of events in flight to " << numberOfWBSlots << endmsg;
    }
  }

  // Get dedicated scheduler for I/O-bound algorithms
  if ( m_useIOBoundAlgScheduler ) {
    m_IOBoundAlgScheduler = serviceLocator()->service( m_IOBoundAlgSchedulerSvcName );
    if ( !m_IOBoundAlgScheduler.isValid() )
      fatal() << "Error retrieving IOBoundSchedulerAlgSvc interface IAccelerator." << endmsg;
  }
  // Align the two quantities
  m_maxEventsInFlight = numberOfWBSlots;

  // Set the number of free slots
  m_freeSlots = m_maxEventsInFlight;

  if ( m_algosDependencies.size() != 0 ) {
    warning() << " ##### Property AlgosDependencies is deprecated and ignored."
              << " FIX your job options #####" << endmsg;
  }

  // Get the list of algorithms
  const std::list<IAlgorithm*>& algos = m_algResourcePool->getFlatAlgList();
  const unsigned int algsNumber       = algos.size();
  info() << "Found " << algsNumber << " algorithms" << endmsg;

  /* Dependencies
   1) Look for handles in algo, if none
   2) Assume none are required
  */

  DataObjIDColl globalInp, globalOutp;

  // figure out all outputs
  for (IAlgorithm* ialgoPtr : algos) {
    Algorithm* algoPtr = dynamic_cast<Algorithm*>(ialgoPtr);
    if (!algoPtr) {
      fatal() << "Could not convert IAlgorithm into Algorithm: this will result in a crash." << endmsg;
    }
    for (auto id : algoPtr->outputDataObjs()) {
        auto r = globalOutp.insert(id);
        if (!r.second) {
           warning() << "multiple algorithms declare " << id << " as output! could be a single instance in multiple paths though, or control flow may guarantee only one runs...!" << endmsg;
        }
    }
  }
  info() << "outputs:\n" ;
  for (const auto& i : globalOutp ) {
      info() << i << '\n' ;
  }
  info() << endmsg;



  info() << "Data Dependencies for Algorithms:";

  std::vector<DataObjIDColl> m_algosDependencies;
  for ( IAlgorithm* ialgoPtr : algos ) {
    Algorithm* algoPtr = dynamic_cast<Algorithm*>( ialgoPtr );
    if ( nullptr == algoPtr )
      fatal() << "Could not convert IAlgorithm into Algorithm: this will result in a crash." << endmsg;

    info() << "\n  " << algoPtr->name();

    // FIXME
    DataObjIDColl algoDependencies;
    if ( !algoPtr->inputDataObjs().empty() || !algoPtr->outputDataObjs().empty() ) {
      for ( auto id : algoPtr->inputDataObjs() ) {
        info() << "\n    o INPUT  " << id;
        if (id.key().find(":")!=std::string::npos) {
            info() << " contains alternatives which require resolution... " << endmsg;
            auto tokens = boost::tokenizer<boost::char_separator<char>>{id.key(),boost::char_separator<char>{":"}};
            auto itok = std::find_if( tokens.begin(), tokens.end(),
                                     [&](const std::string& t) {
                return globalOutp.find( DataObjID{t} ) != globalOutp.end();
            } );
            if (itok!=tokens.end()) {
                info() << "found matching output for " << *itok << " -- updating scheduler info" << endmsg;
                id.updateKey(*itok);
            } else {
                error() << "failed to find alternate in global output list" << endmsg;
            }
        }
        algoDependencies.insert( id );
        globalInp.insert( id );
      }
      for ( auto id : algoPtr->outputDataObjs() ) {
        info() << "\n    o OUTPUT " << id;
        if (id.key().find(":")!=std::string::npos) {
            info() << " alternatives are NOT allowed for outputs..." << endmsg;
        }
      }
    } else {
      info() << "\n      none";
    }
    m_algosDependencies.emplace_back( algoDependencies );
  }
  info() << endmsg;

  // Fill the containers to convert algo names to index
  m_algname_vect.reserve( algsNumber );
  unsigned int index = 0;
  for ( IAlgorithm* algo : algos ) {
    const std::string& name   = algo->name();
    m_algname_index_map[name] = index;
    m_algname_vect.emplace_back( name );
    index++;
  }

  // Check if we have unmet global input dependencies
  if ( m_checkDeps ) {
    DataObjIDColl unmetDep;
    for ( auto o : globalInp ) {
      if ( globalOutp.find( o ) == globalOutp.end() ) {
        unmetDep.insert( o );
      }
    }

    if ( unmetDep.size() > 0 ) {
      fatal() << "The following unmet INPUT data dependencies were found: ";
      for ( auto& o : unmetDep ) {
        fatal() << "\n   o " << o << "    required by Algorithm: ";
        for ( size_t i = 0; i < m_algosDependencies.size(); ++i ) {
          if ( m_algosDependencies[i].find( o ) != m_algosDependencies[i].end() ) {
            fatal() << "\n       * " << m_algname_vect[i];
          }
        }
      }
      fatal() << endmsg;
      return StatusCode::FAILURE;
    } else {
      info() << "No unmet INPUT data dependencies were found" << endmsg;
    }
  }

  // prepare the control flow part
  if ( m_CFNext ) m_DFNext = true; // force usage of new data flow machinery when new control flow is used
  if ( !m_CFNext && !m_optimizationMode.empty() ) {
    fatal() << "Execution optimization is only available with the graph-based execution flow management" << endmsg;
    return StatusCode::FAILURE;
  }
  const AlgResourcePool* algPool = dynamic_cast<const AlgResourcePool*>( m_algResourcePool.get() );
  sc =
      m_efManager.initialize( algPool->getExecutionFlowGraph(), m_algname_index_map, m_eventSlots, m_optimizationMode );
  unsigned int controlFlowNodeNumber = m_efManager.getExecutionFlowGraph()->getControlFlowNodeCounter();

  // Shortcut for the message service
  SmartIF<IMessageSvc> messageSvc( serviceLocator() );
  if ( !messageSvc.isValid() ) error() << "Error retrieving MessageSvc interface IMessageSvc." << endmsg;

  m_eventSlots.assign( m_maxEventsInFlight,
                       EventSlot( m_algosDependencies, algsNumber, controlFlowNodeNumber, messageSvc ) );
  std::for_each( m_eventSlots.begin(), m_eventSlots.end(), []( EventSlot& slot ) { slot.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 Number of algorithms in flight: " << m_maxAlgosInFlight << endmsg;
  info() << " o TBB thread pool size: " << m_threadPoolSize << endmsg;

  // Simulating execution flow by only analyzing the graph topology and logic
  if ( m_simulateExecution ) {
    auto vis = concurrency::RunSimulator( 0 );
    m_efManager.simulateExecutionFlow( vis );
  }

  return sc;
}

StatusCode ForwardSchedulerSvc::isStalled ( int  iSlot )

private

Check if the scheduling is in a stall.

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

This is the case when no actions are present in the actionsQueue, no algorithm is in flight and no algorithm has all of its dependencies satisfied.

Definition at line 804 of file ForwardSchedulerSvc.cpp.

{
  // Get the slot
  EventSlot& thisSlot = m_eventSlots[iSlot];

  if ( m_actionsQueue.empty() && m_algosInFlight == 0 && m_IOBoundAlgosInFlight == 0 &&
       ( !thisSlot.algsStates.algsPresent( AlgsExecutionStates::DATAREADY ) ) ) {

    info() << "About to declare a stall" << endmsg;
    fatal() << "*** Stall detected! ***\n" << endmsg;
    dumpSchedulerState( iSlot );
    // throw GaudiException ("Stall detected",name(),StatusCode::FAILURE);

    return StatusCode::FAILURE;
  }
  return StatusCode::SUCCESS;
}

StatusCode ForwardSchedulerSvc::m_drain ( )

private

Drain the actions present in the queue.

Update the states for all slots until nothing is left to do.

Definition at line 468 of file ForwardSchedulerSvc.cpp.

{

  unsigned int slotNum = 0;
  for ( auto& thisSlot : m_eventSlots ) {
    if ( not thisSlot.algsStates.allAlgsExecuted() and not thisSlot.complete ) {
      updateStates( slotNum );
    }
    slotNum++;
  }
  return StatusCode::SUCCESS;
}

StatusCode ForwardSchedulerSvc::popFinishedEvent ( EventContext *&  eventContext )

override

Blocks until an event is availble.

Get a finished event or block until one becomes available.

Definition at line 485 of file ForwardSchedulerSvc.cpp.

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

StatusCode ForwardSchedulerSvc::promoteToAsyncExecuted ( unsigned int  iAlgo, int  si, IAlgorithm *  algo, EventContext *  eventContext  )

private

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

Definition at line 1113 of file ForwardSchedulerSvc.cpp.

{

  // Put back the instance
  Algorithm* castedAlgo = dynamic_cast<Algorithm*>( algo ); // DP: expose context getter in IAlgo?
  if ( !castedAlgo ) fatal() << "[Asynchronous] The casting did not succeed!" << endmsg;
  // EventContext* eventContext = castedAlgo->getContext();

  // Check if the execution failed
  if (m_algExecStateSvc->eventStatus(*eventContext) != EventStatus::Success)
    eventFailed(eventContext).ignore();

  StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );

  if ( !sc.isSuccess() ) {
    error() << "[Asynchronous]  [Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
            << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
    return StatusCode::FAILURE;
  }

  m_IOBoundAlgosInFlight--;

  EventSlot& thisSlot = m_eventSlots[si];
  // XXX: CF tests
  if ( !m_DFNext ) {
    // Update the catalog: some new products may be there
    m_whiteboard->selectStore( eventContext->slot() ).ignore();

    // update prods in the dataflow
    // DP: Handles could be used. Just update what the algo wrote
    DataObjIDColl new_products;
    m_whiteboard->getNewDataObjects(new_products).ignore();
    for (const auto& new_product : new_products)
      if (msgLevel(MSG::DEBUG))
        debug() << "Found in WB [" << si << "]: " << new_product << endmsg;
    thisSlot.dataFlowMgr.updateDataObjectsCatalog(new_products);
  }

  if (msgLevel(MSG::DEBUG))
    debug() << "[Asynchronous] Algorithm " << algo->name() << " executed in slot " << si
            << ". Algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;

  // Limit number of updates
  if ( m_CFNext )
    m_updateNeeded = true; // XXX: CF tests: with the new CF traversal the if clause below has to be removed
  if ( m_updateNeeded ) {
    // Schedule an update of the status of the algorithms
    auto updateAction = std::bind( &ForwardSchedulerSvc::updateStates, this, -1, algo->name() );
    m_actionsQueue.push( updateAction );
    m_updateNeeded = false;
  }

  if (msgLevel(MSG::DEBUG))
    debug() << "[Asynchronous] Trying to handle execution result of "
            << index2algname(iAlgo) << " on slot " << si << endmsg;
  State state;
  if ( algo->filterPassed() ) {
    state = State::EVTACCEPTED;
  } else {
    state = State::EVTREJECTED;
  }

  sc = thisSlot.algsStates.updateState( iAlgo, state );

  if (sc.isSuccess())
    if (msgLevel(MSG::VERBOSE))
      verbose() << "[Asynchronous] Promoting " << index2algname(iAlgo) << " on slot "
                << si << " to " << AlgsExecutionStates::stateNames[state] << endmsg;

  return sc;
}

StatusCode ForwardSchedulerSvc::promoteToAsyncScheduled ( unsigned int  iAlgo, int  si  )

private

Definition at line 989 of file ForwardSchedulerSvc.cpp.

{

  if ( m_IOBoundAlgosInFlight == m_maxIOBoundAlgosInFlight ) return StatusCode::FAILURE;

  // bool IOBound = m_efManager.getExecutionFlowGraph()->getAlgorithmNode(algName)->isIOBound();

  const std::string& algName( index2algname( iAlgo ) );
  IAlgorithm* ialgoPtr = nullptr;
  StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );

  if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
    EventContext* eventContext( m_eventSlots[si].eventContext );
    if ( !eventContext )
      fatal() << "[Asynchronous] Event context for algorithm " << algName << " is a nullptr (slot " << si << ")"
              << endmsg;

    ++m_IOBoundAlgosInFlight;
    // Can we use tbb-based overloaded new-operator for a "custom" task (an algorithm wrapper, not derived from tbb::task)? it seems it works..
    IOBoundAlgTask* theTask = new( tbb::task::allocate_root() )
      IOBoundAlgTask(ialgoPtr, iAlgo, eventContext, serviceLocator(),
                     this, m_algExecStateSvc);
    m_IOBoundAlgScheduler->push(*theTask);

    if (msgLevel(MSG::DEBUG))
      debug() << "[Asynchronous] Algorithm " << algName << " was submitted on event "
              << eventContext->evt() << " in slot " << si
              << ". algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;

    StatusCode updateSc( m_eventSlots[si].algsStates.updateState( iAlgo, AlgsExecutionStates::SCHEDULED ) );

    if (updateSc.isSuccess())
      if (msgLevel(MSG::VERBOSE))
        verbose() << "[Asynchronous] Promoting " << index2algname(iAlgo)
                  << " to SCHEDULED on slot " << si << endmsg;
    return updateSc;
  } else {
    if ( msgLevel( MSG::DEBUG ) )
      debug() << "[Asynchronous] Could not acquire instance for algorithm " << index2algname( iAlgo ) << " on slot "
              << si << endmsg;
    return sc;
  }
}

StatusCode ForwardSchedulerSvc::promoteToControlReady ( unsigned int  iAlgo, int  si  )

private

Algorithm promotion: Accepted by the control flow.

Definition at line 901 of file ForwardSchedulerSvc.cpp.

{

  // Do the control flow
  StatusCode sc = m_eventSlots[si].algsStates.updateState(iAlgo,AlgsExecutionStates::CONTROLREADY);
  if (sc.isSuccess())
    if (msgLevel(MSG::VERBOSE))
      verbose() << "Promoting " << index2algname(iAlgo) << " to CONTROLREADY on slot "
                << si << endmsg;

  return sc;
}

StatusCode ForwardSchedulerSvc::promoteToDataReady ( unsigned int  iAlgo, int  si  )

private

Definition at line 916 of file ForwardSchedulerSvc.cpp.

{

  StatusCode sc;
  if ( !m_DFNext ) {
    sc = m_eventSlots[si].dataFlowMgr.canAlgorithmRun( iAlgo );
  } else {
    sc = m_efManager.algoDataDependenciesSatisfied( index2algname( iAlgo ), si );
  }

  StatusCode updateSc( StatusCode::FAILURE );
  if ( sc == StatusCode::SUCCESS )
    updateSc = m_eventSlots[si].algsStates.updateState( iAlgo, AlgsExecutionStates::DATAREADY );

  if (updateSc.isSuccess())
    if (msgLevel(MSG::VERBOSE))
      verbose() << "Promoting " << index2algname(iAlgo) << " to DATAREADY on slot "
                << si<< endmsg;

  return updateSc;
}

StatusCode ForwardSchedulerSvc::promoteToExecuted ( unsigned int  iAlgo, int  si, IAlgorithm *  algo, EventContext *  eventContext  )

private

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

Definition at line 1037 of file ForwardSchedulerSvc.cpp.

{

  // Put back the instance
  Algorithm* castedAlgo = dynamic_cast<Algorithm*>( algo ); // DP: expose context getter in IAlgo?
  if ( !castedAlgo ) fatal() << "The casting did not succeed!" << endmsg;
  //  EventContext* eventContext = castedAlgo->getContext();

  // Check if the execution failed
  if (m_algExecStateSvc->eventStatus(*eventContext) != EventStatus::Success)
    eventFailed(eventContext).ignore();

  Gaudi::Hive::setCurrentContext(eventContext);
  StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );

  if ( !sc.isSuccess() ) {
    error() << "[Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
            << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
    return StatusCode::FAILURE;
  }

  m_algosInFlight--;

  EventSlot& thisSlot = m_eventSlots[si];
  // XXX: CF tests
  if ( !m_DFNext ) {
    // Update the catalog: some new products may be there
    m_whiteboard->selectStore( eventContext->slot() ).ignore();

    // update prods in the dataflow
    // DP: Handles could be used. Just update what the algo wrote
    DataObjIDColl new_products;
    m_whiteboard->getNewDataObjects( new_products ).ignore();
    for ( const auto& new_product : new_products )
      if ( msgLevel( MSG::DEBUG ) ) debug() << "Found in WB [" << si << "]: " << new_product << endmsg;
    thisSlot.dataFlowMgr.updateDataObjectsCatalog( new_products );
  }

  if ( msgLevel( MSG::DEBUG ) )
    debug() << "Algorithm " << algo->name() << " executed in slot " << si << ". Algorithms scheduled are "
            << m_algosInFlight << endmsg;

  // Limit number of updates
  if ( m_CFNext )
    m_updateNeeded = true; // XXX: CF tests: with the new CF traversal the if clause below has to be removed
  if ( m_updateNeeded ) {
    // Schedule an update of the status of the algorithms
    auto updateAction = std::bind( &ForwardSchedulerSvc::updateStates, this, -1, algo->name() );
    m_actionsQueue.push( updateAction );
    m_updateNeeded = false;
  }

  if ( msgLevel( MSG::DEBUG ) )
    debug() << "Trying to handle execution result of " << index2algname( iAlgo ) << " on slot " << si << endmsg;
  State state;
  if ( algo->filterPassed() ) {
    state = State::EVTACCEPTED;
  } else {
    state = State::EVTREJECTED;
  }

  sc = thisSlot.algsStates.updateState( iAlgo, state );

  if (sc.isSuccess())
    if (msgLevel(MSG::VERBOSE))
      verbose() << "Promoting " << index2algname(iAlgo) << " on slot " << si << " to "
                << AlgsExecutionStates::stateNames[state] << endmsg;

  return sc;
}

StatusCode ForwardSchedulerSvc::promoteToFinished ( unsigned int  iAlgo, int  si  )

private

StatusCode ForwardSchedulerSvc::promoteToScheduled ( unsigned int  iAlgo, int  si  )

private

Definition at line 940 of file ForwardSchedulerSvc.cpp.

{

  if ( m_algosInFlight == m_maxAlgosInFlight ) return StatusCode::FAILURE;

  const std::string& algName( index2algname( iAlgo ) );
  IAlgorithm* ialgoPtr = nullptr;
  StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );

  if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
    EventContext* eventContext( m_eventSlots[si].eventContext );
    if ( !eventContext )
      fatal() << "Event context for algorithm " << algName << " is a nullptr (slot " << si << ")" << endmsg;

    ++m_algosInFlight;
    // Avoid to use tbb if the pool size is 1 and run in this thread
    if (-100 != m_threadPoolSize) {
      tbb::task* t = new( tbb::task::allocate_root() )
        AlgoExecutionTask(ialgoPtr, iAlgo, eventContext, serviceLocator(),
                          this, m_algExecStateSvc);
      tbb::task::enqueue( *t);
    } else {
      AlgoExecutionTask theTask(ialgoPtr, iAlgo, eventContext,
                                serviceLocator(), this, m_algExecStateSvc);
      theTask.execute();
    }

    if ( msgLevel( MSG::DEBUG ) )
      debug() << "Algorithm " << algName << " was submitted on event " << eventContext->evt() << " in slot " << si
              << ". Algorithms scheduled are " << m_algosInFlight << endmsg;

    StatusCode updateSc( m_eventSlots[si].algsStates.updateState( iAlgo, AlgsExecutionStates::SCHEDULED ) );

    if ( msgLevel( MSG::VERBOSE ) ) dumpSchedulerState( -1 );

    if (updateSc.isSuccess())
      if (msgLevel(MSG::VERBOSE))
        verbose() << "Promoting " << index2algname(iAlgo) << " to SCHEDULED on slot "
                  << si << endmsg;
    return updateSc;
  } else {
    if ( msgLevel( MSG::DEBUG ) )
      debug() << "Could not acquire instance for algorithm " << index2algname( iAlgo ) << " on slot " << si << endmsg;
    return sc;
  }
}

StatusCode ForwardSchedulerSvc::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 400 of file ForwardSchedulerSvc.cpp.

{

  if ( m_first ) {
    m_first = false;
  }

  if ( !eventContext ) {
    fatal() << "Event context is nullptr" << endmsg;
    return StatusCode::FAILURE;
  }

  if ( m_freeSlots.load() == 0 ) {
    if ( msgLevel( MSG::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;
    }

    info() << "Executing event " << eventContext->evt() << " on slot " << thisSlotNum << endmsg;
    thisSlot.reset( eventContext );
    // XXX: CF tests
    if ( m_CFNext ) {
      auto vis = concurrency::Trigger( thisSlotNum );
      m_efManager.touchReadyAlgorithms( vis );
    }

    return this->updateStates( thisSlotNum );
  }; // end of lambda

  // Kick off the scheduling!
  if ( msgLevel( MSG::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;
}

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

override

Definition at line 451 of file ForwardSchedulerSvc.cpp.

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

StatusCode ForwardSchedulerSvc::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 508 of file ForwardSchedulerSvc.cpp.

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

StatusCode ForwardSchedulerSvc::updateStates ( int  si = -1, const std::string &  algo_name = std::string()  )

private

Loop on algorithm in the slots and promote them to successive states (-1 means all slots, while empty string means skipping an update of the Control Flow state)

Update the state of the algorithms.

The oldest events are checked before the newest, in order to reduce the event backlog. To check if the event is finished the algorithm checks if:

• No algorithms have been signed off by the control flow
• No algorithms have been signed off by the data flow
• No algorithms have been scheduled

Definition at line 573 of file ForwardSchedulerSvc.cpp.

{

  m_updateNeeded = true;

  // Fill a map of initial state / action using closures.
  // done to update the states w/o several if/elses
  // Posterchild for constexpr with gcc4.7 onwards!
  /*const std::map<AlgsExecutionStates::State, std::function<StatusCode(unsigned int iAlgo, int si)>>
   statesTransitions = {
  {AlgsExecutionStates::CONTROLREADY, std::bind(&ForwardSchedulerSvc::promoteToDataReady,
                                      this,
                                      std::placeholders::_1,
                                      std::placeholders::_2)},
  {AlgsExecutionStates::DATAREADY, std::bind(&ForwardSchedulerSvc::promoteToScheduled,
                                   this,
                                   std::placeholders::_1,
                                   std::placeholders::_2)}
  };*/

  StatusCode global_sc( StatusCode::FAILURE, true );

  // Sort from the oldest to the newest event
  // Prepare a vector of pointers to the slots to avoid copies
  std::vector<EventSlot*> eventSlotsPtrs;

  // Consider all slots if si <0 or just one otherwise
  if ( si < 0 ) {
    const int eventsSlotsSize( m_eventSlots.size() );
    eventSlotsPtrs.reserve( eventsSlotsSize );
    for ( auto slotIt = m_eventSlots.begin(); slotIt != m_eventSlots.end(); slotIt++ ) {
      if ( !slotIt->complete ) eventSlotsPtrs.push_back( &( *slotIt ) );
    }
    std::sort( eventSlotsPtrs.begin(), eventSlotsPtrs.end(),
               []( EventSlot* a, EventSlot* b ) { return a->eventContext->evt() < b->eventContext->evt(); } );
  } else {
    eventSlotsPtrs.push_back( &m_eventSlots[si] );
  }

  for ( EventSlot* thisSlotPtr : eventSlotsPtrs ) {
    int iSlot = thisSlotPtr->eventContext->slot();

    // Cache the states of the algos to improve readability and performance
    auto& thisSlot                      = m_eventSlots[iSlot];
    AlgsExecutionStates& thisAlgsStates = thisSlot.algsStates;

    // Take care of the control ready update
    // XXX: CF tests
    if ( !m_CFNext ) {
      m_efManager.updateEventState( thisAlgsStates, thisSlot.controlFlowState );
    } else {
      if ( !algo_name.empty() )
        m_efManager.updateDecision( algo_name, iSlot, thisAlgsStates, thisSlot.controlFlowState );
    }

    // DF note: all this this is a loop over all algs and applies CR->DR and DR->SCHD transistions
    /*for (unsigned int iAlgo=0;iAlgo<m_algname_vect.size();++iAlgo){
        const AlgsExecutionStates::State& algState = thisAlgsStates[iAlgo];
        if (algState==AlgsExecutionStates::ERROR)
            error() << " Algo " << index2algname(iAlgo) << " is in ERROR state." << endmsg;
        // Loop on state transitions from the one suited to algo state up to the one for SCHEDULED.
        partial_sc=StatusCode::SUCCESS;
        for (auto state_transition = statesTransitions.find(algState);
                state_transition!=statesTransitions.end() && partial_sc.isSuccess();
                state_transition++){
            partial_sc = state_transition->second(iAlgo,iSlot);
            if (partial_sc.isFailure()){
            verbose() << "Could not apply transition from "
                      << AlgsExecutionStates::stateNames[thisAlgsStates[iAlgo]]
                      << " for algorithm " << index2algname(iAlgo)
                      << " on processing slot " << iSlot << endmsg;
            }
            else{global_sc=partial_sc;}
        } // end loop on transitions
    }*/ // end loop on algos

    StatusCode partial_sc( StatusCode::FAILURE, true );
    // first update CONTROLREADY to DATAREADY
    if ( !m_CFNext ) {
      for ( auto it = thisAlgsStates.begin( AlgsExecutionStates::State::CONTROLREADY );
            it != thisAlgsStates.end( AlgsExecutionStates::State::CONTROLREADY ); ++it ) {

        uint algIndex = *it;
        partial_sc = promoteToDataReady(algIndex, iSlot);
        if (partial_sc.isFailure())
          if (msgLevel(MSG::VERBOSE))
            verbose() << "Could not apply transition from "
                      << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::CONTROLREADY]
                      << " for algorithm " << index2algname(algIndex) << " on processing slot " << iSlot << endmsg;
      }
    }

    // now update DATAREADY to SCHEDULED
    if ( !m_optimizationMode.empty() ) {
      auto comp_nodes = [this]( const uint& i, const uint& j ) {
        return ( m_efManager.getExecutionFlowGraph()->getAlgorithmNode( index2algname( i ) )->getRank() <
                 m_efManager.getExecutionFlowGraph()->getAlgorithmNode( index2algname( j ) )->getRank() );
      };
      std::priority_queue<uint, std::vector<uint>, std::function<bool( const uint&, const uint& )>> buffer(
          comp_nodes, std::vector<uint>() );
      for ( auto it = thisAlgsStates.begin( AlgsExecutionStates::State::DATAREADY );
            it != thisAlgsStates.end( AlgsExecutionStates::State::DATAREADY ); ++it )
        buffer.push( *it );
      /*std::stringstream s;
      auto buffer2 = buffer;
      while (!buffer2.empty()) {
        s << m_efManager.getExecutionFlowGraph()->getAlgorithmNode(index2algname(buffer2.top()))->getRank() << ", ";
        buffer2.pop();
      }
      info() << "DRBuffer is: [ " << s.str() << " ]  <--" << algo_name << " executed" << endmsg;*/

      /*while (!buffer.empty()) {
        partial_sc = promoteToScheduled(buffer.top(), iSlot);
        if (partial_sc.isFailure()) {
          if (msgLevel(MSG::VERBOSE))
            verbose() << "Could not apply transition from "
                      << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
                      << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot << endmsg;
          if (m_useIOBoundAlgScheduler) {
            partial_sc = promoteToAsyncScheduled(buffer.top(), iSlot);
            if (msgLevel(MSG::VERBOSE))
              if (partial_sc.isFailure())
                verbose() << "[Asynchronous] Could not apply transition from "
                          << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
                          << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot << endmsg;
          }
        }
        buffer.pop();
      }*/
      while ( !buffer.empty() ) {
        bool IOBound = false;
        if ( m_useIOBoundAlgScheduler )
          IOBound = m_efManager.getExecutionFlowGraph()->getAlgorithmNode( index2algname( buffer.top() ) )->isIOBound();

        if ( !IOBound )
          partial_sc = promoteToScheduled( buffer.top(), iSlot );
        else
          partial_sc = promoteToAsyncScheduled( buffer.top(), iSlot );

        if (msgLevel(MSG::VERBOSE))
          if (partial_sc.isFailure())
            verbose() << "Could not apply transition from "
                      << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
                      << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot << endmsg;

        buffer.pop();
      }

    } else {
      for ( auto it = thisAlgsStates.begin( AlgsExecutionStates::State::DATAREADY );
            it != thisAlgsStates.end( AlgsExecutionStates::State::DATAREADY ); ++it ) {
        uint algIndex = *it;

        bool IOBound = false;
        if ( m_useIOBoundAlgScheduler )
          IOBound = m_efManager.getExecutionFlowGraph()->getAlgorithmNode( index2algname( algIndex ) )->isIOBound();

        if ( !IOBound )
          partial_sc = promoteToScheduled( algIndex, iSlot );
        else
          partial_sc = promoteToAsyncScheduled( algIndex, iSlot );

        if (msgLevel(MSG::VERBOSE))
          if (partial_sc.isFailure())
            verbose() << "Could not apply transition from "
                      << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
                      << " for algorithm " << index2algname(algIndex) << " on processing slot " << iSlot << endmsg;
      }
    }

    if (m_dumpIntraEventDynamics) {
      auto now = std::chrono::system_clock::now();
      std::stringstream s;
      s << algo_name << ", " << thisAlgsStates.sizeOfSubset(State::CONTROLREADY) << ", "
                     << thisAlgsStates.sizeOfSubset(State::DATAREADY) << ", "
                     << thisAlgsStates.sizeOfSubset(State::SCHEDULED) << ", "
                     << std::chrono::duration_cast<std::chrono::nanoseconds> (now - m_efManager.getExecutionFlowGraph()->getInitTime()).count()
                     << "\n";
      auto threads = (m_threadPoolSize != -1) ? std::to_string(m_threadPoolSize)
                                              : std::to_string(tbb::task_scheduler_init::default_num_threads());
      std::ofstream myfile;
      myfile.open( "IntraEventConcurrencyDynamics_" + threads + "T.csv", std::ios::app );
      myfile << s.str();
      myfile.close();
    }

    // Not complete because this would mean that the slot is already free!
    if ( !thisSlot.complete && m_efManager.rootDecisionResolved( thisSlot.controlFlowState ) &&
         !thisSlot.algsStates.algsPresent( AlgsExecutionStates::CONTROLREADY ) &&
         !thisSlot.algsStates.algsPresent( AlgsExecutionStates::DATAREADY ) &&
         !thisSlot.algsStates.algsPresent( AlgsExecutionStates::SCHEDULED ) ) {

      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 already
      if(m_algExecStateSvc->eventStatus(*thisSlot.eventContext) == EventStatus::Success) {
        m_finishedEvents.push(thisSlot.eventContext);
        if (msgLevel(MSG::DEBUG))
          debug() << "Event " << thisSlot.eventContext->evt() << " finished (slot "
                  << thisSlot.eventContext->slot() << ")." << endmsg;
      }
      // now let's return the fully evaluated result of the control flow
      if ( msgLevel( MSG::DEBUG ) ) {
        std::stringstream ss;
        m_efManager.printEventState( ss, thisSlot.algsStates, thisSlot.controlFlowState, 0 );
        debug() << ss.str() << endmsg;
      }

      thisSlot.eventContext = nullptr;
    } else {
      StatusCode eventStalledSC = isStalled(iSlot);
      if (! eventStalledSC.isSuccess()) {
        m_algExecStateSvc->setEventStatus(EventStatus::AlgStall, *thisSlot.eventContext);
        eventFailed(thisSlot.eventContext).ignore();
      }
    }
  } // end loop on slots

  verbose() << "States Updated." << endmsg;

  return global_sc;
}

Friends And Related Function Documentation

friend class AlgoExecutionTask

friend

Definition at line 244 of file ForwardSchedulerSvc.h.

friend class IOBoundAlgTask

friend

Definition at line 245 of file ForwardSchedulerSvc.h.

Member Data Documentation

tbb::concurrent_bounded_queue<action> ForwardSchedulerSvc::m_actionsQueue

private

Queue where closures are stored and picked for execution.

Definition at line 238 of file ForwardSchedulerSvc.h.

SmartIF<IAlgExecStateSvc> ForwardSchedulerSvc::m_algExecStateSvc

private

Algorithm execution state manager.

Definition at line 193 of file ForwardSchedulerSvc.h.

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

private

Map to bookkeep the information necessary to the name2index conversion.

Definition at line 166 of file ForwardSchedulerSvc.h.

std::vector<std::string> ForwardSchedulerSvc::m_algname_vect

private

Vector to bookkeep the information necessary to the index2name conversion.

Definition at line 172 of file ForwardSchedulerSvc.h.

Gaudi::Property<std::vector<std::vector<std::string> > > ForwardSchedulerSvc::m_algosDependencies

private

Initial value:

{
      this, "AlgosDependencies", {}, "[[deprecated]]"}

Definition at line 144 of file ForwardSchedulerSvc.h.

unsigned int ForwardSchedulerSvc::m_algosInFlight = 0

private

Number of algoritms presently in flight.

Definition at line 198 of file ForwardSchedulerSvc.h.

SmartIF<IAlgResourcePool> ForwardSchedulerSvc::m_algResourcePool

private

Cache for the algorithm resource pool.

Definition at line 228 of file ForwardSchedulerSvc.h.

Gaudi::Property<bool> ForwardSchedulerSvc::m_CFNext

private

Initial value:

{this, "useGraphFlowManagement", false,
                                 "Temporary property to switch between ControlFlow implementations"}

Definition at line 130 of file ForwardSchedulerSvc.h.

Gaudi::Property<bool> ForwardSchedulerSvc::m_checkDeps {this, "CheckDependencies", false, "[[deprecated]]"}

private

Definition at line 146 of file ForwardSchedulerSvc.h.

Gaudi::Property<bool> ForwardSchedulerSvc::m_DFNext

private

Initial value:

{this, "DataFlowManagerNext", false,
                                 "Temporary property to switch between DataFlow implementations"}

Definition at line 133 of file ForwardSchedulerSvc.h.

Gaudi::Property<bool> ForwardSchedulerSvc::m_dumpIntraEventDynamics

private

Initial value:

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

Definition at line 140 of file ForwardSchedulerSvc.h.

concurrency::ExecutionFlowManager ForwardSchedulerSvc::m_efManager

private

Member to take care of the control flow.

Definition at line 241 of file ForwardSchedulerSvc.h.

std::vector<EventSlot> ForwardSchedulerSvc::m_eventSlots

private

Vector of events slots.

Definition at line 181 of file ForwardSchedulerSvc.h.

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

private

Queue of finished events.

Definition at line 187 of file ForwardSchedulerSvc.h.

bool ForwardSchedulerSvc::m_first = true

private

Definition at line 250 of file ForwardSchedulerSvc.h.

std::atomic_int ForwardSchedulerSvc::m_freeSlots

private

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

Definition at line 184 of file ForwardSchedulerSvc.h.

unsigned int ForwardSchedulerSvc::m_IOBoundAlgosInFlight = 0

private

Number of algoritms presently in flight.

Definition at line 201 of file ForwardSchedulerSvc.h.

SmartIF<IAccelerator> ForwardSchedulerSvc::m_IOBoundAlgScheduler

private

A shortcut to IO-bound algorithm scheduler.

Definition at line 178 of file ForwardSchedulerSvc.h.

Gaudi::Property<std::string> ForwardSchedulerSvc::m_IOBoundAlgSchedulerSvcName {this, "IOBoundAlgSchedulerSvc", "IOBoundAlgSchedulerSvc"}

private

Definition at line 123 of file ForwardSchedulerSvc.h.

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

private

Flag to track if the scheduler is active or not.

Definition at line 157 of file ForwardSchedulerSvc.h.

Gaudi::Property<unsigned int> ForwardSchedulerSvc::m_maxAlgosInFlight

private

Initial value:

{this, "MaxAlgosInFlight", 1,
                                                   "[[deprecated]] Taken from the whiteboard"}

Definition at line 124 of file ForwardSchedulerSvc.h.

Gaudi::Property<int> ForwardSchedulerSvc::m_maxEventsInFlight

private

Initial value:

{this, "MaxEventsInFlight", 0,
                                           "Maximum number of event processed simultaneously"}

Definition at line 117 of file ForwardSchedulerSvc.h.

Gaudi::Property<unsigned int> ForwardSchedulerSvc::m_maxIOBoundAlgosInFlight

private

Initial value:

{this, "MaxIOBoundAlgosInFlight", 0,
                                                          "Maximum number of simultaneous I/O-bound algorithms"}

Definition at line 127 of file ForwardSchedulerSvc.h.

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

private

Initial value:

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

Definition at line 138 of file ForwardSchedulerSvc.h.

Gaudi::Property<bool> ForwardSchedulerSvc::m_simulateExecution

private

Initial value:

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

Definition at line 135 of file ForwardSchedulerSvc.h.

std::mutex ForwardSchedulerSvc::m_ssMut

staticprivate

Definition at line 282 of file ForwardSchedulerSvc.h.

std::list< ForwardSchedulerSvc::SchedulerState > ForwardSchedulerSvc::m_sState

staticprivate

Definition at line 281 of file ForwardSchedulerSvc.h.

std::thread ForwardSchedulerSvc::m_thread

private

The thread in which the activate function runs.

Definition at line 160 of file ForwardSchedulerSvc.h.

Gaudi::Property<int> ForwardSchedulerSvc::m_threadPoolSize

private

Initial value:

{
      this, "ThreadPoolSize", -1,
      "Size of the threadpool initialised by TBB; a value of -1 gives TBB the freedom to choose"}

Definition at line 119 of file ForwardSchedulerSvc.h.

SmartIF<IThreadPoolSvc> ForwardSchedulerSvc::m_threadPoolSvc

private

Definition at line 248 of file ForwardSchedulerSvc.h.

bool ForwardSchedulerSvc::m_updateNeeded = true

private

Keep track of update actions scheduled.

Definition at line 224 of file ForwardSchedulerSvc.h.

Gaudi::Property<bool> ForwardSchedulerSvc::m_useIOBoundAlgScheduler

private

Initial value:

{this, "PreemptiveIOBoundTasks", false,
                                                 "Turn on preemptive way of scheduling of I/O-bound algorithms"}

Definition at line 142 of file ForwardSchedulerSvc.h.

SmartIF<IHiveWhiteBoard> ForwardSchedulerSvc::m_whiteboard

private

A shortcut to the whiteboard.

Definition at line 175 of file ForwardSchedulerSvc.h.

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

private

Definition at line 122 of file ForwardSchedulerSvc.h.

---

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

• GaudiHive/src/ForwardSchedulerSvc.h
• GaudiHive/src/ForwardSchedulerSvc.cpp