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 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::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)
 
WorkergetWorker (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)
 
AllWorkers const & unscheduledWorkers () const
 
 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

◆ AllWorkers

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

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

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_

◆ 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.

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

◆ actionTable()

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

Definition at line 90 of file WorkerManager.h.

References actionTable_.

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

90 { return *actionTable_; }
ExceptionToActionTable const * actionTable_

◆ addToAllWorkers()

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

Definition at line 135 of file WorkerManager.cc.

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

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

135  {
136  if (!search_all(allWorkers_, w)) {
137  allWorkers_.push_back(w);
138  }
139  }
T w() const
AllWorkers allWorkers_
bool search_all(ForwardSequence const &s, Datum const &d)
Definition: Algorithms.h:36

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

58  {
59  //Need to
60  // 1) create worker
61  // 2) if it is a WorkerT<EDProducer>, add it to our list
62  auto modType = pset.getParameter<std::string>("@module_edm_type");
63  if (modType == kProducerType || modType == kFilterType) {
64  Worker* newWorker = getWorker(pset, preg, prealloc, processConfiguration, label);
65  assert(newWorker->moduleType() == Worker::kProducer || newWorker->moduleType() == Worker::kFilter);
66  unscheduledLabels.insert(label);
67  unscheduled_.addWorker(newWorker);
68  //add to list so it gets reset each new event
69  addToAllWorkers(newWorker);
70  } else {
71  shouldBeUsedLabels.push_back(label);
72  }
73  }
static const std::string kFilterType("EDFilter")
UnscheduledCallProducer unscheduled_
assert(be >=bs)
Worker * getWorker(ParameterSet &pset, ProductRegistry &preg, PreallocationConfiguration const *prealloc, std::shared_ptr< ProcessConfiguration const > processConfiguration, std::string const &label)
char const * label
static const std::string kProducerType("EDProducer")
void addToAllWorkers(Worker *w)

◆ allWorkers()

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

Definition at line 85 of file WorkerManager.h.

References allWorkers_.

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

85 { return allWorkers_; }
AllWorkers allWorkers_

◆ beginJob()

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, SimL1EmulatorRepack_CalouGT_cff::processName, and edm::ProductRegistry::productLookup().

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

93  {
94  auto const processBlockLookup = iRegistry.productLookup(InProcess);
95  auto const runLookup = iRegistry.productLookup(InRun);
96  auto const lumiLookup = iRegistry.productLookup(InLumi);
97  auto const eventLookup = iRegistry.productLookup(InEvent);
98  if (!allWorkers_.empty()) {
99  auto const& processName = allWorkers_[0]->description()->processName();
100  auto processBlockModuleToIndicies = processBlockLookup->indiciesForModulesInProcess(processName);
101  auto runModuleToIndicies = runLookup->indiciesForModulesInProcess(processName);
102  auto lumiModuleToIndicies = lumiLookup->indiciesForModulesInProcess(processName);
103  auto eventModuleToIndicies = eventLookup->indiciesForModulesInProcess(processName);
104  for (auto& worker : allWorkers_) {
105  worker->updateLookup(InProcess, *processBlockLookup);
106  worker->updateLookup(InRun, *runLookup);
107  worker->updateLookup(InLumi, *lumiLookup);
108  worker->updateLookup(InEvent, *eventLookup);
109  worker->updateLookup(iESIndices);
110  worker->resolvePutIndicies(InProcess, processBlockModuleToIndicies);
111  worker->resolvePutIndicies(InRun, runModuleToIndicies);
112  worker->resolvePutIndicies(InLumi, lumiModuleToIndicies);
113  worker->resolvePutIndicies(InEvent, eventModuleToIndicies);
114  worker->selectInputProcessBlocks(iRegistry, processBlockHelperBase);
115  }
116 
117  for_all(allWorkers_, std::bind(&Worker::beginJob, std::placeholders::_1));
118  }
119  }
AllWorkers allWorkers_
Func for_all(ForwardSequence &s, Func f)
wrapper for std::for_each
Definition: Algorithms.h:14
void beginJob()
Definition: Worker.cc:273

◆ beginStream()

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

121  {
122  for (auto& worker : allWorkers_) {
123  worker->beginStream(iID, streamContext);
124  }
125  }
AllWorkers allWorkers_

◆ deleteModuleIfExists()

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

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

