Speed up runTheMatrix.py by up to 7x

Configuration/PyReleaseValidation/python/MatrixReader.py

@@ -28,6 +28,9 @@ def __init__(self, opt):
 
         self.noRun = opt.noRun
         self.checkInputs = opt.checkInputs
+        # -l restricts the expansion to the requested workflows; the listing, the
+        # runner and the wmagent injection all act on that same selection
+        self.selected = set(opt.testList) if opt.testList else None
         return
 
     def reset(self, what='all'):
@@ -102,23 +105,24 @@ def reset(self, what='all'):
 
     def makeCmd(self, step):
 
-        cmd = ''
+        cmd = []
         cfg = None
         input = None
         for k,v in step.items():
             if 'no_exec' in k : continue  # we want to really run it ...
-            if k.lower() == 'cfg':
+            klow = k.lower()
+            if klow == 'cfg':
                 cfg = v
                 continue # do not append to cmd, return separately
-            if k.lower() == 'input':
-                input = v 
+            if klow == 'input':
+                input = v
                 continue # do not append to cmd, return separately
-            
+
             #chain the configs
             #if k.lower() == '--python':
             #    v = 'step%d_%s'%(index,v)
-            cmd += ' ' + k + ' ' + str(v)
-        return cfg, input, cmd
+            cmd.append(' ' + k + ' ' + str(v))
+        return cfg, input, ''.join(cmd)
 
     def makeStep(self,step,overrides):
         from Configuration.PyReleaseValidation.relval_steps import merge
