Merge pull request #48052 from Dr15Jones/signalDelayedGet

cmsbuild · web-flow · commit 29e8ec79307e · 2025-05-13T14:04:54.000+02:00
Make delayed get signal work with Refs
diff --git a/FWCore/Framework/src/ProductResolvers.cc b/FWCore/Framework/src/ProductResolvers.cc
@@ -11,6 +11,7 @@
 #include "FWCore/Framework/interface/DelayedReader.h"
 #include "FWCore/Framework/interface/TransitionInfoTypes.h"
 #include "FWCore/Framework/interface/ProductProvenanceRetriever.h"
+#include "FWCore/ServiceRegistry/interface/CurrentModuleOnThread.h"
 #include "DataFormats/Provenance/interface/BranchKey.h"
 #include "DataFormats/Provenance/interface/ParentageRegistry.h"
 #include "FWCore/MessageLogger/interface/MessageLogger.h"
@@ -176,11 +177,15 @@ namespace edm {
         // The file may already be closed so the reader is invalid
         return;
       }
-      if (mcc and branchType == InEvent and aux_) {
-        aux_->preModuleDelayedGetSignal_.emit(*(mcc->getStreamContext()), *mcc);
+      auto context = mcc;
+      if (!context) {
+        context = CurrentModuleOnThread::getCurrentModuleOnThread();
+      }
+      if (context and branchType == InEvent and aux_) {
+        aux_->preModuleDelayedGetSignal_.emit(*(context->getStreamContext()), *context);
       }
 
-      auto sentry(make_sentry(mcc, [this, branchType](ModuleCallingContext const* iContext) {
+      auto sentry(make_sentry(context, [this, branchType](ModuleCallingContext const* iContext) {
         if (branchType == InEvent and aux_) {
           aux_->postModuleDelayedGetSignal_.emit(*(iContext->getStreamContext()), *iContext);
         }
@@ -194,13 +199,13 @@ namespace edm {
         if (not productResolved()) {
           try {
             //another thread could have beaten us here
-            setProduct(reader->getProduct(productDescription().branchID(), &principal, mcc));
+            setProduct(reader->getProduct(productDescription().branchID(), &principal, context));
           } catch (cms::Exception& e) {
-            extendException(e, productDescription(), mcc);
+            extendException(e, productDescription(), context);
             throw;
           } catch (std::exception const& e) {
             auto newExcept = edm::Exception(errors::StdException) << e.what();
-            extendException(newExcept, productDescription(), mcc);
+            extendException(newExcept, productDescription(), context);
             throw newExcept;
           }
         }
diff --git a/FWCore/Services/plugins/Tracer.cc b/FWCore/Services/plugins/Tracer.cc
@@ -1448,7 +1448,7 @@ void Tracer::preModuleEventDelayedGet(StreamContext const& sc, ModuleCallingCont
   for (unsigned int i = 0; i < nIndents; ++i) {
     out << indention_;
   }
-  out << " starting: delayed processing event for module: stream = " << sc.streamID() << " label = '"
+  out << " starting: delayed get while processing event for module: stream = " << sc.streamID() << " label = '"
       << mcc.moduleDescription()->moduleLabel() << "' id = " << mcc.moduleDescription()->id();
   if (dumpContextForLabels_.find(mcc.moduleDescription()->moduleLabel()) != dumpContextForLabels_.end()) {
     out << "\n" << sc;
@@ -1463,7 +1463,7 @@ void Tracer::postModuleEventDelayedGet(StreamContext const& sc, ModuleCallingCon
   for (unsigned int i = 0; i < nIndents; ++i) {
     out << indention_;
   }
-  out << " finished: delayed processing event for module: stream = " << sc.streamID() << " label = '"
+  out << " finished: delayed get while processing event for module: stream = " << sc.streamID() << " label = '"
       << mcc.moduleDescription()->moduleLabel() << "' id = " << mcc.moduleDescription()->id();
   if (dumpContextForLabels_.find(mcc.moduleDescription()->moduleLabel()) != dumpContextForLabels_.end()) {
     out << "\n" << sc;
diff --git a/FWCore/Services/scripts/edmTracerCompactLogViewer.py b/FWCore/Services/scripts/edmTracerCompactLogViewer.py
@@ -479,11 +479,11 @@ def textIfTransform(self):
         return ''
     def text(self, context):
         return f'{self.textPrefix(context)} {self.textSpecial()}{self.textIfTransform()}: {self.textPostfix()}'
-    def _preJson(self, activity, counter, data, mayUseTemp = False):
+    def _preJson(self, activity, counter, data, mayUseTemp = False, isSrc = False):
         index = self.index
         found = False
         if mayUseTemp:
-            compare = lambda x: x['type'] == self.transition and x['id'] == self.index and x['mod'] == self.moduleID and x['call'] == self.callID and (x['act'] == Activity.temporary or x['act'] == Activity.externalWork)
+            compare = lambda x: x['type'] == self.transition and x['id'] == self.index and x['mod'] == self.moduleID and x['call'] == self.callID and ((x.get('isSrc') != None) == isSrc ) and (x['act'] == Activity.temporary or x['act'] == Activity.externalWork)
             if transitionIsGlobal(self.transition):
                 item,slot = data.findLastInModGlobals(index, self.moduleID, compare)
             else:
@@ -501,8 +501,8 @@ def _preJson(self, activity, counter, data, mayUseTemp = False):
                 slot = data.findOpenSlotInModStreams(index, self.moduleID)
         slot.append(jsonModuleTransition(type=self.transition, id=self.index, modID=self.moduleID, callID=self.callID, activity=activity, start=self.time))
         return slot[-1]
-    def _postJson(self, counter, data, injectAfter = None):
-        compare = lambda x: x['id'] == self.index and x['mod'] == self.moduleID and x['call'] == self.callID and x['type'] == self.transition
+    def _postJson(self, counter, data, injectAfter = None, isSrc = False):
+        compare = lambda x: x['id'] == self.index and x['mod'] == self.moduleID and x['call'] == self.callID and x['type'] == self.transition and ((x.get('isSrc') != None) == isSrc )
         index = self.index
         if transitionIsGlobal(self.transition):
             item,slot = data.findLastInModGlobals(index, self.moduleID, compare)
@@ -582,28 +582,37 @@ def jsonInfo(self, counter, data):
         return self._postJson(counter,data)
 
 class PreEDModuleEventDelayedGetParser(EDModuleTransitionParser):
-    def __init__(self, payload, names):
+    def __init__(self, payload, names, moduleCentric):
         super().__init__(payload, names)
+        self._moduleCentric = moduleCentric
     def textSpecial(self):
         return "starting delayed get"
     def jsonInfo(self, counter, data):
-        return self._preJson(Activity.delayedGet, counter,data)
+        #a pre delayed get is effectively a post transition as the system assumes one state per module per transition instance
+        if self._moduleCentric:
+            #inject a dummy at end of the same slot to guarantee module run is in that slot
+            return self._postJson(counter, data, jsonModuleTransition(type=self.transition, id=self.index, modID=self.moduleID, callID=self.callID, activity=Activity.temporary, start=self.time))
+        return self._postJson(counter,data)
+        #return self._preJson(Activity.delayedGet, counter,data)
 
 class PostEDModuleEventDelayedGetParser(EDModuleTransitionParser):
-    def __init__(self, payload, names):
+    def __init__(self, payload, names, moduleCentric):
         super().__init__(payload, names)
+        self._moduleCentric = moduleCentric
     def textSpecial(self):
         return "finished delayed get"
     def jsonInfo(self, counter, data):
-        return self._postJson(counter,data)
+        return self._preJson(Activity.process, counter,data, mayUseTemp=self._moduleCentric)
+        #return self._postJson(counter,data)
 
 class PreEventReadFromSourceParser(EDModuleTransitionParser):
-    def __init__(self, payload, names):
+    def __init__(self, payload, names, moduleCentric):
         super().__init__(payload, names)
+        self._moduleCentric = moduleCentric
     def textSpecial(self):
         return "starting read from source"
     def jsonInfo(self, counter, data):
-        slot = self._preJson(Activity.process, counter,data)
+        slot = self._preJson(Activity.process, counter,data, mayUseTemp=self._moduleCentric, isSrc=True)
         slot['isSrc'] = True
         return slot
 
@@ -613,7 +622,7 @@ def __init__(self, payload, names):
     def textSpecial(self):
         return "finished read from source"
     def jsonInfo(self, counter, data):
-        return self._postJson(counter,data)
+        return self._postJson(counter,data, isSrc=True)
 
 class ESModuleTransitionParser(object):
     def __init__(self, payload, moduleNames, esModuleNames, recordNames):
@@ -747,11 +756,11 @@ def lineParserFactory (step, payload, moduleNames, esModuleNames, recordNames, f
     if step == 'a':
         return PostEDModuleAcquireParser(payload, moduleNames, moduleCentric)
     if step == 'D':
-        return PreEDModuleEventDelayedGetParser(payload, moduleNames)
+        return PreEDModuleEventDelayedGetParser(payload, moduleNames, moduleCentric)
     if step == 'd':
-        return PostEDModuleEventDelayedGetParser(payload, moduleNames)
+        return PostEDModuleEventDelayedGetParser(payload, moduleNames, moduleCentric)
     if step == 'R':
-        return PreEventReadFromSourceParser(payload, moduleNames)
+        return PreEventReadFromSourceParser(payload, moduleNames, moduleCentric)
     if step == 'r':
         return PostEventReadFromSourceParser(payload, moduleNames)
     if step == 'N':
