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