@@ -143,8 +147,8 @@ def verifyDefaultInputs(self):
 =============================================================================
                              """.format(sys._getframe(1).f_lineno - 1,wf[0],wf))    
 
-    def readMatrix(self, fileNameIn, useInput=None, refRel=None, fromScratch=None):
-        
+    def readMatrix(self, fileNameIn, useInput=None, refRel=None, fromScratch=None, selected=None):
+
         prefix = self.filesPrefMap[fileNameIn]
 
         print("processing", fileNameIn)
@@ -204,7 +208,12 @@ def readMatrix(self, fileNameIn, useInput=None, refRel=None, fromScratch=None):
                     [(x,refRel) for x in self.relvalModule.baseDataSetRelease]
                     )
 
+        useIBEos = os.getenv("CMSSW_USE_IBEOS","false")=="true"
+        madeCmds = {}
+
         for num, wfInfo in self.relvalModule.workflows.items():
+            if selected is not None and num not in selected:
+                continue
             commands=[]
             wfName = wfInfo[0]
             stepList = wfInfo[1]
@@ -295,8 +304,16 @@ def readMatrix(self, fileNameIn, useInput=None, refRel=None, fromScratch=None):
                     from Configuration.PyReleaseValidation.relval_steps import merge
                     copyStep=merge(addCom+[self.makeStep(self.relvalModule.steps[stepName],stepOverrides)])
                     cfg, input, opts = self.makeCmd(copyStep)
-                else:
+                elif stepOverrides:
                     cfg, input, opts = self.makeCmd(self.makeStep(self.relvalModule.steps[stepName],stepOverrides))
+                else:
+                    # without overrides a step yields the same command in every
+                    # workflow that runs it
+                    made = madeCmds.get(stepName)
+                    if made is None:
+                        made = self.makeCmd(self.relvalModule.steps[stepName])
+                        madeCmds[stepName] = made
+                    cfg, input, opts = made
 
                 if input and cfg :
                     msg = "FATAL ERROR: found both cfg and input for workflow "+str(num)+' step '+stepName
@@ -322,7 +339,7 @@ def readMatrix(self, fileNameIn, useInput=None, refRel=None, fromScratch=None):
                     if self.wm and self.revertDqmio=='yes':
                         cmd=cmd.replace('DQMIO','DQM')
                         cmd=cmd.replace('--filetype DQM','')
-                    if os.getenv("CMSSW_USE_IBEOS","false")=="true":
+                    if useIBEos:
                         cmd="export CMSSW_USE_IBEOS=true; "+cmd
                 commands.append(cmd)
                 ranStepList.append(stepName)
@@ -526,13 +543,13 @@ def prepare(self, useInput=None, refRel='', fromScratch=None):
                 continue
 
             try:
-                self.readMatrix(matrixFile, useInput, refRel, fromScratch)
+                self.readMatrix(matrixFile, useInput, refRel, fromScratch, self.selected)
                 if self.checkInputs:
                     self.verifyDefaultInputs()
             except Exception as e:
                 print("ERROR reading file:", matrixFile, str(e))
                 raise
-            
+
             try:
                 self.createWorkFlows(matrixFile)
             except Exception as e:

Configuration/PyReleaseValidation/python/MatrixUtil.py

@@ -1,3 +1,4 @@
+import copy
 import os
 import subprocess
 
@@ -7,24 +8,131 @@ def __setitem__(self,key,value):
             print("ERROR in Matrix")
             print("overwriting",key,"not allowed")
         else:
-            self.update({float(key):WF(float(key),value)})
+            num=float(key)
+            dict.__setitem__(self,num,WF(num,value))
 
     def addOverride(self,key,override):
         self[key].addOverride(override)
 
+# a fragment step, held as (cfg, howMuch, base) and turned into its dictionary the
+# first time it is read; the upgrade matrix holds a couple of million of them and a
+# run reads a few percent
+class DeferredFragmentStep(tuple):
+    __slots__ = ()
+    # same result, same key order, as merge([{'cfg':cfg},howMuch,base])
+    def expand(self):
+        cfg,howMuch,base = self
+        step = copy.copy(base)
+        step.update(howMuch)
+        step['cfg'] = cfg
+        return step
+
 #the class to collect all possible steps
+# the upgrade matrix holds some two million fragment steps grouped in about 180
+# families, one family per step name spanning every fragment and every upgrade key.
+# every name a family defines ends in '_'+family, so a name identifies its own family,
+# and a family is built the first time one of its names is asked for. anything that
+# walks the whole dictionary builds every family first.
 class Steps(dict):
+    def __init__(self,*args,**kwargs):
+        dict.__init__(self,*args,**kwargs)
+        self.familyBuilder = None
+        self.pendingFamilies = {}
+        self.familiesByTail = {}
+        self.buildingFamily = False
+
+    def deferFamily(self,family,rows):
+        self.pendingFamilies[family] = rows
+        self.familiesByTail.setdefault(family.rsplit('_',1)[-1],[]).append(family)
+
+    def buildFamily(self,rows):
+        self.buildingFamily = True
+        try:
+            for row in rows:
+                self.familyBuilder(*row)
+        finally:
+            self.buildingFamily = False
+
+    def buildFamiliesFor(self,name):
+        # build every pending family that could define name; True when one was built
+        if not self.pendingFamilies:
+            return False
+        if name.endswith('INPUT'):
+            name = name[:-len('INPUT')]
+        built = False
+        for family in self.familiesByTail.get(name.rsplit('_',1)[-1],()):
+            if not name.endswith('_'+family):
+                continue
+            rows = self.pendingFamilies.pop(family,None)
+            if rows is not None:
+                self.buildFamily(rows)
+                built = True
+        return built
+
+    def buildAllFamilies(self):
+        while self.pendingFamilies:
+            family = next(iter(self.pendingFamilies))
+            self.buildFamily(self.pendingFamilies.pop(family))
+
+    def __missing__(self,key):
+        if self.buildFamiliesFor(key):
+            return dict.__getitem__(self,key)
+        raise KeyError(key)
+
+    def __contains__(self,key):
+        if dict.__contains__(self,key):
+            return True
+        # while a family is being built its own names are new, and the only other
+        # membership test made there is for a name that belongs to no family
+        if self.buildingFamily:
+            return False
+        return self.buildFamiliesFor(key) and dict.__contains__(self,key)
+
+    def __iter__(self):
+        self.buildAllFamilies()
+        return dict.__iter__(self)
+
+    def __len__(self):
+        self.buildAllFamilies()
+        return dict.__len__(self)
+
+    def keys(self):
+        self.buildAllFamilies()
+        return dict.keys(self)
+
+    def values(self):
+        self.buildAllFamilies()
+        return dict.values(self)
+
+    def items(self):
+        self.buildAllFamilies()
+        return dict.items(self)
+
     def __setitem__(self,key,value):
         if key in self:
             print("ERROR in Step")
             print("overwriting",key,"not allowed")
             import sys
             sys.exit(-9)
         else:
-            self.update({key:value})
+            dict.__setitem__(self,key,value)
             # make the python file named <step>.py
             #if not '--python' in value:                self[key].update({'--python':'%s.py'%(key,)})
 
+    def __getitem__(self,key):
+        value=dict.__getitem__(self,key)
+        if type(value) is DeferredFragmentStep:
+            value=value.expand()
+            dict.__setitem__(self,key,value)
+        return value
+
+    # dict.get would bypass __getitem__ and hand out an unexpanded step
+    def get(self,key,default=None):
+        try:
+            return self[key]
+        except KeyError:
+            return default
+
     def overwrite(self,keypair):
         value=self[keypair[1]]
         self.update({keypair[0]:value})
@@ -210,23 +318,14 @@ def __str__(self):
 
 
 # merge dictionaries, with priority on the [0] index
+# the result keeps the type and the key order of the last item, extended with the
+# keys each earlier item adds, and the value of the earliest item that defines a key
 def merge(dictlist,TELL=False):
-    import copy
-    last=len(dictlist)-1
-    if TELL: print(last,dictlist)
-    if last==0:
-        # ONLY ONE ITEM LEFT
-        return copy.copy(dictlist[0])
-    else:
-        reducedlist=dictlist[0:max(0,last-1)]
-        if TELL: print(reducedlist)
-        # make a copy of the last item
-        d=copy.copy(dictlist[last])
-        # update with the last but one item
-        d.update(dictlist[last-1])
-        # and recursively do the rest
-        reducedlist.append(d)
-        return merge(reducedlist,TELL)
+    if TELL: print(len(dictlist)-1,dictlist)
+    d=copy.copy(dictlist[-1])
+    for i in range(len(dictlist)-2,-1,-1):
+        d.update(dictlist[i])
+    return d
 
 def remove(d,key,TELL=False):
     import copy