Make new main ProductRegistry when input updates

If the ProductRegistry of the source changes, we now create a new main ProductRegistry. The main Registry is then passed to the Principals which update their data structures.

Author: Dr15Jones

FWCore/Framework/interface/Principal.h

@@ -74,7 +74,8 @@ namespace edm {
 
     ~Principal() override;
 
-    void adjustIndexesAfterProductRegistryAddition();
+    //This should only be called when this Principal is not being actively used
+    void possiblyUpdateAfterAddition(std::shared_ptr<ProductRegistry const>);
 
     void fillPrincipal(DelayedReader* reader);
     void fillPrincipal(ProcessHistoryID const& hist, ProcessHistory const* phr, DelayedReader* reader);
@@ -213,6 +214,8 @@ namespace edm {
     }
 
   private:
+    void adjustIndexesAfterProductRegistryAddition();
+
     //called by adjustIndexesAfterProductRegistryAddition only if an index actually changed
     virtual void changedIndexes_() {}
 

FWCore/Framework/interface/PrincipalCache.h

@@ -53,7 +53,7 @@ namespace edm {
 
     void adjustEventsToNewProductRegistry(std::shared_ptr<ProductRegistry const>);
 
-    void adjustIndexesAfterProductRegistryAddition();
+    void adjustIndexesAfterProductRegistryAddition(std::shared_ptr<ProductRegistry const>);
 
   private:
     std::unique_ptr<ProcessBlockPrincipal> processBlockPrincipal_;

FWCore/Framework/src/EventProcessor.cc

@@ -1022,21 +1022,29 @@ namespace edm {
     FDEBUG(1) << " \treadFile\n";
     SendSourceTerminationSignalIfException sentry(actReg_.get());
 
+    //CDJ: THIS SHOULD NOT BE APPLIED TO THE ACTIVE INTERVALS
     if (streamRunActive_ > 0) {
+      //deals with data structures that allows merged Run products to be split on Lumi boundaries then
+      // in later processes reintegrated.
       streamRunStatus_[0]->runPrincipal()->preReadFile();
-      streamRunStatus_[0]->runPrincipal()->adjustIndexesAfterProductRegistryAddition();
-    }
-
-    if (streamLumiActive_ > 0) {
-      streamLumiStatus_[0]->lumiPrincipal()->adjustIndexesAfterProductRegistryAddition();
     }
 
+    auto sizeBefore = input_->productRegistry().size();
     fb_ = input_->readFile();
     //incase the input's registry changed
-    const size_t size = preg_->size();
-    preg_->merge(input_->productRegistry(), fb_ ? fb_->fileName() : std::string());
-    if (size < preg_->size()) {
-      principalCache_.adjustIndexesAfterProductRegistryAddition();
+    if (input_->productRegistry().size() != sizeBefore) {
+      auto temp = std::make_shared<edm::ProductRegistry>(*preg_);
+      temp->merge(input_->productRegistry(), fb_ ? fb_->fileName() : std::string());
+      preg_ = std::move(temp);
+      //This handles are presently unused Run/Lumis
+      principalCache_.adjustIndexesAfterProductRegistryAddition(edm::get_underlying_safe(preg_));
+      if (streamLumiActive_ > 0) {
+        //Can update the active ones now, even before an `end` transition is called because no OutputModule
+        // supports storing ProductDescriptions for Run/LuminosityBlock products which were dropped. Since only
+        // dropped products can change the ProductRegistry, only changes in Event can cause that.
+        streamRunStatus_[0]->runPrincipal()->possiblyUpdateAfterAddition(edm::get_underlying_safe(preg_));
+        streamLumiStatus_[0]->lumiPrincipal()->possiblyUpdateAfterAddition(edm::get_underlying_safe(preg_));
+      }
     }
     principalCache_.adjustEventsToNewProductRegistry(preg());
     if (preallocations_.numberOfStreams() > 1 and preallocations_.numberOfThreads() > 1) {
@@ -2022,6 +2030,7 @@ namespace edm {
 
   std::shared_ptr<RunPrincipal> EventProcessor::readRun() {
     auto rp = principalCache_.getAvailableRunPrincipalPtr();
+    rp->possiblyUpdateAfterAddition(preg());
     assert(rp);
     rp->setAux(*input_->runAuxiliary());
     {
@@ -2046,6 +2055,7 @@ namespace edm {
 
   std::shared_ptr<LuminosityBlockPrincipal> EventProcessor::readLuminosityBlock(std::shared_ptr<RunPrincipal> rp) {
     auto lbp = principalCache_.getAvailableLumiPrincipalPtr();
+    lbp->possiblyUpdateAfterAddition(preg());
     assert(lbp);
     lbp->setAux(*input_->luminosityBlockAuxiliary());
     {

FWCore/Framework/src/Principal.cc

@@ -31,7 +31,8 @@
 #include <stdexcept>
 #include <typeinfo>
 #include <atomic>
-
+//DEBUG
+#include <iostream>
 namespace edm {
 
   static ProcessHistory const s_emptyProcessHistory;
@@ -149,6 +150,13 @@ namespace edm {
     return size;
   }
 
+  void Principal::possiblyUpdateAfterAddition(std::shared_ptr<ProductRegistry const> iProd) {
+    if (iProd.get() != preg_.get()) {
+      preg_ = iProd;
+      adjustIndexesAfterProductRegistryAddition();
+    }
+  }
+
   void Principal::addDroppedProduct(ProductDescription const& bd) {
     addProductOrThrow(std::make_unique<DroppedDataProductResolver>(std::make_shared<ProductDescription const>(bd)));
   }

FWCore/Framework/src/PrincipalCache.cc

@@ -43,22 +43,22 @@ namespace edm {
   void PrincipalCache::adjustEventsToNewProductRegistry(std::shared_ptr<ProductRegistry const> reg) {
     for (auto& eventPrincipal : eventPrincipals_) {
       if (eventPrincipal) {
-        eventPrincipal->adjustIndexesAfterProductRegistryAddition();
+        eventPrincipal->possiblyUpdateAfterAddition(reg);
       }
     }
   }
 
-  void PrincipalCache::adjustIndexesAfterProductRegistryAddition() {
+  void PrincipalCache::adjustIndexesAfterProductRegistryAddition(std::shared_ptr<ProductRegistry const> iReg) {
     //Need to temporarily hold all the runs to clear out the runHolder_
     std::vector<std::shared_ptr<RunPrincipal>> tempRunPrincipals;
     while (auto p = runHolder_.tryToGet()) {
-      p->adjustIndexesAfterProductRegistryAddition();
+      p->possiblyUpdateAfterAddition(iReg);
       tempRunPrincipals.emplace_back(std::move(p));
     }
     //Need to temporarily hold all the lumis to clear out the lumiHolder_
     std::vector<std::shared_ptr<LuminosityBlockPrincipal>> tempLumiPrincipals;
     while (auto p = lumiHolder_.tryToGet()) {
-      p->adjustIndexesAfterProductRegistryAddition();
+      p->possiblyUpdateAfterAddition(iReg);
       tempLumiPrincipals.emplace_back(std::move(p));
     }
   }

FWCore/TestProcessor/src/TestSourceProcessor.cc

@@ -229,7 +229,7 @@ namespace edm::test {
     const size_t size = preg_->size();
     preg_->merge(source_->productRegistry(), fb_ ? fb_->fileName() : std::string());
     if (size < preg_->size()) {
-      principalCache_.adjustIndexesAfterProductRegistryAddition();
+      principalCache_.adjustIndexesAfterProductRegistryAddition(preg_);
     }
     principalCache_.adjustEventsToNewProductRegistry(preg_);