#include <GlobalSchedule.h>

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::ESRecordsToProductResolverIndices const &, ProcessBlockHelperBase const &, PathsAndConsumesOfModulesBase const &, ProcessContext 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 const > 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...

void	handleException (GlobalContext const *, ServiceWeakToken const &, bool cleaningUpAfterException, std::exception_ptr &)

template<typename T >
void	postScheduleSignal (GlobalContext const *, ServiceWeakToken const &, std::exception_ptr &)

template<typename T >
void	preScheduleSignal (GlobalContext const *, ServiceToken const &)

Private Attributes
std::shared_ptr< ActivityRegistry >	actReg_

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

unsigned int	numberOfConcurrentLumis_

unsigned int	numberOfConcurrentRuns_

ProcessContext const *	processContext_

std::vector< WorkerManager >	workerManagers_

Static Private Attributes
static constexpr unsigned int	numberOfConcurrentJobs_ = 1

static constexpr unsigned int	numberOfConcurrentProcessBlocks_ = 1

Detailed Description

Definition at line 49 of file GlobalSchedule.h.

Member Typedef Documentation

◆ AllWorkers

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

Definition at line 52 of file GlobalSchedule.h.

◆ vstring

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

Definition at line 51 of file GlobalSchedule.h.

◆ WorkerPtr

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

Definition at line 53 of file GlobalSchedule.h.

◆ Workers

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

Definition at line 54 of file GlobalSchedule.h.

Constructor & Destructor Documentation

◆ GlobalSchedule() [1/2]

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 const >	processConfiguration,
		ProcessContext const *	processContext
	)

Definition at line 24 of file GlobalSchedule.cc.

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

       : actReg_(areg),
         processContext_(processContext),
         numberOfConcurrentLumis_(prealloc.numberOfLuminosityBlocks()),
         numberOfConcurrentRuns_(prealloc.numberOfRuns()) {
     unsigned int nManagers = prealloc.numberOfLuminosityBlocks() + prealloc.numberOfRuns() +
                              numberOfConcurrentProcessBlocks_ + numberOfConcurrentJobs_;
     workerManagers_.reserve(nManagers);
     for (unsigned int i = 0; i < nManagers; ++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

◆ GlobalSchedule() [2/2]

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

delete

Member Function Documentation

◆ actionTable()

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

inlineprivate

returns the action table

Definition at line 105 of file GlobalSchedule.h.

References workerManagers_.

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

edm::GlobalSchedule::workerManagers_

std::vector< WorkerManager > workerManagers_

Definition: GlobalSchedule.h:118

◆ allWorkers()

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

inline

returns the collection of pointers to workers

Definition at line 101 of file GlobalSchedule.h.

References workerManagers_.

Referenced by getAllModuleDescriptions().

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

edm::GlobalSchedule::workerManagers_

std::vector< WorkerManager > workerManagers_

Definition: GlobalSchedule.h:118

◆ beginJob()

void edm::GlobalSchedule::beginJob	(	ProductRegistry const &	iRegistry,
		eventsetup::ESRecordsToProductResolverIndices const &	iESIndices,
		ProcessBlockHelperBase const &	processBlockHelperBase,
		PathsAndConsumesOfModulesBase const &	pathsAndConsumesOfModules,
		ProcessContext const &	processContext
	)

Definition at line 98 of file GlobalSchedule.cc.

References actReg_, CMS_SA_ALLOW, edm::exceptionContext(), edm::GlobalContext::kBeginJob, numberOfConcurrentLumis_, numberOfConcurrentProcessBlocks_, numberOfConcurrentRuns_, processContext_, workerManagers_, and edm::convertException::wrap().

                                                                       {
     GlobalContext globalContext(GlobalContext::Transition::kBeginJob, processContext_);
     unsigned int const managerIndex =
         numberOfConcurrentLumis_ + numberOfConcurrentRuns_ + numberOfConcurrentProcessBlocks_;
 
     std::exception_ptr exceptionPtr;
     CMS_SA_ALLOW try {
       try {
         convertException::wrap([this, &pathsAndConsumesOfModules, &processContext]() {
           actReg_->preBeginJobSignal_(pathsAndConsumesOfModules, processContext);
         });
       } catch (cms::Exception& ex) {
         exceptionContext(ex, globalContext, "Handling pre signal, likely in a service function");
         throw;
       }
       workerManagers_[managerIndex].beginJob(iRegistry, iESIndices, processBlockHelperBase, globalContext);
     } catch (...) {
       exceptionPtr = std::current_exception();
     }
 
     try {
       convertException::wrap([this]() { actReg_->postBeginJobSignal_(); });
     } catch (cms::Exception& ex) {
       if (!exceptionPtr) {
         exceptionContext(ex, globalContext, "Handling post signal, likely in a service function");
         exceptionPtr = std::current_exception();
       }
     }
     if (exceptionPtr) {
       std::rethrow_exception(exceptionPtr);
     }
   }

◆ deleteModule()

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

Delete the module with label iLabel.

Definition at line 191 of file GlobalSchedule.cc.

References workerManagers_.

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

◆ endJob()

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

Definition at line 135 of file GlobalSchedule.cc.

References actReg_, edm::ExceptionCollector::call(), CMS_SA_ALLOW, edm::exceptionContext(), edm::GlobalContext::kEndJob, numberOfConcurrentLumis_, numberOfConcurrentProcessBlocks_, numberOfConcurrentRuns_, processContext_, workerManagers_, and edm::convertException::wrap().

                                                            {
     GlobalContext globalContext(GlobalContext::Transition::kEndJob, processContext_);
     unsigned int const managerIndex =
         numberOfConcurrentLumis_ + numberOfConcurrentRuns_ + numberOfConcurrentProcessBlocks_;
 
     std::exception_ptr exceptionPtr;
     CMS_SA_ALLOW try {
       try {
         convertException::wrap([this]() { actReg_->preEndJobSignal_(); });
       } catch (cms::Exception& ex) {
         exceptionContext(ex, globalContext, "Handling pre signal, likely in a service function");
         throw;
       }
       workerManagers_[managerIndex].endJob(collector, globalContext);
     } catch (...) {
       exceptionPtr = std::current_exception();
     }
 
     try {
       convertException::wrap([this]() { actReg_->postEndJobSignal_(); });
     } catch (cms::Exception& ex) {
       if (!exceptionPtr) {
         exceptionContext(ex, globalContext, "Handling post signal, likely in a service function");
         exceptionPtr = std::current_exception();
       }
     }
     if (exceptionPtr) {
       collector.call([&exceptionPtr]() { std::rethrow_exception(exceptionPtr); });
     }
   }

◆ getAllModuleDescriptions()

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

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

                                                                                      {
     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;
   }

◆ handleException()

void edm::GlobalSchedule::handleException	(	GlobalContext const *	globalContext,
		ServiceWeakToken const &	weakToken,
		bool	cleaningUpAfterException,
		std::exception_ptr &	excpt
	)

private

Definition at line 208 of file GlobalSchedule.cc.

References actReg_, edm::addContextAndPrintException(), CMS_SA_ALLOW, cms::Exception::context(), edm::exceptionContext(), edm::ExceptionFromThisContext, edm::ServiceWeakToken::lock(), findAndChange::op, and edm::convertException::wrap().

Referenced by processOneGlobalAsync().

                                                                 {
     //add context information to the exception and print message
     try {
       convertException::wrap([&excpt]() { std::rethrow_exception(excpt); });
     } catch (cms::Exception& ex) {
       std::ostringstream ost;
       // In most cases the exception will already have context at this point,
       // but add some context here in those rare cases where it does not.
       if (ex.context().empty()) {
         exceptionContext(ost, *globalContext);
       }
       ServiceRegistry::Operate op(weakToken.lock());
       addContextAndPrintException(ost.str().c_str(), ex, cleaningUpAfterException);
       excpt = std::current_exception();
     }
     // We are already handling an earlier exception, so ignore it
     // if this signal results in another exception being thrown.
     CMS_SA_ALLOW try {
       if (actReg_) {
         ServiceRegistry::Operate op(weakToken.lock());
         actReg_->preGlobalEarlyTerminationSignal_(*globalContext, TerminationOrigin::ExceptionFromThisContext);
       }
     } catch (...) {
     }
   }

◆ postScheduleSignal()

template<typename T >

void edm::GlobalSchedule::postScheduleSignal	(	GlobalContext const *	globalContext,
		ServiceWeakToken const &	weakToken,
		std::exception_ptr &	excpt
	)

private

Definition at line 204 of file GlobalSchedule.h.

References actReg_, edm::exceptionContext(), edm::ServiceWeakToken::lock(), findAndChange::op, and edm::convertException::wrap().

                                                                    {
     if (actReg_) {
       try {
         convertException::wrap([this, &weakToken, globalContext]() {
           ServiceRegistry::Operate op(weakToken.lock());
           T::postScheduleSignal(actReg_.get(), globalContext);
         });
       } catch (cms::Exception& ex) {
         if (not excpt) {
           exceptionContext(ex, *globalContext, "Handling post signal, likely in a service function");
           excpt = std::current_exception();
         }
       }
     }
   }

◆ preScheduleSignal()

template<typename T >

void edm::GlobalSchedule::preScheduleSignal	(	GlobalContext const *	globalContext,
		ServiceToken const &	token
	)

private

Definition at line 191 of file GlobalSchedule.h.

References actReg_, edm::exceptionContext(), findAndChange::op, unpackBuffers-CaloStage2::token, and edm::convertException::wrap().

                                                                                                       {
     if (actReg_) {
       try {
         ServiceRegistry::Operate op(token);
         convertException::wrap([this, globalContext]() { T::preScheduleSignal(actReg_.get(), globalContext); });
       } catch (cms::Exception& ex) {
         exceptionContext(ex, *globalContext, "Handling pre signal, likely in a service function");
         throw;
       }
     }
   }

◆ processOneGlobalAsync()

template<typename T >

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

Definition at line 133 of file GlobalSchedule.h.

References edm::WorkerManager::allWorkers(), CMS_SA_ALLOW, ALPAKA_ACCELERATOR_NAMESPACE::brokenline::constexpr(), edm::WaitingTaskHolder::doneWaiting(), edm::WaitingTaskHolder::group(), handleException(), edm::InProcess, edm::InRun, edm::StreamID::invalidStreamID(), edm::make_waiting_task(), numberOfConcurrentLumis_, numberOfConcurrentRuns_, processContext_, edm::WorkerManager::resetAll(), groupFilesInBlocks::reverse, edm::WorkerManager::setupResolvers(), unpackBuffers-CaloStage2::token, and workerManagers_.

                                                                             {
     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_));
 
       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
               handleException(globalContext.get(), weakToken, cleaningUpAfterException, excpt);
             }
             postScheduleSignal<T>(globalContext.get(), weakToken, excpt);
             iHolder.doneWaiting(excpt);
           });
 
       //make sure the task doesn't get run until all workers have beens started
       WaitingTaskHolder holdForLoop(*iHolder.group(), doneTask);
 
       CMS_SA_ALLOW try {
         preScheduleSignal<T>(globalContext.get(), token);
 
         unsigned int managerIndex = principal.index();
         if constexpr (T::branchType_ == InRun) {
           managerIndex += numberOfConcurrentLumis_;
         } else if constexpr (T::branchType_ == InProcess) {
           managerIndex += (numberOfConcurrentLumis_ + numberOfConcurrentRuns_);
         }
         WorkerManager& workerManager = workerManagers_[managerIndex];
         workerManager.resetAll();
 
         ParentContext parentContext(globalContext.get());
         // make sure the ProductResolvers know about their
         // workers to allow proper data dependency handling
         workerManager.setupResolvers(transitionInfo.principal());
 
         auto& aw = workerManager.allWorkers();
         for (Worker* worker : boost::adaptors::reverse(aw)) {
           worker->doWorkAsync<T>(
               holdForLoop, transitionInfo, token, StreamID::invalidStreamID(), parentContext, globalContext.get());
         }
       } catch (...) {
         holdForLoop.doneWaiting(std::current_exception());
       }
     } catch (...) {
       iHolder.doneWaiting(std::current_exception());
     }
   }

◆ replaceModule()

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

References newFWLiteAna::found, edm::GlobalContext::kBeginJob, numberOfConcurrentLumis_, numberOfConcurrentProcessBlocks_, numberOfConcurrentRuns_, processContext_, edm::maker::ModuleHolder::replaceModuleFor(), and workerManagers_.

                                                                                      {
     Worker* found = nullptr;
     unsigned int const jobManagerIndex =
         numberOfConcurrentLumis_ + numberOfConcurrentRuns_ + numberOfConcurrentProcessBlocks_;
     unsigned int managerIndex = 0;
     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);
       if (managerIndex == jobManagerIndex) {
         GlobalContext globalContext(GlobalContext::Transition::kBeginJob, processContext_);
         found->beginJob(globalContext);
       }
       ++managerIndex;
     }
   }

◆ terminate()

bool edm::GlobalSchedule::terminate ( ) const

Return whether each output module has reached its maximum count.

Member Data Documentation

◆ actReg_

std::shared_ptr<ActivityRegistry> edm::GlobalSchedule::actReg_

private

Definition at line 119 of file GlobalSchedule.h.

Referenced by beginJob(), endJob(), GlobalSchedule(), handleException(), postScheduleSignal(), and preScheduleSignal().

◆ extraWorkers_

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

private

