Merge pull request #47602 from Dr15Jones/justInTimePrincipalUpdates

Just in time Principal updates

Author: cmsbuild

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/Modules/src/ProvenanceCheckerOutputModule.cc

@@ -172,6 +172,19 @@ namespace edm {
                                                 : "")
           << (!missingFromReg.empty() ? " Have missing info from ProductRegistry.\n" : "");
     }
+
+    //check consistency with all Intervals
+    if (e.productRegistry().cacheIdentifier() != e.getRun().productRegistry().cacheIdentifier()) {
+      throw cms::Exception("ProvenanceError")
+          << "The registry cache id for Event ( " << e.productRegistry().cacheIdentifier()
+          << " ) does not match the one for Run ( " << e.getRun().productRegistry().cacheIdentifier() << " )";
+    }
+    if (e.productRegistry().cacheIdentifier() != e.getLuminosityBlock().productRegistry().cacheIdentifier()) {
+      throw cms::Exception("ProvenanceError")
+          << "The registry cache id for Event ( " << e.productRegistry().cacheIdentifier()
+          << " ) does not match the one for LuminosityBlock ( "
+          << e.getLuminosityBlock().productRegistry().cacheIdentifier() << " )";
+    }
   }
 
   //

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);