1 |
+ |
|
2 |
+ |
__revision__ = "$Id$" |
3 |
+ |
__version__ = "$Revision$" |
4 |
+ |
|
5 |
|
import common |
2 |
– |
from crab_logger import Logger |
6 |
|
from crab_exceptions import * |
7 |
|
from crab_util import * |
8 |
+ |
|
9 |
+ |
from WMCore.DataStructs.File import File |
10 |
+ |
from WMCore.DataStructs.Fileset import Fileset |
11 |
+ |
from WMCore.DataStructs.Run import Run |
12 |
+ |
from WMCore.DataStructs.Subscription import Subscription |
13 |
+ |
from WMCore.DataStructs.Workflow import Workflow |
14 |
+ |
from WMCore.JobSplitting.SplitterFactory import SplitterFactory |
15 |
|
from WMCore.SiteScreening.BlackWhiteListParser import SEBlackWhiteListParser |
16 |
+ |
try: # Can remove when CMSSW 3.7 and earlier are dropped |
17 |
+ |
from FWCore.PythonUtilities.LumiList import LumiList |
18 |
+ |
except ImportError: |
19 |
+ |
from LumiList import LumiList |
20 |
|
|
21 |
|
class JobSplitter: |
22 |
|
def __init__( self, cfg_params, args ): |
23 |
|
self.cfg_params = cfg_params |
24 |
|
self.args=args |
25 |
+ |
|
26 |
+ |
self.lumisPerJob = -1 |
27 |
+ |
self.totalNLumis = 0 |
28 |
+ |
self.theNumberOfJobs = 0 |
29 |
+ |
self.limitNJobs = False |
30 |
+ |
self.limitTotalLumis = False |
31 |
+ |
self.limitJobLumis = False |
32 |
+ |
|
33 |
|
#self.maxEvents |
34 |
|
# init BlackWhiteListParser |
35 |
< |
seWhiteList = cfg_params.get('EDG.se_white_list',[]) |
36 |
< |
seBlackList = cfg_params.get('EDG.se_black_list',[]) |
37 |
< |
self.blackWhiteListParser = SEBlackWhiteListParser(seWhiteList, seBlackList, common.logger) |
35 |
> |
self.seWhiteList = cfg_params.get('GRID.se_white_list',[]) |
36 |
> |
seBlackList = cfg_params.get('GRID.se_black_list',[]) |
37 |
> |
self.blackWhiteListParser = SEBlackWhiteListParser(self.seWhiteList, seBlackList, common.logger()) |
38 |
> |
|
39 |
> |
## check if has been asked for a non default file to store/read analyzed fileBlocks |
40 |
> |
defaultName = common.work_space.shareDir()+'AnalyzedBlocks.txt' |
41 |
> |
self.fileBlocks_FileName = os.path.abspath(self.cfg_params.get('CMSSW.fileblocks_file',defaultName)) |
42 |
|
|
43 |
|
|
44 |
|
def checkUserSettings(self): |
69 |
|
self.total_number_of_events = 0 |
70 |
|
self.selectTotalNumberEvents = 0 |
71 |
|
|
72 |
+ |
return |
73 |
+ |
|
74 |
+ |
def checkLumiSettings(self): |
75 |
+ |
""" |
76 |
+ |
Check to make sure the user has specified enough information to |
77 |
+ |
perform splitting by Lumis to run the job |
78 |
+ |
""" |
79 |
+ |
settings = 0 |
80 |
+ |
if self.cfg_params.has_key('CMSSW.lumis_per_job'): |
81 |
+ |
self.lumisPerJob =int( self.cfg_params['CMSSW.lumis_per_job']) |
82 |
+ |
self.limitJobLumis = True |
83 |
+ |
settings += 1 |
84 |
+ |
|
85 |
+ |
if self.cfg_params.has_key('CMSSW.number_of_jobs'): |
86 |
+ |
self.theNumberOfJobs =int( self.cfg_params['CMSSW.number_of_jobs']) |
87 |
+ |
self.limitNJobs = True |
88 |
+ |
settings += 1 |
89 |
+ |
|
90 |
+ |
if self.cfg_params.has_key('CMSSW.total_number_of_lumis'): |
91 |
+ |
self.totalNLumis = int(self.cfg_params['CMSSW.total_number_of_lumis']) |
92 |
+ |
self.limitTotalLumis = (self.totalNLumis != -1) |
93 |
+ |
settings += 1 |
94 |
+ |
|
95 |
+ |
if settings != 2: |
96 |
+ |
msg = 'When splitting by lumi section you must specify two and only two of:\n' |
97 |
+ |
msg += ' number_of_jobs, lumis_per_job, total_number_of_lumis' |
98 |
+ |
raise CrabException(msg) |
99 |
+ |
if self.limitNJobs and self.limitJobLumis: |
100 |
+ |
self.limitTotalLumis = True |
101 |
+ |
self.totalNLumis = self.lumisPerJob * self.theNumberOfJobs |
102 |
+ |
|
103 |
+ |
# Has the user specified runselection? |
104 |
+ |
if (self.cfg_params.has_key('CMSSW.runselection')): |
105 |
+ |
common.logger.info('You have specified runselection and split by lumi.') |
106 |
+ |
common.logger.info('Good lumi list will be the intersection of runselection and lumimask or ADS (if any).') |
107 |
+ |
return |
108 |
+ |
|
109 |
+ |
def ComputeSubBlockSites( self, blockSites ): |
110 |
+ |
""" |
111 |
+ |
""" |
112 |
+ |
sub_blockSites = {} |
113 |
+ |
for k,v in blockSites.iteritems(): |
114 |
+ |
sites=self.blackWhiteListParser.checkWhiteList(v) |
115 |
+ |
if sites : sub_blockSites[k]=v |
116 |
+ |
if len(sub_blockSites) < 1: |
117 |
+ |
msg = 'WARNING: the sites %s is not hosting any part of data.'%self.seWhiteList |
118 |
+ |
raise CrabException(msg) |
119 |
+ |
return sub_blockSites |
120 |
|
|
121 |
|
######################################################################## |
122 |
|
def jobSplittingByEvent( self ): |
132 |
|
self.list_of_args - File(s) job will run on (a list of lists) |
133 |
|
""" |
134 |
|
|
135 |
< |
jobDestination=[] |
135 |
> |
jobDestination=[] |
136 |
|
self.checkUserSettings() |
137 |
|
if ( (self.selectTotalNumberEvents + self.selectEventsPerJob + self.selectNumberOfJobs) != 2 ): |
138 |
|
msg = 'Must define exactly two of total_number_of_events, events_per_job, or number_of_jobs.' |
139 |
|
raise CrabException(msg) |
140 |
< |
|
141 |
< |
blockSites = self.args['blockSites'] |
140 |
> |
|
141 |
> |
blockSites = self.args['blockSites'] |
142 |
|
pubdata = self.args['pubdata'] |
143 |
|
filesbyblock=pubdata.getFiles() |
144 |
|
|
152 |
|
self.useParent = int(self.cfg_params.get('CMSSW.use_parent',0)) |
153 |
|
noBboundary = int(self.cfg_params.get('CMSSW.no_block_boundary',0)) |
154 |
|
|
155 |
+ |
if noBboundary == 1: |
156 |
+ |
if self.total_number_of_events== -1: |
157 |
+ |
msg = 'You are selecting no_block_boundary=1 which does not allow to set total_number_of_events=-1\n' |
158 |
+ |
msg +='\tYou shoud get the number of event from DBS web interface and use it for your configuration.' |
159 |
+ |
raise CrabException(msg) |
160 |
+ |
if len(self.seWhiteList) == 0 or len(self.seWhiteList.split(',')) != 1: |
161 |
+ |
msg = 'You are selecting no_block_boundary=1 which requires to choose one and only one site.\n' |
162 |
+ |
msg += "\tPlease set se_white_list with the site's storage element name." |
163 |
+ |
raise CrabException(msg) |
164 |
+ |
blockSites = self.ComputeSubBlockSites(blockSites) |
165 |
+ |
|
166 |
|
# ---- Handle the possible job splitting configurations ---- # |
167 |
|
if (self.selectTotalNumberEvents): |
168 |
|
totalEventsRequested = self.total_number_of_events |
177 |
|
# If user requested more events than are in the dataset |
178 |
|
elif (totalEventsRequested > self.maxEvents): |
179 |
|
eventsRemaining = self.maxEvents |
180 |
< |
common.logger.message("Requested "+str(self.total_number_of_events)+ " events, but only "+str(self.maxEvents)+" events are available.") |
180 |
> |
common.logger.info("Requested "+str(self.total_number_of_events)+ " events, but only "+str(self.maxEvents)+" events are available.") |
181 |
|
# If user requested less events than are in the dataset |
182 |
|
else: |
183 |
|
eventsRemaining = totalEventsRequested |
193 |
|
eventsPerJobRequested = int(eventsRemaining/self.theNumberOfJobs) |
194 |
|
|
195 |
|
if (self.selectNumberOfJobs): |
196 |
< |
common.logger.message("May not create the exact number_of_jobs requested.") |
196 |
> |
common.logger.info("May not create the exact number_of_jobs requested.") |
197 |
|
|
198 |
|
# old... to remove Daniele |
199 |
|
totalNumberOfJobs = 999999999 |
210 |
|
jobsOfBlock = {} |
211 |
|
|
212 |
|
parString = "" |
213 |
+ |
pString = "" |
214 |
|
filesEventCount = 0 |
215 |
+ |
msg='' |
216 |
|
|
217 |
|
# ---- Iterate over the blocks in the dataset until ---- # |
218 |
|
# ---- we've met the requested total # of events ---- # |
224 |
|
|
225 |
|
if self.eventsbyblock.has_key(block) : |
226 |
|
numEventsInBlock = self.eventsbyblock[block] |
227 |
< |
common.logger.debug(5,'Events in Block File '+str(numEventsInBlock)) |
227 |
> |
common.logger.debug('Events in Block File '+str(numEventsInBlock)) |
228 |
|
|
229 |
|
files = filesbyblock[block] |
230 |
|
numFilesInBlock = len(files) |
234 |
|
if noBboundary == 0: # DD |
235 |
|
# ---- New block => New job ---- # |
236 |
|
parString = "" |
237 |
+ |
pString="" |
238 |
|
# counter for number of events in files currently worked on |
239 |
|
filesEventCount = 0 |
240 |
|
# flag if next while loop should touch new file |
244 |
|
|
245 |
|
# ---- Iterate over the files in the block until we've met the requested ---- # |
246 |
|
# ---- total # of events or we've gone over all the files in this block ---- # |
247 |
< |
pString='' |
247 |
> |
msg='\n' |
248 |
|
while ( (eventsRemaining > 0) and (fileCount < numFilesInBlock) and (jobCount < totalNumberOfJobs) ): |
249 |
|
file = files[fileCount] |
250 |
|
if self.useParent==1: |
251 |
|
parent = self.parentFiles[file] |
252 |
< |
for f in parent : |
165 |
< |
pString += '\\\"' + f + '\\\"\,' |
166 |
< |
common.logger.debug(6, "File "+str(file)+" has the following parents: "+str(parent)) |
167 |
< |
common.logger.write("File "+str(file)+" has the following parents: "+str(parent)) |
252 |
> |
common.logger.log(10-1, "File "+str(file)+" has the following parents: "+str(parent)) |
253 |
|
if newFile : |
254 |
|
try: |
255 |
|
numEventsInFile = self.eventsbyfile[file] |
256 |
< |
common.logger.debug(6, "File "+str(file)+" has "+str(numEventsInFile)+" events") |
256 |
> |
common.logger.log(10-1, "File "+str(file)+" has "+str(numEventsInFile)+" events") |
257 |
|
# increase filesEventCount |
258 |
|
filesEventCount += numEventsInFile |
259 |
|
# Add file to current job |
260 |
< |
parString += '\\\"' + file + '\\\"\,' |
260 |
> |
parString += file + ',' |
261 |
> |
if self.useParent==1: |
262 |
> |
for f in parent : |
263 |
> |
pString += f + ',' |
264 |
|
newFile = 0 |
265 |
|
except KeyError: |
266 |
< |
common.logger.message("File "+str(file)+" has unknown number of events: skipping") |
266 |
> |
common.logger.info("File "+str(file)+" has unknown number of events: skipping") |
267 |
|
|
268 |
|
eventsPerJobRequested = min(eventsPerJobRequested, eventsRemaining) |
269 |
|
# if less events in file remain than eventsPerJobRequested |
276 |
|
if ( fileCount == numFilesInBlock-1 ) : |
277 |
|
# end job using last file, use remaining events in block |
278 |
|
# close job and touch new file |
279 |
< |
fullString = parString[:-2] |
279 |
> |
fullString = parString[:-1] |
280 |
|
if self.useParent==1: |
281 |
< |
fullParentString = pString[:-2] |
281 |
> |
fullParentString = pString[:-1] |
282 |
|
list_of_lists.append([fullString,fullParentString,str(-1),str(jobSkipEventCount)]) |
283 |
|
else: |
284 |
|
list_of_lists.append([fullString,str(-1),str(jobSkipEventCount)]) |
285 |
< |
common.logger.debug(3,"Job "+str(jobCount+1)+" can run over "+str(filesEventCount - jobSkipEventCount)+" events (last file in block).") |
285 |
> |
msg += "Job %s can run over %s events (last file in block).\n"%(str(jobCount+1), str(filesEventCount - jobSkipEventCount)) |
286 |
|
jobDestination.append(blockSites[block]) |
287 |
< |
common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(jobDestination[jobCount])) |
287 |
> |
msg += "Job %s Destination: %s\n"%(str(jobCount+1),str(SE2CMS(jobDestination[jobCount]))) |
288 |
|
# fill jobs of block dictionary |
289 |
|
jobsOfBlock[block].append(jobCount+1) |
290 |
|
# reset counter |
305 |
|
# if events in file equal to eventsPerJobRequested |
306 |
|
elif ( filesEventCount - jobSkipEventCount == eventsPerJobRequested ) : |
307 |
|
# close job and touch new file |
308 |
< |
fullString = parString[:-2] |
308 |
> |
fullString = parString[:-1] |
309 |
|
if self.useParent==1: |
310 |
< |
fullParentString = pString[:-2] |
310 |
> |
fullParentString = pString[:-1] |
311 |
|
list_of_lists.append([fullString,fullParentString,str(eventsPerJobRequested),str(jobSkipEventCount)]) |
312 |
|
else: |
313 |
|
list_of_lists.append([fullString,str(eventsPerJobRequested),str(jobSkipEventCount)]) |
314 |
< |
common.logger.debug(3,"Job "+str(jobCount+1)+" can run over "+str(eventsPerJobRequested)+" events.") |
314 |
> |
msg += "Job %s can run over %s events.\n"%(str(jobCount+1),str(eventsPerJobRequested)) |
315 |
|
jobDestination.append(blockSites[block]) |
316 |
< |
common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(jobDestination[jobCount])) |
316 |
> |
msg+= "Job %s Destination: %s\n"%(str(jobCount+1),str(SE2CMS(jobDestination[jobCount]))) |
317 |
|
jobsOfBlock[block].append(jobCount+1) |
318 |
|
# reset counter |
319 |
|
jobCount = jobCount + 1 |
330 |
|
# if more events in file remain than eventsPerJobRequested |
331 |
|
else : |
332 |
|
# close job but don't touch new file |
333 |
< |
fullString = parString[:-2] |
333 |
> |
fullString = parString[:-1] |
334 |
|
if self.useParent==1: |
335 |
< |
fullParentString = pString[:-2] |
335 |
> |
fullParentString = pString[:-1] |
336 |
|
list_of_lists.append([fullString,fullParentString,str(eventsPerJobRequested),str(jobSkipEventCount)]) |
337 |
|
else: |
338 |
|
list_of_lists.append([fullString,str(eventsPerJobRequested),str(jobSkipEventCount)]) |
339 |
< |
common.logger.debug(3,"Job "+str(jobCount+1)+" can run over "+str(eventsPerJobRequested)+" events.") |
339 |
> |
msg += "Job %s can run over %s events.\n"%(str(jobCount+1),str(eventsPerJobRequested)) |
340 |
|
jobDestination.append(blockSites[block]) |
341 |
< |
common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(jobDestination[jobCount])) |
341 |
> |
msg+= "Job %s Destination: %s\n"%(str(jobCount+1),str(SE2CMS(jobDestination[jobCount]))) |
342 |
|
jobsOfBlock[block].append(jobCount+1) |
343 |
|
# increase counter |
344 |
|
jobCount = jobCount + 1 |
349 |
|
jobSkipEventCount = eventsPerJobRequested - (filesEventCount - jobSkipEventCount - self.eventsbyfile[file]) |
350 |
|
# remove all but the last file |
351 |
|
filesEventCount = self.eventsbyfile[file] |
352 |
+ |
pString_tmp='' |
353 |
|
if self.useParent==1: |
354 |
< |
for f in parent : pString += '\\\"' + f + '\\\"\,' |
355 |
< |
parString = '\\\"' + file + '\\\"\,' |
354 |
> |
for f in parent : pString_tmp += f + ',' |
355 |
> |
pString = pString_tmp |
356 |
> |
parString = file + ',' |
357 |
|
pass # END if |
358 |
|
pass # END while (iterate over files in the block) |
359 |
|
pass # END while (iterate over blocks in the dataset) |
360 |
+ |
common.logger.debug(msg) |
361 |
|
self.ncjobs = self.total_number_of_jobs = jobCount |
362 |
|
if (eventsRemaining > 0 and jobCount < totalNumberOfJobs ): |
363 |
< |
common.logger.message("Could not run on all requested events because some blocks not hosted at allowed sites.") |
364 |
< |
common.logger.message(str(jobCount)+" job(s) can run on "+str(totalEventCount)+" events.\n") |
365 |
< |
|
363 |
> |
common.logger.info("Could not run on all requested events because some blocks not hosted at allowed sites.") |
364 |
> |
common.logger.info(str(jobCount)+" job(s) can run on "+str(totalEventCount)+" events.\n") |
365 |
> |
|
366 |
|
# skip check on block with no sites DD |
367 |
|
if noBboundary == 0 : self.checkBlockNoSite(blocks,jobsOfBlock,blockSites) |
368 |
|
|
369 |
|
# prepare dict output |
370 |
|
dictOut = {} |
371 |
+ |
dictOut['params']= ['InputFiles','MaxEvents','SkipEvents'] |
372 |
+ |
if self.useParent: dictOut['params']= ['InputFiles','ParentFiles','MaxEvents','SkipEvents'] |
373 |
|
dictOut['args'] = list_of_lists |
374 |
|
dictOut['jobDestination'] = jobDestination |
375 |
|
dictOut['njobs']=self.total_number_of_jobs |
378 |
|
|
379 |
|
# keep trace of block with no sites to print a warning at the end |
380 |
|
|
381 |
< |
def checkBlockNoSite(self,blocks,jobsOfBlock,blockSites): |
381 |
> |
def checkBlockNoSite(self,blocks,jobsOfBlock,blockSites): |
382 |
|
# screen output |
383 |
|
screenOutput = "List of jobs and available destination sites:\n\n" |
384 |
|
noSiteBlock = [] |
385 |
|
bloskNoSite = [] |
386 |
+ |
allBlock = [] |
387 |
|
|
388 |
|
blockCounter = 0 |
389 |
+ |
saveFblocks ='' |
390 |
|
for block in blocks: |
391 |
|
if block in jobsOfBlock.keys() : |
392 |
|
blockCounter += 1 |
393 |
+ |
allBlock.append( blockCounter ) |
394 |
+ |
sites=self.blackWhiteListParser.checkWhiteList(self.blackWhiteListParser.checkBlackList(blockSites[block],[block]),[block]) |
395 |
|
screenOutput += "Block %5i: jobs %20s: sites: %s\n" % (blockCounter,spanRanges(jobsOfBlock[block]), |
396 |
< |
','.join(self.blackWhiteListParser.checkWhiteList(self.blackWhiteListParser.checkBlackList(blockSites[block],[block]),[block]))) |
397 |
< |
if len(self.blackWhiteListParser.checkWhiteList(self.blackWhiteListParser.checkBlackList(blockSites[block],[block]),[block])) == 0: |
396 |
> |
', '.join(SE2CMS(sites))) |
397 |
> |
if len(sites) == 0: |
398 |
|
noSiteBlock.append( spanRanges(jobsOfBlock[block]) ) |
399 |
|
bloskNoSite.append( blockCounter ) |
400 |
+ |
else: |
401 |
+ |
saveFblocks += str(block)+'\n' |
402 |
+ |
writeTXTfile(self, self.fileBlocks_FileName , saveFblocks) |
403 |
|
|
404 |
< |
common.logger.message(screenOutput) |
404 |
> |
common.logger.info(screenOutput) |
405 |
|
if len(noSiteBlock) > 0 and len(bloskNoSite) > 0: |
406 |
|
msg = 'WARNING: No sites are hosting any part of data for block:\n ' |
407 |
|
virgola = "" |
409 |
|
virgola = "," |
410 |
|
for block in bloskNoSite: |
411 |
|
msg += ' ' + str(block) + virgola |
412 |
< |
msg += '\n Related jobs:\n ' |
412 |
> |
msg += '\n\t\tRelated jobs:\n ' |
413 |
|
virgola = "" |
414 |
|
if len(noSiteBlock) > 1: |
415 |
|
virgola = "," |
416 |
|
for range_jobs in noSiteBlock: |
417 |
|
msg += str(range_jobs) + virgola |
418 |
< |
msg += '\n will not be submitted and this block of data can not be analyzed!\n' |
419 |
< |
if self.cfg_params.has_key('EDG.se_white_list'): |
420 |
< |
msg += 'WARNING: SE White List: '+self.cfg_params['EDG.se_white_list']+'\n' |
421 |
< |
msg += '(Hint: By whitelisting you force the job to run at this particular site(s).\n' |
422 |
< |
msg += 'Please check if the dataset is available at this site!)\n' |
423 |
< |
if self.cfg_params.has_key('EDG.ce_white_list'): |
424 |
< |
msg += 'WARNING: CE White List: '+self.cfg_params['EDG.ce_white_list']+'\n' |
425 |
< |
msg += '(Hint: By whitelisting you force the job to run at this particular site(s).\n' |
426 |
< |
msg += 'Please check if the dataset is available at this site!)\n' |
427 |
< |
|
428 |
< |
common.logger.message(msg) |
418 |
> |
msg += '\n\t\twill not be submitted and this block of data can not be analyzed!\n' |
419 |
> |
if self.cfg_params.has_key('GRID.se_white_list'): |
420 |
> |
msg += '\tWARNING: SE White List: '+self.cfg_params['GRID.se_white_list']+'\n' |
421 |
> |
msg += '\t(Hint: By whitelisting you force the job to run at this particular site(s).\n' |
422 |
> |
msg += '\tPlease check if the dataset is available at this site!)' |
423 |
> |
if self.cfg_params.has_key('GRID.ce_white_list'): |
424 |
> |
msg += '\tWARNING: CE White List: '+self.cfg_params['GRID.ce_white_list']+'\n' |
425 |
> |
msg += '\t(Hint: By whitelisting you force the job to run at this particular site(s).\n' |
426 |
> |
msg += '\tPlease check if the dataset is available at this site!)\n' |
427 |
> |
|
428 |
> |
common.logger.info(msg) |
429 |
> |
|
430 |
> |
if bloskNoSite == allBlock: |
431 |
> |
msg += 'Requested jobs cannot be Created! \n' |
432 |
> |
if self.cfg_params.has_key('GRID.se_white_list'): |
433 |
> |
msg += '\tWARNING: SE White List: '+self.cfg_params['GRID.se_white_list']+'\n' |
434 |
> |
msg += '\t(Hint: By whitelisting you force the job to run at this particular site(s).\n' |
435 |
> |
msg += '\tPlease check if the dataset is available at this site!)' |
436 |
> |
if self.cfg_params.has_key('GRID.ce_white_list'): |
437 |
> |
msg += '\tWARNING: CE White List: '+self.cfg_params['GRID.ce_white_list']+'\n' |
438 |
> |
msg += '\t(Hint: By whitelisting you force the job to run at this particular site(s).\n' |
439 |
> |
msg += '\tPlease check if the dataset is available at this site!)\n' |
440 |
> |
raise CrabException(msg) |
441 |
|
|
442 |
|
return |
443 |
|
|
444 |
|
|
445 |
|
######################################################################## |
446 |
< |
def jobSplittingByRun(self): |
446 |
> |
def jobSplittingByRun(self): |
447 |
|
""" |
448 |
|
""" |
337 |
– |
from sets import Set |
338 |
– |
from WMCore.JobSplitting.RunBased import RunBased |
339 |
– |
from WMCore.DataStructs.Workflow import Workflow |
340 |
– |
from WMCore.DataStructs.File import File |
341 |
– |
from WMCore.DataStructs.Fileset import Fileset |
342 |
– |
from WMCore.DataStructs.Subscription import Subscription |
343 |
– |
from WMCore.JobSplitting.SplitterFactory import SplitterFactory |
344 |
– |
from WMCore.DataStructs.Run import Run |
449 |
|
|
450 |
|
self.checkUserSettings() |
451 |
< |
blockSites = self.args['blockSites'] |
451 |
> |
blockSites = self.args['blockSites'] |
452 |
|
pubdata = self.args['pubdata'] |
453 |
|
|
454 |
|
if self.selectNumberOfJobs == 0 : |
455 |
|
self.theNumberOfJobs = 9999999 |
456 |
|
blocks = {} |
457 |
< |
runList = [] |
457 |
> |
runList = [] |
458 |
|
thefiles = Fileset(name='FilesToSplit') |
459 |
|
fileList = pubdata.getListFiles() |
460 |
|
for f in fileList: |
357 |
– |
# print f |
461 |
|
block = f['Block']['Name'] |
462 |
< |
# if not blocks.has_key(block): |
360 |
< |
# blocks[block] = reader.listFileBlockLocation(block) |
361 |
< |
try: |
462 |
> |
try: |
463 |
|
f['Block']['StorageElementList'].extend(blockSites[block]) |
464 |
|
except: |
465 |
|
continue |
467 |
|
[ wmbsFile['locations'].add(x) for x in blockSites[block] ] |
468 |
|
wmbsFile['block'] = block |
469 |
|
runNum = f['RunsList'][0]['RunNumber'] |
470 |
< |
runList.append(runNum) |
470 |
> |
runList.append(runNum) |
471 |
|
myRun = Run(runNumber=runNum) |
472 |
|
wmbsFile.addRun( myRun ) |
473 |
|
thefiles.addFile( |
474 |
|
wmbsFile |
475 |
|
) |
476 |
< |
|
476 |
> |
|
477 |
|
work = Workflow() |
478 |
|
subs = Subscription( |
479 |
|
fileset = thefiles, |
482 |
|
type = "Processing") |
483 |
|
splitter = SplitterFactory() |
484 |
|
jobfactory = splitter(subs) |
485 |
< |
|
486 |
< |
#loop over all runs |
386 |
< |
set = Set(runList) |
485 |
> |
|
486 |
> |
#loop over all runs |
487 |
|
list_of_lists = [] |
488 |
|
jobDestination = [] |
489 |
< |
|
489 |
> |
list_of_blocks = [] |
490 |
|
count = 0 |
491 |
< |
for i in list(set): |
491 |
> |
for jobGroup in jobfactory(): |
492 |
|
if count < self.theNumberOfJobs: |
493 |
< |
res = self.getJobInfo(jobfactory()) |
494 |
< |
parString = '' |
493 |
> |
res = self.getJobInfo(jobGroup) |
494 |
> |
parString = '' |
495 |
|
for file in res['lfns']: |
496 |
< |
parString += '\\\"' + file + '\\\"\,' |
497 |
< |
fullString = parString[:-2] |
498 |
< |
list_of_lists.append([fullString,str(-1),str(0)]) |
496 |
> |
parString += file + ',' |
497 |
> |
fullString = parString[:-1] |
498 |
> |
list_of_lists.append([fullString,str(-1),str(0)]) |
499 |
|
#need to check single file location |
500 |
< |
jobDestination.append(res['locations']) |
500 |
> |
jobDestination.append(res['locations']) |
501 |
> |
list_of_blocks.append(res['block']) |
502 |
|
count +=1 |
503 |
< |
#print jobDestination |
403 |
< |
# prepare dict output |
503 |
> |
# prepare dict output |
504 |
|
dictOut = {} |
505 |
+ |
dictOut['params']= ['InputFiles','MaxEvents','SkipEvents'] |
506 |
|
dictOut['args'] = list_of_lists |
507 |
|
dictOut['jobDestination'] = jobDestination |
508 |
|
dictOut['njobs']=count |
509 |
|
|
510 |
+ |
self.cacheBlocks(list_of_blocks,jobDestination) |
511 |
+ |
|
512 |
|
return dictOut |
513 |
|
|
514 |
|
def getJobInfo( self,jobGroup ): |
515 |
|
res = {} |
516 |
< |
lfns = [] |
517 |
< |
locations = [] |
516 |
> |
lfns = [] |
517 |
> |
locations = [] |
518 |
|
tmp_check=0 |
519 |
|
for job in jobGroup.jobs: |
520 |
|
for file in job.getFiles(): |
521 |
< |
lfns.append(file['lfn']) |
521 |
> |
lfns.append(file['lfn']) |
522 |
|
for loc in file['locations']: |
523 |
|
if tmp_check < 1 : |
524 |
|
locations.append(loc) |
525 |
< |
tmp_check = tmp_check + 1 |
526 |
< |
### qui va messo il check per la locations |
527 |
< |
res['lfns'] = lfns |
528 |
< |
res['locations'] = locations |
529 |
< |
return res |
530 |
< |
|
525 |
> |
res['block']= file['block'] |
526 |
> |
tmp_check = tmp_check + 1 |
527 |
> |
res['lfns'] = lfns |
528 |
> |
res['locations'] = locations |
529 |
> |
return res |
530 |
> |
|
531 |
|
######################################################################## |
532 |
< |
def jobSplittingNoInput(self): |
532 |
> |
def prepareSplittingNoInput(self): |
533 |
|
""" |
431 |
– |
Perform job splitting based on number of event per job |
534 |
|
""" |
433 |
– |
common.logger.debug(5,'Splitting per events') |
434 |
– |
self.checkUserSettings() |
435 |
– |
jobDestination=[] |
436 |
– |
if ( (self.selectTotalNumberEvents + self.selectEventsPerJob + self.selectNumberOfJobs) != 2 ): |
437 |
– |
msg = 'Must define exactly two of total_number_of_events, events_per_job, or number_of_jobs.' |
438 |
– |
raise CrabException(msg) |
439 |
– |
|
440 |
– |
managedGenerators =self.args['managedGenerators'] |
441 |
– |
generator = self.args['generator'] |
442 |
– |
firstRun = self.cfg_params.get('CMSSW.first_run',None) |
443 |
– |
|
535 |
|
if (self.selectEventsPerJob): |
536 |
< |
common.logger.message('Required '+str(self.eventsPerJob)+' events per job ') |
536 |
> |
common.logger.info('Required '+str(self.eventsPerJob)+' events per job ') |
537 |
|
if (self.selectNumberOfJobs): |
538 |
< |
common.logger.message('Required '+str(self.theNumberOfJobs)+' jobs in total ') |
538 |
> |
common.logger.info('Required '+str(self.theNumberOfJobs)+' jobs in total ') |
539 |
|
if (self.selectTotalNumberEvents): |
540 |
< |
common.logger.message('Required '+str(self.total_number_of_events)+' events in total ') |
540 |
> |
common.logger.info('Required '+str(self.total_number_of_events)+' events in total ') |
541 |
|
|
542 |
|
if (self.total_number_of_events < 0): |
543 |
|
msg='Cannot split jobs per Events with "-1" as total number of events' |
554 |
|
self.total_number_of_jobs = self.theNumberOfJobs |
555 |
|
self.eventsPerJob = int(self.total_number_of_events/self.total_number_of_jobs) |
556 |
|
|
557 |
< |
common.logger.debug(5,'N jobs '+str(self.total_number_of_jobs)) |
557 |
> |
|
558 |
> |
def jobSplittingNoInput(self): |
559 |
> |
""" |
560 |
> |
Perform job splitting based on number of event per job |
561 |
> |
""" |
562 |
> |
common.logger.debug('Splitting per events') |
563 |
> |
self.checkUserSettings() |
564 |
> |
jobDestination=[] |
565 |
> |
if ( (self.selectTotalNumberEvents + self.selectEventsPerJob + self.selectNumberOfJobs) != 2 ): |
566 |
> |
msg = 'Must define exactly two of total_number_of_events, events_per_job, or number_of_jobs.' |
567 |
> |
raise CrabException(msg) |
568 |
> |
|
569 |
> |
managedGenerators =self.args['managedGenerators'] |
570 |
> |
generator = self.args['generator'] |
571 |
> |
firstLumi = self.cfg_params.get('CMSSW.first_lumi', 1) |
572 |
> |
|
573 |
> |
self.prepareSplittingNoInput() |
574 |
> |
|
575 |
> |
common.logger.debug('N jobs '+str(self.total_number_of_jobs)) |
576 |
|
|
577 |
|
# is there any remainder? |
578 |
|
check = int(self.total_number_of_events) - (int(self.total_number_of_jobs)*self.eventsPerJob) |
579 |
|
|
580 |
< |
common.logger.debug(5,'Check '+str(check)) |
580 |
> |
common.logger.debug('Check '+str(check)) |
581 |
|
|
582 |
< |
common.logger.message(str(self.total_number_of_jobs)+' jobs can be created, each for '+str(self.eventsPerJob)+' for a total of '+str(self.total_number_of_jobs*self.eventsPerJob)+' events') |
582 |
> |
common.logger.info(str(self.total_number_of_jobs)+' jobs can be created, each for '+str(self.eventsPerJob)+' for a total of '+str(self.total_number_of_jobs*self.eventsPerJob)+' events') |
583 |
|
if check > 0: |
584 |
< |
common.logger.message('Warning: asked '+str(self.total_number_of_events)+' but can do only '+str(int(self.total_number_of_jobs)*self.eventsPerJob)) |
584 |
> |
common.logger.info('Warning: asked '+str(self.total_number_of_events)+' but can do only '+str(int(self.total_number_of_jobs)*self.eventsPerJob)) |
585 |
|
|
586 |
|
# argument is seed number.$i |
587 |
|
self.list_of_args = [] |
588 |
|
for i in range(self.total_number_of_jobs): |
589 |
|
## Since there is no input, any site is good |
590 |
< |
jobDestination.append([""]) #must be empty to write correctly the xml |
590 |
> |
jobDestination.append([""]) # must be empty to correctly write the XML |
591 |
|
args=[] |
592 |
< |
if (firstRun): |
593 |
< |
## pythia first run |
485 |
< |
args.append(str(firstRun)+str(i)) |
592 |
> |
if (firstLumi): # Pythia first lumi |
593 |
> |
args.append(str(int(firstLumi)+i)) |
594 |
|
if (generator in managedGenerators): |
595 |
< |
if (generator == 'comphep' and i == 0): |
595 |
> |
args.append(generator) |
596 |
> |
if (generator == 'comphep' and i == 0): |
597 |
|
# COMPHEP is brain-dead and wants event #'s like 1,100,200,300 |
598 |
|
args.append('1') |
599 |
< |
else: |
599 |
> |
else: |
600 |
|
args.append(str(i*self.eventsPerJob)) |
601 |
|
args.append(str(self.eventsPerJob)) |
602 |
|
self.list_of_args.append(args) |
603 |
|
# prepare dict output |
604 |
+ |
|
605 |
|
dictOut = {} |
606 |
+ |
dictOut['params'] = ['MaxEvents'] |
607 |
+ |
if (firstLumi): |
608 |
+ |
dictOut['params'] = ['FirstLumi','MaxEvents'] |
609 |
+ |
if (generator in managedGenerators): |
610 |
+ |
dictOut['params'] = ['FirstLumi', 'Generator', 'FirstEvent', 'MaxEvents'] |
611 |
+ |
else: |
612 |
+ |
if (generator in managedGenerators) : |
613 |
+ |
dictOut['params'] = ['Generator', 'FirstEvent', 'MaxEvents'] |
614 |
|
dictOut['args'] = self.list_of_args |
615 |
|
dictOut['jobDestination'] = jobDestination |
616 |
|
dictOut['njobs']=self.total_number_of_jobs |
627 |
|
msg = 'must specify number_of_jobs.' |
628 |
|
raise crabexception(msg) |
629 |
|
jobDestination = [] |
630 |
< |
common.logger.debug(5,'Splitting per job') |
631 |
< |
common.logger.message('Required '+str(self.theNumberOfJobs)+' jobs in total ') |
630 |
> |
common.logger.debug('Splitting per job') |
631 |
> |
common.logger.info('Required '+str(self.theNumberOfJobs)+' jobs in total ') |
632 |
> |
|
633 |
> |
# self.total_number_of_jobs = self.theNumberOfJobs |
634 |
|
|
635 |
< |
self.total_number_of_jobs = self.theNumberOfJobs |
635 |
> |
self.prepareSplittingNoInput() |
636 |
|
|
637 |
< |
common.logger.debug(5,'N jobs '+str(self.total_number_of_jobs)) |
637 |
> |
common.logger.debug('N jobs '+str(self.total_number_of_jobs)) |
638 |
|
|
639 |
< |
common.logger.message(str(self.total_number_of_jobs)+' jobs can be created') |
639 |
> |
common.logger.info(str(self.total_number_of_jobs)+' jobs can be created') |
640 |
|
|
641 |
|
# argument is seed number.$i |
642 |
|
self.list_of_args = [] |
643 |
|
for i in range(self.total_number_of_jobs): |
644 |
+ |
args=[] |
645 |
|
jobDestination.append([""]) |
646 |
< |
self.list_of_args.append([str(i)]) |
646 |
> |
if self.eventsPerJob != 0 : |
647 |
> |
args.append(str(self.eventsPerJob)) |
648 |
> |
self.list_of_args.append(args) |
649 |
|
|
650 |
|
# prepare dict output |
651 |
|
dictOut = {} |
652 |
< |
dictOut['args'] = self.list_of_args |
652 |
> |
dictOut['params'] = ['MaxEvents'] |
653 |
> |
dictOut['args'] = self.list_of_args |
654 |
|
dictOut['jobDestination'] = jobDestination |
655 |
|
dictOut['njobs']=self.total_number_of_jobs |
656 |
|
return dictOut |
533 |
– |
|
657 |
|
|
658 |
< |
def jobSplittingByLumi(self): |
658 |
> |
|
659 |
> |
def jobSplittingByLumi(self): |
660 |
|
""" |
661 |
+ |
Split task into jobs by Lumi section paying attention to which |
662 |
+ |
lumis should be run (according to the analysis dataset). |
663 |
+ |
This uses WMBS job splitting which does not split files over jobs |
664 |
+ |
so the job will have AT LEAST as many lumis as requested, perhaps |
665 |
+ |
more |
666 |
|
""" |
667 |
< |
return |
667 |
> |
|
668 |
> |
common.logger.debug('Splitting by Lumi') |
669 |
> |
self.checkLumiSettings() |
670 |
> |
|
671 |
> |
blockSites = self.args['blockSites'] |
672 |
> |
pubdata = self.args['pubdata'] |
673 |
> |
|
674 |
> |
lumisPerFile = pubdata.getLumis() |
675 |
> |
|
676 |
> |
# Make the list of WMBS files for job splitter |
677 |
> |
fileList = pubdata.getListFiles() |
678 |
> |
thefiles = Fileset(name='FilesToSplit') |
679 |
> |
for jobFile in fileList: |
680 |
> |
block = jobFile['Block']['Name'] |
681 |
> |
try: |
682 |
> |
jobFile['Block']['StorageElementList'].extend(blockSites[block]) |
683 |
> |
except: |
684 |
> |
continue |
685 |
> |
wmbsFile = File(jobFile['LogicalFileName']) |
686 |
> |
[ wmbsFile['locations'].add(x) for x in blockSites[block] ] |
687 |
> |
wmbsFile['block'] = block |
688 |
> |
for lumi in lumisPerFile[jobFile['LogicalFileName']]: |
689 |
> |
wmbsFile.addRun(Run(lumi[0], lumi[1])) |
690 |
> |
thefiles.addFile(wmbsFile) |
691 |
> |
|
692 |
> |
# Create the factory and workflow |
693 |
> |
work = Workflow() |
694 |
> |
subs = Subscription(fileset = thefiles, workflow = work, |
695 |
> |
split_algo = 'LumiBased', type = "Processing") |
696 |
> |
splitter = SplitterFactory() |
697 |
> |
jobFactory = splitter(subs) |
698 |
> |
|
699 |
> |
list_of_lists = [] |
700 |
> |
jobDestination = [] |
701 |
> |
jobCount = 0 |
702 |
> |
lumisCreated = 0 |
703 |
> |
list_of_blocks = [] |
704 |
> |
if not self.limitJobLumis: |
705 |
> |
self.lumisPerJob = pubdata.getMaxLumis() // self.theNumberOfJobs + 1 |
706 |
> |
common.logger.info('Each job will process about %s lumis.' % |
707 |
> |
self.lumisPerJob) |
708 |
> |
|
709 |
> |
for jobGroup in jobFactory(lumis_per_job = self.lumisPerJob): |
710 |
> |
for job in jobGroup.jobs: |
711 |
> |
if (self.limitNJobs and jobCount >= self.theNumberOfJobs): |
712 |
> |
common.logger.info('Limit on number of jobs reached.') |
713 |
> |
break |
714 |
> |
if (self.limitTotalLumis and lumisCreated >= self.totalNLumis): |
715 |
> |
common.logger.info('Limit on number of lumis reached.') |
716 |
> |
break |
717 |
> |
lumis = [] |
718 |
> |
lfns = [] |
719 |
> |
locations = [] |
720 |
> |
blocks = [] |
721 |
> |
firstFile = True |
722 |
> |
# Collect information from all the files |
723 |
> |
for jobFile in job.getFiles(): |
724 |
> |
doFile = False |
725 |
> |
if firstFile: # Get locations from first file in the job |
726 |
> |
for loc in jobFile['locations']: |
727 |
> |
locations.append(loc) |
728 |
> |
blocks.append(jobFile['block']) |
729 |
> |
firstFile = False |
730 |
> |
# Accumulate Lumis from all files |
731 |
> |
for lumiList in jobFile['runs']: |
732 |
> |
theRun = lumiList.run |
733 |
> |
for theLumi in list(lumiList): |
734 |
> |
if (not self.limitTotalLumis) or \ |
735 |
> |
(lumisCreated < self.totalNLumis): |
736 |
> |
doFile = True |
737 |
> |
lumisCreated += 1 |
738 |
> |
lumis.append( (theRun, theLumi) ) |
739 |
> |
if doFile: |
740 |
> |
lfns.append(jobFile['lfn']) |
741 |
> |
fileString = ','.join(lfns) |
742 |
> |
lumiLister = LumiList(lumis = lumis) |
743 |
> |
lumiString = lumiLister.getCMSSWString() |
744 |
> |
list_of_lists.append([fileString, str(-1), str(0), lumiString]) |
745 |
> |
list_of_blocks.append(blocks) |
746 |
> |
jobDestination.append(locations) |
747 |
> |
jobCount += 1 |
748 |
> |
common.logger.debug('Job %s will run on %s files and %s lumis ' |
749 |
> |
% (jobCount, len(lfns), len(lumis) )) |
750 |
> |
|
751 |
> |
common.logger.info('%s jobs created to run on %s lumis' % |
752 |
> |
(jobCount, lumisCreated)) |
753 |
> |
|
754 |
> |
# Prepare dict output matching back to non-WMBS job creation |
755 |
> |
dictOut = {} |
756 |
> |
dictOut['params'] = ['InputFiles', 'MaxEvents', 'SkipEvents', 'Lumis'] |
757 |
> |
dictOut['args'] = list_of_lists |
758 |
> |
dictOut['jobDestination'] = jobDestination |
759 |
> |
dictOut['njobs'] = jobCount |
760 |
> |
|
761 |
> |
self.cacheBlocks(list_of_blocks,jobDestination) |
762 |
> |
|
763 |
> |
return dictOut |
764 |
> |
|
765 |
> |
def cacheBlocks(self, blocks,destinations): |
766 |
> |
|
767 |
> |
saveFblocks='' |
768 |
> |
for i in range(len(blocks)): |
769 |
> |
sites=self.blackWhiteListParser.checkWhiteList(self.blackWhiteListParser.checkBlackList(destinations[i])) |
770 |
> |
if len(sites) != 0: |
771 |
> |
for block in blocks[i]: |
772 |
> |
saveFblocks += str(block)+'\n' |
773 |
> |
writeTXTfile(self, self.fileBlocks_FileName , saveFblocks) |
774 |
> |
|
775 |
|
def Algos(self): |
776 |
|
""" |
777 |
|
Define key splittingType matrix |
778 |
|
""" |
779 |
< |
SplitAlogs = { |
780 |
< |
'EventBased' : self.jobSplittingByEvent, |
779 |
> |
SplitAlogs = { |
780 |
> |
'EventBased' : self.jobSplittingByEvent, |
781 |
|
'RunBased' : self.jobSplittingByRun, |
782 |
< |
'LumiBased' : self.jobSplittingByLumi, |
783 |
< |
'NoInput' : self.jobSplittingNoInput, |
782 |
> |
'LumiBased' : self.jobSplittingByLumi, |
783 |
> |
'NoInput' : self.jobSplittingNoInput, |
784 |
|
'ForScript' : self.jobSplittingForScript |
785 |
< |
} |
785 |
> |
} |
786 |
|
return SplitAlogs |
787 |
|
|