#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 > processConfiguration, std::string label, std::set< std::string > &unscheduledLabels, std::vector< std::string > &shouldBeUsedLabels)

AllWorkers const &	allWorkers () const

void	beginJob (ProductRegistry const &iRegistry, eventsetup::ESRecordsToProxyIndices const &)

void	beginStream (StreamID iID, StreamContext &streamContext)

void	endJob ()

void	endJob (ExceptionCollector &collector)

void	endStream (StreamID iID, StreamContext &streamContext)

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

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

template<typename T , typename U >
void	processOneOccurrence (typename T::TransitionInfoType &, StreamID, typename T::Context const topContext, U const context, bool cleaningUpAfterException=false)

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

void	resetAll ()

void	setupOnDemandSystem (EventTransitionInfo const &)

void	setupResolvers (Principal &principal)

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

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

	WorkerManager (WorkerManager &&)=default

Private Attributes
ExceptionToActionTable const *	actionTable_

AllWorkers	allWorkers_

void const *	lastSetupEventPrincipal_

UnscheduledCallProducer	unscheduled_

WorkerRegistry	workerReg_

Detailed Description

Definition at line 34 of file WorkerManager.h.

Member Typedef Documentation

◆ AllWorkers

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

Definition at line 36 of file WorkerManager.h.

Constructor & Destructor Documentation

◆ WorkerManager() [1/3]

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

Definition at line 17 of file WorkerManager.cc.

       : workerReg_(areg),
         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 24 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 88 of file WorkerManager.h.

88 { return *actionTable_; }

References actionTable_.

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

◆ addToAllWorkers()

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

Definition at line 123 of file WorkerManager.cc.

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

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

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

◆ addToUnscheduledWorkers()

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

Definition at line 42 of file WorkerManager.cc.

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

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().

◆ allWorkers()

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

inline

Definition at line 84 of file WorkerManager.h.

84 { return allWorkers_; }

References allWorkers_.

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

◆ beginJob()

void edm::WorkerManager::beginJob	(	ProductRegistry const &	iRegistry,
		eventsetup::ESRecordsToProxyIndices const &	iESIndices
	)

Definition at line 81 of file WorkerManager.cc.

                                                                                     {
     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);
       }
  
       for_all(allWorkers_, std::bind(&Worker::beginJob, std::placeholders::_1));
     }
   }

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

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

◆ beginStream()

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

Definition at line 109 of file WorkerManager.cc.

                                                                             {
     for (auto& worker : allWorkers_) {
       worker->beginStream(iID, streamContext);
     }
   }

References allWorkers_.

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

◆ endJob() [1/2]

void edm::WorkerManager::endJob ( void )

Definition at line 65 of file WorkerManager.cc.

                              {
     for (auto& worker : allWorkers_) {
       worker->endJob();
     }
   }

References allWorkers_.

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

◆ endJob() [2/2]

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

Definition at line 71 of file WorkerManager.cc.

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

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

◆ endStream()

void edm::WorkerManager::endStream	(	StreamID	iID,
		StreamContext &	streamContext
	)

Definition at line 115 of file WorkerManager.cc.

                                                                           {
     for (auto& worker : allWorkers_) {
       worker->endStream(iID, streamContext);
     }
   }

References allWorkers_.

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

◆ 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 33 of file WorkerManager.cc.

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

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

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

◆ processAccumulatorsAsync()

template<typename T >

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

Definition at line 146 of file WorkerManager.h.

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

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

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

◆ processOneOccurrence()

template<typename T , typename U >

void edm::WorkerManager::processOneOccurrence	(	typename T::TransitionInfoType &	info,
		StreamID	streamID,
		typename T::Context const *	topContext,
		U const *	context,
		bool	cleaningUpAfterException = `false`
	)

