#include <WorkerManager.h>

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

Public Member Functions
ExceptionToActionTable const &	actionTable () const

void	addToAllWorkers (Worker *w)

void	addToUnscheduledWorkers (ParameterSet &pset, ProductRegistry &preg, PreallocationConfiguration const *prealloc, std::shared_ptr< ProcessConfiguration const > processConfiguration, std::string label, std::set< std::string > &unscheduledLabels, std::vector< std::string > &shouldBeUsedLabels)

AllWorkers const &	allWorkers () const

void	beginJob (ProductRegistry const &iRegistry, eventsetup::ESRecordsToProductResolverIndices const &, ProcessBlockHelperBase const &, GlobalContext const &)

void	beginStream (StreamID, StreamContext const &)

void	deleteModuleIfExists (std::string const &moduleLabel)

void	endJob (ExceptionCollector &, GlobalContext const &)

void	endStream (StreamID, StreamContext const &, ExceptionCollector &, std::mutex &collectorMutex) noexcept

Worker *	getWorker (ParameterSet &pset, ProductRegistry &preg, PreallocationConfiguration const *prealloc, std::shared_ptr< ProcessConfiguration const > processConfiguration, std::string const &label)

template<typename T >
void	processAccumulatorsAsync (WaitingTaskHolder, typename T::TransitionInfoType const &, ServiceToken const &, StreamID, ParentContext const &, typename T::Context const *)

template<typename T , typename U >
void	processOneOccurrenceAsync (WaitingTaskHolder, typename T::TransitionInfoType &, ServiceToken const &, StreamID, typename T::Context const topContext, U const context) noexcept

void	resetAll ()

void	setupOnDemandSystem (EventTransitionInfo const &)

void	setupResolvers (Principal &principal)

AllWorkers const &	unscheduledWorkers () const

	WorkerManager (std::shared_ptr< ActivityRegistry > actReg, ExceptionToActionTable const &actions, ModuleTypeResolverMaker const *typeResolverMaker)

	WorkerManager (WorkerManager &&)=default

	WorkerManager (std::shared_ptr< ModuleRegistry > modReg, std::shared_ptr< ActivityRegistry > actReg, ExceptionToActionTable const &actions)

Private Attributes
ExceptionToActionTable const *	actionTable_

AllWorkers	allWorkers_

void const *	lastSetupEventPrincipal_

UnscheduledCallProducer	unscheduled_

WorkerRegistry	workerReg_

Detailed Description

Definition at line 31 of file WorkerManager.h.

Member Typedef Documentation

◆ AllWorkers

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

Definition at line 33 of file WorkerManager.h.

Constructor & Destructor Documentation

◆ WorkerManager() [1/3]

edm::WorkerManager::WorkerManager	(	std::shared_ptr< ActivityRegistry >	actReg,
		ExceptionToActionTable const &	actions,
		ModuleTypeResolverMaker const *	typeResolverMaker
	)

Definition at line 23 of file WorkerManager.cc.

       : workerReg_(areg, typeResolverMaker),
         actionTable_(&actions),
         allWorkers_(),
         unscheduled_(*areg),
         lastSetupEventPrincipal_(nullptr) {}  // WorkerManager::WorkerManager

◆ WorkerManager() [2/3]

edm::WorkerManager::WorkerManager ( WorkerManager && )

default

◆ WorkerManager() [3/3]

edm::WorkerManager::WorkerManager	(	std::shared_ptr< ModuleRegistry >	modReg,
		std::shared_ptr< ActivityRegistry >	actReg,
		ExceptionToActionTable const &	actions
	)

Definition at line 32 of file WorkerManager.cc.

       : workerReg_(areg, modReg),
         actionTable_(&actions),
         allWorkers_(),
         unscheduled_(*areg),
         lastSetupEventPrincipal_(nullptr) {}  // WorkerManager::WorkerManager

Member Function Documentation

◆ actionTable()

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

inline

Definition at line 87 of file WorkerManager.h.

References actionTable_.

Referenced by edm::StreamSchedule::actionTable().

87 { return *actionTable_; }

edm::WorkerManager::actionTable_

ExceptionToActionTable const * actionTable_

Definition: WorkerManager.h:99

◆ addToAllWorkers()

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

Definition at line 167 of file WorkerManager.cc.

References allWorkers_, edm::search_all(), and w().

Referenced by edm::StreamSchedule::addToAllWorkers(), addToUnscheduledWorkers(), and edm::StreamSchedule::StreamSchedule().

                                                {
     if (!search_all(allWorkers_, w)) {
       allWorkers_.push_back(w);
     }
   }

