#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 &, eventsetup::ESRecordsToProxyIndices const &, ProcessBlockHelperBase const &)

void	deleteModule (std::string const &iLabel)
	Delete the module with label iLabel. More...

void	endJob (ExceptionCollector &collector)

std::vector< ModuleDescription const * >	getAllModuleDescriptions () 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	processOneGlobalAsync (WaitingTaskHolder holder, typename T::TransitionInfoType &, ServiceToken const &token, 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...

Private Attributes
std::shared_ptr< ActivityRegistry >	actReg_

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

ProcessContext const *	processContext_

std::vector< WorkerManager >	workerManagers_

Detailed Description

Definition at line 81 of file GlobalSchedule.h.

Member Typedef Documentation

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

Definition at line 84 of file GlobalSchedule.h.

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

Definition at line 83 of file GlobalSchedule.h.

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

Definition at line 85 of file GlobalSchedule.h.

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

Definition at line 86 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 22 of file GlobalSchedule.cc.

References edm::actions, actReg_, cms::cuda::assert(), extraWorkers_, edm::get_underlying(), edm::ParameterSet::getPSetForUpdate(), mps_fire::i, eostools::move(), edm::PreallocationConfiguration::numberOfLuminosityBlocks(), and workerManagers_.

       : actReg_(areg), processContext_(processContext) {
     workerManagers_.reserve(prealloc.numberOfLuminosityBlocks());
     for (unsigned int i = 0; i < prealloc.numberOfLuminosityBlocks(); ++i) {
       workerManagers_.emplace_back(modReg, areg, actions);
     }
     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
         for (auto& wm : workerManagers_) {
           wm.addToAllWorkers(wm.getWorker(*modpset, pregistry, &prealloc, processConfiguration, moduleLabel));
         }
       }
     }
     if (inserter) {
       inserter->doPreallocate(prealloc);
       for (auto& wm : workerManagers_) {
         auto results_inserter = WorkerPtr(new edm::WorkerT<TriggerResultInserter::ModuleType>(
             inserter, inserter->moduleDescription(), &actions));  // propagate_const<T> has no reset() function
         results_inserter->setActivityRegistry(actReg_);
         wm.addToAllWorkers(results_inserter.get());
         extraWorkers_.emplace_back(std::move(results_inserter));
       }
     }
 
     for (auto& pathStatusInserter : pathStatusInserters) {
       std::shared_ptr<PathStatusInserter> inserterPtr = get_underlying(pathStatusInserter);
       inserterPtr->doPreallocate(prealloc);
 
       for (auto& wm : workerManagers_) {
         WorkerPtr workerPtr(
             new edm::WorkerT<PathStatusInserter::ModuleType>(inserterPtr, inserterPtr->moduleDescription(), &actions));
         workerPtr->setActivityRegistry(actReg_);
         wm.addToAllWorkers(workerPtr.get());
         extraWorkers_.emplace_back(std::move(workerPtr));
       }
     }
 
     for (auto& endPathStatusInserter : endPathStatusInserters) {
       std::shared_ptr<EndPathStatusInserter> inserterPtr = get_underlying(endPathStatusInserter);
       inserterPtr->doPreallocate(prealloc);
       for (auto& wm : workerManagers_) {
         WorkerPtr workerPtr(new edm::WorkerT<EndPathStatusInserter::ModuleType>(
             inserterPtr, inserterPtr->moduleDescription(), &actions));
         workerPtr->setActivityRegistry(actReg_);
         wm.addToAllWorkers(workerPtr.get());
         extraWorkers_.emplace_back(std::move(workerPtr));
       }
     }
 
   }  // 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 153 of file GlobalSchedule.h.

References workerManagers_.

153 { return workerManagers_[0].actionTable(); }

edm::GlobalSchedule::workerManagers_

std::vector< WorkerManager > workerManagers_

Definition: GlobalSchedule.h:155

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

inline

returns the collection of pointers to workers

Definition at line 129 of file GlobalSchedule.h.

References workerManagers_.

Referenced by getAllModuleDescriptions().

129 { return workerManagers_[0].allWorkers(); }

edm::GlobalSchedule::workerManagers_

std::vector< WorkerManager > workerManagers_

Definition: GlobalSchedule.h:155

void edm::GlobalSchedule::beginJob	(	ProductRegistry const &	iRegistry,
		eventsetup::ESRecordsToProxyIndices const &	iESIndices,
		ProcessBlockHelperBase const &	processBlockHelperBase
	)

