This SchedulerSvc implements the IScheduler interface. More...

#include <src/ParallelSequentialSchedulerSvc.h>

Inheritance diagram for ParallelSequentialSchedulerSvc:

Collaboration diagram for ParallelSequentialSchedulerSvc:

Public Member Functions
	ParallelSequentialSchedulerSvc (const std::string &name, ISvcLocator *svc)
	Constructor. More...

	~ParallelSequentialSchedulerSvc ()
	Destructor. More...

virtual StatusCode	initialize ()
	Initialise. More...

virtual StatusCode	finalize ()
	Finalise. More...

virtual StatusCode	pushNewEvent (EventContext *eventContext)
	Make an event available to the scheduler. More...

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

virtual StatusCode	popFinishedEvent (EventContext *&eventContext)
	Blocks until an event is availble. More...

virtual StatusCode	tryPopFinishedEvent (EventContext *&eventContext)
	Try to fetch an event from the scheduler. More...

virtual unsigned int	freeSlots ()
	Get free slots number. More...

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

StatusCode	setProperty (const Property &p) override

StatusCode	setProperty (const std::string &s) override

StatusCode	setProperty (const std::string &n, const std::string &v) override

StatusCode	getProperty (Property *p) const override

const Property &	getProperty (const std::string &name) const override

StatusCode	getProperty (const std::string &n, std::string &v) const override

const std::vector< Property * > &	getProperties () const override

bool	hasProperty (const std::string &name) const override

template<class TYPE >
StatusCode	setProperty (const std::string &name, const TYPE &value)
	set the property form the value 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 >
Property *	declareProperty (const std::string &name, T &property, const std::string &doc="none") const
	Declare the named property. More...

Property *	declareRemoteProperty (const std::string &name, IProperty *rsvc, const std::string &rname="") const
	Declare remote named properties. 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 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 Attributes
bool	m_useTopAlgList
	Decide if the top alglist or its flat version has to be used. More...

std::list< IAlgorithm * >	m_algList
	Cache the list of algs to be executed. More...

tbb::concurrent_bounded_queue< EventContext * >	m_finishedEvents
	Queue of finished events. More...

std::atomic_int	m_freeSlots
	Atomic to account for asyncronous updates by the scheduler wrt the rest. More...

SmartIF< IHiveWhiteBoard >	m_whiteboard
	A shortcut to the whiteboard. More...

std::string	m_whiteboardSvcName
	The whiteboard name. More...

SmartIF< IAlgResourcePool >	m_algResourcePool
	Cache for the algorithm resource pool. More...

int	m_threadPoolSize
	Size of the threadpool initialised by TBB; a value of -1 gives TBB the freedom to choose. More...

std::unique_ptr< tbb::task_scheduler_init >	m_tbb_sched

concurrency::ExecutionFlowManager	m_controlFlow

std::vector< std::string >	m_algname_vect
	Vector to bookkeep the information necessary to the index2name conversion. More...

std::unordered_map< std::string, unsigned int >	m_algname_index_map
	Map to bookkeep the information necessary to the name2index conversion. More...

Friends
class	SequentialTask

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 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 CommonMessaging< implements< IService, IProperty, IStateful > >
void	updateMsgStreamOutputLevel (int level)
	Update the output level of the cached MsgStream. More...

Protected Attributes inherited from Service
IntegerProperty	m_outputLevel = MSG::NIL
	Service output level. More...

Gaudi::StateMachine::State	m_state = Gaudi::StateMachine::OFFLINE
	Service state. More...

Gaudi::StateMachine::State	m_targetState = Gaudi::StateMachine::OFFLINE
	Service state. More...

Detailed Description

This SchedulerSvc implements the IScheduler interface.

It executes all the algorithms in sequence for several events in flight. It pulls the algorithms from the AlgResourcePool

Author: Daniel Funke

Version: 0.1

Definition at line 41 of file ParallelSequentialSchedulerSvc.h.

