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< BASE, Interfaces >
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...

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 extend_interfaces< Interfaces...>
	~extend_interfaces () override=default
	Virtual destructor. More...

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

std::vector< std::vector< std::string > >	m_algosDependencies
	Ugly, will disappear when the deps are declared only within the C++ code of the algos. More...

Friends
class	SequentialTask

Additional Inherited Members
Public Types inherited from extends< BASE, Interfaces >
using	base_class = extends
	Typedef to this class. More...

using	extend_interfaces_base = extend_interfaces< Interfaces...>
	Typedef to the base of this class. More...

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

using	ext_iids = typename Gaudi::interface_list_cat< typename Interfaces::ext_iids...>::type
	take union of the ext_iids of all Interfaces... 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 27 of file ParallelSequentialSchedulerSvc.cpp.

                                                                                                         :
     base_class(name,svcLoc) {
 
   // Will disappear when dependencies are properly propagated into the C++ code of the algos
   declareProperty("AlgosDependencies", m_algosDependencies);
   declareProperty("UseTopAlgList", m_useTopAlgList = false);
   declareProperty("ThreadPoolSize", m_threadPoolSize = -1);
   declareProperty("WhiteboardSvc", m_whiteboardSvcName = "EventDataSvc");
 
 }

ParallelSequentialSchedulerSvc::~ParallelSequentialSchedulerSvc ( )

Destructor.

Definition at line 39 of file ParallelSequentialSchedulerSvc.cpp.

39 {}

Member Function Documentation

StatusCode ParallelSequentialSchedulerSvc::finalize ( )

virtual

Finalise.

Definition at line 147 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 219 of file ParallelSequentialSchedulerSvc.cpp.

219 {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:84

StatusCode ParallelSequentialSchedulerSvc::initialize ( )

virtual

Initialise.

Definition at line 42 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=m_algosDependencies.size();
   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;
       }
 
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wdeprecated-declarations"
       const std::vector<MinimalDataObjectHandle*>& algoHandles(algoPtr->handles());
 #pragma GCC diagnostic pop
       std::vector<std::string> algoDependencies;
       if (!algoHandles.empty()){
 
         info() << "Algorithm " << algoPtr->name() << " data dependencies:" << endmsg;
         for (MinimalDataObjectHandle* handlePtr : algoHandles ){
           if (handlePtr->accessType() == MinimalDataObjectHandle::AccessType::READ){
             const std::string& productName = rootInTESName + handlePtr->dataProductName();
             info() << "  o READ Handle found for product " << productName << endmsg;
             algoDependencies.emplace_back(productName);
           }
         }
       } else {
         info() << "Algorithm " << algoPtr->name() << " has no data dependencies." << endmsg;
       }
 
       m_algosDependencies.emplace_back(algoDependencies);
     }
   }
 
   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 191 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 159 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 166 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 204 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;
 }