Allow duplicate wfs

Author: AdrianoDee

Configuration/PyReleaseValidation/python/MatrixRunner.py

@@ -1,19 +1,26 @@
 import os, sys, time
 
+from collections import Counter
+
 from Configuration.PyReleaseValidation.WorkFlow import WorkFlow
 from Configuration.PyReleaseValidation.WorkFlowRunner import WorkFlowRunner
-
+from Configuration.PyReleaseValidation.MatrixUtil import check_dups
 # ================================================================================
 
 class MatrixRunner(object):
 
-    def __init__(self, wfIn=None, nThrMax=4, nThreads=1):
+    def __init__(self, wfIn=None, nThrMax=4, nThreads=1, gpu=False):
 
         self.workFlows = wfIn
 
         self.threadList = []
         self.maxThreads = nThrMax
         self.nThreads = nThreads
+        self.gpu = gpu
+
+        if self.gpu:
+            print("Checks for GPU")
+            pass
 
         #the directories in which it happened
         self.runDirs={}
@@ -43,12 +50,19 @@ def runTests(self, opt):
             print('resetting to default number of process threads = %s' %  self.maxThreads)
 
         print('Running %s %s %s, each with %s thread%s per process' % ('up to' if self.maxThreads > 1 else '', self.maxThreads, 'concurrent jobs' if self.maxThreads > 1 else 'job', self.nThreads, 's' if self.nThreads > 1 else ''))
-
-
-        for wf in self.workFlows:
-
-            if testList and float(wf.numId) not in [float(x) for x in testList]: continue
-
+        
+        njob = None
+        wfs_to_run = self.workFlows
+        withDups = False
+        if testList:
+            withDups = len(check_dups(testList))>0
+            wfs_to_run = [wf for wf in self.workFlows if float(wf.numId) in testList for i in range(Counter(testList)[wf.numId])]
+
+        for n,wf in enumerate(wfs_to_run):
+
+            if withDups and opt.nProcs > 1: # to avoid overwriting the work areas
+                njob = n
+        
             item = wf.nameId
             if os.path.islink(item) : continue # ignore symlinks
 
@@ -58,7 +72,7 @@ def runTests(self, opt):
 
             print('\nPreparing to run %s %s' % (wf.numId, item))
             sys.stdout.flush()
-            current = WorkFlowRunner(wf,opt,noRun,dryRun,cafVeto)
+            current = WorkFlowRunner(wf,opt,noRun,dryRun,cafVeto,njob)
             self.threadList.append(current)
             current.start()
             if not dryRun:

Configuration/PyReleaseValidation/python/WorkFlowRunner.py

@@ -7,7 +7,7 @@
 from datetime import datetime
 
 class WorkFlowRunner(Thread):
-    def __init__(self, wf, opt, noRun=False, dryRun=False, cafVeto=True):
+    def __init__(self, wf, opt, noRun=False, dryRun=False, cafVeto=True, jobNumber=None):
         Thread.__init__(self)
         self.wf = wf
 
@@ -29,6 +29,9 @@ def __init__(self, wf, opt, noRun=False, dryRun=False, cafVeto=True):
         self.recycle = opt.recycle
 
         self.wfDir=str(self.wf.numId)+'_'+self.wf.nameId
+        if jobNumber is not None:
+            self.wfDir = self.wfDir + '_job' + str(jobNumber)
+        print(self.wfDir)
         return
 
     def doCmd(self, cmd):

Configuration/PyReleaseValidation/scripts/runTheMatrix.py

@@ -28,8 +28,6 @@ def runSelected(opt):
         testSet = set(opt.testList)
         undefSet = testSet - definedSet
         if len(undefSet)>0: raise ValueError('Undefined workflows: '+', '.join(map(str,list(undefSet))))
-        duplicates = [wf for wf in testSet if definedWf.count(wf)>1 ]
-        if len(duplicates)>0: raise ValueError('Duplicated workflows: '+', '.join(map(str,list(duplicates))))
 
     ret = 0
     if opt.show:
@@ -497,7 +495,7 @@ def stepOrIndex(s):
                 except:
                     print(entry,'is not a possible selected entry')
 
-        opt.testList = list(set(testList))
+        opt.testList = list(testList)
 
     if opt.wmcontrol:
         performInjectionOptionTest(opt)