#include <GlobalSchedule.h>

Classes
class	SendTerminationSignalIfException

Public Types
typedef std::vector< Worker * >	AllWorkers

typedef std::vector< std::string >	vstring

typedef std::shared_ptr< Worker >	WorkerPtr

typedef std::vector< Worker * >	Workers

Public Member Functions
AllWorkers const &	allWorkers () const
	returns the collection of pointers to workers More...

void	beginJob (ProductRegistry const &)

void	endJob (ExceptionCollector &collector)

std::vector< ModuleDescription const * >	getAllModuleDescriptions () const

void	getTriggerReport (TriggerReport &rep) const

	GlobalSchedule (std::shared_ptr< TriggerResultInserter > inserter, std::vector< edm::propagate_const< std::shared_ptr< PathStatusInserter >>> &pathStatusInserters, std::vector< edm::propagate_const< std::shared_ptr< EndPathStatusInserter >>> &endPathStatusInserters, std::shared_ptr< ModuleRegistry > modReg, std::vector< std::string > const &modulesToUse, ParameterSet &proc_pset, ProductRegistry &pregistry, PreallocationConfiguration const &prealloc, ExceptionToActionTable const &actions, std::shared_ptr< ActivityRegistry > areg, std::shared_ptr< ProcessConfiguration > processConfiguration, ProcessContext const *processContext)

	GlobalSchedule (GlobalSchedule const &)=delete

template<typename T >
void	processOneGlobal (typename T::MyPrincipal &principal, EventSetup const &eventSetup, bool cleaningUpAfterException=false)

template<typename T >
void	processOneGlobalAsync (WaitingTaskHolder holder, typename T::MyPrincipal &principal, EventSetup const &eventSetup, bool cleaningUpAfterException=false)

void	replaceModule (maker::ModuleHolder *iMod, std::string const &iLabel)
	clone the type of module with label iLabel but configure with iPSet. More...

bool	terminate () const
	Return whether each output module has reached its maximum count. More...

Private Member Functions
ExceptionToActionTable const &	actionTable () const
	returns the action table More...

void	addToAllWorkers (Worker *w)

template<typename T >
void	runNow (typename T::MyPrincipal const &p, EventSetup const &es, GlobalContext const *context)

Private Attributes
std::shared_ptr< ActivityRegistry >	actReg_

std::vector< edm::propagate_const< WorkerPtr > >	endPathStatusInserterWorkers_

std::vector< edm::propagate_const< WorkerPtr > >	pathStatusInserterWorkers_

ProcessContext const *	processContext_

edm::propagate_const< WorkerPtr >	results_inserter_

WorkerManager	workerManager_

Detailed Description

Definition at line 75 of file GlobalSchedule.h.

Member Typedef Documentation

typedef std::vector<Worker*> edm::GlobalSchedule::AllWorkers

Definition at line 78 of file GlobalSchedule.h.

typedef std::vector<std::string> edm::GlobalSchedule::vstring

Definition at line 77 of file GlobalSchedule.h.

typedef std::shared_ptr<Worker> edm::GlobalSchedule::WorkerPtr

Definition at line 79 of file GlobalSchedule.h.

typedef std::vector<Worker*> edm::GlobalSchedule::Workers

Definition at line 80 of file GlobalSchedule.h.

Constructor & Destructor Documentation

edm::GlobalSchedule::GlobalSchedule	(	std::shared_ptr< TriggerResultInserter >	inserter,
		std::vector< edm::propagate_const< std::shared_ptr< PathStatusInserter >>> &	pathStatusInserters,
		std::vector< edm::propagate_const< std::shared_ptr< EndPathStatusInserter >>> &	endPathStatusInserters,
		std::shared_ptr< ModuleRegistry >	modReg,
		std::vector< std::string > const &	modulesToUse,
		ParameterSet &	proc_pset,
		ProductRegistry &	pregistry,
		PreallocationConfiguration const &	prealloc,
		ExceptionToActionTable const &	actions,
		std::shared_ptr< ActivityRegistry >	areg,
		std::shared_ptr< ProcessConfiguration >	processConfiguration,
		ProcessContext const *	processContext
	)