◆ addToUnscheduledWorkers()

void edm::WorkerManager::addToUnscheduledWorkers	(	ParameterSet &	pset,
		ProductRegistry &	preg,
		PreallocationConfiguration const *	prealloc,
		std::shared_ptr< ProcessConfiguration const >	processConfiguration,
		std::string	label,
		std::set< std::string > &	unscheduledLabels,
		std::vector< std::string > &	shouldBeUsedLabels
	)

Definition at line 60 of file WorkerManager.cc.

References addToAllWorkers(), edm::UnscheduledCallProducer::addWorker(), cms::cuda::assert(), getWorker(), edm::Worker::kFilter, kFilterType(), edm::Worker::kProducer, kProducerType(), label, edm::Worker::moduleType(), muonDTDigis_cfi::pset, AlCaHLTBitMon_QueryRunRegistry::string, and unscheduled_.

Referenced by edm::SecondaryEventProvider::SecondaryEventProvider(), and edm::StreamSchedule::StreamSchedule().

                                                                                         {
     //Need to
     // 1) create worker
     // 2) if it is a WorkerT<EDProducer>, add it to our list
     auto modType = pset.getParameter<std::string>("@module_edm_type");
     if (modType == kProducerType || modType == kFilterType) {
       Worker* newWorker = getWorker(pset, preg, prealloc, processConfiguration, label);
       assert(newWorker->moduleType() == Worker::kProducer || newWorker->moduleType() == Worker::kFilter);
       unscheduledLabels.insert(label);
       unscheduled_.addWorker(newWorker);
       //add to list so it gets reset each new event
       addToAllWorkers(newWorker);
     } else {
       shouldBeUsedLabels.push_back(label);
     }
   }

◆ allWorkers()

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

inline

Definition at line 82 of file WorkerManager.h.

References allWorkers_.

Referenced by edm::StreamSchedule::allWorkersBeginEnd(), edm::StreamSchedule::allWorkersLumisAndEvents(), edm::StreamSchedule::allWorkersRuns(), and edm::GlobalSchedule::processOneGlobalAsync().

82 { return allWorkers_; }

edm::WorkerManager::allWorkers_

AllWorkers allWorkers_

Definition: WorkerManager.h:100

◆ beginJob()

void edm::WorkerManager::beginJob	(	ProductRegistry const &	iRegistry,
		eventsetup::ESRecordsToProductResolverIndices const &	iESIndices,
		ProcessBlockHelperBase const &	processBlockHelperBase,
		GlobalContext const &	globalContext
	)

Definition at line 83 of file WorkerManager.cc.

References allWorkers_, CMS_SA_ALLOW, edm::InEvent, edm::InLumi, edm::InProcess, edm::InRun, SimL1EmulatorRepack_CalouGT_cff::processName, and edm::ProductRegistry::productLookup().

Referenced by edm::SecondaryEventProvider::beginJob().

                                                                    {
     std::exception_ptr exceptionPtr;
     CMS_SA_ALLOW try {
       auto const processBlockLookup = iRegistry.productLookup(InProcess);
       auto const runLookup = iRegistry.productLookup(InRun);
       auto const lumiLookup = iRegistry.productLookup(InLumi);
       auto const eventLookup = iRegistry.productLookup(InEvent);
       if (!allWorkers_.empty()) {
         auto const& processName = allWorkers_[0]->description()->processName();
         auto processBlockModuleToIndicies = processBlockLookup->indiciesForModulesInProcess(processName);
         auto runModuleToIndicies = runLookup->indiciesForModulesInProcess(processName);
         auto lumiModuleToIndicies = lumiLookup->indiciesForModulesInProcess(processName);
         auto eventModuleToIndicies = eventLookup->indiciesForModulesInProcess(processName);
         for (auto& worker : allWorkers_) {
           worker->updateLookup(InProcess, *processBlockLookup);
           worker->updateLookup(InRun, *runLookup);
           worker->updateLookup(InLumi, *lumiLookup);
           worker->updateLookup(InEvent, *eventLookup);
           worker->updateLookup(iESIndices);
           worker->resolvePutIndicies(InProcess, processBlockModuleToIndicies);
           worker->resolvePutIndicies(InRun, runModuleToIndicies);
           worker->resolvePutIndicies(InLumi, lumiModuleToIndicies);
           worker->resolvePutIndicies(InEvent, eventModuleToIndicies);
           worker->selectInputProcessBlocks(iRegistry, processBlockHelperBase);
         }
       }
     } catch (...) {
       exceptionPtr = std::current_exception();
     }
 
     for (auto& worker : allWorkers_) {
       CMS_SA_ALLOW try { worker->beginJob(globalContext); } catch (...) {
         if (!exceptionPtr) {
           exceptionPtr = std::current_exception();
         }
       }
     }
     if (exceptionPtr) {
       std::rethrow_exception(exceptionPtr);
     }
   }

