Context Navigation

tbroadcast.py @ 518

Last change on this file since 518 was 316, checked in by garonne, 18 years ago
fixed a bug
Property svn:executable set to ``*
File size: 24.8 KB

Rev	Line
[232]	1	#----------------------------------#
	2	# -- Author: V.Garonne
[242]	3	# -- Mail: garonne@lal.in2p3.fr
[232]	4	# -- Date: 08/25/2006
	5	# -- Name: tbroadcast
[242]	6	# -- Description: main class
[232]	7	#----------------------------------#
	8
	9	import os
	10	import sys
	11	import time
	12	import string
	13	import random
[243]	14	import os.path
[232]	15	import commands
[239]	16	import traceback
[241]	17	from threading import BoundedSemaphore
[232]	18
	19	from threadpool import WorkRequest
	20	from threadpool import ThreadPool
	21	from threadpool import NoResultsPending
	22	from threadpool import NoWorkersAvailable
	23	from threadpool import makeRequests
	24	from executer import exeCommand
	25
	26	class Scheduler:
	27
[245]	28	def __init__(self, num_workers=20, file=None, ignore_cycles=False, local=False, output=None, error=None, silent = False, perf=False):
[241]	29	self.pool = ThreadPool(num_workers=num_workers)
	30	self.current_package = self.get_current_package()
[243]	31	self.current_project = {'name': None, 'path': None, 'version': None}
[244]	32	self.packages = {}
[241]	33	self.counter = 0
	34	self.semaphore = BoundedSemaphore(1)
[243]	35	self.local = local
	36	self.ignore_cycles = ignore_cycles
	37	self.output = output
[244]	38	self.error = error
	39	self.silent = silent
[245]	40	self.perf = perf
[248]	41	if self.perf is not False:
	42	f = open (self.perf, 'w+')
[245]	43	f.close()
[243]	44	if output is not None:
	45	if not os.path.exists (output):
	46	print "path",output,"no exits"
	47	sys.exit(-1)
	48	if not os.path.isdir(output):
	49	print "path",output,"no a valid directory"
	50	sys.exit(-1)
	51
	52	self.get_current_project()
	53	self.instanciate_packages (file)
	54	if self.local: self.get_local_graph()
[238]	55	self.check_cycles()
[244]	56
	57	# status, output = commands.getstatusoutput("cmt broadcast -local 'echo <package>'")
	58	# lignes = string.split(output, '\n')
	59	# i = 1
	60	# for package in lignes:
	61	# if package!='' and package[0] != '#':
	62	# print i , package
	63	# i =i +1
	64	# if not self.packages.has_key(package):
	65	# print package
	66	# print len(self.packages)
	67	# sys.exit(-1)
[243]	68
	69	def get_current_project(self):
	70	cmd = 'cmt show projects \| grep current'
	71	status, output = commands.getstatusoutput (cmd)
	72	if status != 0:
	73	print output
	74	sys.exit(-1)
	75	lines = string.split(output, '\n')
	76	for line in lines:
	77	if line!='' and line [0] != '#':
	78	item = string.split (line, ' ')
	79	self.current_project ['name'] = item[0]
	80	self.current_project ['version'] = item[1]
	81	self.current_project ['path'] = item[3][:-1]
	82	version = self.current_project ['path'][len(self.current_project ['path'])-len(self.current_project ['version'] ):]
	83	if self.current_project ['version'] == version:
	84	self.current_project ['path'] = os.path.normpath(self.current_project ['path'][:-len(self.current_project ['version'] )])
[316]	85	return
[243]	86	#print self.current_project
[241]	87
	88	def get_counter(self):
	89	self.semaphore.acquire ()
	90	self.counter = self.counter + 1
	91	value = self.counter
	92	self.semaphore.release()
	93	return value
	94
[238]	95	def check_cycles (self):
[243]	96	cmd = 'cmt -private show cycles'
[238]	97	cycle_found = False
	98	status, output = commands.getstatusoutput (cmd)
	99	if status != 0:
	100	print output
	101	sys.exit(-1)
	102	lines = string.split(output, '\n')