Definition at line 21 of file GlobalSchedule.cc.

References actions, actReg_, addToAllWorkers(), endPathStatusInserterWorkers_, edm::propagate_const< T >::get(), edm::get_underlying(), edm::ParameterSet::getPSetForUpdate(), edm::WorkerManager::getWorker(), pathStatusInserterWorkers_, results_inserter_, and workerManager_.

                                                                        :
     workerManager_(modReg,areg,actions),
     actReg_(areg),
     processContext_(processContext)
   {
     for (auto const& moduleLabel : iModulesToUse) {
       bool isTracked;
       ParameterSet* modpset = proc_pset.getPSetForUpdate(moduleLabel, isTracked);
       if (modpset != nullptr) { // It will be null for PathStatusInserters, it should
                                 // be impossible to be null for anything else
         assert(isTracked);
 
         //side effect keeps this module around
         addToAllWorkers(workerManager_.getWorker(*modpset, pregistry, &prealloc,processConfiguration, moduleLabel));
       }
     }
     if(inserter) {
       results_inserter_ = WorkerPtr(new edm::WorkerT<TriggerResultInserter::ModuleType>(inserter, inserter->moduleDescription(), &actions)); // propagate_const<T> has no reset() function
       inserter->doPreallocate(prealloc);
       results_inserter_->setActivityRegistry(actReg_);
       addToAllWorkers(results_inserter_.get());
     }
 
     for(auto & pathStatusInserter : pathStatusInserters) {
       std::shared_ptr<PathStatusInserter> inserterPtr = get_underlying(pathStatusInserter);
       WorkerPtr workerPtr(new edm::WorkerT<PathStatusInserter::ModuleType>(inserterPtr,
                                                                            inserterPtr->moduleDescription(),
                                                                            &actions));
       pathStatusInserterWorkers_.emplace_back(workerPtr);
       inserterPtr->doPreallocate(prealloc);
       workerPtr->setActivityRegistry(actReg_);
       addToAllWorkers(workerPtr.get());
     }
 
     for(auto & endPathStatusInserter : endPathStatusInserters) {
       std::shared_ptr<EndPathStatusInserter> inserterPtr = get_underlying(endPathStatusInserter);
       WorkerPtr workerPtr(new edm::WorkerT<EndPathStatusInserter::ModuleType>(inserterPtr,
                                                                               inserterPtr->moduleDescription(),
                                                                               &actions));
       endPathStatusInserterWorkers_.emplace_back(workerPtr);
       inserterPtr->doPreallocate(prealloc);
       workerPtr->setActivityRegistry(actReg_);
       addToAllWorkers(workerPtr.get());
     }
 
   } // GlobalSchedule::GlobalSchedule

edm::GlobalSchedule::GlobalSchedule ( GlobalSchedule const & )

delete

Member Function Documentation

ExceptionToActionTable const& edm::GlobalSchedule::actionTable ( ) const

inlineprivate

returns the action table

Definition at line 162 of file GlobalSchedule.h.

References w.

                                                       {
       return workerManager_.actionTable();
     }

void edm::GlobalSchedule::addToAllWorkers ( Worker * w )

private

Definition at line 117 of file GlobalSchedule.cc.

References edm::WorkerManager::addToAllWorkers(), and workerManager_.

Referenced by GlobalSchedule().

                                            {
     workerManager_.addToAllWorkers(w);
   }

AllWorkers const& edm::GlobalSchedule::allWorkers ( ) const

inline

returns the collection of pointers to workers

Definition at line 129 of file GlobalSchedule.h.

Referenced by getAllModuleDescriptions(), and replaceModule().

                                          {
       return workerManager_.allWorkers();
     }

void edm::GlobalSchedule::beginJob ( ProductRegistry const & iRegistry )

