Merge pull request #48416 from Dr15Jones/moreRefactorWorker

Refactored how begin/end/Job/Stream are implemented

Author: cmsbuild

FWCore/Framework/interface/GlobalSchedule.h

@@ -73,8 +73,8 @@ namespace edm {
                                ServiceToken const& token,
                                bool cleaningUpAfterException = false);
 
-    void beginJob(ProcessContext const&);
-    void endJob(ExceptionCollector& collector);
+    void beginJob(ModuleRegistry&);
+    void endJob(ExceptionCollector& collector, ModuleRegistry&);
 
     /// Return a vector allowing const access to all the
     /// ModuleDescriptions for this GlobalSchedule.
@@ -112,17 +112,17 @@ namespace edm {
                          std::exception_ptr&);
 
     std::vector<WorkerManager> workerManagers_;
+    std::vector<unsigned int> beginJobFailedForModule_;
     std::shared_ptr<ActivityRegistry> actReg_;  // We do not use propagate_const because the registry itself is mutable.
     std::vector<edm::propagate_const<WorkerPtr>> extraWorkers_;
     ProcessContext const* processContext_;
 
-    // The next 4 variables use the same naming convention, even though we have no intention
-    // to ever have concurrent ProcessBlocks or Jobs. They are all related to the number of
+    // The next 3 variables use the same naming convention, even though we have no intention
+    // to ever have concurrent ProcessBlocks. They are all related to the number of
     // WorkerManagers needed for global transitions.
     unsigned int numberOfConcurrentLumis_;
     unsigned int numberOfConcurrentRuns_;
     static constexpr unsigned int numberOfConcurrentProcessBlocks_ = 1;
-    static constexpr unsigned int numberOfConcurrentJobs_ = 1;
   };
 
   template <typename T>

FWCore/Framework/interface/ModuleRegistry.h

@@ -36,9 +36,6 @@ namespace edm {
   struct MakeModuleParams;
   class ModuleDescription;
   class PreallocationConfiguration;
-  namespace maker {
-    class ModuleHolder;
-  }
 
   class ModuleRegistry {
   public:
@@ -81,6 +78,9 @@ namespace edm {
           holder->finishModuleInitialization(md, iPrealloc, iReg);
           labelToModule_.emplace(md.moduleLabel(), std::move(holder));
 
+          if (maxModuleID_ < module->moduleDescription().id()) {
+            maxModuleID_ = module->moduleDescription().id();
+          }
           // if exception then post will be called in the catch block
           postCalled = true;
           iPost(md);
@@ -104,14 +104,12 @@ namespace edm {
       }
     }
 
-    void finishModulesInitialization(ProductRegistry const& iRegistry,
-                                     eventsetup::ESRecordsToProductResolverIndices const& iESIndices,
-                                     ProcessBlockHelperBase const& processBlockHelperBase,
-                                     std::string const& processName);
+    unsigned int maxModuleID() const { return maxModuleID_; }
 
   private:
     std::map<std::string, edm::propagate_const<std::shared_ptr<maker::ModuleHolder>>> labelToModule_;
     ModuleTypeResolverMaker const* typeResolverMaker_;
+    unsigned int maxModuleID_ = 0;
   };
 }  // namespace edm
 

FWCore/Framework/interface/ModuleRegistryUtilities.h

@@ -0,0 +1,59 @@
+#ifndef FWCore_Framework_ModuleRegistryUtilities_h
+#define FWCore_Framework_ModuleRegistryUtilities_h
+
+#include <string>
+#include <vector>
+#include <mutex>
+namespace edm {
+  class ModuleRegistry;
+  class ActivityRegistry;
+  class ProductRegistry;
+  class StreamContext;
+  class GlobalContext;
+  namespace eventsetup {
+    class ESRecordsToProductResolverIndices;
+  }
+  class ProcessBlockHelperBase;
+  class ExceptionCollector;
+
+  void finishModulesInitialization(ModuleRegistry& iModuleRegistry,
+                                   ProductRegistry const& iProductRegistry,
+                                   eventsetup::ESRecordsToProductResolverIndices const& iESIndices,
+                                   ProcessBlockHelperBase const& processBlockHelperBase,
+                                   std::string const& processName);
+  /** beginJobFailedForModule has the module id of each module which threw an exception during
+     * the call to beginJob function.  The vector should be passed to `runEndJobForModules`.
+     * If an exception is thrown, it will be of type cms::Exception.
+     */
+  void runBeginJobForModules(GlobalContext const& iGlobalContext,
+                             ModuleRegistry& iModuleRegistry,
+                             edm::ActivityRegistry& iActivityRegistry,
+                             std::vector<unsigned int>& beginJobFailedForModule) noexcept(false);
+
+  /// The vector holds module id for modules which should not have their endJob called.
+  void runEndJobForModules(GlobalContext const& iGlobalContext,
+                           ModuleRegistry& iModuleRegistry,
+                           ActivityRegistry& iRegistry,
+                           ExceptionCollector& collector,
+                           std::vector<unsigned int> const& beginJobFailedForModule) noexcept;
+
+  /** beginStreamFailedForModule holds module id for each module which threw an exception during
+     * the call to beginStream function. This vector is used to determine which modules should not
+     * have their endStream called. The vector should be passed to `runEndStreamForModules`.
+     * If an exception is thrown, it will be of type cms::Exception.
+     */
+  void runBeginStreamForModules(StreamContext const& iStreamContext,
+                                ModuleRegistry& iModuleRegistry,
+                                edm::ActivityRegistry& iActivityRegistry,
+                                std::vector<unsigned int>& beginStreamFailedForModule) noexcept(false);
+
+  /// The vector hold module id for modules which should not have their endStream called.
+  void runEndStreamForModules(StreamContext const& iStreamContext,
+                              ModuleRegistry& iModuleRegistry,
+                              ActivityRegistry& iRegistry,
+                              ExceptionCollector& collector,
+                              std::mutex& collectorMutex,
+                              std::vector<unsigned int> const& beginStreamFailedForModule) noexcept;
+
+}  // namespace edm
+#endif  // FWCore_Framework_ModuleRegistryUtilities_h