Constructor & Destructor Documentation

ParallelSequentialSchedulerSvc::ParallelSequentialSchedulerSvc	(	const std::string &	name,
		ISvcLocator *	svc
	)

Constructor.

Definition at line 28 of file ParallelSequentialSchedulerSvc.cpp.

                                                                                                         :
     base_class(name,svcLoc) {
 
   declareProperty("UseTopAlgList", m_useTopAlgList = false);
   declareProperty("ThreadPoolSize", m_threadPoolSize = -1);
   declareProperty("WhiteboardSvc", m_whiteboardSvcName = "EventDataSvc");
 
 }

ParallelSequentialSchedulerSvc::~ParallelSequentialSchedulerSvc ( )

Destructor.

Definition at line 38 of file ParallelSequentialSchedulerSvc.cpp.

38 {}

Member Function Documentation

StatusCode ParallelSequentialSchedulerSvc::finalize ( )

virtual

Finalise.

Definition at line 148 of file ParallelSequentialSchedulerSvc.cpp.

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

unsigned int ParallelSequentialSchedulerSvc::freeSlots ( )

virtual

Get free slots number.

Given that the scheduler is sequential and its methods non reentrant, this is always 1.

Definition at line 220 of file ParallelSequentialSchedulerSvc.cpp.

220 {return m_freeSlots;}

ParallelSequentialSchedulerSvc::m_freeSlots

std::atomic_int m_freeSlots

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

Definition: ParallelSequentialSchedulerSvc.h:85

StatusCode ParallelSequentialSchedulerSvc::initialize ( )

virtual

Initialise.

Definition at line 41 of file ParallelSequentialSchedulerSvc.cpp.

                                                      {
 
   // Initialise mother class (read properties, ...)
   StatusCode sc(Service::initialize());
   if (!sc.isSuccess())
     warning () << "Base class could not be initialized" << endmsg;
 
   // Get the algo resource pool
   m_algResourcePool =  serviceLocator()->service("AlgResourcePool");
   if (!m_algResourcePool.isValid()){
     error() << "Error retrieving AlgResourcePool" << endmsg;
     return StatusCode::FAILURE;
   }
 
   // Get the list of algorithms
   m_algList = m_useTopAlgList ? m_algResourcePool->getTopAlgList() : m_algResourcePool->getFlatAlgList();
   info() << "Found " <<  m_algList.size() << " algorithms" << endmsg;
 
   // Get Whiteboard
   m_whiteboard = serviceLocator()->service(m_whiteboardSvcName);
   if (!m_whiteboard.isValid())
     fatal() << "Error retrieving EventDataSvc interface IHiveWhiteBoard." << endmsg;
 
   // Check the MaxEventsInFlight parameters and react
   // Deprecated for the moment
   size_t numberOfWBSlots = m_whiteboard->getNumberOfStores();
 
   // Set the number of free slots
   m_freeSlots=numberOfWBSlots;
 
   info() << "Allowing " << m_freeSlots << " events in flight" << endmsg;
 
   if(m_threadPoolSize == -1)
     m_threadPoolSize = numberOfWBSlots;
 
   debug() << "Initialising a TBB thread pool of size " << m_threadPoolSize << endmsg;
   m_tbb_sched.reset(new tbb::task_scheduler_init(m_threadPoolSize));
 
   // Fill the containers to convert algo names to index
   m_algname_index_map.reserve(m_algList.size());
   m_algname_vect.reserve(m_algList.size());
   unsigned int index=0;
   for (IAlgorithm* algo : m_algList){
     const std::string& name = algo->name();
     m_algname_index_map[name]=index;
     m_algname_vect.emplace_back(name);
     index++;
   }
 
   //initialize control flow manager
   const AlgResourcePool* algPool = dynamic_cast<const AlgResourcePool*>(m_algResourcePool.get());
 
   m_controlFlow.initialize(algPool->getExecutionFlowGraph(), m_algname_index_map);
 
   const unsigned int algosDependenciesSize=0;
   info() << "Algodependecies size is " << algosDependenciesSize << endmsg;
 
   //get algorithm dependencies
   /* Dependencies
      0) Read deps from config file
      1) Look for handles in algo, if none
      2) Assume none are required
    */
   if (algosDependenciesSize == 0){
     // Get the event root from the IDataManagerSvc interface of the WhiteBoard
     SmartIF<IDataManagerSvc> dataMgrSvc (m_whiteboard);
     std::string rootInTESName(dataMgrSvc->rootName());
     if ("" != rootInTESName && '/'!=rootInTESName[rootInTESName.size()-1]){
       rootInTESName = rootInTESName+"/";
     }
 
     for (IAlgorithm* ialgoPtr : m_algList){
       Algorithm* algoPtr = dynamic_cast<Algorithm*> (ialgoPtr);
       if (nullptr == algoPtr){
         fatal() << "Could not convert IAlgorithm into Algorithm: this will result in a crash." << endmsg;
       }
 
       std::vector<Gaudi::DataHandle*> algoHandles(algoPtr->inputHandles());
       DataObjIDColl algoDependencies;
       if (!algoHandles.empty()){
         info() << "Algorithm " << algoPtr->name() << " data dependencies:" << endmsg;
 
         DataObjIDColl inputObjs, outputObjs;
         DHHVisitor avis(inputObjs, outputObjs);
 
         algoPtr->acceptDHVisitor( &avis );
 
         for (auto id : inputObjs) {
           const std::string& productName = rootInTESName + id.key();
           info() << "  o Input dep for " << productName << endmsg;
           algoDependencies.insert(id);
         }
 
 
       } else {
         info() << "Algorithm " << algoPtr->name() << " has no data dependencies."
                << endmsg;
       }
 
     }
   }
 
   return StatusCode::SUCCESS;
 
 }