Definition at line 93 of file GlobalSchedule.cc.

References workerManagers_.

                                                                                       {
     workerManagers_[0].beginJob(iRegistry, iESIndices, processBlockHelperBase);
   }

void edm::GlobalSchedule::deleteModule ( std::string const & iLabel )

Delete the module with label iLabel.

Definition at line 117 of file GlobalSchedule.cc.

References workerManagers_.

                                                            {
     for (auto& wm : workerManagers_) {
       wm.deleteModuleIfExists(iLabel);
     }
   }

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

Definition at line 91 of file GlobalSchedule.cc.

References workerManagers_.

91 { workerManagers_[0].endJob(collector); }

edm::GlobalSchedule::workerManagers_

std::vector< WorkerManager > workerManagers_

Definition: GlobalSchedule.h:155

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 123 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->description();
       result.push_back(p);
     }
     return result;
   }

template<typename T >

void edm::GlobalSchedule::processOneGlobalAsync	(	WaitingTaskHolder	holder,
		typename T::TransitionInfoType &	transitionInfo,
		ServiceToken const &	token,
		bool	cleaningUpAfterException = `false`
	)

Definition at line 162 of file GlobalSchedule.h.

References actReg_, edm::addContextAndPrintException(), CMS_SA_ALLOW, cms::Exception::context(), edm::WaitingTaskHolder::doneWaiting(), edm::ExceptionFromThisContext, edm::WaitingTaskHolder::group(), edm::StreamID::invalidStreamID(), edm::ServiceWeakToken::lock(), edm::make_waiting_task(), processContext_, unpackBuffers-CaloStage2::token, workerManagers_, and edm::convertException::wrap().

                                                                             {
     auto const& principal = transitionInfo.principal();
 
     // Caught exception is propagated via WaitingTaskHolder
     CMS_SA_ALLOW try {
       //need the doneTask to own the memory
       auto globalContext = std::make_shared<GlobalContext>(T::makeGlobalContext(principal, processContext_));
 
       if (actReg_) {
         //Services may depend upon each other
         ServiceRegistry::Operate op(token);
         T::preScheduleSignal(actReg_.get(), globalContext.get());
       }
 
       ServiceWeakToken weakToken = token;
       auto doneTask = make_waiting_task(
           [this, iHolder, cleaningUpAfterException, globalContext, weakToken](std::exception_ptr const* iPtr) mutable {
             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() << " ";
                 }
                 ServiceRegistry::Operate op(weakToken.lock());
                 addContextAndPrintException(ost.str().c_str(), ex, cleaningUpAfterException);
                 excpt = std::current_exception();
               }
               if (actReg_) {
                 ServiceRegistry::Operate op(weakToken.lock());
                 actReg_->preGlobalEarlyTerminationSignal_(*globalContext, TerminationOrigin::ExceptionFromThisContext);
               }
             }
             if (actReg_) {
               // Caught exception is propagated via WaitingTaskHolder
               CMS_SA_ALLOW try {
                 ServiceRegistry::Operate op(weakToken.lock());
                 T::postScheduleSignal(actReg_.get(), globalContext.get());
               } catch (...) {
                 if (not excpt) {
                   excpt = std::current_exception();
                 }
               }
             }
             iHolder.doneWaiting(excpt);
           });
       WorkerManager& workerManager = workerManagers_[principal.index()];
       workerManager.resetAll();
 
       ParentContext parentContext(globalContext.get());
       //make sure the ProductResolvers know about their
       // workers to allow proper data dependency handling
       workerManager.setupResolvers(transitionInfo.principal());
 
       //make sure the task doesn't get run until all workers have beens started
       WaitingTaskHolder holdForLoop(*iHolder.group(), doneTask);
       auto& aw = workerManager.allWorkers();
       for (Worker* worker : boost::adaptors::reverse(aw)) {
         worker->doWorkAsync<T>(
             holdForLoop, transitionInfo, token, StreamID::invalidStreamID(), parentContext, globalContext.get());
       }
     } catch (...) {
       iHolder.doneWaiting(std::current_exception());
     }
   }

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 99 of file GlobalSchedule.cc.

References edm::Worker::beginJob(), newFWLiteAna::found, edm::maker::ModuleHolder::replaceModuleFor(), and workerManagers_.

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