Update Principal held ProductRegistry just before it is needed

Reading a new file can cause the ProductRegistry to be modified which may require the Principal to update its internal data structure. Now check if the ProductRegistry changed just before the Principal will be used for data processing.

Author: Dr15Jones

DataFormats/Provenance/interface/ProductRegistry.h

@@ -93,6 +93,11 @@ namespace edm {
 
     ProductList::size_type size() const { return productList_.size(); }
 
+    //If the value differs between two versions of the main registry then
+    // one must update any related meta data
+    using CacheID = ProductList::size_type;
+    CacheID cacheIdentifier() const { return size(); }
+
     void print(std::ostream& os) const;
 
     bool anyProducts(BranchType const brType) const;

FWCore/Framework/interface/PrincipalCache.h

@@ -51,10 +51,6 @@ namespace edm {
     void insert(std::unique_ptr<LuminosityBlockPrincipal>);
     void insert(std::shared_ptr<EventPrincipal>);
 
-    void adjustEventsToNewProductRegistry(std::shared_ptr<ProductRegistry const>);
-
-    void adjustIndexesAfterProductRegistryAddition(std::shared_ptr<ProductRegistry const>);
-
   private:
     std::unique_ptr<ProcessBlockPrincipal> processBlockPrincipal_;
     std::unique_ptr<ProcessBlockPrincipal> inputProcessBlockPrincipal_;

FWCore/Framework/src/EventProcessor.cc

@@ -1037,24 +1037,14 @@ namespace edm {
       streamRunStatus_[0]->runPrincipal()->preReadFile();
     }
 
-    auto sizeBefore = input_->productRegistry().size();
+    auto oldCacheID = input_->productRegistry().cacheIdentifier();
     fb_ = input_->readFile();
     //incase the input's registry changed
-    if (input_->productRegistry().size() != sizeBefore) {
+    if (input_->productRegistry().cacheIdentifier() != oldCacheID) {
       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) {
       fb_->setNotFastClonable(FileBlock::ParallelProcesses);
     }
@@ -2038,6 +2028,7 @@ namespace edm {
 
   std::shared_ptr<RunPrincipal> EventProcessor::readRun() {
     auto rp = principalCache_.getAvailableRunPrincipalPtr();
+    //a new file may have been opened since the last use of this Run
     rp->possiblyUpdateAfterAddition(preg());
     assert(rp);
     rp->setAux(*input_->runAuxiliary());
@@ -2052,6 +2043,9 @@ namespace edm {
 
   void EventProcessor::readAndMergeRun(RunProcessingStatus& iStatus) {
     RunPrincipal& runPrincipal = *iStatus.runPrincipal();
+    //If a file open happened and we are continuing the Run we may need
+    // to do the update
+    runPrincipal.possiblyUpdateAfterAddition(preg());
 
     runPrincipal.mergeAuxiliary(*input_->runAuxiliary());
     {
@@ -2063,6 +2057,7 @@ namespace edm {
 
   std::shared_ptr<LuminosityBlockPrincipal> EventProcessor::readLuminosityBlock(std::shared_ptr<RunPrincipal> rp) {
     auto lbp = principalCache_.getAvailableLumiPrincipalPtr();
+    //A new file may have been opened since the last use of the LuminosityBlock
     lbp->possiblyUpdateAfterAddition(preg());
     assert(lbp);
     lbp->setAux(*input_->luminosityBlockAuxiliary());
@@ -2081,6 +2076,9 @@ namespace edm {
            input_->processHistoryRegistry().reducedProcessHistoryID(lumiPrincipal.aux().processHistoryID()) ==
                input_->processHistoryRegistry().reducedProcessHistoryID(
                    input_->luminosityBlockAuxiliary()->processHistoryID()));
+    //If a file was opened and the LuminosityBlock is continuing
+    // we may need to do the update
+    lumiPrincipal.possiblyUpdateAfterAddition(preg());
     lumiPrincipal.mergeAuxiliary(*input_->luminosityBlockAuxiliary());
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
@@ -2386,6 +2384,8 @@ namespace edm {
     //TODO this will have to become per stream
     auto& event = principalCache_.eventPrincipal(iStreamIndex);
     StreamContext streamContext(event.streamID(), &processContext_);
+    // a new file may have been read since the last time this event was used
+    event.possiblyUpdateAfterAddition(preg());
 
     SendSourceTerminationSignalIfException sentry(actReg_.get());
     input_->readEvent(event, streamContext);

FWCore/Framework/src/Principal.cc

@@ -149,7 +149,7 @@ namespace edm {
   }
 
   void Principal::possiblyUpdateAfterAddition(std::shared_ptr<ProductRegistry const> iProd) {
-    if (iProd.get() != preg_.get()) {
+    if (iProd.get() != preg_.get() || iProd->cacheIdentifier() != preg_->cacheIdentifier()) {
       preg_ = iProd;
       adjustIndexesAfterProductRegistryAddition();
     }

FWCore/Framework/src/PrincipalCache.cc

@@ -39,28 +39,4 @@ namespace edm {
     assert(iStreamIndex < eventPrincipals_.size());
     eventPrincipals_[iStreamIndex] = ep;
   }
-
-  void PrincipalCache::adjustEventsToNewProductRegistry(std::shared_ptr<ProductRegistry const> reg) {
-    for (auto& eventPrincipal : eventPrincipals_) {
-      if (eventPrincipal) {
-        eventPrincipal->possiblyUpdateAfterAddition(reg);
-      }
-    }
-  }
-
-  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->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->possiblyUpdateAfterAddition(iReg);
-      tempLumiPrincipals.emplace_back(std::move(p));
-    }
-  }
-
 }  // namespace edm

FWCore/Integration/test/provenance_prod_cfg.py

@@ -3,6 +3,8 @@
 
 parser = ArgumentParser(description='Write streamer output file for provenance read test')
 parser.add_argument("--consumeProd2", help="add an extra producer to the job and drop on output", action="store_true")
+parser.add_argument("--diffRun", help="use a different run number", action="store_true")
+parser.add_argument("--diffLumi", help="use a different LuminosityBlock number", action="store_true")
 args = parser.parse_args()
 
 
@@ -11,11 +13,16 @@
 from FWCore.Modules.modules import EmptySource
 
 runNumber = 1
+lumiNumber = 1
+if args.diffRun:
+    runNumber = 2
+if args.diffLumi:
+    lumiNumber=2
 eventNumber = 1
 if args.consumeProd2:
     eventNumber = 2
 
-process.source = EmptySource(firstRun = runNumber, firstEvent = eventNumber )
+process.source = EmptySource(firstRun = runNumber, firstEvent = eventNumber, firstLuminosityBlock = lumiNumber )
 
 from FWCore.Framework.modules import AddIntsProducer, IntProducer
 

FWCore/Integration/test/provenance_test.sh

@@ -9,4 +9,13 @@ function die { echo $1: status $2 ;  exit $2; }
 # The check makes sure the provenance in the ProductRegistry is properly updated when the new file is read
 cmsRun ${SCRAM_TEST_PATH}/provenance_prod_cfg.py || die 'Failed in provenance_prod_cfg.py' $?
 cmsRun ${SCRAM_TEST_PATH}/provenance_prod_cfg.py  --consumeProd2 || die 'Failed in provenance_prod_cfg.py  --consumeProd2' $?
-cmsRun ${SCRAM_TEST_PATH}/provenance_check_cfg.py || die 'Failed test of provenance' $?
+cmsRun ${SCRAM_TEST_PATH}/provenance_check_cfg.py || die 'Failed test of provenance' $?
+
+cmsRun ${SCRAM_TEST_PATH}/provenance_prod_cfg.py  --consumeProd2 --diffRun || die 'Failed in provenance_prod_cfg.py  --consumeProd2 --diffRun' $?
+cmsRun ${SCRAM_TEST_PATH}/provenance_check_cfg.py || die 'Failed test of provenance with diffRun' $?
+
+cmsRun ${SCRAM_TEST_PATH}/provenance_prod_cfg.py  --consumeProd2 --diffLumi || die 'Failed in provenance_prod_cfg.py  --consumeProd2 --diffLumi' $?
+cmsRun ${SCRAM_TEST_PATH}/provenance_check_cfg.py || die 'Failed test of provenance with diffLumi' $?
+
+cmsRun ${SCRAM_TEST_PATH}/provenance_prod_cfg.py  --consumeProd2 --diffLumi --diffRun|| die 'Failed in provenance_prod_cfg.py  --consumeProd2 --diffLumi --diffRun' $?
+cmsRun ${SCRAM_TEST_PATH}/provenance_check_cfg.py || die 'Failed test of provenance with diffLumi & diffRun' $?

FWCore/TestProcessor/src/TestSourceProcessor.cc

@@ -224,15 +224,12 @@ namespace edm::test {
     //make the services available
     ServiceRegistry::Operate operate(serviceToken_);
 
+    auto oldCacheID = source_->productRegistry().cacheIdentifier();
     fb_ = source_->readFile();
     //incase the input's registry changed
-    const size_t size = preg_->size();
-    preg_->merge(source_->productRegistry(), fb_ ? fb_->fileName() : std::string());
-    if (size < preg_->size()) {
-      principalCache_.adjustIndexesAfterProductRegistryAddition(preg_);
+    if (oldCacheID != source_->productRegistry().cacheIdentifier()) {
+      preg_->merge(source_->productRegistry(), fb_ ? fb_->fileName() : std::string());
     }
-    principalCache_.adjustEventsToNewProductRegistry(preg_);
-
     source_->fillProcessBlockHelper();
     ProcessBlockPrincipal& processBlockPrincipal = principalCache_.inputProcessBlockPrincipal();
     while (source_->nextProcessBlock(processBlockPrincipal)) {
@@ -263,6 +260,7 @@ namespace edm::test {
 
     //NOTE: should probably handle merging as well
     runPrincipal_ = principalCache_.getAvailableRunPrincipalPtr();
+    runPrincipal_->possiblyUpdateAfterAddition(preg_);
     runPrincipal_->setAux(*source_->runAuxiliary());
     source_->readRun(*runPrincipal_, *historyAppender_);
 
@@ -280,6 +278,7 @@ namespace edm::test {
 
     lumiPrincipal_ = principalCache_.getAvailableLumiPrincipalPtr();
     assert(lumiPrincipal_);
+    lumiPrincipal_->possiblyUpdateAfterAddition(preg_);
     lumiPrincipal_->setAux(*source_->luminosityBlockAuxiliary());
     source_->readLuminosityBlock(*lumiPrincipal_, *historyAppender_);
 
@@ -295,6 +294,7 @@ namespace edm::test {
     ServiceRegistry::Operate operate(serviceToken_);
 
     auto& event = principalCache_.eventPrincipal(0);
+    event.possiblyUpdateAfterAddition(preg_);
     StreamContext streamContext(event.streamID(), &processContext_);
 
     source_->readEvent(event, streamContext);