[243]	103	cycles = list ()
[238]	104	for line in lines:
[243]	105	if line!='' and line [0] != '#':
	106	cycles.append (string.split(line))
	107	cercles =list()
	108	for cycle in cycles:
	109	cycleInProject = True
	110	for package in cycle:
	111	if not self.packages.has_key(package):
	112	cycleInProject = False
	113	if cycleInProject:
	114	cercles.append(cycle)
	115	if len(cercles):
	116	if not self.ignore_cycles:
	117	print "# Error: cycles found, not possible to execute broadcast with threads. Please correct the following cycles:"
	118	for cycle in cercles:
	119	loop = ""
	120	for package in cycle:
	121	loop = loop + package + ' -> '
	122	print loop + '...'
	123	sys.exit(-1)
	124	else:
	125	print "# Warning: There are cycles and you have selected the automatic suppress cycles mode"
	126	for cycle in cercles:
	127	loop = ""
	128	for package in cycle:
	129	loop = loop + package + ' -> '
	130	if cycle[0] in self.packages[cycle[len(cycle)-1]]['uses']:
	131	print '## In cycle: '+loop + '..., we suppress the dependency '+ cycle[len(cycle)-1]+'->'+cycle[0]
	132	self.packages[cycle[len(cycle)-1]]['uses'].remove(cycle[0])
	133	# sys.exit(-1)
[242]	134
	135	def format_uses (self, content):
	136	# format variables
	137	lignes = string.split(content, '\n')
	138	lines = list()
	139	for ligne in lignes:
[243]	140	if ligne [0] == '#' and ligne[:5] != "#CMT>" and ligne[:10] != "# Required" and ligne not in ['# Selection :','#']:
[242]	141	lines.append(ligne)
	142	lines.reverse()
	143	return lines
	144
	145	def format_paths (self, content):
	146	# format variables
	147	lignes = string.split(content, '\n')
	148	lines = list()
	149	for ligne in lignes:
	150	if ligne[:4] == "use ":
	151	lines.append(ligne)
	152	return lines
	153
	154	def get_paths (self, content):
	155	lines = self.format_paths(content)
[232]	156	for line in lines:
[237]	157	result = string.split (line[4:len(line)], ' ')
	158	if self.packages.has_key(result[0]):
	159	if len(result)==4:
	160	name, version, offset, path = string.split (line[4:len(line)], " ")
[241]	161	#print name, version, offset, path
	162	#print path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt'
	163	if path == '(no_auto_imports)':
	164	path = offset
	165	offset = ''
