CMS 3D CMS Logo
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Groups Pages
MatrixInjector.py
Go to the documentation of this file.
1 from __future__ import print_function
2 import sys
3 import json
4 import os
5 import copy
6 import multiprocessing
7 import time
8 import re
9 
10 MAXWORKFLOWLENGTH = 81
11 
12 def performInjectionOptionTest(opt):
13  if opt.show:
14  print('Not injecting to wmagent in --show mode. Need to run the worklfows.')
15  sys.exit(-1)
16  if opt.wmcontrol=='init':
17  #init means it'll be in test mode
18  opt.nProcs=0
19  if opt.wmcontrol=='test':
20  #means the wf were created already, and we just dryRun it.
21  opt.dryRun=True
22  if opt.wmcontrol=='submit' and opt.nProcs==0:
23  print('Not injecting to wmagent in -j 0 mode. Need to run the worklfows.')
24  sys.exit(-1)
25  if opt.wmcontrol=='force':
26  print("This is an expert setting, you'd better know what you're doing")
27  opt.dryRun=True
28 
29 def upload_to_couch_oneArg(arguments):
30  from modules.wma import upload_to_couch
31  (filePath,labelInCouch,user,group,where) = arguments
32  cacheId=upload_to_couch(filePath,
33  labelInCouch,
34  user,
35  group,
36  test_mode=False,
37  url=where)
38  return cacheId
39 
40 
41 class MatrixInjector(object):
42 
43  def __init__(self,opt,mode='init',options=''):
44  self.count=1040
45 
46  self.dqmgui=None
47  self.wmagent=None
48  for k in options.split(','):
49  if k.startswith('dqm:'):
50  self.dqmgui=k.split(':',1)[-1]
51  elif k.startswith('wma:'):
52  self.wmagent=k.split(':',1)[-1]
53 
54  self.testMode=((mode!='submit') and (mode!='force'))
55  self.version =1
56  self.keep = opt.keep
57  self.memoryOffset = opt.memoryOffset
58  self.memPerCore = opt.memPerCore
59  self.numberEventsInLuminosityBlock = opt.numberEventsInLuminosityBlock
60  self.numberOfStreams = 0
61  if(opt.nStreams>0):
62  self.numberOfStreams = opt.nStreams
63  self.batchName = ''
64  self.batchTime = str(int(time.time()))
65  if(opt.batchName):
66  self.batchName = '__'+opt.batchName+'-'+self.batchTime
67 
68  ####################################
69  # Checking and setting up GPU attributes
70  ####################################
71  # Mendatory
72  self.RequiresGPU = opt.gpu
73  self.GPUMemoryMB = opt.GPUMemoryMB
74  self.CUDACapabilities = opt.CUDACapabilities
75  self.CUDARuntime = opt.CUDARuntime
76  # optional
77  self.GPUName = opt.GPUName
78  self.CUDADriverVersion = opt.CUDADriverVersion
79  self.CUDARuntimeVersion = opt.CUDARuntimeVersion
80 
81  # WMagent url
82  if not self.wmagent:
83  # Overwrite with env variable
84  self.wmagent = os.getenv('WMAGENT_REQMGR')
85 
86  if not self.wmagent:
87  # Default values
88  if not opt.testbed:
89  self.wmagent = 'cmsweb.cern.ch'
90  else:
91  self.wmagent = 'cmsweb-testbed.cern.ch'
92 
93  # DBSReader url
94  if opt.dbsUrl is not None:
95  self.DbsUrl = opt.dbsUrl
96  elif os.getenv('CMS_DBSREADER_URL') is not None:
97  self.DbsUrl = os.getenv('CMS_DBSREADER_URL')
98  else:
99  # Default values
100  if not opt.testbed:
101  self.DbsUrl = "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader"
102  else:
103  self.DbsUrl = "https://cmsweb-testbed.cern.ch/dbs/int/global/DBSReader"
104 
105  if not self.dqmgui:
106  self.dqmgui="https://cmsweb.cern.ch/dqm/relval"
107  #couch stuff
108  self.couch = 'https://'+self.wmagent+'/couchdb'
109 # self.couchDB = 'reqmgr_config_cache'
110  self.couchCache={} # so that we do not upload like crazy, and recyle cfgs
111  self.user = os.getenv('USER')
112  self.group = 'ppd'
113  self.label = 'RelValSet_'+os.getenv('CMSSW_VERSION').replace('-','')+'_v'+str(self.version)
114  self.speciallabel=''
115  if opt.label:
116  self.speciallabel= '_'+opt.label
117  self.longWFName = []
118 
119  if not os.getenv('WMCORE_ROOT'):
120  print('\n\twmclient is not setup properly. Will not be able to upload or submit requests.\n')
121  if not self.testMode:
122  print('\n\t QUIT\n')
123  sys.exit(-18)
124  else:
125  print('\n\tFound wmclient\n')
126 
127  self.defaultChain={
128  "RequestType" : "TaskChain", #this is how we handle relvals
129  "SubRequestType" : "RelVal", #this is how we handle relvals, now that TaskChain is also used for central MC production
130  "RequestPriority": 500000,
131  "Requestor": self.user, #Person responsible
132  "Group": self.group, #group for the request
133  "CMSSWVersion": os.getenv('CMSSW_VERSION'), #CMSSW Version (used for all tasks in chain)
134  "Campaign": os.getenv('CMSSW_VERSION'), # = AcquisitionEra, will be reset later to the one of first task, will both be the CMSSW_VERSION
135  "ScramArch": os.getenv('SCRAM_ARCH'), #Scram Arch (used for all tasks in chain)
136  "ProcessingVersion": self.version, #Processing Version (used for all tasks in chain)
137  "GlobalTag": None, #Global Tag (overridden per task)
138  "ConfigCacheUrl": self.couch, #URL of CouchDB containing Config Cache
139  "DbsUrl": self.DbsUrl,
140  #- Will contain all configs for all Tasks
141  #"SiteWhitelist" : ["T2_CH_CERN", "T1_US_FNAL"], #Site whitelist
142  "TaskChain" : None, #Define number of tasks in chain.
143  "nowmTasklist" : [], #a list of tasks as we put them in
144  "Multicore" : 1, # do not set multicore for the whole chain
145  "Memory" : 3000,
146  "SizePerEvent" : 1234,
147  "TimePerEvent" : 10,
148  "PrepID": os.getenv('CMSSW_VERSION')
149  }
150 
151  self.defaultHarvest={
152  "EnableHarvesting" : "True",
153  "DQMUploadUrl" : self.dqmgui,
154  "DQMConfigCacheID" : None,
155  "Multicore" : 1 # hardcode Multicore to be 1 for Harvest
156  }
157 
158  self.defaultScratch={
159  "TaskName" : None, #Task Name
160  "ConfigCacheID" : None, #Generator Config id
161  "GlobalTag": None,
162  "SplittingAlgo" : "EventBased", #Splitting Algorithm
163  "EventsPerJob" : None, #Size of jobs in terms of splitting algorithm
164  "EventsPerLumi" : None,
165  "RequestNumEvents" : None, #Total number of events to generate
166  "Seeding" : "AutomaticSeeding", #Random seeding method
167  "PrimaryDataset" : None, #Primary Dataset to be created
168  "nowmIO": {},
169  "Multicore" : opt.nThreads, # this is the per-taskchain Multicore; it's the default assigned to a task if it has no value specified
170  "EventStreams": self.numberOfStreams,
171  "KeepOutput" : False
172  }
173  self.defaultInput={
174  "TaskName" : "DigiHLT", #Task Name
175  "ConfigCacheID" : None, #Processing Config id
176  "GlobalTag": None,
177  "InputDataset" : None, #Input Dataset to be processed
178  "SplittingAlgo" : "LumiBased", #Splitting Algorithm
179  "LumisPerJob" : 10, #Size of jobs in terms of splitting algorithm
180  "nowmIO": {},
181  "Multicore" : opt.nThreads, # this is the per-taskchain Multicore; it's the default assigned to a task if it has no value specified
182  "EventStreams": self.numberOfStreams,
183  "KeepOutput" : False
184  }
185  self.defaultTask={
186  "TaskName" : None, #Task Name
187  "InputTask" : None, #Input Task Name (Task Name field of a previous Task entry)
188  "InputFromOutputModule" : None, #OutputModule name in the input task that will provide files to process
189  "ConfigCacheID" : None, #Processing Config id
190  "GlobalTag": None,
191  "SplittingAlgo" : "LumiBased", #Splitting Algorithm
192  "LumisPerJob" : 10, #Size of jobs in terms of splitting algorithm
193  "nowmIO": {},
194  "Multicore" : opt.nThreads, # this is the per-taskchain Multicore; it's the default assigned to a task if it has no value specified
195  "EventStreams": self.numberOfStreams,
196  "KeepOutput" : False,
197  "RequiresGPU" : None,
198  "GPUParams": None
199  }
200  self.defaultGPUParams={
201  "GPUMemoryMB": self.GPUMemoryMB,
202  "CUDACapabilities": self.CUDACapabilities,
203  "CUDARuntime": self.CUDARuntime
204  }
205  if self.GPUName: self.defaultGPUParams.update({"GPUName": self.GPUName})
206  if self.CUDADriverVersion: self.defaultGPUParams.update({"CUDADriverVersion": self.CUDADriverVersion})
207  if self.CUDARuntimeVersion: self.defaultGPUParams.update({"CUDARuntimeVersion": self.CUDARuntimeVersion})
208 
209  self.chainDicts={}
210 
211  @staticmethod
212  def get_wmsplit():
213  """
214  Return a "wmsplit" dictionary that contain non-default LumisPerJob values
215  """
216  wmsplit = {}
217  try:
218  wmsplit['DIGIHI'] = 5
219  wmsplit['RECOHI'] = 5
220  wmsplit['HLTD'] = 5
221  wmsplit['RECODreHLT'] = 2
222  wmsplit['DIGIPU'] = 4
223  wmsplit['DIGIPU1'] = 4
224  wmsplit['RECOPU1'] = 1
225  wmsplit['DIGIUP15_PU50'] = 1
226  wmsplit['RECOUP15_PU50'] = 1
227  wmsplit['DIGIUP15_PU25'] = 1
228  wmsplit['RECOUP15_PU25'] = 1
229  wmsplit['DIGIUP15_PU25HS'] = 1
230  wmsplit['RECOUP15_PU25HS'] = 1
231  wmsplit['DIGIHIMIX'] = 5
232  wmsplit['RECOHIMIX'] = 5
233  wmsplit['RECODSplit'] = 1
234  wmsplit['SingleMuPt10_UP15_ID'] = 1
235  wmsplit['DIGIUP15_ID'] = 1
236  wmsplit['RECOUP15_ID'] = 1
237  wmsplit['TTbar_13_ID'] = 1
238  wmsplit['SingleMuPt10FS_ID'] = 1
239  wmsplit['TTbarFS_ID'] = 1
240  wmsplit['RECODR2_50nsreHLT'] = 5
241  wmsplit['RECODR2_25nsreHLT'] = 5
242  wmsplit['RECODR2_2016reHLT'] = 5
243  wmsplit['RECODR2_50nsreHLT_HIPM'] = 5
244  wmsplit['RECODR2_25nsreHLT_HIPM'] = 5
245  wmsplit['RECODR2_2016reHLT_HIPM'] = 1
246  wmsplit['RECODR2_2016reHLT_skimSingleMu'] = 1
247  wmsplit['RECODR2_2016reHLT_skimDoubleEG'] = 1
248  wmsplit['RECODR2_2016reHLT_skimMuonEG'] = 1
249  wmsplit['RECODR2_2016reHLT_skimJetHT'] = 1
250  wmsplit['RECODR2_2016reHLT_skimMET'] = 1
251  wmsplit['RECODR2_2016reHLT_skimSinglePh'] = 1
252  wmsplit['RECODR2_2016reHLT_skimMuOnia'] = 1
253  wmsplit['RECODR2_2016reHLT_skimSingleMu_HIPM'] = 1
254  wmsplit['RECODR2_2016reHLT_skimDoubleEG_HIPM'] = 1
255  wmsplit['RECODR2_2016reHLT_skimMuonEG_HIPM'] = 1
256  wmsplit['RECODR2_2016reHLT_skimJetHT_HIPM'] = 1
257  wmsplit['RECODR2_2016reHLT_skimMET_HIPM'] = 1
258  wmsplit['RECODR2_2016reHLT_skimSinglePh_HIPM'] = 1
259  wmsplit['RECODR2_2016reHLT_skimMuOnia_HIPM'] = 1
260  wmsplit['RECODR2_2017reHLT_Prompt'] = 1
261  wmsplit['RECODR2_2017reHLT_skimSingleMu_Prompt_Lumi'] = 1
262  wmsplit['RECODR2_2017reHLT_skimDoubleEG_Prompt'] = 1
263  wmsplit['RECODR2_2017reHLT_skimMET_Prompt'] = 1
264  wmsplit['RECODR2_2017reHLT_skimMuOnia_Prompt'] = 1
265  wmsplit['RECODR2_2017reHLT_Prompt_L1TEgDQM'] = 1
266  wmsplit['RECODR2_2018reHLT_Prompt'] = 1
267  wmsplit['RECODR2_2018reHLT_skimSingleMu_Prompt_Lumi'] = 1
268  wmsplit['RECODR2_2018reHLT_skimDoubleEG_Prompt'] = 1
269  wmsplit['RECODR2_2018reHLT_skimJetHT_Prompt'] = 1
270  wmsplit['RECODR2_2018reHLT_skimMET_Prompt'] = 1
271  wmsplit['RECODR2_2018reHLT_skimMuOnia_Prompt'] = 1
272  wmsplit['RECODR2_2018reHLT_skimEGamma_Prompt_L1TEgDQM'] = 1
273  wmsplit['RECODR2_2018reHLT_skimMuonEG_Prompt'] = 1
274  wmsplit['RECODR2_2018reHLT_skimCharmonium_Prompt'] = 1
275  wmsplit['RECODR2_2018reHLT_skimJetHT_Prompt_HEfail'] = 1
276  wmsplit['RECODR2_2018reHLT_skimJetHT_Prompt_BadHcalMitig'] = 1
277  wmsplit['RECODR2_2018reHLTAlCaTkCosmics_Prompt'] = 1
278  wmsplit['RECODR2_2018reHLT_skimDisplacedJet_Prompt'] = 1
279  wmsplit['RECODR2_2018reHLT_ZBPrompt'] = 1
280  wmsplit['RECODR2_2018reHLT_Offline'] = 1
281  wmsplit['RECODR2_2018reHLT_skimSingleMu_Offline_Lumi'] = 1
282  wmsplit['RECODR2_2018reHLT_skimDoubleEG_Offline'] = 1
283  wmsplit['RECODR2_2018reHLT_skimJetHT_Offline'] = 1
284  wmsplit['RECODR2_2018reHLT_skimMET_Offline'] = 1
285  wmsplit['RECODR2_2018reHLT_skimMuOnia_Offline'] = 1
286  wmsplit['RECODR2_2018reHLT_skimEGamma_Offline_L1TEgDQM'] = 1
287  wmsplit['RECODR2_2018reHLT_skimMuonEG_Offline'] = 1
288  wmsplit['RECODR2_2018reHLT_skimCharmonium_Offline'] = 1
289  wmsplit['RECODR2_2018reHLT_skimJetHT_Offline_HEfail'] = 1
290  wmsplit['RECODR2_2018reHLT_skimJetHT_Offline_BadHcalMitig'] = 1
291  wmsplit['RECODR2_2018reHLTAlCaTkCosmics_Offline'] = 1
292  wmsplit['RECODR2_2018reHLT_skimDisplacedJet_Offline'] = 1
293  wmsplit['RECODR2_2018reHLT_ZBOffline'] = 1
294  wmsplit['HLTDR2_50ns'] = 1
295  wmsplit['HLTDR2_25ns'] = 1
296  wmsplit['HLTDR2_2016'] = 1
297  wmsplit['HLTDR2_2017'] = 1
298  wmsplit['HLTDR2_2018'] = 1
299  wmsplit['HLTDR2_2018_BadHcalMitig'] = 1
300  wmsplit['Hadronizer'] = 1
301  wmsplit['DIGIUP15'] = 1
302  wmsplit['RECOUP15'] = 1
303  wmsplit['RECOAODUP15'] = 5
304  wmsplit['DBLMINIAODMCUP15NODQM'] = 5
305  wmsplit['Digi'] = 5
306  wmsplit['Reco'] = 5
307  wmsplit['DigiPU'] = 1
308  wmsplit['RecoPU'] = 1
309  wmsplit['RECOHID11'] = 1
310  wmsplit['DIGIUP17'] = 1
311  wmsplit['RECOUP17'] = 1
312  wmsplit['DIGIUP17_PU25'] = 1
313  wmsplit['RECOUP17_PU25'] = 1
314  wmsplit['DIGICOS_UP16'] = 1
315  wmsplit['RECOCOS_UP16'] = 1
316  wmsplit['DIGICOS_UP17'] = 1
317  wmsplit['RECOCOS_UP17'] = 1
318  wmsplit['DIGICOS_UP18'] = 1
319  wmsplit['RECOCOS_UP18'] = 1
320  wmsplit['DIGICOS_UP21'] = 1
321  wmsplit['RECOCOS_UP21'] = 1
322  wmsplit['HYBRIDRepackHI2015VR'] = 1
323  wmsplit['HYBRIDZSHI2015'] = 1
324  wmsplit['RECOHID15'] = 1
325  wmsplit['RECOHID18'] = 1
326  # automate for phase 2
327  from .upgradeWorkflowComponents import upgradeKeys
328  for key in upgradeKeys[2026]:
329  if 'PU' not in key:
330  continue
331 
332  wmsplit['DigiTriggerPU_' + key] = 1
333  wmsplit['RecoGlobalPU_' + key] = 1
334 
335  except Exception as ex:
336  print('Exception while building a wmsplit dictionary: %s' % (str(ex)))
337  return {}
338 
339  return wmsplit
340 
341  def prepare(self, mReader, directories, mode='init'):
342  wmsplit = MatrixInjector.get_wmsplit()
343  acqEra=False
344  for (n,dir) in directories.items():
345  chainDict=copy.deepcopy(self.defaultChain)
346  print("inspecting",dir)
347  nextHasDSInput=None
348  for (x,s) in mReader.workFlowSteps.items():
349  #x has the format (num, prefix)
350  #s has the format (num, name, commands, stepList)
351  if x[0]==n:
352  #print "found",n,s[3]
353  #chainDict['RequestString']='RV'+chainDict['CMSSWVersion']+s[1].split('+')[0]
354  index=0
355  splitForThisWf=None
356  thisLabel=self.speciallabel
357  #if 'HARVESTGEN' in s[3]:
358  if len( [step for step in s[3] if "HARVESTGEN" in step] )>0:
359  chainDict['TimePerEvent']=0.01
360  thisLabel=thisLabel+"_gen"
361  # for double miniAOD test
362  if len( [step for step in s[3] if "DBLMINIAODMCUP15NODQM" in step] )>0:
363  thisLabel=thisLabel+"_dblMiniAOD"
364  processStrPrefix=''
365  setPrimaryDs=None
366  nanoedmGT=''
367  for step in s[3]:
368 
369  if 'INPUT' in step or (not isinstance(s[2][index],str)):
370  nextHasDSInput=s[2][index]
371 
372  else:
373 
374  if (index==0):
375  #first step and not input -> gen part
376  chainDict['nowmTasklist'].append(copy.deepcopy(self.defaultScratch))
377  try:
378  chainDict['nowmTasklist'][-1]['nowmIO']=json.loads(open('%s/%s.io'%(dir,step)).read())
379  except:
380  print("Failed to find",'%s/%s.io'%(dir,step),".The workflows were probably not run on cfg not created")
381  return -15
382 
383  chainDict['nowmTasklist'][-1]['PrimaryDataset']='RelVal'+s[1].split('+')[0]
384  if not '--relval' in s[2][index]:
385  print('Impossible to create task from scratch without splitting information with --relval')
386  return -12
387  else:
388  arg=s[2][index].split()
389  ns=list(map(int,arg[len(arg) - arg[-1::-1].index('--relval')].split(',')))
390  chainDict['nowmTasklist'][-1]['RequestNumEvents'] = ns[0]
391  chainDict['nowmTasklist'][-1]['EventsPerJob'] = ns[1]
392  chainDict['nowmTasklist'][-1]['EventsPerLumi'] = ns[1]
393  #overwrite EventsPerLumi if numberEventsInLuminosityBlock is set in cmsDriver
394  if 'numberEventsInLuminosityBlock' in s[2][index]:
395  nEventsInLuminosityBlock = re.findall('process.source.numberEventsInLuminosityBlock=cms.untracked.uint32\(([ 0-9 ]*)\)', s[2][index],re.DOTALL)
396  if nEventsInLuminosityBlock[-1].isdigit() and int(nEventsInLuminosityBlock[-1]) < ns[1]:
397  chainDict['nowmTasklist'][-1]['EventsPerLumi'] = int(nEventsInLuminosityBlock[-1])
398  if(self.numberEventsInLuminosityBlock > 0 and self.numberEventsInLuminosityBlock <= ns[1]):
399  chainDict['nowmTasklist'][-1]['EventsPerLumi'] = self.numberEventsInLuminosityBlock
400  if 'FASTSIM' in s[2][index] or '--fast' in s[2][index]:
401  thisLabel+='_FastSim'
402  if 'lhe' in s[2][index] in s[2][index]:
403  chainDict['nowmTasklist'][-1]['LheInputFiles'] =True
404 
405  elif nextHasDSInput:
406  chainDict['nowmTasklist'].append(copy.deepcopy(self.defaultInput))
407  try:
408  chainDict['nowmTasklist'][-1]['nowmIO']=json.loads(open('%s/%s.io'%(dir,step)).read())
409  except:
410  print("Failed to find",'%s/%s.io'%(dir,step),".The workflows were probably not run on cfg not created")
411  return -15
412  chainDict['nowmTasklist'][-1]['InputDataset']=nextHasDSInput.dataSet
413  if ('DQMHLTonRAWAOD' in step) :
414  chainDict['nowmTasklist'][-1]['IncludeParents']=True
415  splitForThisWf=nextHasDSInput.split
416  chainDict['nowmTasklist'][-1]['LumisPerJob']=splitForThisWf
417  if step in wmsplit:
418  chainDict['nowmTasklist'][-1]['LumisPerJob']=wmsplit[step]
419  # get the run numbers or #events
420  if len(nextHasDSInput.run):
421  chainDict['nowmTasklist'][-1]['RunWhitelist']=nextHasDSInput.run
422  if len(nextHasDSInput.ls):
423  chainDict['nowmTasklist'][-1]['LumiList']=nextHasDSInput.ls
424  #print "what is s",s[2][index]
425  if '--data' in s[2][index] and nextHasDSInput.label:
426  thisLabel+='_RelVal_%s'%nextHasDSInput.label
427  if 'filter' in chainDict['nowmTasklist'][-1]['nowmIO']:
428  print("This has an input DS and a filter sequence: very likely to be the PyQuen sample")
429  processStrPrefix='PU_'
430  setPrimaryDs = 'RelVal'+s[1].split('+')[0]
431  if setPrimaryDs:
432  chainDict['nowmTasklist'][-1]['PrimaryDataset']=setPrimaryDs
433  nextHasDSInput=None
434  if 'GPU' in step and self.RequiresGPU != 'forbidden':
435  chainDict['nowmTasklist'][-1]['RequiresGPU'] = self.RequiresGPU
436  chainDict['nowmTasklist'][-1]['GPUParams']=json.dumps(self.defaultGPUParams)
437  else:
438  #not first step and no inputDS
439  chainDict['nowmTasklist'].append(copy.deepcopy(self.defaultTask))
440  try:
441  chainDict['nowmTasklist'][-1]['nowmIO']=json.loads(open('%s/%s.io'%(dir,step)).read())
442  except:
443  print("Failed to find",'%s/%s.io'%(dir,step),".The workflows were probably not run on cfg not created")
444  return -15
445  if splitForThisWf:
446  chainDict['nowmTasklist'][-1]['LumisPerJob']=splitForThisWf
447  if step in wmsplit:
448  chainDict['nowmTasklist'][-1]['LumisPerJob']=wmsplit[step]
449  if 'GPU' in step and self.RequiresGPU != 'forbidden':
450  chainDict['nowmTasklist'][-1]['RequiresGPU'] = self.RequiresGPU
451  chainDict['nowmTasklist'][-1]['GPUParams']=json.dumps(self.defaultGPUParams)
452 
453  # change LumisPerJob for Hadronizer steps.
454  if 'Hadronizer' in step:
455  chainDict['nowmTasklist'][-1]['LumisPerJob']=wmsplit['Hadronizer']
456 
457  #print step
458  chainDict['nowmTasklist'][-1]['TaskName']=step
459  if setPrimaryDs:
460  chainDict['nowmTasklist'][-1]['PrimaryDataset']=setPrimaryDs
461  chainDict['nowmTasklist'][-1]['ConfigCacheID']='%s/%s.py'%(dir,step)
462  chainDict['nowmTasklist'][-1]['GlobalTag']=chainDict['nowmTasklist'][-1]['nowmIO']['GT'] # copy to the proper parameter name
463  chainDict['GlobalTag']=chainDict['nowmTasklist'][-1]['nowmIO']['GT'] #set in general to the last one of the chain
464  if 'NANOEDM' in step :
465  nanoedmGT = chainDict['nowmTasklist'][-1]['nowmIO']['GT']
466  if 'NANOMERGE' in step :
467  chainDict['GlobalTag'] = nanoedmGT
468  if 'pileup' in chainDict['nowmTasklist'][-1]['nowmIO']:
469  chainDict['nowmTasklist'][-1]['MCPileup']=chainDict['nowmTasklist'][-1]['nowmIO']['pileup']
470  if '--pileup ' in s[2][index]: # catch --pileup (scenarion) and not --pileup_ (dataset to be mixed) => works also making PRE-MIXed dataset
471  pileupString = s[2][index].split()[s[2][index].split().index('--pileup')+1]
472  processStrPrefix='PU_' # take care of pu overlay done with GEN-SIM mixing
473  if pileupString.find('25ns') > 0 :
474  processStrPrefix='PU25ns_'
475  elif pileupString.find('50ns') > 0 :
476  processStrPrefix='PU50ns_'
477  elif 'nopu' in pileupString.lower():
478  processStrPrefix=''
479  if 'premix_stage2' in s[2][index] and '--pileup_input' in s[2][index]: # take care of pu overlay done with DIGI mixing of premixed events
480  if s[2][index].split()[ s[2][index].split().index('--pileup_input')+1 ].find('25ns') > 0 :
481  processStrPrefix='PUpmx25ns_'
482  elif s[2][index].split()[ s[2][index].split().index('--pileup_input')+1 ].find('50ns') > 0 :
483  processStrPrefix='PUpmx50ns_'
484 
485  if acqEra:
486  #chainDict['AcquisitionEra'][step]=(chainDict['CMSSWVersion']+'-PU_'+chainDict['nowmTasklist'][-1]['GlobalTag']).replace('::All','')+thisLabel
487  chainDict['AcquisitionEra'][step]=chainDict['CMSSWVersion']
488  chainDict['ProcessingString'][step]=processStrPrefix+chainDict['nowmTasklist'][-1]['GlobalTag'].replace('::All','').replace('-','_')+thisLabel
489  if 'NANOMERGE' in step :
490  chainDict['ProcessingString'][step]=processStrPrefix+nanoedmGT.replace('::All','').replace('-','_')+thisLabel
491  else:
492  #chainDict['nowmTasklist'][-1]['AcquisitionEra']=(chainDict['CMSSWVersion']+'-PU_'+chainDict['nowmTasklist'][-1]['GlobalTag']).replace('::All','')+thisLabel
493  chainDict['nowmTasklist'][-1]['AcquisitionEra']=chainDict['CMSSWVersion']
494  chainDict['nowmTasklist'][-1]['ProcessingString']=processStrPrefix+chainDict['nowmTasklist'][-1]['GlobalTag'].replace('::All','').replace('-','_')+thisLabel
495  if 'NANOMERGE' in step :
496  chainDict['nowmTasklist'][-1]['ProcessingString']=processStrPrefix+nanoedmGT.replace('::All','').replace('-','_')+thisLabel
497 
498  if (self.batchName):
499  chainDict['nowmTasklist'][-1]['Campaign'] = chainDict['nowmTasklist'][-1]['AcquisitionEra']+self.batchName
500 
501  # specify different ProcessingString for double miniAOD dataset
502  if ('DBLMINIAODMCUP15NODQM' in step):
503  chainDict['nowmTasklist'][-1]['ProcessingString']=chainDict['nowmTasklist'][-1]['ProcessingString']+'_miniAOD'
504 
505  if( chainDict['nowmTasklist'][-1]['Multicore'] ):
506  # the scaling factor of 1.2GB / thread is empirical and measured on a SECOND round of tests with PU samples
507  # the number of threads is NO LONGER assumed to be the same for all tasks
508  # https://hypernews.cern.ch/HyperNews/CMS/get/edmFramework/3509/1/1/1.html
509  # now change to 1.5GB / additional thread according to discussion:
510  # https://hypernews.cern.ch/HyperNews/CMS/get/relval/4817/1/1.html
511 # chainDict['nowmTasklist'][-1]['Memory'] = 3000 + int( chainDict['nowmTasklist'][-1]['Multicore'] -1 )*1500
512  chainDict['nowmTasklist'][-1]['Memory'] = self.memoryOffset + int( chainDict['nowmTasklist'][-1]['Multicore'] -1 ) * self.memPerCore
513 
514  index+=1
515  #end of loop through steps
516  chainDict['RequestString']='RV'+chainDict['CMSSWVersion']+s[1].split('+')[0]
517  if processStrPrefix or thisLabel:
518  chainDict['RequestString']+='_'+processStrPrefix+thisLabel
519  #check candidate WF name
520  self.candidateWFName = self.user+'_'+chainDict['RequestString']
521  if (len(self.candidateWFName)>MAXWORKFLOWLENGTH):
522  self.longWFName.append(self.candidateWFName)
523 
524 ### PrepID
525  chainDict['PrepID'] = chainDict['CMSSWVersion']+'__'+self.batchTime+'-'+s[1].split('+')[0]
526  if(self.batchName):
527  chainDict['PrepID'] = chainDict['CMSSWVersion']+self.batchName+'-'+s[1].split('+')[0]
528  if( 'HIN' in self.batchName ):
529  chainDict['SubRequestType'] = "HIRelVal"
530 
531  #wrap up for this one
532  import pprint
533  #print 'wrapping up'
534  #pprint.pprint(chainDict)
535  #loop on the task list
536  for i_second in reversed(range(len(chainDict['nowmTasklist']))):
537  t_second=chainDict['nowmTasklist'][i_second]
538  #print "t_second taskname", t_second['TaskName']
539  if 'primary' in t_second['nowmIO']:
540  #print t_second['nowmIO']['primary']
541  primary=t_second['nowmIO']['primary'][0].replace('file:','')
542  for i_input in reversed(range(0,i_second)):
543  t_input=chainDict['nowmTasklist'][i_input]
544  for (om,o) in t_input['nowmIO'].items():
545  if primary in o:
546  #print "found",primary,"procuced by",om,"of",t_input['TaskName']
547  #ad-hoc fix due to restriction in TaskName of 50 characters
548  if (len(t_input['TaskName'])>50):
549  if (t_input['TaskName'].find('GenSim') != -1):
550  t_input['TaskName'] = 'GenSimFull'
551  if (t_input['TaskName'].find('Hadronizer') != -1):
552  t_input['TaskName'] = 'HadronizerFull'
553  t_second['InputTask'] = t_input['TaskName']
554  t_second['InputFromOutputModule'] = om
555  #print 't_second',pprint.pformat(t_second)
556  if t_second['TaskName'].startswith('HARVEST'):
557  chainDict.update(copy.deepcopy(self.defaultHarvest))
558  if "_RD" in t_second['TaskName']:
559  chainDict['DQMHarvestUnit'] = "multiRun"
560  chainDict['DQMConfigCacheID']=t_second['ConfigCacheID']
561  ## the info are not in the task specific dict but in the general dict
562  #t_input.update(copy.deepcopy(self.defaultHarvest))
563  #t_input['DQMConfigCacheID']=t_second['ConfigCacheID']
564  break
565 
566  # agreed changes for wm injection:
567  # - Campaign: *optional* string during creation. It will default to AcqEra value if possible.
568  # Otherwise it will be empty.
569  # - AcquisitionEra: *mandatory* string at request level during creation. *optional* string
570  # at task level during creation. "optional" during assignment.
571  # - ProcessingString: *mandatory* string at request level during creation. *optional* string
572  # at task level during creation. "optional" during assignment.
573  # - ProcessingVersion: *optional* during creation (default 1). *optional* during assignment.
574  #
575  # Which requires following changes here:
576  # - reset Global AcuisitionEra, ProcessingString to be the one in the first task
577  # - and also Campaign to be always the same as the AcquisitionEra
578 
579  if acqEra:
580  chainDict['AcquisitionEra'] = chainDict['AcquisitionEra'].values()[0]
581  chainDict['ProcessingString'] = chainDict['ProcessingString'].values()[0]
582  else:
583  chainDict['AcquisitionEra'] = chainDict['nowmTasklist'][0]['AcquisitionEra']
584  chainDict['ProcessingString'] = chainDict['nowmTasklist'][0]['ProcessingString']
585 
586 ##### batch name appended to Campaign name
587 # chainDict['Campaign'] = chainDict['AcquisitionEra']
588  chainDict['Campaign'] = chainDict['AcquisitionEra']+self.batchName
589 
590  ## clean things up now
591  itask=0
592  if self.keep:
593  for i in self.keep:
594  if isinstance(i, int) and i < len(chainDict['nowmTasklist']):
595  chainDict['nowmTasklist'][i]['KeepOutput']=True
596  for (i,t) in enumerate(chainDict['nowmTasklist']):
597  if t['TaskName'].startswith('HARVEST'):
598  continue
599  if not self.keep:
600  t['KeepOutput']=True
601  elif t['TaskName'] in self.keep:
602  t['KeepOutput']=True
603  if t['TaskName'].startswith('HYBRIDRepackHI2015VR'):
604  t['KeepOutput']=False
605  t.pop('nowmIO')
606  itask+=1
607  chainDict['Task%d'%(itask)]=t
608 
609 
610  ##
611 
612 
613  ## provide the number of tasks
614  chainDict['TaskChain']=itask#len(chainDict['nowmTasklist'])
615 
616  chainDict.pop('nowmTasklist')
617  self.chainDicts[n]=chainDict
618 
619 
620  return 0
621 
622  def uploadConf(self,filePath,label,where):
623  labelInCouch=self.label+'_'+label
624  cacheName=filePath.split('/')[-1]
625  if self.testMode:
626  self.count+=1
627  print('\tFake upload of',filePath,'to couch with label',labelInCouch)
628  return self.count
629  else:
630  try:
631  from modules.wma import upload_to_couch,DATABASE_NAME
632  except:
633  print('\n\tUnable to find wmcontrol modules. Please include it in your python path\n')
634  print('\n\t QUIT\n')
635  sys.exit(-16)
636 
637  if cacheName in self.couchCache:
638  print("Not re-uploading",filePath,"to",where,"for",label)
639  cacheId=self.couchCache[cacheName]
640  else:
641  print("Loading",filePath,"to",where,"for",label)
642  ## totally fork the upload to couch to prevent cross loading of process configurations
643  pool = multiprocessing.Pool(1)
644  cacheIds = pool.map( upload_to_couch_oneArg, [(filePath,labelInCouch,self.user,self.group,where)] )
645  cacheId = cacheIds[0]
646  self.couchCache[cacheName]=cacheId
647  return cacheId
648 
649  def upload(self):
650  for (n,d) in self.chainDicts.items():
651  for it in d:
652  if it.startswith("Task") and it!='TaskChain':
653  #upload
654  couchID=self.uploadConf(d[it]['ConfigCacheID'],
655  str(n)+d[it]['TaskName'],
656  d['ConfigCacheUrl']
657  )
658  print(d[it]['ConfigCacheID']," uploaded to couchDB for",str(n),"with ID",couchID)
659  d[it]['ConfigCacheID']=couchID
660  if it =='DQMConfigCacheID':
661  couchID=self.uploadConf(d['DQMConfigCacheID'],
662  str(n)+'harvesting',
663  d['ConfigCacheUrl']
664  )
665  print(d['DQMConfigCacheID'],"uploaded to couchDB for",str(n),"with ID",couchID)
666  d['DQMConfigCacheID']=couchID
667 
668 
669  def submit(self):
670  try:
671  from modules.wma import makeRequest,approveRequest
672  from wmcontrol import random_sleep
673  print('\n\tFound wmcontrol\n')
674  except:
675  print('\n\tUnable to find wmcontrol modules. Please include it in your python path\n')
676  if not self.testMode:
677  print('\n\t QUIT\n')
678  sys.exit(-17)
679 
680  import pprint
681  for (n,d) in self.chainDicts.items():
682  if self.testMode:
683  print("Only viewing request",n)
684  print(pprint.pprint(d))
685  else:
686  #submit to wmagent each dict
687  print("For eyes before submitting",n)
688  print(pprint.pprint(d))
689  print("Submitting",n,"...........")
690  workFlow=makeRequest(self.wmagent,d,encodeDict=True)
691  print("...........",n,"submitted")
692  random_sleep()
693  if self.testMode and len(self.longWFName)>0:
694  print("\n*** WARNING: "+str(len(self.longWFName))+" workflows have too long names for submission (>"+str(MAXWORKFLOWLENGTH)+ "characters) ***")
695  print('\n'.join(self.longWFName))
bitset_utilities::append
boost::dynamic_bitset append(const boost::dynamic_bitset<> &bs1, const boost::dynamic_bitset<> &bs2)
this method takes two bitsets bs1 and bs2 and returns result of bs2 appended to the end of bs1 ...
Definition: bitset_append.cc:10
spr::find
void find(edm::Handle< EcalRecHitCollection > &hits, DetId thisDet, std::vector< EcalRecHitCollection::const_iterator > &hit, bool debug=false)
Definition: FindCaloHit.cc:19
sistrip::SpyUtilities::range
const uint16_t range(const Frame &aFrame)
Definition: SiStripSpyUtilities.cc:76
if
if(conf_.getParameter< bool >("UseStripCablingDB"))
Definition: SiStripLAProfileBooker.cc:72
MatrixInjector.MatrixInjector.uploadConf
def uploadConf
provide the number of tasks
Definition: MatrixInjector.py:622
print
void print(TMatrixD &m, const char *label=nullptr, bool mathematicaFormat=false)
Definition: Utilities.cc:47
SiPixelLorentzAngle_cfi.read
tuple read
Definition: SiPixelLorentzAngle_cfi.py:7
MatrixInjector.MatrixInjector.RequiresGPU
RequiresGPU
Checking and setting up GPU attributes.
Definition: MatrixInjector.py:72
mps_monitormerge.items
list items
Definition: mps_monitormerge.py:29
join
static std::string join(char **cmd)
Definition: RemoteFile.cc:19
MatrixInjector.performInjectionOptionTest
def performInjectionOptionTest
Definition: MatrixInjector.py:12
MatrixInjector.upload_to_couch_oneArg
def upload_to_couch_oneArg
Definition: MatrixInjector.py:29
str
#define str(s)
Definition: TestProcessor.cc:53
Generated for CMSSW Reference Manual by doxygen 1.8.5