CMS 3D CMS Logo

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