Definition at line 83 of file GlobalSchedule.cc.

References edm::WorkerManager::beginJob(), and workerManager_.

                                                                 {
     workerManager_.beginJob(iRegistry);
   }

void edm::GlobalSchedule::endJob ( ExceptionCollector & collector )

Definition at line 79 of file GlobalSchedule.cc.

References edm::WorkerManager::endJob(), and workerManager_.

                                                             {
     workerManager_.endJob(collector);
   }

std::vector< ModuleDescription const * > edm::GlobalSchedule::getAllModuleDescriptions ( ) const

Return a vector allowing const access to all the ModuleDescriptions for this GlobalSchedule. *** N.B. *** Ownership of the ModuleDescriptions is not *** passed to the caller. Do not call delete on these *** pointers!

Definition at line 105 of file GlobalSchedule.cc.

References allWorkers(), AlCaHLTBitMon_ParallelJobs::p, mps_fire::result, and findQualityFiles::size.

                                                  {
     std::vector<ModuleDescription const*> result;
     result.reserve(allWorkers().size());
 
     for (auto const& worker : allWorkers()) {
       ModuleDescription const* p = worker->descPtr();
       result.push_back(p);
     }
     return result;
   }

void edm::GlobalSchedule::getTriggerReport ( TriggerReport & rep ) const

Return the trigger report information on paths, modules-in-path, modules-in-endpath, and modules.

template<typename T >

void edm::GlobalSchedule::processOneGlobal	(	typename T::MyPrincipal &	principal,
		EventSetup const &	eventSetup,
		bool	cleaningUpAfterException = `false`
	)

Definition at line 180 of file GlobalSchedule.h.

References edm::addContextAndPrintException(), cms::Exception::context(), edm::StreamID::invalidStreamID(), and edm::convertException::wrap().

                                                                 {
     GlobalContext globalContext = T::makeGlobalContext(ep, processContext_);
 
     GlobalScheduleSignalSentry<T> sentry(actReg_.get(), &globalContext);
     
     SendTerminationSignalIfException terminationSentry(actReg_.get(), &globalContext);
 
     //If we are in an end transition, we need to reset failed items since they might
     // be set this time around
     if( not T::begin_) {
       ep.resetFailedFromThisProcess();
     }
     // This call takes care of the unscheduled processing.
     workerManager_.processOneOccurrence<T>(ep, es, StreamID::invalidStreamID(), &globalContext, &globalContext, cleaningUpAfterException);
 
     try {
       convertException::wrap([&]() {
         runNow<T>(ep,es,&globalContext);
       });
     }
     catch(cms::Exception& ex) {
       if (ex.context().empty()) {
         addContextAndPrintException("Calling function GlobalSchedule::processOneGlobal", ex, cleaningUpAfterException);
       } else {
         addContextAndPrintException("", ex, cleaningUpAfterException);
       }
       throw;
     }
     terminationSentry.completedSuccessfully();
     
     //If we got here no other exception has happened so we can propogate any Service related exceptions
     sentry.allowThrow();
   }

template<typename T >

void edm::GlobalSchedule::processOneGlobalAsync	(	WaitingTaskHolder	holder,
		typename T::MyPrincipal &	principal,
		EventSetup const &	eventSetup,
		bool	cleaningUpAfterException = `false`
	)

Definition at line 218 of file GlobalSchedule.h.

