#include <src/ParallelSequentialSchedulerSvc.h>

Inheritance diagram for SequentialTask:

Collaboration diagram for SequentialTask:

Public Member Functions
	SequentialTask (ISvcLocator svcLocator, EventContext eventContext, ParallelSequentialSchedulerSvc scheduler, IAlgResourcePool algPool)

virtual tbb::task *	execute ()

Private Attributes
SmartIF< ISvcLocator >	m_serviceLocator

EventContext *	m_eventContext

SmartIF< ParallelSequentialSchedulerSvc >	m_scheduler

SmartIF< IAlgResourcePool >	m_algPool

Detailed Description

Definition at line 116 of file ParallelSequentialSchedulerSvc.h.

Constructor & Destructor Documentation

SequentialTask::SequentialTask	(	ISvcLocator *	svcLocator,
		EventContext *	eventContext,
		ParallelSequentialSchedulerSvc *	scheduler,
		IAlgResourcePool *	algPool
	)

inline

Definition at line 118 of file ParallelSequentialSchedulerSvc.h.

                                            :
 
                    m_serviceLocator(svcLocator),
                    m_eventContext(eventContext),
                    m_scheduler(scheduler),
                    m_algPool(algPool){
 
   };

Member Function Documentation

tbb::task * SequentialTask::execute ( )

virtual

