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 ModuleTypeResolverMaker;
0033   class PreallocationConfiguration;
0034   namespace eventsetup {
0035     class ESRecordsToProxyIndices;
0036   }
0037   class WorkerManager {
0038   public:
0039     typedef std::vector<Worker*> AllWorkers;
0040
0041     WorkerManager(std::shared_ptr<ActivityRegistry> actReg,
0042                   ExceptionToActionTable const& actions,
0043                   ModuleTypeResolverMaker const* typeResolverMaker);
0044     WorkerManager(WorkerManager&&) = default;
0045
0046     WorkerManager(std::shared_ptr<ModuleRegistry> modReg,
0047                   std::shared_ptr<ActivityRegistry> actReg,
0048                   ExceptionToActionTable const& actions);
0049
0050     void deleteModuleIfExists(std::string const& moduleLabel);
0051
0052     void addToUnscheduledWorkers(ParameterSet& pset,
0053                                  ProductRegistry& preg,
0054                                  PreallocationConfiguration const* prealloc,
0055                                  std::shared_ptr<ProcessConfiguration const> processConfiguration,
0056                                  std::string label,
0057                                  std::set<std::string>& unscheduledLabels,
0058                                  std::vector<std::string>& shouldBeUsedLabels);
0059
0060     template <typename T, typename U>
0061     void processOneOccurrenceAsync(WaitingTaskHolder,
0062                                    typename T::TransitionInfoType&,
0063                                    ServiceToken const&,
0064                                    StreamID,
0065                                    typename T::Context const* topContext,
0066                                    U const* context);
0067
0068     template <typename T>
0069     void processAccumulatorsAsync(WaitingTaskHolder,
0070                                   typename T::TransitionInfoType const&,
0071                                   ServiceToken const&,
0072                                   StreamID,
0073                                   ParentContext const&,
0074                                   typename T::Context const*);
0075
0076     void setupResolvers(Principal& principal);
0077     void setupOnDemandSystem(EventTransitionInfo const&);
0078
0079     void beginJob(ProductRegistry const& iRegistry,
0080                   eventsetup::ESRecordsToProxyIndices const&,
0081                   ProcessBlockHelperBase const&);
0082     void endJob();
0083     void endJob(ExceptionCollector& collector);
0084
0085     void beginStream(StreamID iID, StreamContext& streamContext);
0086     void endStream(StreamID iID, StreamContext& streamContext);
0087
0088     AllWorkers const& allWorkers() const { return allWorkers_; }
0089     AllWorkers const& unscheduledWorkers() const { return unscheduled_.workers(); }
0090
0091     void addToAllWorkers(Worker* w);
0092
0093     ExceptionToActionTable const& actionTable() const { return *actionTable_; }
0094
0095     Worker* getWorker(ParameterSet& pset,
0096                       ProductRegistry& preg,
0097                       PreallocationConfiguration const* prealloc,
0098                       std::shared_ptr<ProcessConfiguration const> processConfiguration,
0099                       std::string const& label);
0100
0101     void resetAll();
0102
0103   private:
0104     WorkerRegistry workerReg_;
0105     ExceptionToActionTable const* actionTable_;
0106     AllWorkers allWorkers_;
0107     UnscheduledCallProducer unscheduled_;
0108     void const* lastSetupEventPrincipal_;
0109   };
0110
0111   template <typename T, typename U>
0112   void WorkerManager::processOneOccurrenceAsync(WaitingTaskHolder task,
0113                                                 typename T::TransitionInfoType& info,
0114                                                 ServiceToken const& token,
0115                                                 StreamID streamID,
0116                                                 typename T::Context const* topContext,
0117                                                 U const* context) {
0118     //make sure the unscheduled items see this run or lumi transition
0119     unscheduled_.runNowAsync<T, U>(std::move(task), info, token, streamID, topContext, context);
0120   }
0121
0122   template <typename T>
0123   void WorkerManager::processAccumulatorsAsync(WaitingTaskHolder task,
0124                                                typename T::TransitionInfoType const& info,
0125                                                ServiceToken const& token,
0126                                                StreamID streamID,
0127                                                ParentContext const& parentContext,
0128                                                typename T::Context const* context) {
0129     unscheduled_.runAccumulatorsAsync<T>(std::move(task), info, token, streamID, parentContext, context);
0130   }
0131 }  // namespace edm
0132
0133 #endif