[237]	166	if os.path.exists(path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt'):
	167	full_path = path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt'
	168	elif os.path.exists(path[1:-1] + '/' + offset + '/' +name + '/cmt'):
	169	full_path = path[1:-1] + '/' + offset + '/' +name + '/cmt'
	170	else:
	171	print '# error path not found for', name
	172	sys.exit(-1)
	173	elif len(result)==5:
	174	name, version, offset, path, importation = string.split (line[4:len(line)], " ")
	175	if os.path.exists(path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt'):
	176	full_path = path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt'
	177	elif os.path.exists(path[1:-1] + '/' + offset + '/' +name + '/cmt'):
	178	full_path = path[1:-1] + '/' + offset + '/' +name + '/cmt'
	179	else:
	180	print '# error path not found for', name
	181	sys.exit(-1)
	182	elif len(result)==3:
	183	name, version, path = string.split (line[4:len(line)], " ")
	184	if os.path.exists(path[1:-1] + '/' +name + '/' + version + '/cmt'):
	185	full_path = path[1:-1] + '/' +name + '/' + version + '/cmt'
	186	elif os.path.exists(path[1:-1] + '/' +name + + '/cmt'):
	187	full_path = path[1:-1] + '/' +name + + '/cmt'
	188	else:
	189	print '# error path not found for', name
	190	sys.exit(-1)
	191	else:
	192	print "error:",line
	193	print str(result)
	194	sys.exit(-1)
[243]	195	self.packages[result[0]]['path'] = os.path.normpath(full_path)
	196	commonprefix = os.path.commonprefix([self.packages[result[0]]['path'], self.current_project ['path']])
	197	if os.path.normpath(commonprefix) == self.current_project ['path']:
	198	#print result[0], ' belong to project', self.current_project ['name']
	199	self.packages[result[0]]['current_project'] = True
	200
[242]	201	def get_uses(self, content):
	202	# initiates variables
	203	lignes = self.format_uses(content)
[243]	204	if not len(lignes): return
	205	self.packages [self.current_package] = {'version': '*', 'client': list(),
	206	'uses': list(), 'status': 'waiting',
	207	'current_project': True, 'path': os.getcwd()}
[242]	208	previous_client = self.current_package
	209	previous_level = 0
	210	level_stack = [{'name':previous_client,'level':previous_level},]
	211	ligne = lignes.pop()
	212	while len(lignes)!=0:
	213	current_level = string.find(ligne, 'use')
	214	while current_level > previous_level:
	215	name = string.split (ligne)[2]
	216	version = string.split (ligne)[3]
	217	if not self.packages.has_key (name):
[243]	218	self.packages [name] = {'version': version, 'uses': list(),
	219	'client': list(), 'status': 'waiting',
	220	'current_project': False, 'path': None}
	221	if name not in self.packages[previous_client]['uses']:# and name != previous_client:
[242]	222	self.packages[previous_client]['uses'].append (name)
	223	level_stack.append({'name':previous_client,'level':previous_level})
	224	previous_client = name
	225	previous_level = current_level
	226	if len(lignes):
	227	ligne = lignes.pop()
	228	#print ligne
	229	current_level = string.find(ligne, 'use')
[243]	230
[242]	231	#self.packages [previous_client]['status'] ='queued'
	232	# restore the level
[243]	233	if len(lignes):
	234	if len(level_stack):
	235	item = level_stack.pop()
	236	while item['level'] >= current_level and len(level_stack):
	237	item = level_stack.pop()
	238	previous_client = item['name']
	239	previous_level = item['level']
[242]	240	#print previous_client, '-->',string.split (ligne)[2]
[232]	241
[242]	242	def instanciate_packages(self, file=None):
	243	# We create the schedule of the work units
[243]	244	print '# First, we initialize the DAG by parsing "cmt show uses"'
[242]	245	if file is None:
	246	cmd = 'cmt show uses'
	247	else:
	248	cmd = 'cat ' + file
	249	status, output = commands.getstatusoutput (cmd)
	250	if status != 0:
	251	print output
	252	sys.exit(-1)
	253	self.get_uses(output)
	254	self.get_paths(output)
	255	#self.check_execution (package=self.current_package)
	256	#self.simulate_execution()
	257
[243]	258	def get_local_graph(self):
	259	To_remove = list()
	260	for key in self.packages:
	261	if self.packages[key]['current_project']== False:
	262	for selected in self.packages:
	263	if key in self.packages[selected]['uses']:
	264	self.packages[selected]['uses'].remove(key)
	265	To_remove.append (key)
	266	for item in To_remove:
[244]	267	del self.packages[item]
[243]	268
[242]	269	def simulate_execution(self):
	270	ok = True
[243]	271	indice = 1
[242]	272	while ok:
	273	runnable = list()
	274	for key in self.packages:
	275	if self.packages[key]['status']!='done':
	276	if len(self.packages[key]['uses']) == 0:
[243]	277	runnable.append(key)
[242]	278	if len(runnable):
	279	print '\n#--------------------------------------------------------------'
	280	print "# Execute parallel actions within packages " + str(runnable)
	281	for selected in runnable:
	282	print '#--------------------------------------------------------------'
	283	print '# ('+str(indice)+'/'+str(len(self.packages))+') Now trying [] in '+ self.packages[selected]['path']
	284	print '#--------------------------------------------------------------'
	285	self.packages[selected]['status']='done'
	286	indice = indice + 1
	287	for key in self.packages:
	288	if selected in self.packages[key]['uses']:
	289	self.packages[key]['uses'].remove(selected)
	290	#print 'remove', selected, 'from',key
	291	if len(runnable)==0:
	292	ok = False
	293
	294	def check_execution(self, package, path=list(), cycles=list()):
	295	#print package,'-->',self.packages[package]['uses']
	296	#print path
	297	if package in path:
	298	if path[path.index(package):] not in cycles:
	299	print 'Cycles:',path[path.index(package):], package
	300	cycles = cycles + path[path.index(package):]
	301	sys.exit(-1)
	302	path.append(package)
	303	for item in self.packages[package]['uses']:
	304	self.check_execution(package=item, path=path, cycles=cycles)
	305	path.pop()
	306
[232]	307	def get_current_package(self):
	308	cmd = 'cmt show macro package'
	309	status, output = commands.getstatusoutput (cmd)
	310	if status != 0:
	311	print output
	312	sys.exit(-1)
	313	lines = string.split(output, '\n')
	314	for line in lines:
	315	if line [0] != '#':
	316	start = string.find(line,"'")
	317	end = string.find(line[start+1:len(line)],"'")
	318	return line [start+1:start+end+1]
	319
	320	def get_work_area_path (self, name):
	321	return self.packages [name]['path']
	322
	323	def get_package_path (self, name):
[237]	324	#return os.getcwd ()
[232]	325	cmd = 'cmt -use='+name+' run pwd'
	326	status, output = commands.getstatusoutput (cmd)
	327	if status != 0:
	328	print output
	329	sys.exit(-1)
	330	lines = string.split(output, '\n')
	331	for line in lines:
[237]	332	if line [0] != '#' and line[:5] != "#CMT>":
	333	print line
[232]	334	return line
	335
	336	def print_dependencies(self):
	337	print '# ------------------------'
	338	print '# package --> dependencies'
	339	print '# ------------------------'
	340	for key in self.packages.keys():
[242]	341	print key, '-->', self.packages[key] ['uses'],',', self.packages[key] ['status']
[232]	342
[241]	343	def print_status(self, status):
	344	print '# ------------------------'
	345	print '# package --> dependencies'
	346	print '# ------------------------'
	347	i = 1
	348	for key in self.packages.keys():
	349	if self.packages[key] ['status'] == status:
[242]	350	print i , key, '-->', self.packages[key] ['uses'],',', self.packages[key] ['status']
[241]	351	i = i + 1
	352
[232]	353	def is_work_unit_waiting (self, name):
	354	return self.packages[name] ['status'] == 'waiting'
	355
	356	def set_work_unit_status (self, name, status):
	357	self.packages[name] ['status'] = status
	358
	359	def get_dependencies (self, name):
[242]	360	return self.packages[name] ['uses']
[232]	361
	362	def get_next_work_units (self):
	363	result = list ()
	364	for key in self.packages.keys():
	365	if len(self.get_dependencies (key)) == 0 and self.is_work_unit_waiting(key) :
	366	result.append(key)
	367	return result
	368
[238]	369	def is_work_units (self):
	370	result = list ()
	371	for key in self.packages.keys():
	372	if self.is_work_unit_waiting(key) :
	373	return True
	374	return False
	375
[232]	376	def suppress_work_unit (self, name):
[239]	377	#print '# remove', name, 'from schedule'
[232]	378	for key in self.packages.keys():
[242]	379	if name in self.packages[key]['uses']:
	380	self.packages[key]['uses'].remove(name)
[241]	381
[232]	382	def add_work_unit (self, name, cmd):
	383	if self.is_work_unit_waiting (name):
	384	# we create requests
	385	arg = {'cmd': cmd , 'package':name}
	386	req = WorkRequest(self.do_execute, [arg] , None, callback=self.result_callback, exc_callback=self.handle_exception)
	387	# then we put the work request in the queue...
	388	self.set_work_unit_status (name, 'queued')
	389	self.pool.putRequest(req)
	390	#print "# Work request #%s added on %s." % (req.requestID, str(arg['package']))
	391
	392	def execute (self, command):
[238]	393	#self.print_dependencies ()
[232]	394	packages = self.get_next_work_units()
	395	if len(packages) !=0:
[240]	396	print '\n#--------------------------------------------------------------'
	397	print '# Execute parallel actions within packages', packages
[238]	398	for package in packages:
	399	self.add_work_unit (package, command)
[232]	400
	401	def execute_all(self,command):
[241]	402	#self.print_dependencies ()
[232]	403	self.execute (command)
	404	self.wait()
[241]	405	#self.print_dependencies ()
	406	#self.print_status (status='waiting')
	407	#while self.is_work_units():
[242]	408	#self.wait()
[232]	409
	410	def wait (self):
	411	self.pool.wait()
	412
	413	# this will be called each time a result is available
	414	def result_callback(self, request, result):
	415	#print "**Result: %s from request #%s" % (str(result), request.requestID)
[239]	416	#print "# Result: %s from request #%s" % (result['package'], request.requestID)
[245]	417	#if result['package'] == 'CodeCheck':
	418	# sys.exit(-1)
	419	self.execute (result['cmd'])
[232]	420
	421	# the work the threads will have to do
	422	def do_execute(self, arg):
[240]	423	path = self.get_work_area_path (arg['package'])
[245]	424	if path == None or not os.path.exists(path):
[240]	425	raise RuntimeError('Path to package '+ arg['package'] +' not found')
	426	self.set_work_unit_status (arg['package'], 'running')
[245]	427	#cmd = "cmt -use="+ arg['package'] + " run '"+ arg['cmd'] + "'"
	428	#os.chdir(path)
	429	#arg['cmd'] = "cd "+ path +";"+ arg['cmd']
[260]	430	header = '#--------------------------------------------------------------\n'
	431	header = header + '# ('+str(self.get_counter())+'/'+str(len(self.packages))+') Now trying ['+ arg['cmd']+'] in ' + path + '\n'
	432	header = header + '#--------------------------------------------------------------\n'
	433	print header
[261]	434	project_path = self.current_project['path']+'/'+self.current_project['version']+'/'
	435	log_name = string.replace(path, project_path, '')
	436	log_name = string.replace(log_name, '/cmt', '')
	437	log_name = string.replace(log_name, '/', '_')
	438	log_name = log_name+'.loglog'
	439	arg['log'] = log_name
[245]	440	cmd = "cd "+ path +";"+ arg['cmd']
[244]	441	#status, output= commands.getstatusoutput(cmd)
	442	# init output file
[243]	443	if self.output is not None:
[261]	444	f = open (self.output+'/'+ log_name, 'w+')
[260]	445	f.write (header)
[244]	446	f.close()
	447	if self.error is not None:
[261]	448	f = open (self.error+'/error'+log_name, 'w+')
[245]	449	f.close()
[261]	450	self.packages[arg['package']] ['startTime'] = time.time ()
[244]	451	status, output, error, pythonError = exeCommand(sCmd=cmd, oLineCallback=self.redirectOutput, arg=arg)#,iTimeout = 3600)
[245]	452	self.packages[arg['package']] ['endTime'] = time.time ()
	453	if self.perf:
	454	self.semaphore.acquire ()
[248]	455	f = open (self.perf, 'a')
[245]	456	f.write (arg['package']+" "+str(self.packages[arg['package']] ['startTime'])+" "+str(self.packages[arg['package']] ['endTime'] )+'\n')
	457	f.close()
	458	self.semaphore.release()
[240]	459	self.suppress_work_unit (arg['package'])
	460	self.set_work_unit_status (arg['package'], 'done')
	461	# status, output= commands.getstatusoutput(cmd)
	462	#if status != 0:
	463	# raise RuntimeError(output)
	464	return {'output':output, 'cmd': arg['cmd'], 'package':arg['package']}
[244]	465
	466	def redirectOutput(self, index, buffer, arg):
	467	"""Filter function to redirect the std output and error of the job
	468	executable for real-time debugging
	469	"""
	470	if self.output is not None:
	471	if index==0:
[261]	472	f = open (self.output+'/'+arg['log'], 'a')
[244]	473	f.write (buffer+'\n')
	474	f.close()
	475	elif index==1:
	476	if self.error is not None:
[261]	477	f = open (self.error+'/error'+arg['log'], 'a')
[244]	478	else:
[261]	479	f = open (self.output+'/'+arg['log'], 'a')
[244]	480	f.write (buffer+'\n')
	481	f.close()
	482	if not self.silent:
	483	print buffer
[239]	484
[232]	485	# this will be called when an exception occurs within a thread
	486	def handle_exception(self, request, exc_info):
[239]	487	#traceback.print_stack()
[240]	488	print '#--------------------------------------------------------------'
	489	#print "# Exception occured in request #%s: %s" %(request.requestID, exc_info[1])
	490	print "# Exception occured: %s" %(exc_info[1])
[245]	491	print exc_info
[240]	492	print '#--------------------------------------------------------------'
[245]	493	#sys.exit(-1)
[244]	494
[306]	495
	496	def generate_make (self, file, command):
	497	makefile = open (file, 'w+')
[315]	498	makefile.write ('MAKE=make\n')
[312]	499	#MFLAGS= -j10
[310]	500	self.counter = len(self.packages)
[306]	501	self.recursive_make (self.current_package, command, makefile, len(self.packages))
	502	makefile.close ()
	503
	504	def recursive_make (self, package, command, makefile, indice,actions=list()):
	505	lines = self.generate_action_make (package, command, indice)
	506	makefile.write (lines)
[307]	507	#print lines
[306]	508	for pkg in self.packages[package] ['uses']:
	509	if pkg not in actions:
	510	actions.append(pkg)
	511	indice = indice - 1
[310]	512	self.counter = self.counter - 1
[306]	513	self.recursive_make(pkg, command,makefile, indice, actions)
	514
	515	def generate_action_make (self, package, command, indice):
	516	lines = package + ' :: '
	517	# add dependencies
	518	for pkg in self.packages[package] ['uses']:
	519	lines = lines + ' ' + pkg
	520
	521	# add the action itself
[310]	522	newcommand = string.replace (command, '<package>', package)
	523	if command =='':
[312]	524	newcommand='$(MAKE)'
[306]	525	lines = lines + '\n'
	526	lines = lines + '\t@echo "#--------------------------------------------------------------"\n'
[310]	527	lines = lines + '\t@echo "# ('+str(self.counter)+'/'+str(len(self.packages))+') Now trying ['+newcommand+'] in '+ self.packages[package]['path']+'"\n'
[306]	528	lines = lines + '\t@echo "#--------------------------------------------------------------"\n'
[314]	529	lines = lines + 'ifdef LOCATION\n'
[315]	530	lines = lines + '\t@echo "#--------------------------------------------------------------"> $(LOCATION)/'+ package +'.loglog\n'
	531	lines = lines + '\t@echo "# ('+str(self.counter)+'/'+str(len(self.packages))+') Now trying ['+newcommand+'] in '+ self.packages[package]['path']+'">> $(LOCATION)/'+ package +'.loglog\n'
	532	lines = lines + '\t@echo "#--------------------------------------------------------------">> $(LOCATION)/'+ package +'.loglog\n'
[311]	533	lines = lines + '\t+@cd ' + self.packages[package]['path']
[315]	534	lines = lines + ' && ' + newcommand + ' >> $(LOCATION)/'+ package +'.loglog 2>&1\n'
[314]	535	lines = lines + 'else\n'
	536	lines = lines + '\t+@cd ' + self.packages[package]['path']
	537	lines = lines + ' && ' + newcommand + '\n'
	538	lines = lines + 'endif\n\n'
[306]	539	return lines
	540
[241]	541	#--------- EoF --------#

Note: See TracBrowser for help on using the repository browser.

Context Navigation

source: tbroadcast/v2.0.3/python/tbroadcast.py @ 518

Download in other formats: