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

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 task, typename T::MyPrincipal const &ep, EventSetup const &es, ServiceToken const &token, StreamID streamID, ParentContext const &parentContext, typename T::Context const context)

template<typename T , typename U >
void	processOneOccurrence (typename T::MyPrincipal &principal, EventSetup const &eventSetup, StreamID streamID, typename T::Context const topContext, U const context, bool cleaningUpAfterException=false)

template<typename T , typename U >
void	processOneOccurrenceAsync (WaitingTask task, typename T::MyPrincipal &principal, EventSetup const &eventSetup, ServiceToken const &token, StreamID streamID, typename T::Context const topContext, U const *context)

void	resetAll ()

void	setOnDemandProducts (ProductRegistry &pregistry, std::set< std::string > const &unscheduledLabels) const

void	setupOnDemandSystem (Principal &principal, EventSetup const &es)

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

Member Typedef Documentation

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

Definition at line 33 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 27 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 90 of file WorkerManager.h.

References actionTable_, getWorker(), diffTwoXMLs::label, muonDTDigis_cfi::pset, resetAll(), and AlCaHLTBitMon_QueryRunRegistry::string.

90 {return *actionTable_;}

edm::WorkerManager::actionTable_

ExceptionToActionTable const * actionTable_

Definition: WorkerManager.h:103

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

Definition at line 142 of file WorkerManager.cc.

References allWorkers_, and edm::search_all().

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

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

References addToAllWorkers(), edm::UnscheduledCallProducer::addWorker(), 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 86 of file WorkerManager.h.

References addToAllWorkers(), allWorkers_, and w.

86 {return allWorkers_;}

edm::WorkerManager::allWorkers_

AllWorkers allWorkers_

Definition: WorkerManager.h:104

void edm::WorkerManager::beginJob ( ProductRegistry const & iRegistry )

Definition at line 100 of file WorkerManager.cc.

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

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

                                                                {
     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 runModuleToIndicies = runLookup->indiciesForModulesInProcess(processName);
       auto lumiModuleToIndicies = lumiLookup->indiciesForModulesInProcess(processName);
       auto eventModuleToIndicies = eventLookup->indiciesForModulesInProcess(processName);
       for(auto& worker : allWorkers_) {
         worker->updateLookup(InRun,*runLookup);
         worker->updateLookup(InLumi,*lumiLookup);
         worker->updateLookup(InEvent,*eventLookup);
         worker->resolvePutIndicies(InRun,runModuleToIndicies);
         worker->resolvePutIndicies(InLumi,lumiModuleToIndicies);
         worker->resolvePutIndicies(InEvent,eventModuleToIndicies);
       }
       
       for_all(allWorkers_, std::bind(&Worker::beginJob, std::placeholders::_1));
     }
   }

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

Definition at line 123 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::endJob ( void )

Definition at line 80 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 86 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 130 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 38 of file WorkerManager.cc.

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

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

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

template<typename T >

void edm::WorkerManager::processAccumulatorsAsync	(	WaitingTask *	task,
		typename T::MyPrincipal const &	ep,
		EventSetup const &	es,
		ServiceToken const &	token,
		StreamID	streamID,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 154 of file WorkerManager.h.

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

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

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

template<typename T , typename U >

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

Definition at line 111 of file WorkerManager.h.

References edm::addContextAndPrintException(), cms::Exception::context(), 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().

                                                                        {
     this->resetAll();
 
     auto waitTask = make_empty_waiting_task();
     waitTask->increment_ref_count();
     processOneOccurrenceAsync<T,U>(waitTask.get(), ep, es, 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;
       }
     }
   }

template<typename T , typename U >

void edm::WorkerManager::processOneOccurrenceAsync	(	WaitingTask *	task,
		typename T::MyPrincipal &	principal,
		EventSetup const &	eventSetup,
		ServiceToken const &	token,
		StreamID	streamID,
		typename T::Context const *	topContext,
		U const *	context
	)

Definition at line 141 of file WorkerManager.h.

References edm::UnscheduledCallProducer::runNowAsync(), mitigatedMETSequence_cff::U, and unscheduled_.

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

void edm::WorkerManager::resetAll ( )

Definition at line 137 of file WorkerManager.cc.

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

Referenced by actionTable(), processOneOccurrence(), and setupOnDemandSystem().

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

void edm::WorkerManager::setOnDemandProducts	(	ProductRegistry &	pregistry,
		std::set< std::string > const &	unscheduledLabels
	)		const

Definition at line 70 of file WorkerManager.cc.

References edm::InEvent, parseEventContent::prod, and edm::ProductRegistry::productListUpdator().

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

                                                                                                                       {
     for(auto& prod : pregistry.productListUpdator()) {
       if(prod.second.produced() &&
           prod.second.branchType() == InEvent &&
           unscheduledLabels.end() != unscheduledLabels.find(prod.second.moduleLabel())) {
         prod.second.setOnDemand(true);
       }
     }
   }

void edm::WorkerManager::setupOnDemandSystem	(	Principal &	principal,
		EventSetup const &	es
	)

Definition at line 149 of file WorkerManager.cc.

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

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

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