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

void	beginStream (StreamID iID, StreamContext &streamContext)

void	deleteModuleIfExists (std::string const &moduleLabel)

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

void	resetAll ()

void	setupOnDemandSystem (EventTransitionInfo const &)

void	setupResolvers (Principal &principal)

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

	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 36 of file WorkerManager.h.

Member Typedef Documentation

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

Definition at line 38 of file WorkerManager.h.

Constructor & Destructor Documentation

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

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

default

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

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

inline

Definition at line 89 of file WorkerManager.h.

References actionTable_.

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

89 { return *actionTable_; }

edm::WorkerManager::actionTable_

ExceptionToActionTable const * actionTable_

Definition: WorkerManager.h:101

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

Definition at line 135 of file WorkerManager.cc.

References allWorkers_, and edm::search_all().

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

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

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

References addToAllWorkers(), edm::UnscheduledCallProducer::addWorker(), cms::cuda::assert(), edm::ParameterSet::getParameter(), getWorker(), edm::Worker::kFilter, kFilterType(), edm::Worker::kProducer, kProducerType(), edm::Worker::moduleType(), 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 const& edm::WorkerManager::allWorkers ( ) const

inline

Definition at line 85 of file WorkerManager.h.

References allWorkers_.

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

85 { return allWorkers_; }

edm::WorkerManager::allWorkers_

AllWorkers allWorkers_

Definition: WorkerManager.h:102

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

Definition at line 91 of file WorkerManager.cc.

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

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

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

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

Definition at line 121 of file WorkerManager.cc.

References allWorkers_.

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

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

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

Definition at line 33 of file WorkerManager.cc.

References allWorkers_, edm::WorkerRegistry::deleteModule(), edm::WorkerRegistry::get(), 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);
     }
   }

void edm::WorkerManager::endJob ( void )

Definition at line 75 of file WorkerManager.cc.

References allWorkers_.

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

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

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

Definition at line 81 of file WorkerManager.cc.

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

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

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

Definition at line 127 of file WorkerManager.cc.

References allWorkers_.

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

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

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

Definition at line 43 of file WorkerManager.cc.

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

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

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

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 119 of file WorkerManager.h.

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

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

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

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
	)

Definition at line 108 of file WorkerManager.h.

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

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

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

void edm::WorkerManager::resetAll ( )

Definition at line 133 of file WorkerManager.cc.

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

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

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

edm::WorkerManager::allWorkers_

AllWorkers allWorkers_

Definition: WorkerManager.h:102

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:178

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

Definition at line 150 of file WorkerManager.cc.

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

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

                                                                          {
     unscheduled_.setEventTransitionInfo(info);
   }

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

Definition at line 141 of file WorkerManager.cc.

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

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

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