Fix using unscheduled execution with a SubProcess

FWCore/Framework/src/ProductResolvers.h

@@ -267,7 +267,7 @@ namespace edm {
                                  bool skipCurrentProcess,
                                  SharedResourcesAcquirer* sra,
                                  ModuleCallingContext const* mcc) const override {
-      realProduct_->prefetchAsync( waitTask, principal, skipCurrentProcess, sra, mcc);
+      realProduct_->prefetchAsync( waitTask, *parentPrincipal_, skipCurrentProcess, sra, mcc);
     }
     virtual bool unscheduledWasNotRun_() const override {return realProduct_->unscheduledWasNotRun();}
     virtual bool productUnavailable_() const override {return realProduct_->productUnavailable();}

FWCore/Framework/src/SystemTimeKeeper.cc

@@ -122,6 +122,16 @@ SystemTimeKeeper::pathTiming(StreamContext const& iStream,
   return m_streamPathTiming[iStream.streamID().value()][iPath.pathID()+offset];
 }
 
+//NOTE: Have to check bounds rather than ProcessContext on the
+// module callbacks because the ProcessContext could be for a
+// SubProcess which requested an usncheduled execution of a
+// module in a parent process. In that case the ProcessContext
+// is for the SubProcess but the module is is for the parent process.
+inline bool
+SystemTimeKeeper::checkBounds(unsigned int id) const {
+  return id >= m_minModuleID and id <m_modules.size()+ m_minModuleID;
+}
+
 
 
 void
@@ -164,7 +174,7 @@ SystemTimeKeeper::stopPath(StreamContext const& iStream,
 
 void
 SystemTimeKeeper::startModuleEvent(StreamContext const& iStream, ModuleCallingContext const& iModule) {
-  if(m_processContext == iStream.processContext()) {
+  if(checkBounds(iModule.moduleDescription()->id())) {
     auto& mod =
     m_streamModuleTiming[iStream.streamID().value()][iModule.moduleDescription()->id()-m_minModuleID];
     mod.m_timer.start();
@@ -173,7 +183,7 @@ SystemTimeKeeper::startModuleEvent(StreamContext const& iStream, ModuleCallingCo
 }
 void SystemTimeKeeper::stopModuleEvent(StreamContext const& iStream,
                                        ModuleCallingContext const& iModule) {
-  if(m_processContext == iStream.processContext()) {
+  if(checkBounds(iModule.moduleDescription()->id())) {
     auto& mod =
     m_streamModuleTiming[iStream.streamID().value()][iModule.moduleDescription()->id()-m_minModuleID];
     auto times = mod.m_timer.stop();
@@ -188,7 +198,7 @@ void SystemTimeKeeper::stopModuleEvent(StreamContext const& iStream,
 }
 void SystemTimeKeeper::pauseModuleEvent(StreamContext const& iStream,
                                         ModuleCallingContext const& iModule) {
-  if(m_processContext == iStream.processContext()) {
+  if(checkBounds(iModule.moduleDescription()->id())) {
     auto& mod =
     m_streamModuleTiming[iStream.streamID().value()][iModule.moduleDescription()->id()-m_minModuleID];
     auto times = mod.m_timer.stop();
@@ -204,7 +214,7 @@ void SystemTimeKeeper::pauseModuleEvent(StreamContext const& iStream,
 void
 SystemTimeKeeper::restartModuleEvent(StreamContext const& iStream,
                                      ModuleCallingContext const& iModule) {
-  if(m_processContext == iStream.processContext()) {
+  if(checkBounds(iModule.moduleDescription()->id())) {
     auto& mod =
     m_streamModuleTiming[iStream.streamID().value()][iModule.moduleDescription()->id()-m_minModuleID];
     mod.m_timer.start();

FWCore/Framework/src/SystemTimeKeeper.h

@@ -92,6 +92,7 @@ namespace edm {
     const SystemTimeKeeper& operator=(const SystemTimeKeeper&) = delete; // stop default
 
     PathTiming& pathTiming(StreamContext const&, PathContext const&);
+    bool checkBounds(unsigned int id) const;
 
     // ---------- member data --------------------------------
     std::vector<WallclockTimer> m_streamEventTimer;

FWCore/Integration/test/run_SubProcess.sh

@@ -34,6 +34,9 @@ pushd ${LOCAL_TMP_DIR}
   grep "Sharing" testSubProcessEventSetup1.log >> testSubProcessEventSetup1.grep.txt
   diff ${LOCAL_TEST_DIR}/unit_test_outputs/testSubProcessEventSetup1.grep.txt  testSubProcessEventSetup1.grep.txt || die "comparing testSubProcessEventSetup1.grep.txt" $?
 
+  echo cmsRun testSubProcessUnscheduled_cfg.py
+  cmsRun -p ${LOCAL_TEST_DIR}/testSubProcessUnscheduled_cfg.py || die "cmsRun testSubProcessUnscheduled_cfg.py" $?
+
 popd
 
 exit 0

FWCore/Integration/test/testSubProcessUnscheduled_cfg.py

@@ -0,0 +1,38 @@
+import FWCore.ParameterSet.Config as cms
+
+process = cms.Process( "TEST" )
+
+process.options = cms.untracked.PSet(
+    allowUnscheduled = cms.untracked.bool( True ),
+    wantSummary = cms.untracked.bool(True)
+)
+
+process.source = cms.Source( "EmptySource" )
+
+process.maxEvents = cms.untracked.PSet(
+    input = cms.untracked.int32( 10 )
+)
+
+# producer
+process.two = cms.EDProducer("BusyWaitIntProducer", ivalue = cms.int32(2), iterations=cms.uint32(5*1000))
+
+# producer
+process.four = cms.EDProducer("BusyWaitIntProducer", ivalue = cms.int32(4), iterations=cms.uint32(10*1000))
+
+# producer
+process.ten = cms.EDProducer("BusyWaitIntProducer", ivalue = cms.int32(10), iterations=cms.uint32(2*1000))
+
+process.adder = cms.EDProducer("AddIntsProducer", labels = cms.vstring('two','ten'))
+
+subprocess = cms.Process("SUB")
+process.addSubProcess( cms.SubProcess(
+    process = subprocess, 
+    SelectEvents = cms.untracked.PSet(), 
+    outputCommands = cms.untracked.vstring()
+) )
+
+subprocess.options = cms.untracked.PSet( wantSummary = cms.untracked.bool(True))
+# module, reads products from 'adder' in the parent process
+subprocess.final = cms.EDProducer("AddIntsProducer", labels = cms.vstring('adder'))
+
+subprocess.subpath = cms.Path( subprocess.final )