Framework/interface/WorkerManager.h

0001 #ifndef FWCore_Framework_WorkerManager_h
0002 #define FWCore_Framework_WorkerManager_h
0003
0004 /*
0005
0006 */
0007
0008 #include "FWCore/Common/interface/FWCoreCommonFwd.h"
0009 #include "FWCore/Framework/interface/ExceptionHelpers.h"
0010 #include "FWCore/Framework/interface/Frameworkfwd.h"
0011 #include "FWCore/Framework/interface/UnscheduledCallProducer.h"
0012 #include "FWCore/Framework/interface/maker/Worker.h"
0013 #include "FWCore/Framework/interface/WorkerRegistry.h"
0014 #include "FWCore/ServiceRegistry/interface/ServiceRegistry.h"
0015 #include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
0016 #include "FWCore/Utilities/interface/ConvertException.h"
0017 #include "FWCore/Utilities/interface/Exception.h"
0018 #include "FWCore/Utilities/interface/get_underlying_safe.h"
0019
0020 #include <memory>
0021
0022 #include <set>
0023 #include <string>
0024 #include <vector>
0025
0026 namespace edm {
0027   class EventTransitionInfo;
0028   class ExceptionCollector;
0029   class StreamID;
0030   class StreamContext;
0031   class ModuleRegistry;
0032   class PreallocationConfiguration;
0033   namespace eventsetup {
0034     class ESRecordsToProxyIndices;
0035   }
0036   class WorkerManager {
0037   public:
0038     typedef std::vector<Worker*> AllWorkers;
0039
0040     WorkerManager(std::shared_ptr<ActivityRegistry> actReg, ExceptionToActionTable const& actions);
0041     WorkerManager(WorkerManager&&) = default;
0042
0043     WorkerManager(std::shared_ptr<ModuleRegistry> modReg,
0044                   std::shared_ptr<ActivityRegistry> actReg,
0045                   ExceptionToActionTable const& actions);
0046
0047     void deleteModuleIfExists(std::string const& moduleLabel);
0048
0049     void addToUnscheduledWorkers(ParameterSet& pset,
0050                                  ProductRegistry& preg,
0051                                  PreallocationConfiguration const* prealloc,
0052                                  std::shared_ptr<ProcessConfiguration const> processConfiguration,
0053                                  std::string label,
0054                                  std::set<std::string>& unscheduledLabels,
0055                                  std::vector<std::string>& shouldBeUsedLabels);
0056
0057     template <typename T, typename U>
0058     void processOneOccurrenceAsync(WaitingTaskHolder,
0059                                    typename T::TransitionInfoType&,
0060                                    ServiceToken const&,
0061                                    StreamID,
0062                                    typename T::Context const* topContext,
0063                                    U const* context);
0064
0065     template <typename T>
0066     void processAccumulatorsAsync(WaitingTaskHolder,
0067                                   typename T::TransitionInfoType const&,
0068                                   ServiceToken const&,
0069                                   StreamID,
0070                                   ParentContext const&,
0071                                   typename T::Context const*);
0072
0073     void setupResolvers(Principal& principal);
0074     void setupOnDemandSystem(EventTransitionInfo const&);
0075
0076     void beginJob(ProductRegistry const& iRegistry,
0077                   eventsetup::ESRecordsToProxyIndices const&,
0078                   ProcessBlockHelperBase const&);
0079     void endJob();
0080     void endJob(ExceptionCollector& collector);
0081
0082     void beginStream(StreamID iID, StreamContext& streamContext);
0083     void endStream(StreamID iID, StreamContext& streamContext);
0084
0085     AllWorkers const& allWorkers() const { return allWorkers_; }
0086     AllWorkers const& unscheduledWorkers() const { return unscheduled_.workers(); }
0087
0088     void addToAllWorkers(Worker* w);
0089
0090     ExceptionToActionTable const& actionTable() const { return *actionTable_; }
0091
0092     Worker* getWorker(ParameterSet& pset,
0093                       ProductRegistry& preg,
0094                       PreallocationConfiguration const* prealloc,
0095                       std::shared_ptr<ProcessConfiguration const> processConfiguration,
0096                       std::string const& label);
0097
0098     void resetAll();
0099
0100   private:
0101     WorkerRegistry workerReg_;
0102     ExceptionToActionTable const* actionTable_;
0103     AllWorkers allWorkers_;
0104     UnscheduledCallProducer unscheduled_;
0105     void const* lastSetupEventPrincipal_;
0106   };
0107
0108   template <typename T, typename U>
0109   void WorkerManager::processOneOccurrenceAsync(WaitingTaskHolder task,
0110                                                 typename T::TransitionInfoType& info,
0111                                                 ServiceToken const& token,
0112                                                 StreamID streamID,
0113                                                 typename T::Context const* topContext,
0114                                                 U const* context) {
0115     //make sure the unscheduled items see this run or lumi transition
0116     unscheduled_.runNowAsync<T, U>(std::move(task), info, token, streamID, topContext, context);
0117   }
0118
0119   template <typename T>
0120   void WorkerManager::processAccumulatorsAsync(WaitingTaskHolder task,
0121                                                typename T::TransitionInfoType const& info,
0122                                                ServiceToken const& token,
0123                                                StreamID streamID,
0124                                                ParentContext const& parentContext,
0125                                                typename T::Context const* context) {
0126     unscheduled_.runAccumulatorsAsync<T>(std::move(task), info, token, streamID, parentContext, context);
0127   }
0128 }  // namespace edm
0129
0130 #endif