21 #include <unordered_set> 24 #include "boost/thread.hpp" 25 #include "boost/tokenizer.hpp" 26 #include "boost/algorithm/string.hpp" 28 #include "tbb/task_scheduler_init.h" 48 if ( !sc.
isSuccess() ) warning() <<
"Base class could not be initialized" <<
endmsg;
51 m_threadPoolSvc = serviceLocator()->service(
"ThreadPoolSvc" );
52 if ( !m_threadPoolSvc.isValid() ) {
53 fatal() <<
"Error retrieving ThreadPoolSvc" <<
endmsg;
58 info() <<
"Activating scheduler in a separate thread" <<
endmsg;
61 while ( m_isActive != ACTIVE ) {
63 fatal() <<
"Terminating initialization" <<
endmsg;
66 info() <<
"Waiting for ForwardSchedulerSvc to activate" <<
endmsg;
72 m_algResourcePool = serviceLocator()->service(
"AlgResourcePool" );
73 if ( !m_algResourcePool.isValid() ) {
74 fatal() <<
"Error retrieving AlgoResourcePool" <<
endmsg;
78 m_algExecStateSvc = serviceLocator()->service(
"AlgExecStateSvc");
79 if (!m_algExecStateSvc.isValid()) {
80 fatal() <<
"Error retrieving AlgExecStateSvc" <<
endmsg;
85 m_whiteboard = serviceLocator()->service( m_whiteboardSvcName );
86 if ( !m_whiteboard.isValid() ) {
87 fatal() <<
"Error retrieving EventDataSvc interface IHiveWhiteBoard." <<
endmsg;
93 size_t numberOfWBSlots = m_whiteboard->getNumberOfStores();
94 if ( m_maxEventsInFlight != 0 ) {
95 warning() <<
"Property MaxEventsInFlight was set. This works but it's deprecated. " 96 <<
"Please migrate your code options files." <<
endmsg;
98 if ( m_maxEventsInFlight != (
int)numberOfWBSlots ) {
99 warning() <<
"In addition, the number of events in flight (" << m_maxEventsInFlight
100 <<
") differs from the slots in the whiteboard (" << numberOfWBSlots
101 <<
"). Setting the number of events in flight to " << numberOfWBSlots <<
endmsg;
109 m_maxEventsInFlight = numberOfWBSlots;
112 m_freeSlots = m_maxEventsInFlight;
114 if ( m_algosDependencies.size() != 0 ) {
115 warning() <<
" ##### Property AlgosDependencies is deprecated and ignored." 116 <<
" FIX your job options #####" <<
endmsg;
121 const unsigned int algsNumber = algos.
size();
122 info() <<
"Found " << algsNumber <<
" algorithms" <<
endmsg;
135 fatal() <<
"Could not convert IAlgorithm into Algorithm: this will result in a crash." <<
endmsg;
138 auto r = globalOutp.
insert(
id);
140 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;
144 info() <<
"outputs:\n" ;
145 for (
const auto& i : globalOutp ) {
146 info() << i <<
'\n' ;
152 info() <<
"Data Dependencies for Algorithms:";
157 if (
nullptr == algoPtr )
158 fatal() <<
"Could not convert IAlgorithm into Algorithm: this will result in a crash." <<
endmsg;
160 info() <<
"\n " << algoPtr->
name();
166 info() <<
"\n o INPUT " << id;
167 if (
id.key().find(
":")!=std::string::npos) {
168 info() <<
" contains alternatives which require resolution... " <<
endmsg;
169 auto tokens = boost::tokenizer<boost::char_separator<char>>{
id.key(),boost::char_separator<char>{
":"}};
172 return globalOutp.find(
DataObjID{t} ) != globalOutp.end();
174 if (itok!=tokens.end()) {
175 info() <<
"found matching output for " << *itok <<
" -- updating scheduler info" <<
endmsg;
178 error() <<
"failed to find alternate in global output list" <<
endmsg;
181 algoDependencies.
insert(
id );
185 info() <<
"\n o OUTPUT " << id;
186 if (
id.key().find(
":")!=std::string::npos) {
187 info() <<
" alternatives are NOT allowed for outputs..." <<
endmsg;
198 m_algname_vect.reserve( algsNumber );
199 unsigned int index = 0;
203 m_algname_index_map[
name] = index;
204 m_algname_vect.emplace_back( name );
205 if (algo->name() == m_useDataLoader) {
206 dataLoaderAlg = algo;
214 for (
auto o : globalInp ) {
215 if ( globalOutp.find( o ) == globalOutp.end() ) {
220 if ( unmetDep.
size() > 0 ) {
223 for (
auto& o : unmetDep ) {
224 ost <<
"\n o " << o <<
" required by Algorithm: ";
225 for (
size_t i = 0; i < m_algosDependencies.
size(); ++i ) {
226 if ( m_algosDependencies[i].find( o ) != m_algosDependencies[i].
end() ) {
227 ost <<
"\n * " << m_algname_vect[i];
232 if ( m_useDataLoader !=
"" ) {
234 if (dataLoaderAlg ==
nullptr) {
235 fatal() <<
"No DataLoader Algorithm \"" << m_useDataLoader.value()
236 <<
"\" found, and unmet INPUT dependencies " 241 info() <<
"Will attribute the following unmet INPUT dependencies to \"" 242 << dataLoaderAlg->
type() <<
"/" << dataLoaderAlg->name()
249 fatal() <<
"Unable to dcast DataLoader \"" << m_useDataLoader.value()
250 <<
"\" IAlg to Algorithm" <<
endmsg;
254 for (
auto&
id : unmetDep) {
255 debug() <<
"adding OUTPUT dep \"" <<
id <<
"\" to " 256 << dataLoaderAlg->
type() <<
"/" << dataLoaderAlg->name()
262 fatal() <<
"Auto DataLoading not requested, " 263 <<
"and the following unmet INPUT dependencies were found:" 269 info() <<
"No unmet INPUT data dependencies were found" <<
endmsg;
274 sc = m_efManager.initialize( algPool->
getPRGraph(), m_algname_index_map);
275 unsigned int controlFlowNodeNumber = m_efManager.getPrecedenceRulesGraph()->getControlFlowNodeCounter();
279 if ( !messageSvc.
isValid() ) error() <<
"Error retrieving MessageSvc interface IMessageSvc." <<
endmsg;
281 m_eventSlots.assign( m_maxEventsInFlight,
282 EventSlot( m_algosDependencies, algsNumber, controlFlowNodeNumber, messageSvc ) );
283 std::for_each( m_eventSlots.begin(), m_eventSlots.end(), [](
EventSlot& slot ) { slot.complete =
true; } );
286 info() <<
"Concurrency level information:" <<
endmsg;
287 info() <<
" o Number of events in flight: " << m_maxEventsInFlight <<
endmsg;
288 info() <<
" o Number of algorithms in flight: " << m_maxAlgosInFlight <<
endmsg;
289 info() <<
" o TBB thread pool size: " << m_threadPoolSize <<
endmsg;
306 info() <<
"Joining Scheduler thread" <<
endmsg;
311 error() <<
"problems in scheduler thread" <<
endmsg;
333 debug() <<
"ForwardSchedulerSvc::activate()" <<
endmsg;
336 error() <<
"problems initializing ThreadPoolSvc" <<
endmsg;
348 info() <<
"Start checking the actionsQueue" <<
endmsg;
353 verbose() <<
"Action did not succeed (which is not bad per se)." <<
endmsg;
358 info() <<
"Terminating thread-pool resources" <<
endmsg;
360 error() <<
"Problems terminating thread pool" <<
endmsg;
418 if ( !eventContext ) {
433 const unsigned int thisSlotNum = eventContext->
slot();
436 fatal() <<
"The slot " << thisSlotNum <<
" is supposed to be a finished event but it's not" <<
endmsg;
440 info() <<
"Executing event " << eventContext->
evt() <<
" on slot " << thisSlotNum <<
endmsg;
441 thisSlot.
reset( eventContext );
448 verbose() <<
"Pushing the action to update the scheduler for slot " << eventContext->
slot() <<
endmsg;
459 for (
auto context : eventContexts ) {
476 unsigned int slotNum = 0;
478 if ( not thisSlot.algsStates.allAlgsExecuted() and not thisSlot.complete ) {
502 debug() <<
"Popped slot " << eventContext->
slot() <<
"(event " 515 debug() <<
"Try Pop successful slot " << eventContext->
slot() <<
"(event " << eventContext->
evt() <<
")" 534 fatal() <<
"*** Event " << eventContext->
evt() <<
" on slot " 535 << eventContext->
slot() <<
" failed! ***" <<
endmsg;
540 info() <<
"Dumping Alg Exec State for slot " << eventContext->
slot()
603 eventSlotsPtrs.
reserve( eventsSlotsSize );
605 if ( !slotIt->complete ) eventSlotsPtrs.
push_back( &( *slotIt ) );
613 for (
EventSlot* thisSlotPtr : eventSlotsPtrs ) {
646 for (
auto it = thisAlgsStates.
begin( AlgsExecutionStates::State::CONTROLREADY );
647 it != thisAlgsStates.
end( AlgsExecutionStates::State::CONTROLREADY ); ++it ) {
653 verbose() <<
"Could not apply transition from " 655 <<
" for algorithm " <<
index2algname(algIndex) <<
" on processing slot " << iSlot <<
endmsg;
659 for (
auto it = thisAlgsStates.
begin( AlgsExecutionStates::State::DATAREADY );
660 it != thisAlgsStates.
end( AlgsExecutionStates::State::DATAREADY ); ++it ) {
667 verbose() <<
"Could not apply transition from " 669 <<
" for algorithm " <<
index2algname(algIndex) <<
" on processing slot " << iSlot <<
endmsg;
678 thisSlot.complete =
true;
684 debug() <<
"Event " << thisSlot.eventContext->evt() <<
" finished (slot " 685 << thisSlot.eventContext->slot() <<
")." <<
endmsg;
694 thisSlot.eventContext =
nullptr;
724 info() <<
"About to declare a stall" <<
endmsg;
746 outputMessageStream <<
"============================== Execution Task State =============================" 750 outputMessageStream << std::endl
751 <<
"============================== Scheduler State =================================" 757 if ( thisSlot.complete )
continue;
759 outputMessageStream <<
"----------- slot: " << thisSlot.eventContext->slot()
760 <<
" event: " << thisSlot.eventContext->evt() <<
" -----------" <<
std::endl;
762 if ( 0 > iSlot or iSlot == slotCount ) {
763 outputMessageStream <<
"Algorithms states:" <<
std::endl;
765 const DataObjIDColl& wbSlotContent( thisSlot.dataFlowMgr.content() );
766 for (
unsigned int algoIdx = 0; algoIdx < thisSlot.algsStates.size(); ++algoIdx ) {
767 outputMessageStream <<
" o " <<
index2algname( algoIdx ) <<
" [" 769 DataObjIDColl deps( thisSlot.dataFlowMgr.dataDependencies( algoIdx ) );
770 const int depsSize = deps.
size();
771 if ( depsSize == 0 ) outputMessageStream <<
" none";
774 for (
auto d : deps ) {
775 outputMessageStream << d <<
" ";
776 if ( wbSlotContent.find( d ) == wbSlotContent.end() ) {
782 if ( !missing.
empty() ) {
783 outputMessageStream <<
". The following are missing: ";
784 for (
auto d : missing ) {
785 outputMessageStream << d <<
" ";
793 outputMessageStream <<
"\nWhiteboard contents: " <<
std::endl;
794 for (
auto& product : wbSlotContent ) outputMessageStream <<
" o " << product <<
std::endl;
797 outputMessageStream <<
"\nControl Flow:" <<
std::endl;
801 outputMessageStream << cFlowStateStringStream.
str() <<
std::endl;
805 outputMessageStream <<
"=================================== END ======================================" <<
std::endl;
807 info() <<
"Dumping Scheduler State " << std::endl << outputMessageStream.
str() <<
endmsg;
852 if ( sc.isSuccess() ) {
855 fatal() <<
"Event context for algorithm " << algName <<
" is a nullptr (slot " << si <<
")" <<
endmsg;
862 eventContext->
slot(),
870 tbb::task* triggerAlgoStateUpdate =
new(tbb::task::allocate_root())
874 triggerAlgoStateUpdate->set_ref_count(1);
876 tbb::task* algoTask =
new(triggerAlgoStateUpdate->allocate_child())
879 tbb::task::enqueue( *algoTask);
884 promote2ExecutedClosure();
888 debug() <<
"Algorithm " << algName <<
" was submitted on event " << eventContext->
evt() <<
" in slot " << si
895 if (updateSc.isSuccess())
914 if ( !castedAlgo )
fatal() <<
"The casting did not succeed!" <<
endmsg;
925 error() <<
"[Event " << eventContext->
evt() <<
", Slot " << eventContext->
slot() <<
"] " 926 <<
"Instance of algorithm " << algo->name() <<
" could not be properly put back." <<
endmsg;
941 for (
const auto& new_product : new_products )
946 debug() <<
"Algorithm " << algo->name() <<
" executed in slot " << si <<
". Algorithms scheduled are " 961 state = State::EVTACCEPTED;
963 state = State::EVTREJECTED;
995 error() <<
"could not find Alg " << a->
name() <<
" in Scheduler!" <<
endmsg;
virtual StatusCode initPool(const int &poolSize)=0
Initializes the thread pool.
StatusCode deactivate()
Deactivate scheduler.
bool algsPresent(State state) const
StatusCode initialize() override
virtual StatusCode acquireAlgorithm(const std::string &name, IAlgorithm *&algo, bool blocking=false)=0
Acquire a certain algorithm using its name.
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
void updateEventState(AlgsExecutionStates &algo_states, std::vector< int > &node_decisions) const
Update the state of algorithms to controlready, where possible.
void updateDataObjectsCatalog(const DataObjIDColl &newProducts)
Update the catalog of available products in the slot.
void printEventState(std::stringstream &ss, AlgsExecutionStates &states, const std::vector< int > &node_decisions, const unsigned int &recursionLevel) const
Print the state of the control flow for a given event.
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
const std::string & name() const override
The identifying name of the algorithm object.
virtual concurrency::PrecedenceRulesGraph * getPRGraph() const
StatusCode finalize() override
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
AlgsExecutionStates algsStates
Vector of algorithms states.
virtual void dump(std::ostringstream &ost, const EventContext &ctx) const =0
const DataObjIDColl & outputDataObjs() const override
bool isSuccess() const
Test for a status code of SUCCESS.
StatusCode isStalled(int si)
Check if the scheduling is in a stall.
EventContext * eventContext
Cache for the eventContext.
Header file for class GaudiAlgorithm.
StatusCode updateStates(int si=-1)
Loop on algorithm in the slots and promote them to successive states (-1 means all slots...
StatusCode finalize() override
Finalise.
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
virtual bool filterPassed() const =0
Did this algorithm pass or fail its filter criterion for the last event?
SmartIF< IThreadPoolSvc > m_threadPoolSvc
The SchedulerSvc implements the IScheduler interface.
StatusCode pushNewEvent(EventContext *eventContext) override
Make an event available to the scheduler.
The AlgResourcePool is a concrete implementation of the IAlgResourcePool interface.
This class represents an entry point to all the event specific data.
bool isFailure() const
Test for a status code of FAILURE.
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
StatusCode m_drain()
Drain the actions present in the queue.
Gaudi::Property< unsigned int > m_maxAlgosInFlight
virtual const std::string & type() const =0
The type of the algorithm.
tbb::task * execute() override
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
virtual StatusCode terminatePool()=0
Finalize the thread pool.
void addAlg(Algorithm *, EventContext *, pthread_t)
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
bool rootDecisionResolved(const std::vector< int > &node_decisions) const
Check whether root decision was resolved.
DataFlowManager dataFlowMgr
DataFlowManager of this slot.
virtual StatusCode selectStore(size_t partitionIndex)=0
Activate an given 'slot' for all subsequent calls within the same thread id.
The AlgsExecutionStates encodes the state machine for the execution of algorithms within a single eve...
unsigned int m_algosInFlight
Number of algoritms presently in flight.
MsgStream & warning() const
shortcut for the method msgStream(MSG::WARNING)
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
std::thread m_thread
The thread in which the activate function runs.
virtual StatusCode getNewDataObjects(DataObjIDColl &products)=0
Get the latest new data objects registred in store.
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
static std::list< SchedulerState > m_sState
std::vector< EventSlot > m_eventSlots
Vector of events slots.
StatusCode promoteToScheduled(unsigned int iAlgo, int si)
This class is used for returning status codes from appropriate routines.
const DataObjIDColl & inputDataObjs() const override
unsigned int algname2index(const std::string &algoname)
Convert a name to an integer.
StatusCode promoteToExecuted(unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
virtual void setEventStatus(const EventStatus::Status &sc, const EventContext &ctx)=0
unsigned int freeSlots() override
Get free slots number.
StatusCode pushNewEvents(std::vector< EventContext * > &eventContexts) override
#define DECLARE_SERVICE_FACTORY(x)
bool complete
Flags completion of the event.
Gaudi::Property< int > m_maxEventsInFlight
The IAlgorithm is the interface implemented by the Algorithm base class.
GAUDI_API void setCurrentContext(const EventContext *ctx)
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
bool m_updateNeeded
Keep track of update actions scheduled.
void addDependency(const DataObjID &id, const Gaudi::DataHandle::Mode &mode) override
StatusCode tryPopFinishedEvent(EventContext *&eventContext) override
Try to fetch an event from the scheduler.
Base class from which all concrete algorithm classes should be derived.
std::unordered_map< std::string, unsigned int > m_algname_index_map
Map to bookkeep the information necessary to the name2index conversion.
void activate()
Activate scheduler.
void reset(EventContext *theeventContext)
Reset all resources in order to reuse the slot.
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
bool isValid() const
Allow for check if smart pointer is valid.
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Iterator begin(State kind)
StatusCode promoteToControlReady(unsigned int iAlgo, int si)
Algorithm promotion: Accepted by the control flow.
concurrency::ExecutionFlowManager m_efManager
Member to take care of the control flow.
virtual const EventStatus::Status & eventStatus(const EventContext &ctx) const =0
StatusCode popFinishedEvent(EventContext *&eventContext) override
Blocks until an event is availble.
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Class representing the event slot.
static std::mutex m_ssMut
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
State
Execution states of the algorithms.
virtual StatusCode releaseAlgorithm(const std::string &name, IAlgorithm *&algo)=0
Release a certain algorithm.
Gaudi::Property< int > m_threadPoolSize
void dumpState() override
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator.
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
StatusCode promoteToDataReady(unsigned int iAlgo, int si)
static GAUDI_API void setNumConcEvents(const std::size_t &nE)
static std::map< State, std::string > stateNames
StatusCode updateState(unsigned int iAlgo, State newState)