Definition at line 120 of file GlobalSchedule.h.

Referenced by GlobalSchedule().

◆ numberOfConcurrentJobs_

constexpr unsigned int edm::GlobalSchedule::numberOfConcurrentJobs_ = 1

staticprivate

Definition at line 129 of file GlobalSchedule.h.

Referenced by GlobalSchedule().

◆ numberOfConcurrentLumis_

unsigned int edm::GlobalSchedule::numberOfConcurrentLumis_

private

Definition at line 126 of file GlobalSchedule.h.

Referenced by beginJob(), endJob(), processOneGlobalAsync(), and replaceModule().

◆ numberOfConcurrentProcessBlocks_

constexpr unsigned int edm::GlobalSchedule::numberOfConcurrentProcessBlocks_ = 1

staticprivate

Definition at line 128 of file GlobalSchedule.h.

Referenced by beginJob(), endJob(), GlobalSchedule(), and replaceModule().

◆ numberOfConcurrentRuns_

unsigned int edm::GlobalSchedule::numberOfConcurrentRuns_

private

Definition at line 127 of file GlobalSchedule.h.

Referenced by beginJob(), endJob(), processOneGlobalAsync(), and replaceModule().

◆ processContext_

ProcessContext const* edm::GlobalSchedule::processContext_

private

Definition at line 121 of file GlobalSchedule.h.

Referenced by beginJob(), endJob(), processOneGlobalAsync(), and replaceModule().

◆ workerManagers_

std::vector<WorkerManager> edm::GlobalSchedule::workerManagers_

private

Definition at line 118 of file GlobalSchedule.h.

Referenced by actionTable(), allWorkers(), beginJob(), deleteModule(), endJob(), GlobalSchedule(), processOneGlobalAsync(), and replaceModule().

Public Types

Public Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Member Typedef Documentation

◆ AllWorkers

◆ vstring

◆ WorkerPtr

◆ Workers

Constructor & Destructor Documentation

◆ GlobalSchedule() [1/2]

◆ GlobalSchedule() [2/2]

Member Function Documentation

◆ actionTable()

◆ allWorkers()

◆ beginJob()

◆ deleteModule()

◆ endJob()

◆ getAllModuleDescriptions()

◆ handleException()

◆ postScheduleSignal()

◆ preScheduleSignal()

◆ processOneGlobalAsync()

◆ replaceModule()

◆ terminate()

Member Data Documentation

◆ actReg_

◆ extraWorkers_

◆ numberOfConcurrentJobs_

◆ numberOfConcurrentLumis_

◆ numberOfConcurrentProcessBlocks_

◆ numberOfConcurrentRuns_

◆ processContext_

◆ workerManagers_