◆ beginStream()

void edm::WorkerManager::beginStream	(	StreamID	streamID,
		StreamContext const &	streamContext
	)

Definition at line 138 of file WorkerManager.cc.

References allWorkers_, and CMS_SA_ALLOW.

Referenced by edm::SecondaryEventProvider::beginStream(), and edm::StreamSchedule::beginStream().

                                                                                        {
     std::exception_ptr exceptionPtr;
     for (auto& worker : allWorkers_) {
       CMS_SA_ALLOW try { worker->beginStream(streamID, streamContext); } catch (...) {
         if (!exceptionPtr) {
           exceptionPtr = std::current_exception();
         }
       }
     }
     if (exceptionPtr) {
       std::rethrow_exception(exceptionPtr);
     }
   }

◆ deleteModuleIfExists()

void edm::WorkerManager::deleteModuleIfExists ( std::string const & moduleLabel )

Definition at line 41 of file WorkerManager.cc.

References allWorkers_, edm::WorkerRegistry::deleteModule(), edm::WorkerRegistry::get(), HerwigMaxPtPartonFilter_cfi::moduleLabel, MatrixUtil::remove(), edm::UnscheduledCallProducer::removeWorker(), unscheduled_, and workerReg_.

Referenced by edm::StreamSchedule::deleteModule().

                                                                        {
     auto worker = workerReg_.get(moduleLabel);
     if (worker != nullptr) {
       auto eraseBeg = std::remove(allWorkers_.begin(), allWorkers_.end(), worker);
       allWorkers_.erase(eraseBeg, allWorkers_.end());
       unscheduled_.removeWorker(worker);
       workerReg_.deleteModule(moduleLabel);
     }
   }

◆ endJob()

void edm::WorkerManager::endJob	(	ExceptionCollector &	collector,
		GlobalContext const &	globalContext
	)

Definition at line 128 of file WorkerManager.cc.

References edm::ExceptionCollector::addException(), allWorkers_, and edm::convertException::wrap().

Referenced by edm::SecondaryEventProvider::endJob().

                                                                                               {
     for (auto& worker : allWorkers_) {
       try {
         convertException::wrap([&worker, &globalContext]() { worker->endJob(globalContext); });
       } catch (cms::Exception const& ex) {
         collector.addException(ex);
       }
     }
   }

◆ endStream()

void edm::WorkerManager::endStream	(	StreamID	streamID,
		StreamContext const &	streamContext,
		ExceptionCollector &	collector,
		std::mutex &	collectorMutex
	)

noexcept

Definition at line 152 of file WorkerManager.cc.

References CMS_SA_ALLOW.

Referenced by edm::SecondaryEventProvider::endStream(), and edm::StreamSchedule::endStream().

                                                                    {
     for (auto& worker : allWorkers_) {
       CMS_SA_ALLOW try { worker->endStream(streamID, streamContext); } catch (...) {
         std::exception_ptr exceptionPtr = std::current_exception();
         std::lock_guard<std::mutex> collectorLock(collectorMutex);
         collector.call([&exceptionPtr]() { std::rethrow_exception(exceptionPtr); });
       }
     }
   }

◆ getWorker()

Worker * edm::WorkerManager::getWorker	(	ParameterSet &	pset,
		ProductRegistry &	preg,
		PreallocationConfiguration const *	prealloc,
		std::shared_ptr< ProcessConfiguration const >	processConfiguration,
		std::string const &	label
	)

Definition at line 51 of file WorkerManager.cc.

References actionTable_, edm::WorkerRegistry::getWorker(), label, submitPVValidationJobs::params, muonDTDigis_cfi::pset, and workerReg_.

Referenced by addToUnscheduledWorkers().

                                                            {
     WorkerParams params(&pset, preg, prealloc, processConfiguration, *actionTable_);
     return workerReg_.getWorker(params, label);
   }

◆ processAccumulatorsAsync()

template<typename T >

