#----------------------------------# # -- Author: V.Garonne # -- Mail: garonne@lal.in2p3.fr # -- Date: 08/25/2006 # -- Name: tbroadcast # -- Description: main class #----------------------------------# import os import sys import time import string import random import commands import traceback from threading import BoundedSemaphore from threadpool import WorkRequest from threadpool import ThreadPool from threadpool import NoResultsPending from threadpool import NoWorkersAvailable from threadpool import makeRequests from executer import exeCommand class Scheduler: def __init__(self, num_workers=20): self.pool = ThreadPool(num_workers=num_workers) self.current_package = self.get_current_package() self.packages = {} self.counter = 0 self.semaphore = BoundedSemaphore(1) self.check_cycles() self.instanciate_packages () def get_counter(self): self.semaphore.acquire () self.counter = self.counter + 1 value = self.counter self.semaphore.release() return value def check_cycles (self): cmd = 'cmt show cycles' cycle_found = False status, output = commands.getstatusoutput (cmd) if status != 0: print output sys.exit(-1) lines = string.split(output, '\n') for line in lines: if line!='' and line [0] != '#': if not cycle_found: cycle_found = True print "# Error: cycles found, not possible to execute broadcast with threads. See the followings packages:" print line if cycle_found: sys.exit(-1) def format_uses (self, content): # format variables lignes = string.split(content, '\n') lines = list() for ligne in lignes: if ligne [0] == '#' and ligne[:5] != "#CMT>" and ligne not in ['# Selection :','#']: lines.append(ligne) lines.reverse() return lines def format_paths (self, content): # format variables lignes = string.split(content, '\n') lines = list() for ligne in lignes: if ligne[:4] == "use ": lines.append(ligne) return lines def get_paths (self, content): lines = self.format_paths(content) for line in lines: result = string.split (line[4:len(line)], ' ') if self.packages.has_key(result[0]): if len(result)==4: name, version, offset, path = string.split (line[4:len(line)], " ") #print name, version, offset, path #print path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt' if path == '(no_auto_imports)': path = offset offset = '' if os.path.exists(path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt'): full_path = path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt' elif os.path.exists(path[1:-1] + '/' + offset + '/' +name + '/cmt'): full_path = path[1:-1] + '/' + offset + '/' +name + '/cmt' else: print '# error path not found for', name sys.exit(-1) elif len(result)==5: name, version, offset, path, importation = string.split (line[4:len(line)], " ") if os.path.exists(path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt'): full_path = path[1:-1] + '/' + offset + '/' +name + '/' + version + '/cmt' elif os.path.exists(path[1:-1] + '/' + offset + '/' +name + '/cmt'): full_path = path[1:-1] + '/' + offset + '/' +name + '/cmt' else: print '# error path not found for', name sys.exit(-1) elif len(result)==3: name, version, path = string.split (line[4:len(line)], " ") if os.path.exists(path[1:-1] + '/' +name + '/' + version + '/cmt'): full_path = path[1:-1] + '/' +name + '/' + version + '/cmt' elif os.path.exists(path[1:-1] + '/' +name + + '/cmt'): full_path = path[1:-1] + '/' +name + + '/cmt' else: print '# error path not found for', name sys.exit(-1) else: print "error:",line print str(result) sys.exit(-1) self.packages[result[0]]['path'] = full_path def get_uses(self, content): # initiates variables lignes = self.format_uses(content) self.packages [self.current_package] = {'version': '*', 'client': list(),'uses': list(), 'status': 'waiting', 'path': os.getcwd()} previous_client = self.current_package previous_level = 0 level_stack = [{'name':previous_client,'level':previous_level},] ligne = lignes.pop() while len(lignes)!=0: current_level = string.find(ligne, 'use') while current_level > previous_level: name = string.split (ligne)[2] version = string.split (ligne)[3] if not self.packages.has_key (name): self.packages [name] = {'version': version, 'uses': list(), 'client': list(), 'status': 'waiting', 'path': None} if name not in self.packages[previous_client]['uses']: self.packages[previous_client]['uses'].append (name) level_stack.append({'name':previous_client,'level':previous_level}) previous_client = name previous_level = current_level if len(lignes): ligne = lignes.pop() #print ligne current_level = string.find(ligne, 'use') #self.packages [previous_client]['status'] ='queued' # restore the level item = level_stack.pop() while item['level'] >= current_level: item = level_stack.pop() previous_client = item['name'] previous_level = item['level'] #print previous_client, '-->',string.split (ligne)[2] def instanciate_packages(self, file=None): # We create the schedule of the work units print '# First, we initialize the DAG by parsing cmt show uses' if file is None: cmd = 'cmt show uses' else: cmd = 'cat ' + file status, output = commands.getstatusoutput (cmd) if status != 0: print output sys.exit(-1) self.get_uses(output) self.get_paths(output) #self.check_execution (package=self.current_package) #self.simulate_execution() def simulate_execution(self): ok = True indice = 1 while ok: runnable = list() checkable = list() for key in self.packages: if self.packages[key]['status']!='done': if len(self.packages[key]['uses']) == 0: runnable.append(key) if len(runnable): print '\n#--------------------------------------------------------------' print "# Execute parallel actions within packages " + str(runnable) #print 'checkable:',checkable for selected in runnable: print '#--------------------------------------------------------------' print '# ('+str(indice)+'/'+str(len(self.packages))+') Now trying [] in '+ self.packages[selected]['path'] print '#--------------------------------------------------------------' self.packages[selected]['status']='done' indice = indice + 1 for key in self.packages: if selected in self.packages[key]['uses']: self.packages[key]['uses'].remove(selected) #print 'remove', selected, 'from',key if len(runnable)==0: ok = False def check_execution(self, package, path=list(), cycles=list()): #print package,'-->',self.packages[package]['uses'] #print path if package in path: if path[path.index(package):] not in cycles: print 'Cycles:',path[path.index(package):], package cycles = cycles + path[path.index(package):] sys.exit(-1) path.append(package) for item in self.packages[package]['uses']: self.check_execution(package=item, path=path, cycles=cycles) path.pop() def get_current_package(self): cmd = 'cmt show macro package' status, output = commands.getstatusoutput (cmd) if status != 0: print output sys.exit(-1) lines = string.split(output, '\n') for line in lines: if line [0] != '#': start = string.find(line,"'") end = string.find(line[start+1:len(line)],"'") return line [start+1:start+end+1] def get_work_area_path (self, name): return self.packages [name]['path'] def get_package_path (self, name): #return os.getcwd () cmd = 'cmt -use='+name+' run pwd' status, output = commands.getstatusoutput (cmd) if status != 0: print output sys.exit(-1) lines = string.split(output, '\n') for line in lines: if line [0] != '#' and line[:5] != "#CMT>": print line return line def print_dependencies(self): print '# ------------------------' print '# package --> dependencies' print '# ------------------------' for key in self.packages.keys(): print key, '-->', self.packages[key] ['uses'],',', self.packages[key] ['status'] def print_status(self, status): print '# ------------------------' print '# package --> dependencies' print '# ------------------------' i = 1 for key in self.packages.keys(): if self.packages[key] ['status'] == status: print i , key, '-->', self.packages[key] ['uses'],',', self.packages[key] ['status'] i = i + 1 def is_work_unit_waiting (self, name): return self.packages[name] ['status'] == 'waiting' def set_work_unit_status (self, name, status): self.packages[name] ['status'] = status def get_dependencies (self, name): return self.packages[name] ['uses'] def get_next_work_units (self): result = list () for key in self.packages.keys(): if len(self.get_dependencies (key)) == 0 and self.is_work_unit_waiting(key) : result.append(key) return result def is_work_units (self): result = list () for key in self.packages.keys(): if self.is_work_unit_waiting(key) : return True return False def suppress_work_unit (self, name): #print '# remove', name, 'from schedule' for key in self.packages.keys(): if name in self.packages[key]['uses']: self.packages[key]['uses'].remove(name) def add_work_unit (self, name, cmd): if self.is_work_unit_waiting (name): # we create requests arg = {'cmd': cmd , 'package':name} req = WorkRequest(self.do_execute, [arg] , None, callback=self.result_callback, exc_callback=self.handle_exception) # then we put the work request in the queue... self.set_work_unit_status (name, 'queued') self.pool.putRequest(req) #print "# Work request #%s added on %s." % (req.requestID, str(arg['package'])) def execute (self, command): #self.print_dependencies () packages = self.get_next_work_units() if len(packages) !=0: print '\n#--------------------------------------------------------------' print '# Execute parallel actions within packages', packages for package in packages: self.add_work_unit (package, command) def execute_all(self,command): #self.print_dependencies () self.execute (command) self.wait() #self.print_dependencies () #self.print_status (status='waiting') #while self.is_work_units(): #self.wait() def wait (self): self.pool.wait() # this will be called each time a result is available def result_callback(self, request, result): #print "**Result: %s from request #%s" % (str(result), request.requestID) #print "# Result: %s from request #%s" % (result['package'], request.requestID) self.execute (result['cmd']) # the work the threads will have to do def do_execute(self, arg): path = self.get_work_area_path (arg['package']) if path == None: raise RuntimeError('Path to package '+ arg['package'] +' not found') self.set_work_unit_status (arg['package'], 'running') cmd = "cmt -use="+ arg['package'] + " run '"+ arg['cmd'] + "'" os.chdir(path) print '#--------------------------------------------------------------' print '# ('+str(self.get_counter())+'/'+str(len(self.packages))+') Now trying ['+ arg['cmd']+'] in ' + path print '#--------------------------------------------------------------' cmd = arg['cmd'] status, output, error, pythonError = exeCommand(cmd, iTimeout = 10) self.suppress_work_unit (arg['package']) self.set_work_unit_status (arg['package'], 'done') # status, output= commands.getstatusoutput(cmd) #if status != 0: # raise RuntimeError(output) return {'output':output, 'cmd': arg['cmd'], 'package':arg['package']} # this will be called when an exception occurs within a thread def handle_exception(self, request, exc_info): #traceback.print_stack() print '#--------------------------------------------------------------' #print "# Exception occured in request #%s: %s" %(request.requestID, exc_info[1]) print "# Exception occured: %s" %(exc_info[1]) print '#--------------------------------------------------------------' sys.exit(-1) #--------- EoF --------#