Definition at line 224 of file ParallelSequentialSchedulerSvc.cpp.

                                  {
 
   // Get the IProperty interface of the ApplicationMgr to pass it to RetCodeGuard
   const SmartIF<IProperty> appmgr(m_serviceLocator);
   SmartIF<IMessageSvc> messageSvc (m_serviceLocator);
   MsgStream log(messageSvc, "SequentialAlgoExecutionTask");
   log.activate();
 
   StatusCode sc;
 
   //initialize control algorithm states and decisions
   AlgsExecutionStates algStates(m_scheduler->m_algList.size(), messageSvc);
   const AlgResourcePool* algPool = dynamic_cast<const AlgResourcePool*>(m_scheduler->m_algResourcePool.get());
   std::vector<int> nodeDecisions(algPool->getExecutionFlowGraph()->getControlFlowNodeCounter(), -1);
 
   m_scheduler->m_controlFlow.updateEventState(algStates, nodeDecisions);
   m_scheduler->m_controlFlow.promoteToControlReadyState(algStates, nodeDecisions);
 
   //initialize data flow manager
   //DataFlowManager dataFlow(m_scheduler->m_algosDependencies);
 
   //intitialize context
   //    m_eventContext->m_thread_id = pthread_self();
   bool eventFailed = false;
   Gaudi::Hive::setCurrentContextId(m_eventContext->slot());
 
   // loop while algorithms are controlFlowReady and event has not failed
   while(!eventFailed && algStates.algsPresent(AlgsExecutionStates::State::CONTROLREADY) ){
 
     //std::cout << "[" << m_eventContext->evt() << "] algorithms left" << std::endl;
 
     //std::for_each(m_scheduler->m_algList.begin(), m_scheduler->m_algList.end(),
 
     //[&] (IAlgorithm* ialgorithm) {
     for(auto it = algStates.begin(AlgsExecutionStates::State::CONTROLREADY); it != algStates.end(AlgsExecutionStates::State::CONTROLREADY); ++it){
 
       uint algIndex = *it;
 
       std::string algName = m_scheduler->m_algname_vect[algIndex];
 
       //promote algorithm to data ready
       algStates.updateState(algIndex,AlgsExecutionStates::DATAREADY);
 
       //std::cout << "Running algorithm [" << algIndex << "] " << ialgorithm->name() << " for event " << m_eventContext->evt() << std::endl;
       log << MSG::DEBUG << "Running algorithm [" << algIndex << "] " << algName << " for event " << m_eventContext->evt() << endmsg;
 
       IAlgorithm* ialgoPtr=nullptr;
       sc = m_algPool->acquireAlgorithm(algName,ialgoPtr, true); //blocking call
 
       if(sc.isFailure() || ialgoPtr == nullptr){
         log << MSG::ERROR << "Could not acquire algorithm " << algName << endmsg;
         m_eventContext->setFail(true);
       } else { // we got an algorithm
 
         //promote algorithm to scheduled
         algStates.updateState(algIndex,AlgsExecutionStates::SCHEDULED);
 
         Algorithm* algoPtr = dynamic_cast<Algorithm*> (ialgoPtr); // DP: expose the setter of the context?
         algoPtr->setContext(m_eventContext);
 
         // Call the execute() method
         try {
           RetCodeGuard rcg(appmgr, Gaudi::ReturnCode::UnhandledException);
           sc = algoPtr->sysExecute();
           if (UNLIKELY(!sc.isSuccess()))  {
             log << MSG::WARNING  << "Execution of algorithm " << algName << " failed" << endmsg;
             eventFailed = true;
           }
           rcg.ignore(); // disarm the guard
         } catch ( const GaudiException& Exception ) {
           log << MSG::ERROR << ".executeEvent(): Exception with tag=" << Exception.tag()
                                                                                                        << " thrown by " << algName << endmsg;
           log << MSG::ERROR << Exception << endmsg;
         } catch ( const std::exception& Exception ) {
           log << MSG::FATAL << ".executeEvent(): Standard std::exception thrown by "
               << algName << endmsg;
           log << MSG::ERROR <<  Exception.what()  << endmsg;
         } catch(...) {
           log << MSG::FATAL << ".executeEvent(): UNKNOWN Exception thrown by "
               << algName << endmsg;
         }
 
         if(sc.isFailure()){
           eventFailed = true;
         }
 
         //std::cout << "Algorithm [" << algIndex << "] " << ialgorithm->name() << " for event " << m_eventContext->evt()
         //          << (eventFailed ? " failed" : " succeeded") << std::endl;
         log << MSG::DEBUG << "Algorithm [" << algIndex << "] " << algName << " for event " << m_eventContext->evt()
             << (eventFailed ? " failed" : " succeded") << endmsg;
 
         AlgsExecutionStates::State state;
         if (ialgoPtr->filterPassed()){
           state = AlgsExecutionStates::State::EVTACCEPTED;
         } else {
           state = AlgsExecutionStates::State::EVTREJECTED;
         }
 
         //std::cout << "Algorithm [" << algIndex << "] " << ialgorithm->name() << " for event " << m_eventContext->evt()
         //          << (ialgoPtr->filterPassed() ? " passed" : " rejected") << std::endl;
         log << MSG::DEBUG << "Algorithm [" << algIndex << "] " << algName << " for event " << m_eventContext->evt()
             << (ialgoPtr->filterPassed() ? " passed" : " rejected") << endmsg;
 
         sc = m_algPool->releaseAlgorithm(algName,ialgoPtr);
 
         algStates.updateState(algIndex,state);
 
         //just for debug: look at products -- not thread safe
         // Update the catalog: some new products may be there
         /*m_scheduler->m_whiteboard->selectStore(m_eventContext->slot()).ignore();
 
           // update prods in the dataflow
           // DP: Handles could be used. Just update what the algo wrote
           std::vector<std::string> new_products;
           m_scheduler->m_whiteboard->getNewDataObjects(new_products).ignore();
           for (const auto& new_product : new_products)
           std::cout << "Found in WB: " << new_product << std::endl;
           //dataFlow.updateDataObjectsCatalog(new_products);*/
       }
 
     }
     //);
 
     m_scheduler->m_controlFlow.updateEventState(algStates, nodeDecisions);
     m_scheduler->m_controlFlow.promoteToControlReadyState(algStates, nodeDecisions);
 
     if(eventFailed){
       m_eventContext->setFail(eventFailed);
       //std::cout << "ERROR: " << "event " << m_eventContext->evt() << " failed" << std::endl;
       break;
     }
 
     if(!algStates.algsPresent(AlgsExecutionStates::State::CONTROLREADY) && !algStates.allAlgsExecuted()){
       //std::cout << "WARNING: " << " not all algorithms executed for event " << m_eventContext->evt() << std::endl;
 
       /*std::for_each(m_scheduler->m_algList.begin(), m_scheduler->m_algList.end(),
 
           [&] (IAlgorithm* ialgorithm) {
             uint algIndex = m_scheduler->m_algname_index_map[ialgorithm->name()];
 
             if(AlgsExecutionStates::State::SCHEDULED >= algStates.algorithmState(algIndex))
               std::cout << "Event [" << m_eventContext->evt() << "] algorithm " << ialgorithm->name()
                 << " NOT executed" << std::endl;
 
           });*/
     }
   }
 
   m_scheduler->m_finishedEvents.push(m_eventContext);
 
   return nullptr;
 
 }