Changeset 244 for tbroadcast

Timestamp:

Sep 7, 2006, 10:31:17 AM (18 years ago)

Author:

garonne

Message:

MàJ

Location:

tbroadcast/v2

Files:

• python/executer.py (modified) (3 diffs)
• python/tbroadcast.py (modified) (7 diffs)
• scripts/tbroadcast (modified) (4 diffs)

tbroadcast/v2/python/executer.py

@@ -131 +131 @@
             return self.__generateSystemCommandError( "Exceeded maximum buffer size ( %d bytes ) timeout for '%s' call" % ( self.iBufferLimit, self.sCmd ) )
 
-    def systemCall( self, sCmd, oCallbackFunction = None ):
+    def systemCall( self, sCmd, oCallbackFunction = None, arg=None ):
         self.sCmd = sCmd
         self.oCallback = oCallbackFunction
+        self.arg    = arg
         self.oChild = popen2.Popen3( self.sCmd, True )
         self.lBuffers = [ [ "", 0 ], [ "", 0 ] ]
@@ -182 +183 @@
         iNextLine = self.lBuffers[ iIndex ][0][ self.lBuffers[ iIndex ][1]: ].find( "\n" )
         if iNextLine > -1:
-            self.oCallback( iIndex, self.lBuffers[ iIndex ][0][ self.lBuffers[ iIndex ][1]: self.lBuffers[ iIndex ][1] + iNextLine ] )
+            self.oCallback( iIndex, self.lBuffers[ iIndex ][0][ self.lBuffers[ iIndex ][1]: self.lBuffers[ iIndex ][1] + iNextLine ], arg=self.arg )
             self.lBuffers[ iIndex ][1] += iNextLine + 1 
             return True
@@ -196 +197 @@
 
 ############################################################################      
-def exeCommand( sCmd, iTimeout = 0, oLineCallback = redirectOutput):
+def exeCommand( sCmd, iTimeout = 0, oLineCallback = redirectOutput, arg=None):
     """Return ( status, output, error, pythonError ) of executing cmd in a shell."""
     oSPE = SubprocessExecuter( iTimeout )
-    retVal = oSPE.systemCall( sCmd, oLineCallback )
+    retVal = oSPE.systemCall( sCmd, oLineCallback, arg=arg)
     if retVal[ 'OK' ]:
         return retVal[ 'Value' ][0], retVal[ 'Value' ][1], retVal[ 'Value' ][2], 0

tbroadcast/v2/python/tbroadcast.py

@@ -26 +26 @@
 class Scheduler:
 
-    def __init__(self, num_workers=20, file=None, ignore_cycles=False, local=False, output=None):
+    def __init__(self, num_workers=20, file=None, ignore_cycles=False, local=False, output=None, error=None, silent = False):
         self.pool            = ThreadPool(num_workers=num_workers)
         self.current_package = self.get_current_package()
         self.current_project = {'name': None, 'path': None, 'version': None}
-        self.packages        = {}  
+        self.packages        = {}
         self.counter         = 0
         self.semaphore       = BoundedSemaphore(1)
@@ -36 +36 @@
         self.ignore_cycles   = ignore_cycles
         self.output          = output
+        self.error           = error
+        self.silent          = silent
         if output is not None:
             if not os.path.exists (output):
@@ -48 +50 @@
         if self.local: self.get_local_graph()
         self.check_cycles()
+        
+#        status, output = commands.getstatusoutput("cmt broadcast -local 'echo <package>'")
+#        lignes = string.split(output, '\n')
+#        i = 1
+#        for package in lignes:
+#            if package!='' and package[0] != '#':                            
+#                print i , package
+#                i =i +1
+#                if not self.packages.has_key(package):
+#                                    print package        
+#        print len(self.packages)
+#        sys.exit(-1)
 
     def get_current_project(self):
@@ -246 +260 @@
                 To_remove.append (key) 
         for item in To_remove:
-            self.packages.pop(item)
+             del self.packages[item]
 
     def simulate_execution(self):
@@ -409 +423 @@
       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 = 3600)
+      cmd = arg['cmd'] 
+      #status, output= commands.getstatusoutput(cmd)
+      # init output file
       if self.output is not None:
-         f = open (self.output+'/'+arg['package']+'_output.log', 'w+')
-         f.write (output)
-         f.close()
-         f = open (self.output+'/'+arg['package']+'_error.log', 'w+') 
-         f.write (str(error))
-         f.close() 
+           f = open (self.output+'/'+arg['package']+'_output.log', 'w+')
+           f.close()      
+           if self.error is not None:
+               f = open (self.error+'/'+arg['package']+'_error.log', 'w+')
+               f.close()            
+      status, output, error, pythonError  = exeCommand(sCmd=cmd, oLineCallback=self.redirectOutput, arg=arg)#,iTimeout = 3600)
       self.suppress_work_unit (arg['package'])
       self.set_work_unit_status (arg['package'], 'done')
@@ -424 +439 @@
       #   raise RuntimeError(output)
       return {'output':output, 'cmd': arg['cmd'], 'package':arg['package']}
+
+    def redirectOutput(self, index, buffer, arg):
+        """Filter function to redirect the std output and error of the job
+           executable for real-time debugging 
+        """
+        if self.output is not None:
+           if index==0:   
+               f = open (self.output+'/'+arg['package']+'_output.log', 'a')
+               f.write (buffer+'\n')
+               f.close()
+           elif index==1: 
+               if self.error is not None:
+                   f = open (self.error+'/'+arg['package']+'_error.log', 'a')
+               else:
+                   f = open (self.output+'/'+arg['package']+'_output.log', 'a')                   
+               f.write (buffer+'\n')                   
+               f.close()                               
+        if not self.silent: 
+            print buffer
               
     # this will be called when an exception occurs within a thread
@@ -433 +467 @@
       print '#--------------------------------------------------------------'    
       sys.exit(-1)
+
+
 #--------- EoF --------#

tbroadcast/v2/scripts/tbroadcast

@@ -25 +25 @@
     print '#   -ignore_cycles        : '
     print '#   -nb=<num_worker>      : Total number of threads'
-    print '#   -output=<location>    : Output directory to store output files with the form <package>.log'
+    print '#   -output=<location>    : Output directory to store output files with the form <package>_output.log'
+    print '#   -error=<location>     : Output directory to store error output with the form <package>_error.log'
     print '#   -print                : Print dependency graph' 
     print '#   -test                 : Simulate execution'
@@ -38 +39 @@
     local         = False
     ignore_cycles = False
+    silent        = False
     output        = None
+    error         = None
     file          = None
     for arg in sys.argv[1:len(sys.argv)]:            
@@ -49 +52 @@
              if option == '-output':
                 output = string.split(arg,'=')[1]
+             if option == '-error':
+                error = string.split(arg,'=')[1]
              if option == '-local':
                  local= True
              if option == '-ignore_cycles':             
                  ignore_cycles = True
+             if option == '-silent':             
+                 silent = True
+                 
              if option == '-help':    
                   usage()
@@ -66 +74 @@
               command = arg
 
-    master = Scheduler (num_workers=num_worker, file=file, ignore_cycles=ignore_cycles, local=local, output=output)
+    master = Scheduler (num_workers=num_worker, file=file, ignore_cycles=ignore_cycles, local=local, output=output, error=error, silent=silent)
     if test:
         master.simulate_execution()