void edm::WorkerManager::processAccumulatorsAsync	(	WaitingTaskHolder	task,
		typename T::TransitionInfoType const &	info,
		ServiceToken const &	token,
		StreamID	streamID,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 135 of file WorkerManager.h.

References visDQMUpload::context, info(), eostools::move(), edm::UnscheduledCallProducer::runAccumulatorsAsync(), TrackValidation_cff::task, unpackBuffers-CaloStage2::token, and unscheduled_.

Referenced by edm::StreamSchedule::processOneEventAsync().

                                                                                  {
     unscheduled_.runAccumulatorsAsync<T>(std::move(task), info, token, streamID, parentContext, context);
   }

◆ processOneOccurrenceAsync()

template<typename T , typename U >

void edm::WorkerManager::processOneOccurrenceAsync	(	WaitingTaskHolder	task,
		typename T::TransitionInfoType &	info,
		ServiceToken const &	token,
		StreamID	streamID,
		typename T::Context const *	topContext,
		U const *	context
	)

noexcept

Definition at line 106 of file WorkerManager.h.

References visDQMUpload::context, edm::Worker::doWorkNoPrefetchingAsync(), info(), ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, TrackValidation_cff::task, and unpackBuffers-CaloStage2::token.

Referenced by edm::StreamSchedule::processOneStreamAsync().

                                                                            {
     static_assert(!T::isEvent_);
 
     // Spawn them in reverse order. At least in the single threaded case that makes
     // them run in forward order (and more likely to with multiple threads).
     for (auto it = allWorkers_.rbegin(), itEnd = allWorkers_.rend(); it != itEnd; ++it) {
       Worker* worker = *it;
 
       ParentContext parentContext(context);
 
       // We do not need to run prefetching here because this only handles
       // stream begin/end transitions for runs and lumis. There are no products
       // put into the runs or lumis in stream transitions, so there can be
       // no data dependencies which require prefetching. Prefetching is
       // needed for global transitions, but they are run elsewhere.
       // (One exception, the SecondaryEventProvider (used for mixing) sends
       // global begin/end run/lumi transitions through here. They shouldn't
       // need prefetching either and for some years nothing has been using
       // that part of the code anyway...)
       worker->doWorkNoPrefetchingAsync<T>(task, info, token, streamID, parentContext, topContext);
     }
   }

◆ resetAll()

void edm::WorkerManager::resetAll ( )

Definition at line 165 of file WorkerManager.cc.

References allWorkers_, edm::for_all(), and edm::Worker::reset().

Referenced by edm::GlobalSchedule::processOneGlobalAsync(), edm::StreamSchedule::processOneStreamAsync(), and setupResolvers().

165 { for_all(allWorkers_, std::bind(&Worker::reset, std::placeholders::_1)); }

edm::WorkerManager::allWorkers_

AllWorkers allWorkers_

Definition: WorkerManager.h:100

edm::for_all

Func for_all(ForwardSequence &s, Func f)

wrapper for std::for_each

Definition: Algorithms.h:14

edm::Worker::reset

void reset()

Definition: Worker.h:190

◆ setupOnDemandSystem()

void edm::WorkerManager::setupOnDemandSystem ( EventTransitionInfo const & info )

Definition at line 182 of file WorkerManager.cc.

References info(), edm::UnscheduledCallProducer::setEventTransitionInfo(), and unscheduled_.

Referenced by edm::StreamSchedule::processOneEventAsync(), and edm::SecondaryEventProvider::setupPileUpEvent().

                                                                          {
     unscheduled_.setEventTransitionInfo(info);
   }

◆ setupResolvers()

void edm::WorkerManager::setupResolvers ( Principal & principal )

Definition at line 173 of file WorkerManager.cc.

References allWorkers_, edm::UnscheduledCallProducer::auxiliary(), DiMuonV_cfg::config, ecalHexDisplay_cfg::ep, lastSetupEventPrincipal_, resetAll(), and unscheduled_.

Referenced by edm::StreamSchedule::processOneEventAsync(), edm::GlobalSchedule::processOneGlobalAsync(), and edm::SecondaryEventProvider::setupPileUpEvent().

                                                   {
     this->resetAll();
     if (&ep != lastSetupEventPrincipal_) {
       UnscheduledConfigurator config(allWorkers_.begin(), allWorkers_.end(), &(unscheduled_.auxiliary()));
       ep.setupUnscheduled(config);
       lastSetupEventPrincipal_ = &ep;
     }
   }