CMS 3D CMS Logo

List of all members | Public Types | Public Member Functions | Private Attributes
edm::WorkerManager Class Reference

#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)
 
WorkergetWorker (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, EventSetupImpl 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, EventSetupImpl 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, EventSetupImpl const &eventSetup, ServiceToken const &token, StreamID streamID, typename T::Context const *topContext, U const *context)
 
void resetAll ()
 
void setupOnDemandSystem (Principal &principal, EventSetupImpl 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 33 of file WorkerManager.h.

Member Typedef Documentation

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

Definition at line 35 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.

18  : workerReg_(areg),
20  allWorkers_(),
21  unscheduled_(*areg),
22  lastSetupEventPrincipal_(nullptr) {} // WorkerManager::WorkerManager
roAction_t actions[nactions]
Definition: GenABIO.cc:181
UnscheduledCallProducer unscheduled_
AllWorkers allWorkers_
ExceptionToActionTable const * actionTable_
void const * lastSetupEventPrincipal_
WorkerRegistry workerReg_
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.

27  : workerReg_(areg, modReg),
29  allWorkers_(),
30  unscheduled_(*areg),
31  lastSetupEventPrincipal_(nullptr) {} // WorkerManager::WorkerManager
roAction_t actions[nactions]
Definition: GenABIO.cc:181
UnscheduledCallProducer unscheduled_
AllWorkers allWorkers_
ExceptionToActionTable const * actionTable_
void const * lastSetupEventPrincipal_
WorkerRegistry workerReg_

Member Function Documentation

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

Definition at line 89 of file WorkerManager.h.

References label, muonDTDigis_cfi::pset, and AlCaHLTBitMon_QueryRunRegistry::string.

89 { return *actionTable_; }
ExceptionToActionTable const * actionTable_
void edm::WorkerManager::addToAllWorkers ( Worker w)

Definition at line 119 of file WorkerManager.cc.

References allWorkers_, and edm::search_all().

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

119  {
120  if (!search_all(allWorkers_, w)) {
121  allWorkers_.push_back(w);
122  }
123  }
const double w
Definition: UKUtility.cc:23
AllWorkers allWorkers_
bool search_all(ForwardSequence const &s, Datum const &d)
Definition: Algorithms.h:36
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.

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

48  {
49  //Need to
50  // 1) create worker
51  // 2) if it is a WorkerT<EDProducer>, add it to our list
52  auto modType = pset.getParameter<std::string>("@module_edm_type");
53  if (modType == kProducerType || modType == kFilterType) {
54  Worker* newWorker = getWorker(pset, preg, prealloc, processConfiguration, label);
55  assert(newWorker->moduleType() == Worker::kProducer || newWorker->moduleType() == Worker::kFilter);
56  unscheduledLabels.insert(label);
57  unscheduled_.addWorker(newWorker);
58  //add to list so it gets reset each new event
59  addToAllWorkers(newWorker);
60  } else {
61  shouldBeUsedLabels.push_back(label);
62  }
63  }
static const std::string kFilterType("EDFilter")
UnscheduledCallProducer unscheduled_
char const * label
static const std::string kProducerType("EDProducer")
Worker * getWorker(ParameterSet &pset, ProductRegistry &preg, PreallocationConfiguration const *prealloc, std::shared_ptr< ProcessConfiguration const > processConfiguration, std::string const &label)
void addToAllWorkers(Worker *w)
AllWorkers const& edm::WorkerManager::allWorkers ( ) const
inline

Definition at line 85 of file WorkerManager.h.

References w.

85 { return allWorkers_; }
AllWorkers allWorkers_
void edm::WorkerManager::beginJob ( ProductRegistry const &  iRegistry,
eventsetup::ESRecordsToProxyIndices const &  iESIndices 
)

Definition at line 81 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().

82  {
83  auto const runLookup = iRegistry.productLookup(InRun);
84  auto const lumiLookup = iRegistry.productLookup(InLumi);
85  auto const eventLookup = iRegistry.productLookup(InEvent);
86  if (!allWorkers_.empty()) {
87  auto const& processName = allWorkers_[0]->description().processName();
88  auto runModuleToIndicies = runLookup->indiciesForModulesInProcess(processName);
89  auto lumiModuleToIndicies = lumiLookup->indiciesForModulesInProcess(processName);
90  auto eventModuleToIndicies = eventLookup->indiciesForModulesInProcess(processName);
91  for (auto& worker : allWorkers_) {
92  worker->updateLookup(InRun, *runLookup);
93  worker->updateLookup(InLumi, *lumiLookup);
94  worker->updateLookup(InEvent, *eventLookup);
95  worker->updateLookup(iESIndices);
96  worker->resolvePutIndicies(InRun, runModuleToIndicies);
97  worker->resolvePutIndicies(InLumi, lumiModuleToIndicies);
98  worker->resolvePutIndicies(InEvent, eventModuleToIndicies);
99  }
100 
101  for_all(allWorkers_, std::bind(&Worker::beginJob, std::placeholders::_1));
102  }
103  }
AllWorkers allWorkers_
Func for_all(ForwardSequence &s, Func f)
wrapper for std::for_each
Definition: Algorithms.h:14
void beginJob()
Definition: Worker.cc:284
void edm::WorkerManager::beginStream ( StreamID  iID,
StreamContext streamContext 
)

Definition at line 105 of file WorkerManager.cc.

References allWorkers_.

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

105  {
106  for (auto& worker : allWorkers_) {
107  worker->beginStream(iID, streamContext);
108  }
109  }
AllWorkers allWorkers_
void edm::WorkerManager::endJob ( void  )

Definition at line 65 of file WorkerManager.cc.

References allWorkers_.

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

65  {
66  for (auto& worker : allWorkers_) {
67  worker->endJob();
68  }
69  }
AllWorkers allWorkers_
void edm::WorkerManager::endJob ( ExceptionCollector collector)

Definition at line 71 of file WorkerManager.cc.

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

71  {
72  for (auto& worker : allWorkers_) {
73  try {
74  convertException::wrap([&]() { worker->endJob(); });
75  } catch (cms::Exception const& ex) {
76  collector.addException(ex);
77  }
78  }
79  }
AllWorkers allWorkers_
auto wrap(F iFunc) -> decltype(iFunc())
void edm::WorkerManager::endStream ( StreamID  iID,
StreamContext streamContext 
)

Definition at line 111 of file WorkerManager.cc.

References allWorkers_.

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

111  {
112  for (auto& worker : allWorkers_) {
113  worker->endStream(iID, streamContext);
114  }
115  }
AllWorkers allWorkers_
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.

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

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

37  {
38  WorkerParams params(&pset, preg, prealloc, processConfiguration, *actionTable_);
39  return workerReg_.getWorker(params, label);
40  }
ExceptionToActionTable const * actionTable_
char const * label
WorkerRegistry workerReg_
Worker * getWorker(WorkerParams const &p, std::string const &moduleLabel)
Retrieve the particular instance of the worker.
template<typename T >
void edm::WorkerManager::processAccumulatorsAsync ( WaitingTask task,
typename T::MyPrincipal const &  ep,
EventSetupImpl const &  es,
ServiceToken const &  token,
StreamID  streamID,
ParentContext const &  parentContext,
typename T::Context const *  context 
)

Definition at line 149 of file WorkerManager.h.

References TrackValidation_cff::task.

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

155  {
156  unscheduled_.runAccumulatorsAsync<T>(task, ep, es, token, streamID, parentContext, context);
157  }
UnscheduledCallProducer unscheduled_
void runAccumulatorsAsync(WaitingTask *task, typename T::MyPrincipal const &ep, EventSetupImpl const &es, ServiceToken const &token, StreamID streamID, ParentContext const &parentContext, typename T::Context const *context)
long double T
template<typename T , typename U >
void edm::WorkerManager::processOneOccurrence ( typename T::MyPrincipal &  principal,
EventSetupImpl const &  eventSetup,
StreamID  streamID,
typename T::Context const *  topContext,
U const *  context,
bool  cleaningUpAfterException = false 
)

Definition at line 108 of file WorkerManager.h.

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

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

113  {
114  this->resetAll();
115 
116  auto waitTask = make_empty_waiting_task();
117  waitTask->increment_ref_count();
118  processOneOccurrenceAsync<T, U>(
119  waitTask.get(), ep, es, ServiceRegistry::instance().presentToken(), streamID, topContext, context);
120  waitTask->wait_for_all();
121  if (waitTask->exceptionPtr() != nullptr) {
122  try {
123  convertException::wrap([&]() { std::rethrow_exception(*(waitTask->exceptionPtr())); });
124  } catch (cms::Exception& ex) {
125  if (ex.context().empty()) {
127  "Calling function WorkerManager::processOneOccurrence", ex, cleaningUpAfterException);
128  } else {
129  addContextAndPrintException("", ex, cleaningUpAfterException);
130  }
131  throw;
132  }
133  }
134  }
void addContextAndPrintException(char const *context, cms::Exception &ex, bool disablePrint)
ServiceToken presentToken() const
std::unique_ptr< edm::EmptyWaitingTask, waitingtask::TaskDestroyer > make_empty_waiting_task()
Create an EmptyWaitingTask which will properly be destroyed.
std::list< std::string > const & context() const
Definition: Exception.cc:147
static ServiceRegistry & instance()
auto wrap(F iFunc) -> decltype(iFunc())
template<typename T , typename U >
void edm::WorkerManager::processOneOccurrenceAsync ( WaitingTask task,
typename T::MyPrincipal &  principal,
EventSetupImpl const &  eventSetup,
ServiceToken const &  token,
StreamID  streamID,
typename T::Context const *  topContext,
U const *  context 
)

