Merge pull request #48358 from Dr15Jones/refactorWorker

Refactor Worker/Module system

Author: cmsbuild

FWCore/Framework/interface/GlobalSchedule.h

@@ -13,7 +13,6 @@
 #include "FWCore/Framework/interface/RunPrincipal.h"
 #include "FWCore/Framework/interface/WorkerManager.h"
 #include "FWCore/Framework/interface/maker/Worker.h"
-#include "FWCore/Framework/interface/WorkerRegistry.h"
 #include "FWCore/Framework/interface/SignallingProductRegistryFiller.h"
 #include "FWCore/MessageLogger/interface/ExceptionMessages.h"
 #include "FWCore/ServiceRegistry/interface/GlobalContext.h"
@@ -74,10 +73,7 @@ namespace edm {
                                ServiceToken const& token,
                                bool cleaningUpAfterException = false);
 
-    void beginJob(ProductRegistry const&,
-                  eventsetup::ESRecordsToProductResolverIndices const&,
-                  ProcessBlockHelperBase const&,
-                  ProcessContext const&);
+    void beginJob(ProcessContext const&);
     void endJob(ExceptionCollector& collector);
 
     /// Return a vector allowing const access to all the
@@ -100,8 +96,6 @@ namespace edm {
     /// returns the collection of pointers to workers
     AllWorkers const& allWorkers() const { return workerManagers_[0].allWorkers(); }
 
-    void releaseMemoryPostLookupSignal();
-
   private:
     /// returns the action table
     ExceptionToActionTable const& actionTable() const { return workerManagers_[0].actionTable(); }

FWCore/Framework/interface/ModuleRegistry.h

@@ -26,7 +26,9 @@
 // user include files
 #include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
 #include "FWCore/Utilities/interface/propagate_const.h"
+#include "FWCore/Utilities/interface/Exception.h"
 #include "FWCore/Framework/interface/ModuleTypeResolverMaker.h"
+#include "FWCore/Framework/interface/maker/ModuleHolder.h"
 
 // forward declarations
 namespace edm {
@@ -46,6 +48,8 @@ namespace edm {
                                                    std::string const& moduleLabel,
                                                    signalslot::Signal<void(ModuleDescription const&)>& iPre,
                                                    signalslot::Signal<void(ModuleDescription const&)>& iPost);
+    //returns a null if module not found
+    std::shared_ptr<maker::ModuleHolder> getExistingModule(std::string const& moduleLabel);
 
     maker::ModuleHolder* replaceModule(std::string const& iModuleLabel,
                                        edm::ParameterSet const& iPSet,
@@ -55,6 +59,43 @@ namespace edm {
                       signalslot::Signal<void(ModuleDescription const&)>& iPre,
                       signalslot::Signal<void(ModuleDescription const&)>& iPost);
 
+    template <typename T, typename... Args>
+    std::shared_ptr<T> makeExplicitModule(ModuleDescription const& md,
+                                          PreallocationConfiguration const& iPrealloc,
+                                          SignallingProductRegistryFiller* iReg,
+                                          signalslot::Signal<void(ModuleDescription const&)>& iPre,
+                                          signalslot::Signal<void(ModuleDescription const&)>& iPost,
+                                          Args&&... args) {
+      bool postCalled = false;
+      if (labelToModule_.find(md.moduleLabel()) != labelToModule_.end()) {
+        throw cms::Exception("InsertError") << "Module with label '" << md.moduleLabel() << "' already exists.";
+      }
+
+      try {
+        std::shared_ptr<T> module;
+        convertException::wrap([&]() {
+          iPre(md);
+          module = std::make_shared<T>(std::forward<Args>(args)...);
+
+          auto holder = std::make_shared<maker::ModuleHolderT<typename T::ModuleType>>(module);
+          holder->finishModuleInitialization(md, iPrealloc, iReg);
+          labelToModule_.emplace(md.moduleLabel(), std::move(holder));
+
+          // if exception then post will be called in the catch block
+          postCalled = true;
+          iPost(md);
+        });
+        return module;
+      } catch (cms::Exception& iException) {
+        if (!postCalled) {
+          CMS_SA_ALLOW try { iPost(md); } catch (...) {
+            // If post throws an exception ignore it because we are already handling another exception
+          }
+        }
+        throw;
+      }
+    }
+
     template <typename F>
     void forAllModuleHolders(F iFunc) {
       for (auto& labelMod : labelToModule_) {
@@ -63,6 +104,11 @@ namespace edm {
       }
     }
 
+    void finishModulesInitialization(ProductRegistry const& iRegistry,
+                                     eventsetup::ESRecordsToProductResolverIndices const& iESIndices,
+                                     ProcessBlockHelperBase const& processBlockHelperBase,
+                                     std::string const& processName);
+
   private:
     std::map<std::string, edm::propagate_const<std::shared_ptr<maker::ModuleHolder>>> labelToModule_;
     ModuleTypeResolverMaker const* typeResolverMaker_;

FWCore/Framework/interface/Schedule.h

@@ -66,7 +66,6 @@
 #include "FWCore/Framework/interface/OccurrenceTraits.h"
 #include "FWCore/Framework/interface/WorkerManager.h"
 #include "FWCore/Framework/interface/maker/Worker.h"
-#include "FWCore/Framework/interface/WorkerRegistry.h"
 #include "FWCore/Framework/interface/GlobalSchedule.h"
 #include "FWCore/Framework/interface/StreamSchedule.h"
 #include "FWCore/Framework/interface/SystemTimeKeeper.h"
@@ -305,10 +304,10 @@ namespace edm {
     std::shared_ptr<ModuleRegistry const> moduleRegistry() const { return get_underlying_safe(moduleRegistry_); }
     std::shared_ptr<ModuleRegistry>& moduleRegistry() { return get_underlying_safe(moduleRegistry_); }
 
+    edm::propagate_const<std::shared_ptr<ModuleRegistry>> moduleRegistry_;
     edm::propagate_const<std::shared_ptr<TriggerResultInserter>> resultsInserter_;
     std::vector<edm::propagate_const<std::shared_ptr<PathStatusInserter>>> pathStatusInserters_;
     std::vector<edm::propagate_const<std::shared_ptr<EndPathStatusInserter>>> endPathStatusInserters_;
-    edm::propagate_const<std::shared_ptr<ModuleRegistry>> moduleRegistry_;
     std::vector<edm::propagate_const<std::shared_ptr<StreamSchedule>>> streamSchedules_;
     //In the future, we will have one GlobalSchedule per simultaneous transition
     edm::propagate_const<std::unique_ptr<GlobalSchedule>> globalSchedule_;

FWCore/Framework/interface/StreamSchedule.h

@@ -69,7 +69,6 @@
 #include "FWCore/Framework/interface/Path.h"
 #include "FWCore/Framework/interface/TransitionInfoTypes.h"
 #include "FWCore/Framework/interface/maker/Worker.h"
-#include "FWCore/Framework/interface/WorkerRegistry.h"
 #include "FWCore/Framework/interface/EarlyDeleteHelper.h"
 #include "FWCore/MessageLogger/interface/ExceptionMessages.h"
 #include "FWCore/MessageLogger/interface/JobReport.h"
@@ -129,7 +128,6 @@ namespace edm {
     typedef std::vector<Path> TrigPaths;
     typedef std::shared_ptr<HLTGlobalStatus> TrigResPtr;
     typedef std::shared_ptr<HLTGlobalStatus const> TrigResConstPtr;
-    typedef std::shared_ptr<Worker> WorkerPtr;
     typedef std::vector<Worker*> AllWorkers;
 
     typedef std::vector<Worker*> Workers;
@@ -263,16 +261,15 @@ namespace edm {
         SignallingProductRegistryFiller& preg,
         PreallocationConfiguration const* prealloc,
         std::shared_ptr<ProcessConfiguration const> processConfiguration);
-    void fillWorkers(ParameterSet& proc_pset,
-                     SignallingProductRegistryFiller& preg,
-                     PreallocationConfiguration const* prealloc,
-                     std::shared_ptr<ProcessConfiguration const> processConfiguration,
-                     std::string const& name,
-                     bool ignoreFilters,
-                     PathWorkers& out,
-                     std::vector<std::string> const& endPathNames,
-                     ConditionalTaskHelper const& conditionalTaskHelper,
-                     std::unordered_set<std::string>& allConditionalModules);
+    PathWorkers fillWorkers(ParameterSet& proc_pset,
+                            SignallingProductRegistryFiller& preg,
+                            PreallocationConfiguration const* prealloc,
+                            std::shared_ptr<ProcessConfiguration const> processConfiguration,
+                            std::string const& name,
+                            bool ignoreFilters,
+                            std::vector<std::string> const& endPathNames,
+                            ConditionalTaskHelper const& conditionalTaskHelper,
+                            std::unordered_set<std::string>& allConditionalModules);
     void fillTrigPath(ParameterSet& proc_pset,
                       SignallingProductRegistryFiller& preg,
                       PreallocationConfiguration const* prealloc,
@@ -320,9 +317,9 @@ namespace edm {
 
     edm::propagate_const<TrigResPtr> results_;
 
-    edm::propagate_const<WorkerPtr> results_inserter_;
-    std::vector<edm::propagate_const<WorkerPtr>> pathStatusInserterWorkers_;
-    std::vector<edm::propagate_const<WorkerPtr>> endPathStatusInserterWorkers_;
+    edm::propagate_const<Worker*> results_inserter_;
+    std::vector<edm::propagate_const<Worker*>> pathStatusInserterWorkers_;
+    std::vector<edm::propagate_const<Worker*>> endPathStatusInserterWorkers_;
 
     TrigPaths trig_paths_;
     TrigPaths end_paths_;

FWCore/Framework/interface/WorkerManager.h

@@ -32,9 +32,6 @@ namespace edm {
   public:
     typedef std::vector<Worker*> AllWorkers;
 
-    WorkerManager(std::shared_ptr<ActivityRegistry> actReg,
-                  ExceptionToActionTable const& actions,
-                  ModuleTypeResolverMaker const* typeResolverMaker);
     WorkerManager(WorkerManager&&) = default;
 
     WorkerManager(std::shared_ptr<ModuleRegistry> modReg,
@@ -70,10 +67,7 @@ namespace edm {
     void setupResolvers(Principal& principal);
     void setupOnDemandSystem(EventTransitionInfo const&);
 
-    void beginJob(ProductRegistry const& iRegistry,
-                  eventsetup::ESRecordsToProductResolverIndices const&,
-                  ProcessBlockHelperBase const&,
-                  GlobalContext const&);
+    void beginJob(GlobalContext const&);
     void endJob(ExceptionCollector&, GlobalContext const&);
 
     void beginStream(StreamID, StreamContext const&);
@@ -90,13 +84,21 @@ namespace edm {
                       SignallingProductRegistryFiller& preg,
                       PreallocationConfiguration const* prealloc,
                       std::shared_ptr<ProcessConfiguration const> processConfiguration,
-                      std::string const& label);
+                      std::string const& label,
+                      bool addToAllWorkers = true);
 
+    template <typename T>
+    Worker* getWorkerForModule(T const& module) {
+      auto* worker = getWorkerForExistingModule(module.moduleDescription().moduleLabel());
+      assert(worker != nullptr);
+      assert(worker->matchesBaseClassPointer(static_cast<typename T::ModuleType const*>(&module)));
+      return worker;
+    }
     void resetAll();
 
-    void releaseMemoryPostLookupSignal();
-
   private:
+    Worker* getWorkerForExistingModule(std::string const& label);
+
     WorkerRegistry workerReg_;
     ExceptionToActionTable const* actionTable_;
     AllWorkers allWorkers_;

FWCore/Framework/interface/WorkerRegistry.h

@@ -24,21 +24,21 @@ namespace edm {
   class ModuleRegistry;
   class ModuleTypeResolverMaker;
   class ParameterSet;
+  class ExceptionToActionTable;
   namespace maker {
     class ModuleHolder;
   }
 
   /**
      \class WorkerRegistry WorkerRegistry.h "edm/WorkerRegistry.h"
 
-     \brief The Registry of all workers that where requested
+     \brief The Registry of all workers that were requested
      Holds all instances of workers. In this implementation, Workers 
      are owned.
   */
 
   class WorkerRegistry {
   public:
-    explicit WorkerRegistry(std::shared_ptr<ActivityRegistry> areg, ModuleTypeResolverMaker const* resolverMaker);
     WorkerRegistry(std::shared_ptr<ActivityRegistry> areg, std::shared_ptr<ModuleRegistry> iModReg);
     ~WorkerRegistry();
 
@@ -56,6 +56,9 @@ namespace edm {
     /// If one doesn't exist, returns nullptr
     Worker const* get(std::string const& moduleLabel) const;
 
+    //Creates worker if needed
+    Worker* getWorkerFromExistingModule(std::string const& moduleLabel, ExceptionToActionTable const* actions);
+
     /// Deletes the module of the Worker, but the Worker continues to exist.
     void deleteModule(std::string const& moduleLabel);
 

FWCore/Framework/interface/maker/ModuleHolder.h

@@ -23,14 +23,24 @@
 
 // user include files
 #include "FWCore/Framework/interface/maker/WorkerT.h"
-#include "FWCore/Framework/interface/OutputModuleCommunicatorT.h"
 #include "FWCore/Framework/interface/SignallingProductRegistryFiller.h"
+#include "FWCore/Framework/interface/OutputModuleCommunicator.h"
+
+#include "FWCore/Utilities/interface/BranchType.h"
+#include "FWCore/Utilities/interface/ProductResolverIndex.h"
+
+#include <unordered_map>
+#include <string>
 // forward declarations
 namespace edm {
   class ModuleDescription;
   class SignallingProductRegistryFiller;
   class ExceptionToActionTable;
   class PreallocationConfiguration;
+  class ProductResolverIndexHelper;
+  class ProductResolverIndexAndSkipBit;
+  class ProductRegistry;
+  class ThinnedAssociationsHelper;
 
   namespace maker {
     class ModuleHolder {
@@ -46,6 +56,22 @@ namespace edm {
       virtual void replaceModuleFor(Worker*) const = 0;
 
       virtual std::unique_ptr<OutputModuleCommunicator> createOutputModuleCommunicator() = 0;
+
+      void registerThinnedAssociations(ProductRegistry const& registry, ThinnedAssociationsHelper& helper);
+      //Used to make EDGetToken work
+      virtual void updateLookup(BranchType iBranchType, ProductResolverIndexHelper const&) = 0;
+      virtual void updateLookup(eventsetup::ESRecordsToProductResolverIndices const&) = 0;
+      virtual void releaseMemoryPostLookupSignal() = 0;
+      virtual void selectInputProcessBlocks(ProductRegistry const&, ProcessBlockHelperBase const&) = 0;
+      virtual void resolvePutIndicies(
+          BranchType iBranchType,
+          std::unordered_multimap<std::string, std::tuple<TypeID const*, const char*, edm::ProductResolverIndex>> const&
+              iIndicies) = 0;
+      virtual void convertCurrentProcessAlias(std::string const& processName) = 0;
+
+    private:
+      virtual void implRegisterThinnedAssociations(ProductRegistry const& registry,
+                                                   ThinnedAssociationsHelper& helper) = 0;
     };
 
     template <typename T>
@@ -59,7 +85,7 @@ namespace edm {
         assert(nullptr != w);
         w->setModule(m_mod);
       }
-      std::unique_ptr<Worker> makeWorker(ExceptionToActionTable const* actions) const override {
+      std::unique_ptr<Worker> makeWorker(ExceptionToActionTable const* actions) const final {
         return std::make_unique<edm::WorkerT<T>>(module(), moduleDescription(), actions);
       }
 
@@ -73,18 +99,30 @@ namespace edm {
           iModule.registerProductsAndCallbacks(&iModule, iReg);
         }
       };
-      ModuleDescription const& moduleDescription() const override { return m_mod->moduleDescription(); }
+      ModuleDescription const& moduleDescription() const final { return m_mod->moduleDescription(); }
 
       void finishModuleInitialization(ModuleDescription const& iDesc,
                                       PreallocationConfiguration const& iPrealloc,
                                       SignallingProductRegistryFiller* iReg) override {
         finishModuleInitialization(*m_mod, iDesc, iPrealloc, iReg);
       }
-      std::unique_ptr<OutputModuleCommunicator> createOutputModuleCommunicator() override {
-        return OutputModuleCommunicatorT<T>::createIfNeeded(m_mod.get());
+      std::unique_ptr<OutputModuleCommunicator> createOutputModuleCommunicator() final;
+
+      void updateLookup(BranchType iBranchType, ProductResolverIndexHelper const&) final;
+      void updateLookup(eventsetup::ESRecordsToProductResolverIndices const&) final;
+      void releaseMemoryPostLookupSignal() final;
+      void selectInputProcessBlocks(ProductRegistry const&, ProcessBlockHelperBase const&) final;
+      void resolvePutIndicies(
+          BranchType iBranchType,
+          std::unordered_multimap<std::string, std::tuple<TypeID const*, const char*, edm::ProductResolverIndex>> const&
+              iIndicies) final;
+      void convertCurrentProcessAlias(std::string const& processName) final {
+        m_mod->convertCurrentProcessAlias(processName);
       }
 
     private:
+      void implRegisterThinnedAssociations(ProductRegistry const& registry, ThinnedAssociationsHelper& helper) final;
+
       std::shared_ptr<T> m_mod;
     };
   }  // namespace maker