FWCore/Framework/interface/Schedule.h

@@ -170,7 +170,7 @@ namespace edm {
     void beginJob(ProductRegistry const&,
                   eventsetup::ESRecordsToProductResolverIndices const&,
                   ProcessBlockHelperBase const&,
-                  ProcessContext const&);
+                  std::string const& processName);
     void endJob(ExceptionCollector& collector);
     void sendFwkSummaryToMessageLogger() const;
 

FWCore/Framework/interface/StreamSchedule.h

@@ -162,8 +162,8 @@ namespace edm {
                                ServiceToken const& token,
                                bool cleaningUpAfterException = false);
 
-    void beginStream();
-    void endStream(ExceptionCollector& collector, std::mutex& collectorMutex) noexcept;
+    void beginStream(ModuleRegistry& iModuleRegistry);
+    void endStream(ModuleRegistry& iModuleRegistry, ExceptionCollector& collector, std::mutex& collectorMutex) noexcept;
 
     StreamID streamID() const { return streamID_; }
 
@@ -222,7 +222,6 @@ namespace edm {
                                edm::ProductRegistry const& preg);
 
     /// returns the collection of pointers to workers
-    AllWorkers const& allWorkersBeginEnd() const { return workerManagerBeginEnd_.allWorkers(); }
     AllWorkers const& allWorkersRuns() const { return workerManagerRuns_.allWorkers(); }
     AllWorkers const& allWorkersLumisAndEvents() const { return workerManagerLumisAndEvents_.allWorkers(); }
 
@@ -310,7 +309,7 @@ namespace edm {
 
     void handleException(StreamContext const&, bool cleaningUpAfterException, std::exception_ptr&) const noexcept;
 
-    WorkerManager workerManagerBeginEnd_;
+    std::vector<unsigned int> moduleBeginStreamFailed_;
     WorkerManager workerManagerRuns_;
     WorkerManager workerManagerLumisAndEvents_;
     std::shared_ptr<ActivityRegistry> actReg_;  // We do not use propagate_const because the registry itself is mutable.

FWCore/Framework/interface/WorkerManager.h

@@ -67,12 +67,6 @@ namespace edm {
     void setupResolvers(Principal& principal);
     void setupOnDemandSystem(EventTransitionInfo const&);
 
-    void beginJob(GlobalContext const&);
-    void endJob(ExceptionCollector&, GlobalContext const&);
-
-    void beginStream(StreamID, StreamContext const&);
-    void endStream(StreamID, StreamContext const&, ExceptionCollector&, std::mutex& collectorMutex) noexcept;
-
     AllWorkers const& allWorkers() const { return allWorkers_; }
     AllWorkers const& unscheduledWorkers() const { return unscheduled_.workers(); }
 

FWCore/Framework/interface/global/EDAnalyzerBase.h

@@ -105,7 +105,6 @@ namespace edm {
       void doRegisterThinnedAssociations(ProductRegistry const&, ThinnedAssociationsHelper&) {}
 
       void registerProductsAndCallbacks(EDAnalyzerBase* module, SignallingProductRegistryFiller* reg);
-      std::string workerType() const { return "WorkerT<EDAnalyzer>"; }
 
       virtual void analyze(StreamID, Event const&, EventSetup const&) const = 0;
       virtual void beginJob() {}

FWCore/Framework/interface/global/EDFilterBase.h

@@ -114,7 +114,6 @@ namespace edm {
       void registerProductsAndCallbacks(EDFilterBase* module, SignallingProductRegistryFiller* reg) {
         registerProducts(module, reg, moduleDescription_);
       }
-      std::string workerType() const { return "WorkerT<EDProducer>"; }
 
       virtual bool filter(StreamID, Event&, EventSetup const&) const = 0;
       virtual void beginJob() {}

FWCore/Framework/interface/global/EDProducerBase.h

@@ -111,7 +111,6 @@ namespace edm {
       void registerProductsAndCallbacks(EDProducerBase* module, SignallingProductRegistryFiller* reg) {
         registerProducts(module, reg, moduleDescription_);
       }
-      std::string workerType() const { return "WorkerT<EDProducer>"; }
 
       virtual void produce(StreamID, Event&, EventSetup const&) const = 0;
       //For now this is a placeholder

FWCore/Framework/interface/global/OutputModuleBase.h

@@ -66,8 +66,6 @@ namespace edm {
                                                     Principal const& iPrincipal) const noexcept {}
 
     private:
-      std::string workerType() const { return "WorkerT<edm::global::OutputModuleBase>"; }
-
       virtual void preallocStreams(unsigned int) {}
       virtual void preallocate(PreallocationConfiguration const&) {}
       virtual void doBeginStream_(StreamID) {}