Definition at line 137 of file WorkerManager.h.

References TrackValidation_cff::task, and mitigatedMETSequence_cff::U.

143  {
144  //make sure the unscheduled items see this run or lumi transition
145  unscheduled_.runNowAsync<T, U>(task, ep, es, token, streamID, topContext, context);
146  }
UnscheduledCallProducer unscheduled_
void runNowAsync(WaitingTask *task, typename T::MyPrincipal &p, EventSetupImpl const &es, ServiceToken const &token, StreamID streamID, typename T::Context const *topContext, U const *context) const
long double T
void edm::WorkerManager::resetAll ( )

Definition at line 117 of file WorkerManager.cc.

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

Referenced by setupOnDemandSystem().

117 { for_all(allWorkers_, std::bind(&Worker::reset, std::placeholders::_1)); }
AllWorkers allWorkers_
Func for_all(ForwardSequence &s, Func f)
wrapper for std::for_each
Definition: Algorithms.h:14
void reset()
Definition: Worker.h:180
void edm::WorkerManager::setupOnDemandSystem ( Principal principal,
EventSetupImpl const &  es 
)

Member Data Documentation

ExceptionToActionTable const* edm::WorkerManager::actionTable_
private

Definition at line 101 of file WorkerManager.h.

Referenced by getWorker().

AllWorkers edm::WorkerManager::allWorkers_
private
void const* edm::WorkerManager::lastSetupEventPrincipal_
private

Definition at line 104 of file WorkerManager.h.

Referenced by setupOnDemandSystem().

UnscheduledCallProducer edm::WorkerManager::unscheduled_
private

Definition at line 103 of file WorkerManager.h.

Referenced by addToUnscheduledWorkers(), and setupOnDemandSystem().

WorkerRegistry edm::WorkerManager::workerReg_
private

Definition at line 100 of file WorkerManager.h.

Referenced by getWorker().