Merge pull request #49170 from gartung/gartung-PerTools-AllocMonitorAnalyze

PerfTools/ModuleAllocMonitor: Add source c++type to moduleAllocMonitor.log, sum all source activity for each event in source module

Author: cmsbuild

PerfTools/AllocMonitor/plugins/moduleAlloc_setupFile.cc

@@ -451,9 +451,12 @@ namespace edm::service::moduleAlloc {
 
     auto sourceCtrPtr = std::make_shared<ModuleCtrDtr>();
     auto& sourceCtr = *sourceCtrPtr;
-    iRegistry.watchPreSourceConstruction([&sourceCtr, beginTime, iFilter](auto const&) {
+    auto sourceTypePtr = std::make_shared<std::string>("Unknown");
+    iRegistry.watchPreSourceConstruction([&sourceCtr, sourceTypePtr, beginTime, iFilter](auto const& md) {
       auto const t = duration_cast<duration_t>(now() - beginTime).count();
       sourceCtr.beginConstruction = t;
+      // Capture source module type information
+      *sourceTypePtr = md.moduleName();
       iFilter->startOnThread();
     });
     iRegistry.watchPostSourceConstruction([&sourceCtr, beginTime, iFilter](auto const&) {
@@ -495,6 +498,7 @@ namespace edm::service::moduleAlloc {
                                 esModuleTypesPtr,
                                 moduleCtrDtrPtr,
                                 sourceCtrPtr,
+                                sourceTypePtr = std::move(sourceTypePtr),
                                 beginTime,
                                 beginModuleAlloc,
                                 addDataInDtr](auto&) mutable {
@@ -515,6 +519,14 @@ namespace edm::service::moduleAlloc {
         esModuleLabelsPtr.reset();
         esModuleTypesPtr.reset();
       }
+      {
+        std::ostringstream oss;
+        std::vector<std::string> sourceTypeList{1, *sourceTypePtr};
+        std::vector<std::string> sourceNames{1, "source"};
+        moduleIdToLabelAndType(oss, sourceNames, sourceTypeList, 'S', "Source ID", "Source label", "Source type");
+        logFile->write(oss.str());
+        sourceTypePtr.reset();
+      }
       {
         auto const moduleAllocStart = duration_cast<duration_t>(beginModuleAlloc - beginTime).count();
         auto msg = assembleMessage<Step::preFrameworkTransition>(

PerfTools/AllocMonitor/scripts/edmModuleAllocJsonToCircles.py

@@ -38,15 +38,46 @@ def processModuleTransition(moduleLabel, moduleType, moduleInfo, transitionType,
     Any missing field defaults to 0.
 
     Note: Entries with record names are excluded as they belong to EventSetup transition only.
+    
+    For the "source" module and "event" transition, sum all alloc records with the same 
+    run, lumi, and event number.
     """
     moduleKey = UniqueKey(moduleLabel, moduleType, "")
     moduleTransition[moduleKey] = {"cpptype": moduleType, "allocs": []}
-    for entry in moduleInfo:
-        # Only process entries that match the transition type AND don't have record names
-        # (entries with record names are EventSetup only)
-        if (entry.get("transition", None) == transitionType and
-            not ("record" in entry and "name" in entry["record"])):
-            moduleTransition[moduleKey]["allocs"].append(entry.get("alloc", {}))
+    
+    # Special handling for "source" module with "event" transition
+    if moduleLabel == "source" and transitionType == "event":
+        # Group entries by (run, lumi, event)
+        event_groups = {}
+        for entry in moduleInfo:
+            if (entry.get("transition", None) == transitionType and
+                not ("record" in entry and "name" in entry["record"])):
+                sync = entry.get("sync", {})
+                key = (sync.get("run", 0), sync.get("lumi", 0), sync.get("event", 0))
+                if key not in event_groups:
+                    event_groups[key] = []
+                event_groups[key].append(entry.get("alloc", {}))
+        
+        # Sum allocations for each event group
+        for event_key, allocs in event_groups.items():
+            summed_alloc = {
+                "added": sum(a.get("added", 0) for a in allocs),
+                "nAlloc": sum(a.get("nAlloc", 0) for a in allocs),
+                "nDealloc": sum(a.get("nDealloc", 0) for a in allocs),
+                "maxTemp": sum(a.get("maxTemp", 0) for a in allocs),
+                "max1Alloc": sum(a.get("max1Alloc", 0) for a in allocs)
+            }
+            moduleTransition[moduleKey]["allocs"].append(summed_alloc)
+    else:
+        # Original processing for other modules/transitions
+        for entry in moduleInfo:
+            # Only process entries that match the transition type AND don't have record names
+            # (entries with record names are EventSetup only)
+            if (entry.get("transition", None) == transitionType and
+                not ("record" in entry and "name" in entry["record"]) and
+                entry.get("activity") == "process"):
+                moduleTransition[moduleKey]["allocs"].append(entry.get("alloc", {}))
+    
     moduleTransition[moduleKey]["nTransitions"] = len(moduleTransition[moduleKey]["allocs"])
 
 def processESModuleTransition(moduleLabel, moduleType, moduleInfo, moduleTransition):
@@ -206,6 +237,7 @@ def formatToCircles(moduleTransitions):
         eventCount = moduleTransitions['event'].get(eventKey, {}).get("nTransitions", 0)
         # Set events to 1 if it's 0 to prevent NaNs in Circles visualization
         module["events"] = max(eventCount, 1)
+        doc["total"]["events"] = max(doc["total"]["events"], module["events"])
         doc["modules"].append(module)
 
     return doc
@@ -237,8 +269,9 @@ def main(args):
             for moduleLabel, moduleInfo in doc["modules"].items():
                 processESModuleTransition(moduleLabel, moduleTypes[moduleLabel], moduleInfo, moduleTransition)
         else:
+            # Process the "source" module
+            processModuleTransition("source", moduleTypes["source"], doc["source"], transition, moduleTransition)
             # Regular transition processing
-            processModuleTransition("source", "PoolSource", doc["source"], transition, moduleTransition)
             for moduleLabel, moduleInfo in doc["modules"].items():
                 processModuleTransition(moduleLabel, moduleTypes[moduleLabel], moduleInfo, transition, moduleTransition)
         moduleTransitions[transition] = moduleTransition

PerfTools/AllocMonitor/scripts/edmModuleAllocMonitorAnalyze.py

@@ -19,29 +19,29 @@ def printHelp():
   This script will output a more human readable form of the data in the log file.'''
     return s
 
-#these values come from moduleALloc_setupFile.cc
-#enum class Step : char {
-#  preSourceTransition = 'S',
-#  postSourceTransition = 's',
-#  preModulePrefetching = 'P',
-#  postModulePrefetching = 'p',
-#  preModuleEventAcquire = 'A',
-#  postModuleEventAcquire = 'a',
-#  preModuleTransition = 'M',
-#  preEventReadFromSource = 'R',
-#  postEventReadFromSource = 'r',
-#  preModuleEventDelayedGet = 'D',
-#  postModuleEventDelayedGet = 'd',
-#  postModuleTransition = 'm',
-#  preESModulePrefetching = 'Q',
-#  postESModulePrefetching = 'q',
-#  preESModule = 'N',
-#  postESModule = 'n',
-#  preESModuleAcquire = 'B',
-#  postESModuleAcquire = 'b',
-#  preFrameworkTransition = 'F',
-#  postFrameworkTransition = 'f'
-#};
+# These values come from moduleAlloc_setupFile.cc
+# enum class Step : char {
+#   preSourceTransition = 'S',
+#   postSourceTransition = 's',
+#   preModulePrefetching = 'P',
+#   postModulePrefetching = 'p',
+#   preModuleEventAcquire = 'A',
+#   postModuleEventAcquire = 'a',
+#   preModuleTransition = 'M',
+#   preEventReadFromSource = 'R',
+#   postEventReadFromSource = 'r',
+#   preModuleEventDelayedGet = 'D',
+#   postModuleEventDelayedGet = 'd',
+#   postModuleTransition = 'm',
+#   preESModulePrefetching = 'Q',
+#   postESModulePrefetching = 'q',
+#   preESModule = 'N',
+#   postESModule = 'n',
+#   preESModuleAcquire = 'B',
+#   postESModuleAcquire = 'b',
+#   preFrameworkTransition = 'F',
+#   postFrameworkTransition = 'f'
+# };
 
 
 kMicroToSec = 0.000001
@@ -192,6 +192,16 @@ def textPrefix_(time, indentLevel):
     return f'{time:>11} '+"++"*indentLevel
 
 class AllocInfo(object):
+    """Container for memory allocation information from CMSSW module transitions.
+    
+    Attributes:
+        nAllocs: Number of memory allocations
+        nDeallocs: Number of memory deallocations  
+        added: Net memory added (in bytes)
+        minTemp: Minimum temporary memory usage
+        maxTemp: Maximum temporary memory usage
+        max1Alloc: Largest single allocation
+    """
     def __init__(self,payload):
         self.nAllocs = int(payload[0])
         self.nDeallocs = int(payload[1])
@@ -333,6 +343,7 @@ def toSimpleDict(self):
             l = None
             if m == 'source':
                 l = dct['source']
+                cpptypes[m]=self._cpptypes[m]
             elif m == 'clearEvent':
                 l = dct['clearEvent']
             else:
@@ -526,7 +537,7 @@ def jsonVisInfo(self,  data):
 
 
 class SourceTransitionParser(object):
-    def __init__(self, payload):
+    def __init__(self, payload, sourceInfos):
         self.transition = int(payload[0])
         if self.transition == Phase.getNextTransition:
             self.time = int(payload[1])
@@ -556,9 +567,10 @@ def text(self, context):
         return f'{self.textPrefix()} {self.textSpecial()}: {self.textPostfix()}'
 
 class PreSourceTransitionParser(SourceTransitionParser):
-    def __init__(self, payload, moduleCentric):
+    def __init__(self, payload, sourceInfos, moduleCentric):
         self._moduleCentric = moduleCentric
-        super().__init__(payload)
+        super().__init__(payload, sourceInfos)
+        self._sourceInfo = sourceInfos[0]
     def textSpecial(self):
         return "starting"
     def jsonInfo(self, syncs, temp, data):
@@ -607,22 +619,23 @@ def jsonVisInfo(self,  data):
                 data.findOpenSlotInModGlobals(index,0).append(container[-1])
 
 class PostSourceTransitionParser(SourceTransitionParser):
-    def __init__(self, payload, moduleCentric):
-        super().__init__(payload)
+    def __init__(self, payload, sourceInfos, moduleCentric):
+        super().__init__(payload, sourceInfos)
         if self.index == -1:
             self.allocInfo = AllocInfo(payload[2:])
         else:
             self.allocInfo = AllocInfo(payload[3:])
         self._moduleCentric = moduleCentric
+        self._sourceInfo = sourceInfos[0]
     def textSpecial(self):
         return "finished"
     def jsonInfo(self, syncs, temp, data):
         start = temp.findTime("source", self.transition, self.index)
         #we do not know the sync yet so have to wait until the framework transition
         if self.transition in [ Phase.construction, Phase.getNextTransition, Phase.destruction, Phase.openFile]:
-            data.insert( "source" , "PoolSource", start, self.time, self.transition, self.index, (0,) , Activity.process, self.allocInfo)
+            data.insert( "source" , self._sourceInfo._cpptype, start, self.time, self.transition, self.index, (0,) , Activity.process, self.allocInfo)
         else:
-            data.insert( "source" , "PoolSource", start, self.time, self.transition, self.index, self.index , Activity.process, self.allocInfo)
+            data.insert( "source" , self._sourceInfo._cpptype, start, self.time, self.transition, self.index, self.index , Activity.process, self.allocInfo)
     def jsonVisInfo(self,  data):
         index = self.index
         if self.transition == Phase.Event:
@@ -812,7 +825,7 @@ def jsonVisInfo(self,  data):
         return self._postJsonVis(data, self.allocInfo)
     def jsonInfo(self, syncs, temp, data):
         start = temp.findTime(self.moduleInfo._name+'source', self.transition, self.index)
-        data.insert( "source" , "PoolSource", start, self.time, self.transition, self.index, syncs.get(self.transition, self.index) , Activity.delayedGet, self.allocInfo)
+        data.insert( "source" , self.moduleInfo._cpptype, start, self.time, self.transition, self.index, syncs.get(self.transition, self.index) , Activity.delayedGet, self.allocInfo)
 
 class ESModuleTransitionParser(object):
     def __init__(self, payload, moduleInfos, esModuleInfos, recordNames):
@@ -900,16 +913,16 @@ def jsonVisInfo(self,  data):
         return self._postJsonVis(data, self.allocInfo)
 
 
-def lineParserFactory (step, payload, moduleInfos, esModuleInfos, recordNames, moduleCentric):
+def lineParserFactory (step, payload, moduleInfos, esModuleInfos, sourceInfos, recordNames, moduleCentric):
     if step == 'F':
         parser = PreFrameworkTransitionParser(payload)
         return parser
     if step == 'f':
         return PostFrameworkTransitionParser(payload)
     if step == 'S':
-        return PreSourceTransitionParser(payload, moduleCentric)
+        return PreSourceTransitionParser(payload, sourceInfos, moduleCentric)
     if step == 's':
-        return PostSourceTransitionParser(payload, moduleCentric)
+        return PostSourceTransitionParser(payload, sourceInfos, moduleCentric)
     if step == 'M':
         return PreEDModuleTransitionParser(payload, moduleInfos, moduleCentric)
     if step == 'm':
@@ -934,18 +947,18 @@ def lineParserFactory (step, payload, moduleInfos, esModuleInfos, recordNames, m
         return PreESModuleAcquireParser(payload, moduleInfos, esModuleInfos, recordNames)
     if step == 'b':
         return PostESModuleAcquireParser(payload, moduleInfos, esModuleInfos, recordNames)
-    raise LogicError("Unknown step '{}'".format(step))
+    raise ValueError("Unknown step '{}'".format(step))
 
 #----------------------------------------------
-def processingStepsFromFile(f, moduleInfos, esModuleInfos, recordNames, moduleCentric):
+def processingStepsFromFile(f, moduleInfos, esModuleInfos, sourceInfos, recordNames, moduleCentric):
     for rawl in f:
         l = rawl.strip()
         if not l or l[0] == '#':
             continue
         (step,payload) = tuple(l.split(None,1))
         payload=payload.split()
 
-        parser = lineParserFactory(step, payload, moduleInfos, esModuleInfos, recordNames, moduleCentric)
+        parser = lineParserFactory(step, payload, moduleInfos, esModuleInfos, sourceInfos, recordNames, moduleCentric)
         if parser:
             yield parser
     return
@@ -958,6 +971,7 @@ def __init__(self,f, moduleCentric):
         moduleInfos = {}
         esModuleInfos = {}
         recordNames = {}
+        sourceInfos = {}
         for rawl in f:
             l = rawl.strip()
             if l and l[0] == 'M':
@@ -978,6 +992,10 @@ def __init__(self,f, moduleCentric):
                 (id,name,mType)=tuple(l[2:].split())
                 esModuleInfos[int(id)] = ModuleInfo(name,mType)
                 continue
+            if len(l) > 5 and l[0:2] == "#S":
+                (id,name,sType)=tuple(l[2:].split())
+                sourceInfos[0] = ModuleInfo(name,sType)
+                continue
             if len(l) > 5 and l[0:2] == "#R":
                 (id,name)=tuple(l[2:].split())
                 recordNames[int(id)] = name
@@ -990,6 +1008,7 @@ def __init__(self,f, moduleCentric):
         self.numStreams =numStreams
         self._moduleInfos = moduleInfos
         self._esModuleInfos = esModuleInfos
+        self._sourceInfos = sourceInfos
         self._recordNames = recordNames
         self.maxNameSize =0
         for n in moduleInfos.items():
@@ -1004,12 +1023,21 @@ def processingSteps(self):
         Using a generator reduces the memory overhead when parsing a large file.
             """
         self._f.seek(0)
-        return processingStepsFromFile(self._f,self._moduleInfos, self._esModuleInfos, self._recordNames, self._moduleCentric)
+        return processingStepsFromFile(self._f,self._moduleInfos, self._esModuleInfos, self._sourceInfos, self._recordNames, self._moduleCentric)
 
 def textOutput( parser ):
     context = {}
     for p in parser.processingSteps():
         print(p.text(context))
+
+def showSourceTypes( parser ):
+    print("Source Module Types:")
+    print("===================")
+    if hasattr(parser, '_sourceInfos') and parser._sourceInfos:
+        for id, info in parser._sourceInfos.items():
+            print(f"  Source ID {id}: {info._name} (type: {info._cpptype})")
+    else:
+        print("  No source module types found in log file.")
 
 class VisualizationContainers(object):
     def __init__(self):
@@ -1281,7 +1309,7 @@ def jsonVisualizationInfo(parser):
                         time += t["finish"]-t["start"]
             modules[-1]['time']=time
         modules.sort(key= lambda x : x['time'], reverse=True)
-        final['transitions'].append({"name": "source", "slots":sourceSlot})
+        final['transitions'].append({"name": "source", "cpptype": parser._sourceInfos[0]._cpptype, "slots":sourceSlot})
         for m in modules:
             final['transitions'].append(m)
 
@@ -1327,6 +1355,7 @@ def incr(t):
             '#R 1 Record',
             '#M 1 Module ModuleType',
             '#N 1 ESModule ESModuleType',
+            '#S 1 Source SourceType',
              f'F {Phase.startTracing} 0 0 0 0 {incr(t)}',
              f'S {Phase.construction} 0 {incr(t)}',
              f's {Phase.construction} 0 {incr(t)} 1 1 10 0 10 10',
@@ -1493,6 +1522,7 @@ def runTests():
                                      formatter_class=argparse.RawDescriptionHelpFormatter,
                                      epilog=printHelp())
     parser.add_argument('filename',
+                        nargs='?',  # Make filename optional
                         type=argparse.FileType('r'), # open file
                         help='file to process')
     parser.add_argument('-j', '--json',
@@ -1501,7 +1531,7 @@ def runTests():
     parser.add_argument('-s', '--sortBy',
                         default = '',
                         type = str,
-                        help="sort modules by attribute. Alloed values 'nAllocs', 'nDeallocs', 'added', 'minTemp', maxTemp', and 'max1Alloc'")
+                        help="sort modules by attribute. Allowed values 'nAllocs', 'nDeallocs', 'added', 'minTemp', 'maxTemp', and 'max1Alloc'")
 #    parser.add_argument('-w', '--web',
 #                        action='store_true',
 #                        help='''Writes data.js file that can be used with the web based inspector. To use, copy directory ${CMSSW_RELEASE_BASE}/src/FWCore/Services/template/web to a web accessible area and move data.js into that directory.''')
@@ -1511,11 +1541,17 @@ def runTests():
     parser.add_argument('-T', '--test',
                         action='store_true',
                         help='''Run internal tests.''')
+    parser.add_argument('--showSourceTypes',
+                        action='store_true', 
+                        help='''Display source module types.''')
 
     args = parser.parse_args()
 
     if args.test:
         runTests()
+    elif args.filename is None:
+        parser.print_help()
+        sys.exit(1)
     else :
         parser = ModuleAllocCompactFileParser(args.filename, not args.timeOrdered)
 #        if args.json or args.web:
@@ -1528,5 +1564,7 @@ def runTests():
 #                f.close()
         elif args.sortBy:
             print(json.dumps(sortByAttribute(parser, args.sortBy), indent=2))
+        elif args.showSourceTypes:
+            showSourceTypes(parser)
         else:
             textOutput(parser)