References edm::addContextAndPrintException(), cms::Exception::context(), edm::WaitingTaskHolder::doneWaiting(), edm::ExceptionFromThisContext, edm::ServiceRegistry::instance(), edm::StreamID::invalidStreamID(), edm::make_waiting_task(), edm::ServiceRegistry::presentToken(), groupFilesInBlocks::reverse, and edm::convertException::wrap().

                                                                        {
     ServiceToken token = ServiceRegistry::instance().presentToken();
     
     //need the doneTask to own the memory
     auto globalContext = std::make_shared<GlobalContext>(T::makeGlobalContext(ep, processContext_));
     
     if(actReg_) {
       T::preScheduleSignal(actReg_.get(), globalContext.get());
     }
     
     
     //If we are in an end transition, we need to reset failed items since they might
     // be set this time around
     if( not T::begin_) {
       ep.resetFailedFromThisProcess();
     }
     
     auto doneTask = make_waiting_task(tbb::task::allocate_root(),
                                       [this,iHolder, cleaningUpAfterException, globalContext, token](std::exception_ptr const* iPtr) mutable
                                       {
                                         ServiceRegistry::Operate op(token);
                                         std::exception_ptr excpt;
                                         if(iPtr) {
                                           excpt = *iPtr;
                                           //add context information to the exception and print message
                                           try {
                                             convertException::wrap([&]() {
                                               std::rethrow_exception(excpt);
                                             });
                                           } catch(cms::Exception& ex) {
                                             //TODO: should add the transition type info
                                             std::ostringstream ost;
                                             if(ex.context().empty()) {
                                               ost<<"Processing "<<T::transitionName()<<" ";
                                             }
                                             addContextAndPrintException(ost.str().c_str(), ex, cleaningUpAfterException);
                                             excpt = std::current_exception();
                                           }
                                           if(actReg_) {
                                             actReg_->preGlobalEarlyTerminationSignal_(*globalContext,TerminationOrigin::ExceptionFromThisContext);
                                           }
                                         }
                                         if(actReg_) {
                                           try {
                                             T::postScheduleSignal(actReg_.get(), globalContext.get());
                                           } catch(...) {
                                             if(not excpt) {
                                               excpt = std::current_exception();
                                             }
                                           }
                                         }
                                         iHolder.doneWaiting(excpt);
                                         
                                       });
     workerManager_.resetAll();
     
     ParentContext parentContext(globalContext.get());
     //make sure the ProductResolvers know about their
     // workers to allow proper data dependency handling
     workerManager_.setupOnDemandSystem(ep,es);
     
     //make sure the task doesn't get run until all workers have beens started
     WaitingTaskHolder holdForLoop(doneTask);
     for(auto& worker: boost::adaptors::reverse((allWorkers()))) {
       worker->doWorkAsync<T>(doneTask,ep,es,StreamID::invalidStreamID(),parentContext,globalContext.get());
     }
 
   }

void edm::GlobalSchedule::replaceModule	(	maker::ModuleHolder *	iMod,
		std::string const &	iLabel
	)

clone the type of module with label iLabel but configure with iPSet.

Definition at line 87 of file GlobalSchedule.cc.

References allWorkers(), edm::Worker::beginJob(), runEdmFileComparison::found, and edm::maker::ModuleHolder::replaceModuleFor().

                                                               {
     Worker* found = nullptr;
     for (auto const& worker : allWorkers()) {
       if (worker->description().moduleLabel() == iLabel) {
         found = worker;
         break;
       }
     }
     if (nullptr == found) {
       return;
     }
     
     iMod->replaceModuleFor(found);
     found->beginJob();
   }

template<typename T >

void edm::GlobalSchedule::runNow	(	typename T::MyPrincipal const &	p,
		EventSetup const &	es,
		GlobalContext const *	context
	)

private

Definition at line 292 of file GlobalSchedule.h.

References cms::Exception::addContext(), cms::Exception::context(), edm::StreamID::invalidStreamID(), and AlCaHLTBitMon_ParallelJobs::p.

                                             {
     //do nothing for event since we will run when requested
     for(auto & worker: allWorkers()) {
       try {
         ParentContext parentContext(context);
         worker->doWork<T>(p, es,StreamID::invalidStreamID(), parentContext, context);
       }
       catch (cms::Exception & ex) {
         if(ex.context().empty()) {
           std::ostringstream ost;
           ost << "Processing " <<T::transitionName()<<" "<< p.id();
           ex.addContext(ost.str());
         }
         throw;
       }
     }
   }