Definition at line 107 of file WorkerManager.h.

                                                                           {
     this->resetAll();
  
     auto waitTask = make_empty_waiting_task();
     waitTask->increment_ref_count();
     processOneOccurrenceAsync<T, U>(
         waitTask.get(), info, ServiceRegistry::instance().presentToken(), streamID, topContext, context);
     waitTask->wait_for_all();
     if (waitTask->exceptionPtr() != nullptr) {
       try {
         convertException::wrap([&]() { std::rethrow_exception(*(waitTask->exceptionPtr())); });
       } catch (cms::Exception& ex) {
         if (ex.context().empty()) {
           addContextAndPrintException(
               "Calling function WorkerManager::processOneOccurrence", ex, cleaningUpAfterException);
         } else {
           addContextAndPrintException("", ex, cleaningUpAfterException);
         }
         throw;
       }
     }
   }

References edm::addContextAndPrintException(), cms::Exception::context(), info(), edm::ServiceRegistry::instance(), edm::make_empty_waiting_task(), edm::ServiceRegistry::presentToken(), resetAll(), and edm::convertException::wrap().

Referenced by edm::SecondaryEventProvider::beginLuminosityBlock(), edm::SecondaryEventProvider::beginRun(), edm::SecondaryEventProvider::endLuminosityBlock(), and edm::SecondaryEventProvider::endRun().

◆ processOneOccurrenceAsync()

template<typename T , typename U >

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

Definition at line 135 of file WorkerManager.h.

                                                                   {
     //make sure the unscheduled items see this run or lumi transition
     unscheduled_.runNowAsync<T, U>(task, info, token, streamID, topContext, context);
   }

References info(), edm::UnscheduledCallProducer::runNowAsync(), TrackValidation_cff::task, unpackBuffers-CaloStage2::token, mitigatedMETSequence_cff::U, and unscheduled_.

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

◆ resetAll()

void edm::WorkerManager::resetAll ( )

Definition at line 121 of file WorkerManager.cc.

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

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

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

◆ setupOnDemandSystem()

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

Definition at line 138 of file WorkerManager.cc.

                                                                          {
     unscheduled_.setEventTransitionInfo(info);
   }

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

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

◆ setupResolvers()

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

Definition at line 129 of file WorkerManager.cc.

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

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

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

Member Data Documentation

◆ actionTable_

ExceptionToActionTable const* edm::WorkerManager::actionTable_

private

Definition at line 100 of file WorkerManager.h.

Referenced by actionTable(), and getWorker().

◆ allWorkers_

AllWorkers edm::WorkerManager::allWorkers_

private

Definition at line 101 of file WorkerManager.h.

Referenced by addToAllWorkers(), allWorkers(), beginJob(), beginStream(), endJob(), endStream(), resetAll(), and setupResolvers().

◆ lastSetupEventPrincipal_

void const* edm::WorkerManager::lastSetupEventPrincipal_

private

Definition at line 103 of file WorkerManager.h.

Referenced by setupResolvers().

◆ unscheduled_

UnscheduledCallProducer edm::WorkerManager::unscheduled_

private

Definition at line 102 of file WorkerManager.h.

Referenced by addToUnscheduledWorkers(), processAccumulatorsAsync(), processOneOccurrenceAsync(), setupOnDemandSystem(), and setupResolvers().

◆ workerReg_

WorkerRegistry edm::WorkerManager::workerReg_

private

Definition at line 99 of file WorkerManager.h.

Referenced by getWorker().

Public Types

Public Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ AllWorkers

Constructor & Destructor Documentation

◆ WorkerManager() [1/3]

◆ WorkerManager() [2/3]

◆ WorkerManager() [3/3]

Member Function Documentation

◆ actionTable()

◆ addToAllWorkers()

◆ addToUnscheduledWorkers()

◆ allWorkers()

◆ beginJob()

◆ beginStream()

◆ endJob() [1/2]

◆ endJob() [2/2]

◆ endStream()

◆ getWorker()

◆ processAccumulatorsAsync()

◆ processOneOccurrence()

◆ processOneOccurrenceAsync()

◆ resetAll()

◆ setupOnDemandSystem()

◆ setupResolvers()

Member Data Documentation

◆ actionTable_

◆ allWorkers_

◆ lastSetupEventPrincipal_

◆ unscheduled_

◆ workerReg_