ViewVC Help

View File | Revision Log | Show Annotations | Root Listing

root/cvsroot/COMP/CRAB/python/Splitter.py

Comparing COMP/CRAB/python/Splitter.py (file contents):
Revision 1.4 by spiga, Mon Feb 9 18:20:20 2009 UTC vs.
Revision 1.19 by slacapra, Wed Jun 10 11:31:33 2009 UTC

#	Line 1 | Line 1
1		import common
2	–	from crab_logger import Logger
2		from crab_exceptions import *
3		from crab_util import *
4		from WMCore.SiteScreening.BlackWhiteListParser import SEBlackWhiteListParser
#	Line 10 | Line 9 | class JobSplitter:
9		self.args=args
10		#self.maxEvents
11		# init BlackWhiteListParser
12	<	seWhiteList = cfg_params.get('EDG.se_white_list',[])
13	<	seBlackList = cfg_params.get('EDG.se_black_list',[])
14	<	self.blackWhiteListParser = SEBlackWhiteListParser(seWhiteList, seBlackList, common.logger)
12	>	seWhiteList = cfg_params.get('GRID.se_white_list',[])
13	>	seBlackList = cfg_params.get('GRID.se_black_list',[])
14	>	self.blackWhiteListParser = SEBlackWhiteListParser(seWhiteList, seBlackList, common.logger())
15
16
17		def checkUserSettings(self):
#	Line 92 | Line 91 | class JobSplitter:
91		# If user requested more events than are in the dataset
92		elif (totalEventsRequested > self.maxEvents):
93		eventsRemaining = self.maxEvents
94	<	common.logger.message("Requested "+str(self.total_number_of_events)+ " events, but only "+str(self.maxEvents)+" events are available.")
94	>	common.logger.info("Requested "+str(self.total_number_of_events)+ " events, but only "+str(self.maxEvents)+" events are available.")
95		# If user requested less events than are in the dataset
96		else:
97		eventsRemaining = totalEventsRequested
#	Line 108 | Line 107 | class JobSplitter:
107		eventsPerJobRequested = int(eventsRemaining/self.theNumberOfJobs)
108
109		if (self.selectNumberOfJobs):
110	<	common.logger.message("May not create the exact number_of_jobs requested.")
110	>	common.logger.info("May not create the exact number_of_jobs requested.")
111
112		# old... to remove Daniele
113		totalNumberOfJobs = 999999999
#	Line 125 | Line 124 | class JobSplitter:
124		jobsOfBlock = {}
125
126		parString = ""
127	+	pString = ""
128		filesEventCount = 0
129
130		# ---- Iterate over the blocks in the dataset until ---- #
#	Line 137 | Line 137 | class JobSplitter:
137
138		if self.eventsbyblock.has_key(block) :
139		numEventsInBlock = self.eventsbyblock[block]
140	<	common.logger.debug(5,'Events in Block File '+str(numEventsInBlock))
140	>	common.logger.debug('Events in Block File '+str(numEventsInBlock))
141
142		files = filesbyblock[block]
143		numFilesInBlock = len(files)
#	Line 147 | Line 147 | class JobSplitter:
147		if noBboundary == 0: # DD
148		# ---- New block => New job ---- #
149		parString = ""
150	+	pString=""
151		# counter for number of events in files currently worked on
152		filesEventCount = 0
153		# flag if next while loop should touch new file
#	Line 156 | Line 157 | class JobSplitter:
157
158		# ---- Iterate over the files in the block until we've met the requested ---- #
159		# ---- total # of events or we've gone over all the files in this block  ---- #
160	<	pString=''
160	>	msg='\n'
161		while ( (eventsRemaining > 0) and (fileCount < numFilesInBlock) and (jobCount < totalNumberOfJobs) ):
162		file = files[fileCount]
163		if self.useParent==1:
164		parent = self.parentFiles[file]
165	<	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))
165	>	common.logger.log(10-1, "File "+str(file)+" has the following parents: "+str(parent))
166		if newFile :
167		try:
168		numEventsInFile = self.eventsbyfile[file]
169	<	common.logger.debug(6, "File "+str(file)+" has "+str(numEventsInFile)+" events")
169	>	common.logger.log(10-1, "File "+str(file)+" has "+str(numEventsInFile)+" events")
170		# increase filesEventCount
171		filesEventCount += numEventsInFile
172		# Add file to current job
173	<	parString += '\\\"' + file + '\\\"\,'
173	>	parString +=  file + ','
174	>	if self.useParent==1:
175	>	for f in parent :
176	>	pString += f  + ','
177		newFile = 0
178		except KeyError:
179	<	common.logger.message("File "+str(file)+" has unknown number of events: skipping")
179	>	common.logger.info("File "+str(file)+" has unknown number of events: skipping")
180
181		eventsPerJobRequested = min(eventsPerJobRequested, eventsRemaining)
182		# if less events in file remain than eventsPerJobRequested
#	Line 188 | Line 189 | class JobSplitter:
189		if ( fileCount == numFilesInBlock-1 ) :
190		# end job using last file, use remaining events in block
191		# close job and touch new file
192	<	fullString = parString[:-2]
192	>	fullString = parString[:-1]
193		if self.useParent==1:
194	<	fullParentString = pString[:-2]
194	>	fullParentString = pString[:-1]
195		list_of_lists.append([fullString,fullParentString,str(-1),str(jobSkipEventCount)])
196		else:
197		list_of_lists.append([fullString,str(-1),str(jobSkipEventCount)])
198	<	common.logger.debug(3,"Job "+str(jobCount+1)+" can run over "+str(filesEventCount - jobSkipEventCount)+" events (last file in block).")
198	>	msg += "Job %s can run over %s  events (last file in block).\n"%(str(jobCount+1), str(filesEventCount - jobSkipEventCount))
199		jobDestination.append(blockSites[block])
200	<	common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(jobDestination[jobCount]))
200	>	msg += "Job %s Destination: %s\n"%(str(jobCount+1),str(destinationCMS(jobDestination[jobCount])))
201	>	from ProdCommon.SiteDB.CmsSiteMapper import CmsSEMap
202	>	cms_se = CmsSEMap()
203	>	SEDestination = [cms_se[dest] for dest in jobDestination[jobCount]]
204	>	msg+="\t  CMSDestination: %s "%(str(SEDestination))
205		# fill jobs of block dictionary
206		jobsOfBlock[block].append(jobCount+1)
207		# reset counter
#	Line 217 | Line 222 | class JobSplitter:
222		# if events in file equal to eventsPerJobRequested
223		elif ( filesEventCount - jobSkipEventCount == eventsPerJobRequested ) :
224		# close job and touch new file
225	<	fullString = parString[:-2]
225	>	fullString = parString[:-1]
226		if self.useParent==1:
227	<	fullParentString = pString[:-2]
227	>	fullParentString = pString[:-1]
228		list_of_lists.append([fullString,fullParentString,str(eventsPerJobRequested),str(jobSkipEventCount)])
229		else:
230		list_of_lists.append([fullString,str(eventsPerJobRequested),str(jobSkipEventCount)])
231	<	common.logger.debug(3,"Job "+str(jobCount+1)+" can run over "+str(eventsPerJobRequested)+" events.")
231	>	msg += "Job %s can run over %s events.\n"%(str(jobCount+1),str(eventsPerJobRequested))
232		jobDestination.append(blockSites[block])
233	<	common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(jobDestination[jobCount]))
233	>	msg+= "Job %s Destination: %s\n"%(str(jobCount+1),str(destinationCMS(jobDestination[jobCount])))
234		jobsOfBlock[block].append(jobCount+1)
235		# reset counter
236		jobCount = jobCount + 1
#	Line 242 | Line 247 | class JobSplitter:
247		# if more events in file remain than eventsPerJobRequested
248		else :
249		# close job but don't touch new file
250	<	fullString = parString[:-2]
250	>	fullString = parString[:-1]
251		if self.useParent==1:
252	<	fullParentString = pString[:-2]
252	>	fullParentString = pString[:-1]
253		list_of_lists.append([fullString,fullParentString,str(eventsPerJobRequested),str(jobSkipEventCount)])
254		else:
255		list_of_lists.append([fullString,str(eventsPerJobRequested),str(jobSkipEventCount)])
256	<	common.logger.debug(3,"Job "+str(jobCount+1)+" can run over "+str(eventsPerJobRequested)+" events.")
256	>	msg += "Job %s can run over %s events.\n"%(str(jobCount+1),str(eventsPerJobRequested))
257		jobDestination.append(blockSites[block])
258	<	common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(jobDestination[jobCount]))
258	>	msg+= "Job %s Destination: %s\n"%(str(jobCount+1),str(destinationCMS(jobDestination[jobCount])))
259		jobsOfBlock[block].append(jobCount+1)
260		# increase counter
261		jobCount = jobCount + 1
#	Line 261 | Line 266 | class JobSplitter:
266		jobSkipEventCount = eventsPerJobRequested - (filesEventCount - jobSkipEventCount - self.eventsbyfile[file])
267		# remove all but the last file
268		filesEventCount = self.eventsbyfile[file]
269	+	pString_tmp=''
270		if self.useParent==1:
271	<	for f in parent : pString += '\\\"' + f + '\\\"\,'
272	<	parString = '\\\"' + file + '\\\"\,'
271	>	for f in parent : pString_tmp +=  f + ','
272	>	pString =  pString_tmp
273	>	parString =  file + ','
274		pass # END if
275		pass # END while (iterate over files in the block)
276		pass # END while (iterate over blocks in the dataset)
277	+	common.logger.debug(msg)
278		self.ncjobs = self.total_number_of_jobs = jobCount
279		if (eventsRemaining > 0 and jobCount < totalNumberOfJobs ):
280	<	common.logger.message("Could not run on all requested events because some blocks not hosted at allowed sites.")
281	<	common.logger.message(str(jobCount)+" job(s) can run on "+str(totalEventCount)+" events.\n")
280	>	common.logger.info("Could not run on all requested events because some blocks not hosted at allowed sites.")
281	>	common.logger.info(str(jobCount)+" job(s) can run on "+str(totalEventCount)+" events.\n")
282
283		# skip check on  block with no sites  DD
284	<	if noBboundary == 0 : self.checkBlockNoSite(blocks,jobsOfBlock)
284	>	if noBboundary == 0 : self.checkBlockNoSite(blocks,jobsOfBlock,blockSites)
285
286		# prepare dict output
287		dictOut = {}
288	+	dictOut['params']= ['InputFiles','MaxEvents','SkipEvents']
289	+	if self.useParent: dictOut['params']= ['InputFiles','ParentFiles','MaxEvents','SkipEvents']
290		dictOut['args'] = list_of_lists
291		dictOut['jobDestination'] = jobDestination
292		dictOut['njobs']=self.total_number_of_jobs
#	Line 285 | Line 295 | class JobSplitter:
295
296		# keep trace of block with no sites to print a warning at the end
297
298	<	def checkBlockNoSite(self,blocks,jobsOfBlock):
298	>	def checkBlockNoSite(self,blocks,jobsOfBlock,blockSites):
299		# screen output
300		screenOutput = "List of jobs and available destination sites:\n\n"
301		noSiteBlock = []
302		bloskNoSite = []
303	+	allBlock = []
304
305		blockCounter = 0
306		for block in blocks:
307		if block in jobsOfBlock.keys() :
308		blockCounter += 1
309	+	allBlock.append( blockCounter )
310	+	sites=self.blackWhiteListParser.checkWhiteList(self.blackWhiteListParser.checkBlackList(blockSites[block],[block]),[block])
311		screenOutput += "Block %5i: jobs %20s: sites: %s\n" % (blockCounter,spanRanges(jobsOfBlock[block]),
312	<	','.join(self.blackWhiteListParser.checkWhiteList(self.blackWhiteListParser.checkBlackList(blockSites[block],[block]),[block])))
313	<	if len(self.blackWhiteListParser.checkWhiteList(self.blackWhiteListParser.checkBlackList(blockSites[block],[block]),[block])) == 0:
312	>	', '.join(destinationCMS(sites)))
313	>	if len(sites) == 0:
314		noSiteBlock.append( spanRanges(jobsOfBlock[block]) )
315		bloskNoSite.append( blockCounter )
316
317	<	common.logger.message(screenOutput)
317	>	common.logger.info(screenOutput)
318		if len(noSiteBlock) > 0 and len(bloskNoSite) > 0:
319		msg = 'WARNING: No sites are hosting any part of data for block:\n                '
320		virgola = ""
#	Line 316 | Line 329 | class JobSplitter:
329		for range_jobs in noSiteBlock:
330		msg += str(range_jobs) + virgola
331		msg += '\n               will not be submitted and this block of data can not be analyzed!\n'
332	<	if self.cfg_params.has_key('EDG.se_white_list'):
333	<	msg += 'WARNING: SE White List: '+self.cfg_params['EDG.se_white_list']+'\n'
332	>	if self.cfg_params.has_key('GRID.se_white_list'):
333	>	msg += 'WARNING: SE White List: '+self.cfg_params['GRID.se_white_list']+'\n'
334		msg += '(Hint: By whitelisting you force the job to run at this particular site(s).\n'
335		msg += 'Please check if the dataset is available at this site!)\n'
336	<	if self.cfg_params.has_key('EDG.ce_white_list'):
337	<	msg += 'WARNING: CE White List: '+self.cfg_params['EDG.ce_white_list']+'\n'
336	>	if self.cfg_params.has_key('GRID.ce_white_list'):
337	>	msg += 'WARNING: CE White List: '+self.cfg_params['GRID.ce_white_list']+'\n'
338		msg += '(Hint: By whitelisting you force the job to run at this particular site(s).\n'
339		msg += 'Please check if the dataset is available at this site!)\n'
340
341	<	common.logger.message(msg)
341	>	common.logger.info(msg)
342	>
343	>	if bloskNoSite == allBlock:
344	>	raise CrabException('No jobs created')
345
346		return
347
#	Line 354 | Line 370 | class JobSplitter:
370		thefiles = Fileset(name='FilesToSplit')
371		fileList = pubdata.getListFiles()
372		for f in fileList:
357	–	# print f
373		block = f['Block']['Name']
359	–	#  if not blocks.has_key(block):
360	–	#      blocks[block] = reader.listFileBlockLocation(block)
374		try:
375		f['Block']['StorageElementList'].extend(blockSites[block])
376		except:
#	Line 386 | Line 399 | class JobSplitter:
399		set = Set(runList)
400		list_of_lists = []
401		jobDestination = []
389	–
402		count = 0
403	<	for i in list(set):
403	>	for jobGroup in  jobfactory():
404		if count <  self.theNumberOfJobs:
405	<	res = self.getJobInfo(jobfactory())
405	>	res = self.getJobInfo(jobGroup)
406		parString = ''
407		for file in res['lfns']:
408	<	parString += '\\\"' + file + '\\\"\,'
409	<	fullString = parString[:-2]
408	>	parString += file + ','
409	>	fullString = parString[:-1]
410		list_of_lists.append([fullString,str(-1),str(0)])
411		#need to check single file location
412		jobDestination.append(res['locations'])
413		count +=1
402	–	#print jobDestination
414		# prepare dict output
415		dictOut = {}
416	+	dictOut['params']= ['InputFiles','MaxEvents','SkipEvents']
417		dictOut['args'] = list_of_lists
418		dictOut['jobDestination'] = jobDestination
419		dictOut['njobs']=count
#	Line 419 | Line 431 | class JobSplitter:
431		for loc in file['locations']:
432		if tmp_check < 1 :
433		locations.append(loc)
434	<	tmp_check = tmp_check + 1
434	>	tmp_check = tmp_check + 1
435		### qui va messo il check per la locations
436		res['lfns'] = lfns
437		res['locations'] = locations
#	Line 430 | Line 442 | class JobSplitter:
442		"""
443		Perform job splitting based on number of event per job
444		"""
445	<	common.logger.debug(5,'Splitting per events')
445	>	common.logger.debug('Splitting per events')
446		self.checkUserSettings()
447		jobDestination=[]
448	<	if (self.selectNumberOfJobs == 0):
449	<	msg = 'Must specify  number_of_jobs.'
448	>	if ( (self.selectTotalNumberEvents + self.selectEventsPerJob + self.selectNumberOfJobs) != 2 ):
449	>	msg = 'Must define exactly two of total_number_of_events, events_per_job, or number_of_jobs.'
450		raise CrabException(msg)
451
452		managedGenerators =self.args['managedGenerators']
#	Line 442 | Line 454 | class JobSplitter:
454		firstRun = self.cfg_params.get('CMSSW.first_run',None)
455
456		if (self.selectEventsPerJob):
457	<	common.logger.message('Required '+str(self.eventsPerJob)+' events per job ')
457	>	common.logger.info('Required '+str(self.eventsPerJob)+' events per job ')
458		if (self.selectNumberOfJobs):
459	<	common.logger.message('Required '+str(self.theNumberOfJobs)+' jobs in total ')
459	>	common.logger.info('Required '+str(self.theNumberOfJobs)+' jobs in total ')
460		if (self.selectTotalNumberEvents):
461	<	common.logger.message('Required '+str(self.total_number_of_events)+' events in total ')
461	>	common.logger.info('Required '+str(self.total_number_of_events)+' events in total ')
462
463		if (self.total_number_of_events < 0):
464		msg='Cannot split jobs per Events with "-1" as total number of events'
#	Line 463 | Line 475 | class JobSplitter:
475		self.total_number_of_jobs = self.theNumberOfJobs
476		self.eventsPerJob = int(self.total_number_of_events/self.total_number_of_jobs)
477
478	<	common.logger.debug(5,'N jobs  '+str(self.total_number_of_jobs))
478	>	common.logger.debug('N jobs  '+str(self.total_number_of_jobs))
479
480		# is there any remainder?
481		check = int(self.total_number_of_events) - (int(self.total_number_of_jobs)*self.eventsPerJob)
482
483	<	common.logger.debug(5,'Check  '+str(check))
483	>	common.logger.debug('Check  '+str(check))
484
485	<	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')
485	>	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')
486		if check > 0:
487	<	common.logger.message('Warning: asked '+str(self.total_number_of_events)+' but can do only '+str(int(self.total_number_of_jobs)*self.eventsPerJob))
487	>	common.logger.info('Warning: asked '+str(self.total_number_of_events)+' but can do only '+str(int(self.total_number_of_jobs)*self.eventsPerJob))
488
489		# argument is seed number.$i
490		self.list_of_args = []
#	Line 489 | Line 501 | class JobSplitter:
501		args.append('1')
502		else:
503		args.append(str(i*self.eventsPerJob))
504	+	args.append(str(self.eventsPerJob))
505		self.list_of_args.append(args)
506		# prepare dict output
507	+
508		dictOut = {}
509	+	dictOut['params'] = ['MaxEvents']
510	+	if (firstRun):
511	+	dictOut['params'] = ['FirstRun','MaxEvents']
512	+	if ( generator in managedGenerators ) : dictOut['params'] = ['FirstRun', 'FirstEvent', 'MaxEvents']
513	+	else:
514	+	if (generator in managedGenerators) : dictOut['params'] = ['FirstEvent', 'MaxEvents']
515		dictOut['args'] = self.list_of_args
516		dictOut['jobDestination'] = jobDestination
517		dictOut['njobs']=self.total_number_of_jobs
#	Line 508 | Line 528 | class JobSplitter:
528		msg = 'must specify  number_of_jobs.'
529		raise crabexception(msg)
530		jobDestination = []
531	<	common.logger.debug(5,'Splitting per job')
532	<	common.logger.message('Required '+str(self.theNumberOfJobs)+' jobs in total ')
531	>	common.logger.debug('Splitting per job')
532	>	common.logger.info('Required '+str(self.theNumberOfJobs)+' jobs in total ')
533
534		self.total_number_of_jobs = self.theNumberOfJobs
535
536	<	common.logger.debug(5,'N jobs  '+str(self.total_number_of_jobs))
536	>	common.logger.debug('N jobs  '+str(self.total_number_of_jobs))
537
538	<	common.logger.message(str(self.total_number_of_jobs)+' jobs can be created')
538	>	common.logger.info(str(self.total_number_of_jobs)+' jobs can be created')
539
540		# argument is seed number.$i
541	<	self.list_of_args = []
541	>	#self.list_of_args = []
542		for i in range(self.total_number_of_jobs):
543		jobDestination.append([""])
544	<	self.list_of_args.append([str(i)])
544	>	#   self.list_of_args.append([str(i)])
545
546		# prepare dict output
547		dictOut = {}
548	<	dictOut['args'] = self.list_of_args
548	>	dictOut['args'] = [] # self.list_of_args
549		dictOut['jobDestination'] = jobDestination
550		dictOut['njobs']=self.total_number_of_jobs
551		return dictOut

Diff Legend

–	Removed lines
+	Added lines
<	Changed lines
>	Changed lines