StatusCode ParallelSequentialSchedulerSvc::popFinishedEvent ( EventContext *& eventContext )

virtual

Blocks until an event is availble.

Get a finished event or block until one becomes available.

Definition at line 192 of file ParallelSequentialSchedulerSvc.cpp.

                                                                                       {
 
   m_finishedEvents.pop(eventContext);
   debug() << "Popped slot " << eventContext->slot() << "(event "
       << eventContext->evt() << ")" << endmsg;
   m_freeSlots++;
   return StatusCode::SUCCESS;
 }

StatusCode ParallelSequentialSchedulerSvc::pushNewEvent ( EventContext * eventContext )

virtual

Make an event available to the scheduler.

Immediately the algortihms are executed.

Definition at line 160 of file ParallelSequentialSchedulerSvc.cpp.

                                                                                  {
   std::vector<EventContext*> eventContexts;
   eventContexts.push_back(eventContext);
   eventContext->setFail(false);
   return pushNewEvents(eventContexts);
 }

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

virtual

Definition at line 167 of file ParallelSequentialSchedulerSvc.cpp.

                                                                                                {
 
   for(auto evt : eventContexts){
     if(m_freeSlots.load() > 0){
       //only one thread executes scheduler --> m_freeSlots can only grow if other thread finishes
       m_freeSlots--;
 
       debug() << "Enqueuing event " << evt->evt() << " @ " << evt->slot() << endmsg;
 
       tbb::task* t = new( tbb::task::allocate_root() )
                  SequentialTask(serviceLocator(), evt, this, m_algResourcePool);
       tbb::task::enqueue( *t);
     } else {
       return StatusCode::FAILURE;
     }
   }
 
   return StatusCode::SUCCESS;
 
 }

StatusCode ParallelSequentialSchedulerSvc::tryPopFinishedEvent ( EventContext *& eventContext )

virtual

Try to fetch an event from the scheduler.

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

Definition at line 205 of file ParallelSequentialSchedulerSvc.cpp.

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