33  {
34  auto worker = workerReg_.get(moduleLabel);
35  if (worker != nullptr) {
36  auto eraseBeg = std::remove(allWorkers_.begin(), allWorkers_.end(), worker);
37  allWorkers_.erase(eraseBeg, allWorkers_.end());
38  unscheduled_.removeWorker(worker);
40  }
41  }
void removeWorker(Worker const *worker)
UnscheduledCallProducer unscheduled_
AllWorkers allWorkers_
Worker const * get(std::string const &moduleLabel) const
void deleteModule(std::string const &moduleLabel)
Deletes the module of the Worker, but the Worker continues to exist.
def remove(d, key, TELL=False)
Definition: MatrixUtil.py:223
WorkerRegistry workerReg_

◆ endJob() [1/2]

void edm::WorkerManager::endJob ( void  )

Definition at line 75 of file WorkerManager.cc.

References allWorkers_.

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

75  {
76  for (auto& worker : allWorkers_) {
77  worker->endJob();
78  }
79  }
AllWorkers allWorkers_

◆ endJob() [2/2]

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

Definition at line 81 of file WorkerManager.cc.

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

81  {
82  for (auto& worker : allWorkers_) {
83  try {
84  convertException::wrap([&]() { worker->endJob(); });
85  } catch (cms::Exception const& ex) {
86  collector.addException(ex);
87  }
88  }
89  }
AllWorkers allWorkers_
auto wrap(F iFunc) -> decltype(iFunc())

◆ endStream()

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

127  {
128  for (auto& worker : allWorkers_) {
129  worker->endStream(iID, streamContext);
130  }
131  }
AllWorkers allWorkers_

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

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

Referenced by addToUnscheduledWorkers().

47  {
48  WorkerParams params(&pset, preg, prealloc, processConfiguration, *actionTable_);
50  }
ExceptionToActionTable const * actionTable_
char const * label
WorkerRegistry workerReg_
Worker * getWorker(WorkerParams const &p, std::string const &moduleLabel)
Retrieve the particular instance of the worker.

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

125  {
126  unscheduled_.runAccumulatorsAsync<T>(std::move(task), info, token, streamID, parentContext, context);
127  }
static const TGPicture * info(bool iBackgroundIsBlack)
UnscheduledCallProducer unscheduled_
long double T
def move(src, dest)
Definition: eostools.py:511
void runAccumulatorsAsync(WaitingTaskHolder task, typename T::TransitionInfoType const &info, ServiceToken const &token, StreamID streamID, ParentContext const &parentContext, typename T::Context const *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 
)

Definition at line 109 of file WorkerManager.h.

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

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

114  {
115  //make sure the unscheduled items see this run or lumi transition
116  unscheduled_.runNowAsync<T, U>(std::move(task), info, token, streamID, topContext, context);
117  }
static const TGPicture * info(bool iBackgroundIsBlack)
UnscheduledCallProducer unscheduled_
void runNowAsync(WaitingTaskHolder task, typename T::TransitionInfoType const &info, ServiceToken const &token, StreamID streamID, typename T::Context const *topContext, U const *context) const
long double T
def move(src, dest)
Definition: eostools.py:511

◆ resetAll()

void edm::WorkerManager::resetAll ( )

Definition at line 133 of file WorkerManager.cc.

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

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

133 { 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:188

◆ setupOnDemandSystem()

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

Definition at line 150 of file WorkerManager.cc.

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

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

150  {
152  }
static const TGPicture * info(bool iBackgroundIsBlack)
UnscheduledCallProducer unscheduled_
void setEventTransitionInfo(EventTransitionInfo const &info)

◆ setupResolvers()

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

◆ unscheduledWorkers()

AllWorkers const& edm::WorkerManager::unscheduledWorkers ( ) const
inline

Definition at line 86 of file WorkerManager.h.

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

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

86 { return unscheduled_.workers(); }
worker_container const & workers() const
UnscheduledCallProducer unscheduled_

Member Data Documentation

◆ actionTable_

ExceptionToActionTable const* edm::WorkerManager::actionTable_
private

Definition at line 102 of file WorkerManager.h.

Referenced by actionTable(), and getWorker().

◆ allWorkers_

AllWorkers edm::WorkerManager::allWorkers_
private

◆ lastSetupEventPrincipal_

void const* edm::WorkerManager::lastSetupEventPrincipal_
private

Definition at line 105 of file WorkerManager.h.

Referenced by setupResolvers().

◆ unscheduled_

UnscheduledCallProducer edm::WorkerManager::unscheduled_
private

◆ workerReg_

WorkerRegistry edm::WorkerManager::workerReg_
private

Definition at line 101 of file WorkerManager.h.

Referenced by deleteModuleIfExists(), and getWorker().