Index: tbroadcast/HEAD/Changelog
===================================================================
--- tbroadcast/HEAD/Changelog	(revision 511)
+++ tbroadcast/HEAD/Changelog	(revision 517)
@@ -1,11 +1,28 @@
+2009-08-11 Igor Kachaev <Igor.Katchaev@cern.ch>
+	* python/threadpool.py update to version 1.2.5 (2008-11-19)
+	* python/tbroadcast.py implement package sort by use count;
+			       protect common structures with semaphore
+	* python/__init__.py   remove from the package
+	* scripts/tbroadcast   better help; option '-sort' to try sorting
+
+2009-06-25 Igor Kachaev <Igor.Katchaev@cern.ch>
+
+        * scripts/tbroadcast    Check correctness of options supplied;
+                                do '-help' if empty command line
+        * python/tbroadcast.py  add wait() after communicate() to prevent loop at the
+                                end of job; disable -no_keep_going internally as it blocks;
+                                use local copy of getstatusoutput() - else blocks at nightly
+        * cmt/requirements      remove python dir from the path
+
 2009-05-28 Vincent Garonne <vincent.garonne@cern.ch>
 
         * scripts/tbroadcast    Added the python version checking (>= 2.5)
         * python/threadpool.py  Upgraded version to 1.1
-        * python/tbroadcast.py: Replaced the execute method by communicate from the Popen py25 module to avoid deadlocks 
-                                due to any of the other OS pipe buffers filling up and blocking the child process
+        * python/tbroadcast.py: Replaced the execute method by communicate from the Popen py25
+                                module to avoid deadlocks due to any of the other OS pipe buffers
+                                filling up and blocking the child process
         * python/executer.py    Removed from the package 
 
-2007-03-22 Vincent Garonne <garonne@lal.in2p3.fr> 1
+2007-03-22 Vincent Garonne <garonne@lal.in2p3.fr>
 
 	* scripts/tbroadcast, python/tbroadcast.py: Added the -no_keep_going option
Index: tbroadcast/HEAD/cmt/requirements
===================================================================
--- tbroadcast/HEAD/cmt/requirements	(revision 511)
+++ tbroadcast/HEAD/cmt/requirements	(revision 517)
@@ -3,5 +3,5 @@
 author Vincent Garonne <garonne@lal.in2p3.fr>
 
-path_append PATH  "${tbroadcast_root}/python"
+# path_append PATH  "${tbroadcast_root}/python"
 
 path_append PATH  "${tbroadcast_root}/scripts"
Index: tbroadcast/HEAD/python/executer.py
===================================================================
--- tbroadcast/HEAD/python/executer.py	(revision 511)
+++ 	(revision )
@@ -1,228 +1,0 @@
-# -*- coding: iso-8859-1 -*-
-
-import os
-import sys
-import string
-import popen2
-import stat
-import re
-import time
-import commands
-import select
-
-from os.path import join
-
-############################################################################
-## Error and success return function
-def S_ERROR( sMessage = '' ):
-  return { 'Status': -1, 'OK' : 0, 'Message' : sMessage  }
-    
-def S_OK( sValue = None, sPname = 'Value' ):
-  dResult = { 'Status': 'OK', 'OK' : 1 }
-  if sValue is not None:
-    dResult[ sPname ] = sValue
-  return dResult
-  
-############################################################################  
-class SubprocessExecuter:
-
-    def __init__( self, iTimeout = False ):
-        self.changeTimeout( iTimeout )
-        self.iBufferLimit = 5242880 # 5MB limit for data
-
-    def changeTimeout( self, iTimeout ):
-        self.iTimeout = iTimeout
-        if self.iTimeout == 0:
-            self.iTimeout = False
-        
-    def __readFromPipe( self, oPipe, iBaseLength = 0 ):
-        sData = ""
-        iMaxSliceLength = 8192
-        iLastSliceLength = 8192
-        
-        while iLastSliceLength == iMaxSliceLength:
-            sReadBuffer = os.read( oPipe, iMaxSliceLength )
-            iLastSliceLength = len( sReadBuffer )
-            sData += sReadBuffer
-            if len( sData ) + iBaseLength > self.iBufferLimit:
-                dRetVal = S_ERROR( "Reached maximum allowed length (%d bytes) for called function return value" % self.iBufferLimit )
-                dRetVal[ 'ReadData' ] = sData
-                return dRetVal
-            
-        return S_OK( sData )
-                    
-    def __executePythonFunction( self, oFunc, stArgs, oWritePipe ):
-        try:
-            os.write( oWritePipe, "%s\n" % str( S_OK( oFunc( *stArgs ) ) ) )
-        except Exception, v:
-            os.write( oWritePipe, "%s\n" % str( S_ERROR( str( v ) ) ) )
-        try:
-            os.close( oWritePipe )
-        finally:
-            os._exit(0)
-    
-    def __selectFD( self, lR, iTimeout = False ):
-        if self.iTimeout and not iTimeout:
-            iTimeout = self.iTimeout
-        if not iTimeout: 
-            return select.select( lR , [], [] )[0]
-        else:
-            return select.select( lR , [], [], iTimeout )[0]
-    
-    def pythonCall( self, oFunction, stArgs ):
-        oReadPipe, oWritePipe = os.pipe()
-        iPid = os.fork()
-        if iPid == 0:
-            os.close( oReadPipe )
-            self.__executePythonFunction( oFunction, stArgs, oWritePipe )
-            os.close( oWritePipe )
-        else:
-            os.close( oWritePipe )
-            lReadable = self.__selectFD( [ oReadPipe ] )
-            if len( lReadable ) == 0:
-                os.close( oReadPipe )
-                os.kill( iPid, 9 )
-                os.waitpid( iPid, 0 )
-                return S_ERROR( "%d seconds timeout for '%s' call" % ( self.iTimeout, oFunction.__name__ ) )
-            elif lReadable[0] == oReadPipe:
-                dData = self.__readFromPipe( oReadPipe )
-                os.close( oReadPipe )
-                os.waitpid( iPid, 0 )
-                if dData[ 'OK' ]:
-                    return eval( dData[ 'Value' ] )
-                return dData
-            
-    def __generateSystemCommandError( self, sMessage ):
-        retVal = S_ERROR( sMessage )
-        retVal[ 'stdout' ] = self.lBuffers[0][0]
-        retVal[ 'stderr' ] = self.lBuffers[1][0]
-        return retVal
-        
-    def __readFromFile( self, oFile, iBaseLength, bAll ):
-        try:
-            if bAll:
-                sData = "".join( oFile.readlines() )
-            else:
-                sData = oFile.readline()
-        except Exception, v:
-            pass 
-        if sData == "":
-            #self.checkAlive()
-            self.bAlive = False
-        if len( sData ) + iBaseLength > self.iBufferLimit:
-            dRetVal = S_ERROR( "Reached maximum allowed length (%d bytes) for called function return value" % self.iBufferLimit )
-            dRetVal[ 'ReadData' ] = sData
-            return dRetVal
-            
-        return S_OK( sData )
-
-    def __readFromSystemCommandOutput( self, oFile, iDataIndex, bAll = False ):
-        retVal = self.__readFromFile( oFile, len( self.lBuffers[ iDataIndex ][0] ), bAll )
-        if retVal[ 'OK' ]:
-            self.lBuffers[ iDataIndex ][0] += retVal[ 'Value' ]
-            if not self.oCallback == None:
-                while self.__callLineCallback( iDataIndex ):
-                    pass
-            return S_OK()
-        else:
-            self.lBuffers[ iDataIndex ][0] += retVal[ 'ReadData' ]
-            os.kill( self.oChild.pid, 9 )
-            self.oChild.wait()
-            return self.__generateSystemCommandError( "Exceeded maximum buffer size ( %d bytes ) timeout for '%s' call" % ( self.iBufferLimit, self.sCmd ) )
-
-    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 ] ]
-        iInitialTime = time.time()
-        iExitStatus = self.oChild.poll()
-
-        while iExitStatus == -1:
-            retVal = self.__readFromCommand()
-            if not retVal[ 'OK' ]:
-                return retVal
-            if self.iTimeout and time.time() - iInitialTime > self.iTimeout:
-                os.kill( self.oChild.pid, 9 )
-                self.oChild.wait()
-                self.__readFromCommand( True )
-                self.oChild.fromchild.close()
-                self.oChild.childerr.close()
-                return self.__generateSystemCommandError( "Timeout (%d seconds) for '%s' call" % ( self.iTimeout, sCmd ) )
-            iExitStatus = self.oChild.poll()
-  
-        self.__readFromCommand(True )
-
-        self.oChild.fromchild.close()
-        self.oChild.childerr.close() 
-        return S_OK( ( iExitStatus / 256, self.lBuffers[0][0], self.lBuffers[1][0] ) )
-
-    def __readFromCommand( self, bLast = False ):
-        if bLast:
-            retVal = self.__readFromSystemCommandOutput( self.oChild.fromchild, 0, True )
-            if not retVal[ 'OK' ]:
-                return retVal
-            retVal = self.__readFromSystemCommandOutput( self.oChild.childerr, 1, True )
-            if not retVal[ 'OK' ]:
-                return retVal
-        else:
-            lReadable = self.__selectFD( [ self.oChild.fromchild, self.oChild.childerr ], 1 )
-            if self.oChild.fromchild in lReadable:
-                retVal = self.__readFromSystemCommandOutput( self.oChild.fromchild, 0 )
-                if not retVal[ 'OK' ]:
-                    return retVal
-            if self.oChild.childerr in lReadable:
-                retVal = self.__readFromSystemCommandOutput( self.oChild.childerr, 1 )
-                if not retVal[ 'OK' ]:
-                    return retVal
-        return S_OK()
-
-    
-    def __callLineCallback( self, iIndex ):
-        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 ], arg=self.arg )
-            self.lBuffers[ iIndex ][1] += iNextLine + 1 
-            return True
-        return False
-
-############################################################################    
-def redirectOutput(index, buffer):
-    """Filter function to redirect the std output and error of the job
-       executable for real-time debugging 
-    """
-    print buffer
-
-
-############################################################################      
-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, arg=arg)
-    if retVal[ 'OK' ]:
-        return retVal[ 'Value' ][0], retVal[ 'Value' ][1], retVal[ 'Value' ][2], 0
-    else:
-        if re.search("Timeout",retVal['Message']):
-          return 1, retVal['stdout'], retVal['Message'], 2
-        else:  
-          return 1, retVal['stdout'], retVal['stderr'], 1
-    
-############################################################################      
-def getstatusoutput(cmd):
-    """Return (status, stdout) of executing cmd in a shell.
-    
-    A trailing line separator is removed from the output string. The exit status of the command is encoded in the format specified for wait(), when the exit status is zero (termination without errors), 0 is returned.
-    """
-    import os
-    p = os.popen(cmd, 'r')
-    out = p.read()
-    sts = p.close()
-    if sts is None: sts = 0
-    if out.endswith(os.linesep):
-        out = out[:out.rindex(os.linesep)]
-    return sts, out
-
-#################################################################################
-###################################### EoF ######################################
-#################################################################################
Index: tbroadcast/HEAD/python/tbroadcast.py
===================================================================
--- tbroadcast/HEAD/python/tbroadcast.py	(revision 511)
+++ tbroadcast/HEAD/python/tbroadcast.py	(revision 517)
@@ -2,8 +2,10 @@
 # -- Author: V.Garonne
 # -- Mail: garonne@lal.in2p3.fr
-# -- Date: 08/25/2006 
+# -- Date: 08/25/2006
 # -- Name: tbroadcast
 # -- Description: main class
 #----------------------------------#
+
+# 21-Jul-2009 compile most used packages first; protect critical sections
 
 import os
@@ -11,7 +13,6 @@
 import time
 import string
-import random
 import os.path
-import commands
+# import commands
 import traceback
 import exceptions
@@ -20,7 +21,4 @@
 from threadpool  import WorkRequest
 from threadpool  import ThreadPool
-from threadpool  import NoResultsPending
-from threadpool  import NoWorkersAvailable
-from threadpool  import  makeRequests
 
 from  subprocess import Popen
@@ -28,7 +26,8 @@
 class Scheduler:
 
-    def __init__(self, num_workers=20, file=None, ignore_cycles=False, local=False, output=None, error=None, silent = False, perf=False, keep_going=True):
-        self.pool            = ThreadPool(num_workers=num_workers)
-        self.current_package = self.get_current_package()
+    def __init__(self, num_workers=20, file=None, ignore_cycles=False, local=False, sort=False,
+                 output=None, error=None, silent = False, perf=False, keep_going=True):
+        self.pool            = ThreadPool(num_workers=num_workers, poll_timeout=3)
+        self.num_workers     = num_workers
         self.current_project = {'name': None, 'path': None, 'version': None}
         self.packages        = {}
@@ -36,4 +35,5 @@
         self.semaphore       = BoundedSemaphore(1)
         self.local           = local
+        self.sort            = sort
         self.ignore_cycles   = ignore_cycles
         self.output          = output
@@ -42,4 +42,6 @@
         self.perf            = perf
         self.keep_going      = keep_going
+        if self.sort:
+            print "Compile packages sorted according to use count"
         if self.perf is not False:
             f = open (self.perf, 'w+')
@@ -47,36 +49,28 @@
         if output is not None:
             if not os.path.exists (output):
-                print "path",output,"no exits"        
+                print "path",output,"does not exists"
                 sys.exit(-1)
             if not os.path.isdir(output):
-                print "path",output,"no a valid directory"        
-                sys.exit(-1)            
-                
+                print "path",output,"is not a valid directory"
+                sys.exit(-1)
+
+        # init cmt stuff
         self.get_current_project()
+        self.current_package = self.get_current_package()
         self.instanciate_packages (file)
         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)
+        self.get_use_count()
 
     def get_current_project(self):
         cmd = 'cmt show projects | grep current'
-        status, output = commands.getstatusoutput (cmd)
+        status, output = getstatusoutput (cmd)
+        #status, output = commands.getstatusoutput (cmd)
         if status != 0:
             print output
-            sys.exit(-1)    
+            sys.exit(-1)
         lines = string.split(output, '\n')
-        for line in lines:            
-            if line!='' and line [0] != '#':                            
+        for line in lines:
+            if line!='' and line [0] != '#':
                 item  = string.split (line, ' ')
                 self.current_project ['name']    = item[0]
@@ -87,32 +81,32 @@
                     self.current_project ['path'] =  os.path.normpath(self.current_project ['path'][:-len(self.current_project ['version'] )])
                 return
-                #print self.current_project     
-        
+
     def get_counter(self):
-        self.semaphore.acquire ()        
+        self.semaphore.acquire ()
         self.counter = self.counter + 1
-        value = self.counter 
+        value = self.counter
         self.semaphore.release()
         return value
-        
+
     def check_cycles (self):
         cmd = 'cmt -private show cycles'
         cycle_found = False
-        status, output = commands.getstatusoutput (cmd)
+        status, output = getstatusoutput (cmd)
+        #status, output = commands.getstatusoutput (cmd)
         if status != 0:
             print output
-            sys.exit(-1)    
+            sys.exit(-1)
         lines = string.split(output, '\n')
-        cycles = list ()
-        for line in lines:            
-            if line!='' and line [0] != '#':                    
-               cycles.append (string.split(line)) 
-        cercles =list()        
+        cycles = list()
+        for line in lines:
+            if line!='' and line [0] != '#':
+               cycles.append (string.split(line))
+        cercles =list()
         for cycle in cycles:
             cycleInProject = True
-            for package in cycle:            
+            for package in cycle:
                 if not self.packages.has_key(package):
-                    cycleInProject = False        
-            if cycleInProject: 
+                    cycleInProject = False
+            if cycleInProject:
               cercles.append(cycle)
         if len(cercles):
@@ -120,15 +114,15 @@
                 print "# Error: cycles found, not possible to execute broadcast with threads. Please correct the following cycles:"
                 for cycle in cercles:
-                    loop = ""                    
+                    loop = ""
                     for package in cycle:
                         loop = loop + package + ' -> '
                     print loop + '...'
-                sys.exit(-1)       
+                sys.exit(-1)
             else:
                 print "# Warning: There are cycles and you have selected the automatic suppress cycles mode"
                 for cycle in cercles:
-                    loop = ""                    
+                    loop = ""
                     for package in cycle:
-                        loop = loop + package + ' -> '                     
+                        loop = loop + package + ' -> '
                     if cycle[0] in self.packages[cycle[len(cycle)-1]]['uses']:
                         print '## In cycle: '+loop + '..., we suppress the dependency '+ cycle[len(cycle)-1]+'->'+cycle[0]
@@ -151,5 +145,5 @@
         lines   = list()
         for ligne in lignes:
-            if ligne[:4] == "use ":              
+            if ligne[:4] == "use ":
                lines.append(ligne)
         return lines
@@ -169,23 +163,23 @@
                         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'):    
+                        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)   
+                            sys.exit(-1)
                     elif len(result)==5:
-                        name, version, offset, path, importation = string.split (line[4:len(line)], " ")                                        
+                        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'):    
+                        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)                                                                                                   
+                            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'):    
+                        elif os.path.exists(path[1:-1] + '/' +name + + '/cmt'):
                             full_path = path[1:-1] + '/' +name + + '/cmt'
                         else:
@@ -195,46 +189,45 @@
                         print "error:",line
                         print str(result)
-                        sys.exit(-1) 
+                        sys.exit(-1)
                     self.packages[result[0]]['path'] = os.path.normpath(full_path)
                     commonprefix = os.path.commonprefix([self.packages[result[0]]['path'], self.current_project ['path']])
-                    if os.path.normpath(commonprefix) == self.current_project ['path']:                    
+                    if os.path.normpath(commonprefix) == self.current_project ['path']:
                         #print result[0], ' belong to project', self.current_project ['name']
                         self.packages[result[0]]['current_project'] = True
 
-    def get_uses(self, content):        
+    def get_uses(self, content):
         # initiates variables
         lignes = self.format_uses(content)
         if not len(lignes): return
-        self.packages [self.current_package] = {'version': '*', 'client': list(),
-                                                'uses': list(), 'status': 'waiting', 
+        self.packages [self.current_package] = {'version': '*', 'use_count': 0,
+                                                'uses': list(), 'status': 'waiting',
                                                 'current_project': True, '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:   
+        ligne = lignes.pop()
+        while len(lignes)!=0:
             current_level = string.find(ligne, 'use')
-            while current_level > previous_level:               
+            while current_level > previous_level:
                 name    = string.split (ligne)[2]
-                version = string.split (ligne)[3]                              
+                version = string.split (ligne)[3]
                 if not self.packages.has_key (name):
-                  self.packages [name] = {'version': version, 'uses': list(), 
-                                          'client': list(), 'status': 'waiting', 
+                  self.packages [name] = {'version': version, 'use_count': 0,
+                                          'uses': list(), 'status': 'waiting',
                                           'current_project': False, 'path': None}
                 if name not in self.packages[previous_client]['uses']:# and name != previous_client:
-                   self.packages[previous_client]['uses'].append (name)                
+                   self.packages[previous_client]['uses'].append (name)
                 level_stack.append({'name':previous_client,'level':previous_level})
-                previous_client = name  
-                previous_level = current_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
             if len(lignes):
-                if len(level_stack):                        
-                    item = level_stack.pop()                
+                if len(level_stack):
+                    item = level_stack.pop()
                     while item['level'] >= current_level and len(level_stack):
                              item = level_stack.pop()
@@ -248,14 +241,22 @@
         if file is None:
             cmd  = 'cmt show uses'
-        else:    
-            cmd = 'cat ' + file        
-        status, output = commands.getstatusoutput (cmd)
+        else:
+            cmd = 'cat ' + file
+        status, output = getstatusoutput (cmd)
+        #status, output = commands.getstatusoutput (cmd)
         if status != 0:
             print output
             sys.exit(-1)
-        self.get_uses(output)    
+        self.get_uses(output)
         self.get_paths(output)
-        #self.check_execution (package=self.current_package)
         #self.simulate_execution()
+
+    def get_use_count(self):
+        for key in self.packages:
+            count = 0
+            for parent in self.packages:
+                if key in self.packages[parent]['uses']: count += 1
+            self.packages[key]['use_count'] = count
+            # print "Package",key,"use_count",count
 
     def get_local_graph(self):
@@ -266,52 +267,35 @@
                     if key in self.packages[selected]['uses']:
                        self.packages[selected]['uses'].remove(key)
-                To_remove.append (key) 
+                To_remove.append (key)
         for item in To_remove:
-             del self.packages[item]
+            del self.packages[item]
 
     def simulate_execution(self):
-       ok = True
-       indice = 1                     
-       while ok:
-           runnable  = 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) 
-           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):    
+        while True:
+            ndone = self.simulate_requests()
+            if ndone == 0: break
+
+    def simulate_requests(self):
+        runnable = self.get_next_work_units()
+        if len(runnable):
+            print '\n#--------------------------------------------------------------'
+            print "# Execute parallel actions within packages - total", len(runnable)
+            print '#--------------------------------------------------------------'
+        for selected in runnable:
+            use_count = self.packages[selected]['use_count']
+            path = self.packages[selected]['path']
+            print '#--------------------------------------------------------------'
+            print '# (%d/%d %d) Now trying [] in %s' % (self.get_counter(), len(self.packages), use_count, path)
+            print '#--------------------------------------------------------------'
+            self.suppress_work_unit(selected)
+        return len(runnable)
+
+    def get_current_package(self):
         cmd            = 'cmt show macro package'
-        status, output = commands.getstatusoutput (cmd)
+        status, output = getstatusoutput (cmd)
+        #status, output = commands.getstatusoutput (cmd)
         if status != 0:
             print output
-            sys.exit(-1)    
+            sys.exit(-1)
         lines = string.split(output, '\n')
         for line in lines:
@@ -321,37 +305,21 @@
                 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 '# ------------------------' 
+        print '# -------------------------------------------'
+        print '# package --> dependencies, status, use_count'
+        print '# -------------------------------------------'
         for key in self.packages.keys():
-            print key, '-->', self.packages[key] ['uses'],',', self.packages[key] ['status']                       
+            print key, '-->', self.packages[key] ['uses'],',', self.packages[key] ['status'],',', self.packages[key] ['use_count']
 
     def print_status(self, status):
-        print '# ------------------------' 
-        print '# package --> dependencies' 
-        print '# ------------------------' 
+        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']                 
+                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'
@@ -360,45 +328,51 @@
         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)
+        # by default returned list is in arbitrary order - may be this is better
+        # if self.sort is set returned list is sorted - most used packages first
+        runnable = list()
+        for key in self.packages:
+            if len(self.packages[key]['uses']) == 0 and self.packages[key]['status'] == 'waiting':
+                use_count = self.packages[key]['use_count']
+                runnable.append((use_count,key))
+        if self.sort:
+            runnable.sort()
+            runnable.reverse()
+        result = [ pair[1] for pair in runnable ]
         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' 
+        #print '# remove', name, 'from schedule'
+        self.semaphore.acquire()
+        self.packages[name]['status']='done'
         for key in self.packages.keys():
             if name in self.packages[key]['uses']:
                 self.packages[key]['uses'].remove(name)
+        self.semaphore.release()
 
     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) 
+            arg = {'cmd': cmd , 'package': name}
+            req = WorkRequest(self.do_execute, [arg] , None, callback=self.result_callback)
+#           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']))
+            # print "# Work request #%s added on %s." % (req.requestID, str(arg['package']))
 
     def execute (self, command):
         #self.print_dependencies ()
+        self.semaphore.acquire()
         packages = self.get_next_work_units()
-        if len(packages) !=0:
-            print '\n#--------------------------------------------------------------'   
-            print '# Execute parallel actions within packages', packages                      
+        if len(packages):
+            print '\n#--------------------------------------------------------------'
+            print '# Execute parallel actions within packages (total',len(packages),')',packages
+            print '\n#--------------------------------------------------------------'
             for package in packages:
                 self.add_work_unit (package, command)
+            sys.stdout.flush()
+        self.semaphore.release()
 
     def execute_all(self,command):
@@ -406,11 +380,12 @@
         self.execute (command)
         self.wait()
+        if self.counter != len(self.packages):
+            print 'tbroadcast: warning: compiled',self.counter,'out of',len(self.packages),'packages'
+        self.pool.dismissWorkers(self.num_workers, do_join=True)
         #self.print_dependencies ()
-        #self.print_status (status='waiting')        
-        #while self.is_work_units(): 
-        #self.wait()            
-        
+        #self.print_status (status='waiting')
+
     def wait (self):
-       self.pool.wait()    
+       self.pool.wait()
 
     # this will be called each time a result is available
@@ -418,21 +393,18 @@
       #print "**Result: %s from request #%s" % (str(result), request.requestID)
       #print "# Result: %s from request #%s" % (result['package'], request.requestID)
-      #if result['package'] == 'CodeCheck':
-      #    sys.exit(-1)
-      self.execute (result['cmd'])    
-
-    # the work the threads will have to do 
+      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'])
+      package = arg['package']
+      path = self.packages[package]['path']
       if path == None or not os.path.exists(path):
-          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)
-      #arg['cmd'] = "cd "+ path +";"+ arg['cmd']
-      header = '#--------------------------------------------------------------\n'
-      header = header + '# ('+str(self.get_counter())+'/'+str(len(self.packages))+') Now trying ['+ arg['cmd']+'] in ' + path + '\n'
+          raise RuntimeError('Path to package '+ package +' not found')
+      self.set_work_unit_status(package, 'running')
+      header =          '#--------------------------------------------------------------\n'
+      header = header + '# ('+str(self.get_counter())+'/'+str(len(self.packages))+') Now trying ['+arg['cmd']+'] in '+path+'\n'
       header = header + '#--------------------------------------------------------------\n'
       print header
+      sys.stdout.flush()
       project_path = self.current_project['path']+'/'+self.current_project['version']+'/'
       log_name     = string.replace(path, project_path, '')
@@ -440,59 +412,64 @@
       log_name     = string.replace(log_name, '/', '_')
       log_name     = log_name+'.loglog'
-      arg['log']   = log_name
-      cmd = "cd "+ path +";"+ arg['cmd'] 
-      #status, output= commands.getstatusoutput(cmd)
+      # arg['log']   = log_name
+      cmd = "cd "+ path +";"+ arg['cmd']
+      # status, output = commands.getstatusoutput(cmd)
       # init output file
 
-      self.packages[arg['package']] ['startTime'] = time.time ()                           
+      self.packages[package] ['startTime'] = time.time()
 
       if self.output is not None:
-           f1 = open (self.output+'/'+ log_name, 'w+')           
+           f1 = open (self.output+'/'+ log_name, 'w+')
            f1.write (header)
+           f1.flush()
+           if self.error is not None:
+               f2 = open (self.error+'/error'+log_name, 'w+')
+               fp = Popen(cmd, shell=True, stdout=f1, stderr=f2)
+               fp.communicate()
+               status = fp.wait()
+               f2.close()
+           else:
+               fp = Popen(cmd, shell=True, stdout=f1, stderr=f1)
+               fp.communicate()
+               status = fp.wait()
            f1.close()
-           f1 = open (self.output+'/'+ log_name, 'a')
-           if self.error is not None:        
-               f2 = open (self.error+'/error'+log_name, 'w+')               
-               Popen(cmd, shell=True, stdout=f1, stderr=f2).communicate()
-               f2.close()  
-           else:
-               Popen(cmd, shell=True, stdout=f1, stderr=f1).communicate()
-           f1.close()               
       else:
-          Popen(cmd, shell=True).communicate()
-      
+           fp = Popen(cmd, shell=True)
+           fp.communicate()
+           status = fp.wait()
+      sys.stdout.flush()
+      sys.stderr.flush()
+
+      # Error is not handled - exit() is forbidden here
       if not self.keep_going and status > 0:
-        sys.exit(status)   
-                     
-      self.packages[arg['package']] ['endTime'] = time.time ()
+          print 'Error',status,'for package',package
+          # sys.exit(status)
+
+      self.packages[package] ['endTime'] = time.time()
       if self.perf:
-          self.semaphore.acquire ()        
+          self.semaphore.acquire()
           f = open (self.perf, 'a')
-          f.write (arg['package']+" "+str(self.packages[arg['package']] ['startTime'])+" "+str(self.packages[arg['package']] ['endTime'] )+'\n')  
+          f.write (package+" "+str(self.packages[package]['startTime'])+" "+str(self.packages[package]['endTime'] )+'\n')
           f.close()
           self.semaphore.release()
-      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)
+      self.suppress_work_unit(package)
       return {'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 '#--------------------------------------------------------------'        
+      #traceback.print_stack()
+      print '#--------------------------------------------------------------'
       #print "# Exception occured in request #%s: %s" %(request.requestID, exc_info[1])
       if exc_info[0]== exceptions.SystemExit:
-        print "Stop execution (No_keep_going option enabled): exit code == %s " %(exc_info[1])  
-        print '#--------------------------------------------------------------'    
+        print "Stop execution (No_keep_going option enabled): exit code == %s " %(exc_info[1])
+        print '#--------------------------------------------------------------'
         sys.exit(exc_info[1])
-      print "# Exception occured: %s" %(exc_info[1])      
+      print "# Exception occured: %s" %(exc_info[1])
       print exc_info
-      print '#--------------------------------------------------------------'    
+      print '#--------------------------------------------------------------'
       #sys.exit(-1)
 
-    
+
     def generate_make (self, file, command):
         makefile = open (file, 'w+')
@@ -502,9 +479,9 @@
         self.recursive_make (self.current_package, command, makefile, len(self.packages))
         makefile.close ()
-        
+
     def recursive_make (self, package, command, makefile, indice,actions=list()):
         lines = self.generate_action_make (package, command, indice)
         makefile.write (lines)
-        #print lines                
+        #print lines
         for pkg in self.packages[package] ['uses']:
             if pkg not in actions:
@@ -512,14 +489,14 @@
                 indice = indice - 1
                 self.counter = self.counter - 1
-                self.recursive_make(pkg, command,makefile, indice, actions)        
-        
+                self.recursive_make(pkg, command,makefile, indice, actions)
+
     def generate_action_make (self, package, command, indice):
-        lines = package + ' :: '        
+        lines = package + ' :: '
         # add dependencies
         for pkg in self.packages[package] ['uses']:
-            lines = lines + ' ' + pkg            
+            lines = lines + ' ' + pkg
 
         # add the action itself
-        newcommand = string.replace (command, '<package>', package)        
+        newcommand = string.replace (command, '<package>', package)
         if command =='':
             newcommand='$(MAKE)'
@@ -529,14 +506,32 @@
         lines = lines +  '\t@echo "#--------------------------------------------------------------"\n'
         lines = lines +  'ifdef LOCATION\n'
-        lines = lines +  '\t@echo "#--------------------------------------------------------------"> $(LOCATION)/'+ package +'.loglog\n'        
+        lines = lines +  '\t@echo "#--------------------------------------------------------------"> $(LOCATION)/'+ package +'.loglog\n'
         lines = lines +  '\t@echo "# ('+str(self.counter)+'/'+str(len(self.packages))+') Now trying ['+newcommand+'] in '+ self.packages[package]['path']+'">> $(LOCATION)/'+ package +'.loglog\n'
         lines = lines +  '\t@echo "#--------------------------------------------------------------">> $(LOCATION)/'+ package +'.loglog\n'
-        lines = lines + '\t+@cd ' + self.packages[package]['path']            
+        lines = lines + '\t+@cd ' + self.packages[package]['path']
         lines = lines + ' && ' + newcommand + ' >> $(LOCATION)/'+ package +'.loglog 2>&1\n'
         lines = lines + 'else\n'
-        lines = lines + '\t+@cd ' + self.packages[package]['path']            
+        lines = lines + '\t+@cd ' + self.packages[package]['path']
         lines = lines + ' && ' + newcommand + '\n'
-        lines = lines + 'endif\n\n'                
+        lines = lines + 'endif\n\n'
         return lines
-        
+
+
+# own copy of getstatusoutput
+def getstatusoutput(cmd):
+    """Return (status, stdout) of executing cmd in a shell.
+
+    A trailing line separator is removed from the output string.
+    The exit status of the command is encoded in the format specified for wait(),
+    when the exit status is zero (termination without errors), 0 is returned.
+    """
+    import os
+    p = os.popen(cmd, 'r')
+    out = p.read()
+    sts = p.close()
+    if sts is None: sts = 0
+    if out.endswith(os.linesep):
+        out = out[:out.rindex(os.linesep)]
+    return sts, out
+
 #--------- EoF --------#
Index: tbroadcast/HEAD/python/threadpool.py
===================================================================
--- tbroadcast/HEAD/python/threadpool.py	(revision 511)
+++ tbroadcast/HEAD/python/threadpool.py	(revision 517)
@@ -1,54 +1,114 @@
-"""
-
-@author: Christopher Arndt
-@version: 1.1
-
-@see: http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/435883
-
-Easy to use object-oriented thread pool framework.
-
-A thread pool is a class that maintains a pool of worker threads to perform
+# -*- coding: UTF-8 -*-
+"""Easy to use object-oriented thread pool framework.
+
+A thread pool is an object that maintains a pool of worker threads to perform
 time consuming operations in parallel. It assigns jobs to the threads
 by putting them in a work request queue, where they are picked up by the
 next available thread. This then performs the requested operation in the
-background and puts the results in a another queue.
-
-The thread pool class can then collect the results from all threads from
+background and puts the results in another queue.
+
+The thread pool object can then collect the results from all threads from
 this queue as soon as they become available or after all threads have
 finished their work. It's also possible, to define callbacks to handle
 each result as it comes in.
 
-The basic concept and some code was taken from the book "Python in a Nutshell"
-by Alex Martelli, copyright 2003, ISBN 0-596-00188-6, from section 14.5
-"Threaded Program Architecture". I wrapped the main program logic in the
+The basic concept and some code was taken from the book "Python in a Nutshell,
+2nd edition" by Alex Martelli, O'Reilly 2006, ISBN 0-596-10046-9, from section
+14.5 "Threaded Program Architecture". I wrapped the main program logic in the
 ThreadPool class, added the WorkRequest class and the callback system and
-tweaked the code here and there.
-
-Basic usage:
-
->>> main = ThreadPool(poolsize)
->>> requests = makeRequests(some_callable, list_of_args, callback)
->>> [main.putRequests(req) for req in requests]
->>> main.wait()
+tweaked the code here and there. Kudos also to Florent Aide for the exception
+handling mechanism.
+
+Basic usage::
+
+    >>> pool = ThreadPool(poolsize)
+    >>> requests = makeRequests(some_callable, list_of_args, callback)
+    >>> [pool.putRequest(req) for req in requests]
+    >>> pool.wait()
 
 See the end of the module code for a brief, annotated usage example.
+
+Website : http://chrisarndt.de/projects/threadpool/
+
 """
-
-__all__ = ['makeRequests', 'NoResultsPending', 'NoWorkersAvailable',
-  'ThreadPool', 'WorkRequest', 'WorkerThread']
-
-__author__ = 'Christopher Arndt'
-__version__ = '1.1'
-__date__ = '2005-07-19'
-
-import threading, Queue
-
+__docformat__ = "restructuredtext en"
+
+__all__ = [
+    'makeRequests',
+    'NoResultsPending',
+    'NoWorkersAvailable',
+    'ThreadPool',
+    'WorkRequest',
+    'WorkerThread'
+]
+
+__author__ = "Christopher Arndt"
+__version__ = "1.2.5"
+__revision__ = "$Revision: 354 $"
+__date__ = "$Date: 2008-11-19 18:34:46 +0100 (Wed, 19 Nov 2008) $"
+__license__ = 'MIT license'
+
+
+# standard library modules
+import sys
+import threading
+import Queue
+import traceback
+
+
+# exceptions
 class NoResultsPending(Exception):
     """All work requests have been processed."""
     pass
+
 class NoWorkersAvailable(Exception):
     """No worker threads available to process remaining requests."""
     pass
 
+
+# internal module helper functions
+def _handle_thread_exception(request, exc_info):
+    """Default exception handler callback function.
+
+    This just prints the exception info via ``traceback.print_exception``.
+
+    """
+    traceback.print_exception(*exc_info)
+
+
+# utility functions
+def makeRequests(callable_, args_list, callback=None,
+        exc_callback=_handle_thread_exception):
+    """Create several work requests for same callable with different arguments.
+
+    Convenience function for creating several work requests for the same
+    callable where each invocation of the callable receives different values
+    for its arguments.
+
+    ``args_list`` contains the parameters for each invocation of callable.
+    Each item in ``args_list`` should be either a 2-item tuple of the list of
+    positional arguments and a dictionary of keyword arguments or a single,
+    non-tuple argument.
+
+    See docstring for ``WorkRequest`` for info on ``callback`` and
+    ``exc_callback``.
+
+    """
+    requests = []
+    for item in args_list:
+        if isinstance(item, tuple):
+            requests.append(
+                WorkRequest(callable_, item[0], item[1], callback=callback,
+                    exc_callback=exc_callback)
+            )
+        else:
+            requests.append(
+                WorkRequest(callable_, [item], None, callback=callback,
+                    exc_callback=exc_callback)
+            )
+    return requests
+
+
+# classes
 class WorkerThread(threading.Thread):
     """Background thread connected to the requests/results queues.
@@ -56,39 +116,50 @@
     A worker thread sits in the background and picks up work requests from
     one queue and puts the results in another until it is dismissed.
+
     """
 
-    def __init__ (self, requestsQueue, resultsQueue, **kwds):
-        """Set up thread in damonic mode and start it immediatedly.
-
-        requestsQueue and resultQueue are instances of Queue.Queue passed
-        by the ThreadPool class when it creates a new worker thread.
+    def __init__(self, requests_queue, results_queue, poll_timeout=5, **kwds):
+        """Set up thread in daemonic mode and start it immediatedly.
+
+        ``requests_queue`` and ``results_queue`` are instances of
+        ``Queue.Queue`` passed by the ``ThreadPool`` class when it creates a new
+        worker thread.
+
         """
         threading.Thread.__init__(self, **kwds)
         self.setDaemon(1)
-        self.workRequestQueue = requestsQueue
-        self.resultQueue = resultsQueue
+        self._requests_queue = requests_queue
+        self._results_queue = results_queue
+        self._poll_timeout = poll_timeout
         self._dismissed = threading.Event()
         self.start()
 
     def run(self):
-        """Repeatedly process the job queue until told to exit.
-        """
-
-        while not self._dismissed.isSet():
-            # thread blocks here, if queue empty
-            request = self.workRequestQueue.get()
+        """Repeatedly process the job queue until told to exit."""
+        while True:
             if self._dismissed.isSet():
-                # return the work request we just picked up
-                self.workRequestQueue.put(request)
-                break # and exit
-            # XXX catch exceptions here and stick them to request object
-            self.resultQueue.put(
-                (request, request.callable(*request.args, **request.kwds))
-            )
+                # we are dismissed, break out of loop
+                break
+            # get next work request. If we don't get a new request from the
+            # queue after self._poll_timout seconds, we jump to the start of
+            # the while loop again, to give the thread a chance to exit.
+            try:
+                request = self._requests_queue.get(True, self._poll_timeout)
+            except Queue.Empty:
+                continue
+            else:
+                if self._dismissed.isSet():
+                    # we are dismissed, put back request in queue and exit loop
+                    self._requests_queue.put(request)
+                    break
+                try:
+                    result = request.callable(*request.args, **request.kwds)
+                    self._results_queue.put((request, result))
+                except:
+                    request.exception = True
+                    self._results_queue.put((request, sys.exc_info()))
 
     def dismiss(self):
-        """Sets a flag to tell the thread to exit when done with current job.
-        """
-
+        """Sets a flag to tell the thread to exit when done with current job."""
         self._dismissed.set()
 
@@ -97,89 +168,150 @@
     """A request to execute a callable for putting in the request queue later.
 
-    See the module function makeRequests() for the common case
-    where you want to build several work requests for the same callable
-    but different arguments for each call.
+    See the module function ``makeRequests`` for the common case
+    where you want to build several ``WorkRequest`` objects for the same
+    callable but with different arguments for each call.
+
     """
 
-    def __init__ (self, callable, args=None, kwds=None, requestID=None,
-      callback=None):
-        """A work request consists of the a callable to be executed by a
+    def __init__(self, callable_, args=None, kwds=None, requestID=None,
+            callback=None, exc_callback=_handle_thread_exception):
+        """Create a work request for a callable and attach callbacks.
+
+        A work request consists of the a callable to be executed by a
         worker thread, a list of positional arguments, a dictionary
         of keyword arguments.
 
-        A callback function can be specified, that is called when the results
-        of the request are picked up from the result queue. It must accept
-        two arguments, the request object and it's results in that order.
-        If you want to pass additional information to the callback, just stick
-        it on the request object.
-
-        requestID, if given, must be hashable as it is used by the ThreadPool
-        class to store the results of that work request in a dictionary.
-        It defaults to the return value of id(self).
+        A ``callback`` function can be specified, that is called when the
+        results of the request are picked up from the result queue. It must
+        accept two anonymous arguments, the ``WorkRequest`` object and the
+        results of the callable, in that order. If you want to pass additional
+        information to the callback, just stick it on the request object.
+
+        You can also give custom callback for when an exception occurs with
+        the ``exc_callback`` keyword parameter. It should also accept two
+        anonymous arguments, the ``WorkRequest`` and a tuple with the exception
+        details as returned by ``sys.exc_info()``. The default implementation
+        of this callback just prints the exception info via
+        ``traceback.print_exception``. If you want no exception handler
+        callback, just pass in ``None``.
+
+        ``requestID``, if given, must be hashable since it is used by
+        ``ThreadPool`` object to store the results of that work request in a
+        dictionary. It defaults to the return value of ``id(self)``.
+
         """
         if requestID is None:
             self.requestID = id(self)
         else:
-            self.requestID = requestID
+            try:
+                self.requestID = hash(requestID)
+            except TypeError:
+                raise TypeError("requestID must be hashable.")
+        self.exception = False
         self.callback = callback
-        self.callable = callable
+        self.exc_callback = exc_callback
+        self.callable = callable_
         self.args = args or []
         self.kwds = kwds or {}
 
+    def __str__(self):
+        return "<WorkRequest id=%s args=%r kwargs=%r exception=%s>" % \
+            (self.requestID, self.args, self.kwds, self.exception)
 
 class ThreadPool:
     """A thread pool, distributing work requests and collecting results.
 
-    See the module doctring for more information.
+    See the module docstring for more information.
+
     """
 
-    def __init__ (self, num_workers, q_size=0):
+    def __init__(self, num_workers, q_size=0, resq_size=0, poll_timeout=5):
         """Set up the thread pool and start num_workers worker threads.
 
-        num_workers is the number of worker threads to start initialy.
-        If q_size > 0 the size of the work request is limited and the
-        thread pool blocks when queue is full and it tries to put more
-        work requests in it.
+        ``num_workers`` is the number of worker threads to start initially.
+
+        If ``q_size > 0`` the size of the work *request queue* is limited and
+        the thread pool blocks when the queue is full and it tries to put
+        more work requests in it (see ``putRequest`` method), unless you also
+        use a positive ``timeout`` value for ``putRequest``.
+
+        If ``resq_size > 0`` the size of the *results queue* is limited and the
+        worker threads will block when the queue is full and they try to put
+        new results in it.
+
+        .. warning:
+            If you set both ``q_size`` and ``resq_size`` to ``!= 0`` there is
+            the possibilty of a deadlock, when the results queue is not pulled
+            regularly and too many jobs are put in the work requests queue.
+            To prevent this, always set ``timeout > 0`` when calling
+            ``ThreadPool.putRequest()`` and catch ``Queue.Full`` exceptions.
+
         """
-
-        self.requestsQueue = Queue.Queue(q_size)
-        self.resultsQueue = Queue.Queue()
+        self._requests_queue = Queue.Queue(q_size)
+        self._results_queue = Queue.Queue(resq_size)
         self.workers = []
+        self.dismissedWorkers = []
         self.workRequests = {}
-        self.createWorkers(num_workers)
-
-    def createWorkers(self, num_workers):
-        """Add num_workers worker threads to the pool."""
-
+        self.createWorkers(num_workers, poll_timeout)
+
+    def createWorkers(self, num_workers, poll_timeout=5):
+        """Add num_workers worker threads to the pool.
+
+        ``poll_timout`` sets the interval in seconds (int or float) for how
+        ofte threads should check whether they are dismissed, while waiting for
+        requests.
+
+        """
         for i in range(num_workers):
-            self.workers.append(WorkerThread(self.requestsQueue,
-              self.resultsQueue))
-
-    def dismissWorkers(self, num_workers):
-        """Tell num_workers worker threads to to quit when they're done."""
-
+            self.workers.append(WorkerThread(self._requests_queue,
+                self._results_queue, poll_timeout=poll_timeout))
+
+    def dismissWorkers(self, num_workers, do_join=False):
+        """Tell num_workers worker threads to quit after their current task."""
+        dismiss_list = []
         for i in range(min(num_workers, len(self.workers))):
             worker = self.workers.pop()
             worker.dismiss()
-
-    def putRequest(self, request):
-        """Put work request into work queue and save for later."""
-        self.requestsQueue.put(request)
+            dismiss_list.append(worker)
+
+        if do_join:
+            for worker in dismiss_list:
+                worker.join()
+        else:
+            self.dismissedWorkers.extend(dismiss_list)
+
+    def joinAllDismissedWorkers(self):
+        """Perform Thread.join() on all worker threads that have been dismissed.
+        """
+        for worker in self.dismissedWorkers:
+            worker.join()
+        self.dismissedWorkers = []
+
+    def putRequest(self, request, block=True, timeout=0):
+        """Put work request into work queue and save its id for later."""
+        assert isinstance(request, WorkRequest)
+        # don't reuse old work requests
+        assert not getattr(request, 'exception', None)
+        self._requests_queue.put(request, block, timeout)
         self.workRequests[request.requestID] = request
 
     def poll(self, block=False):
         """Process any new results in the queue."""
-        while 1:
+        while True:
+            # still results pending?
+            if not self.workRequests:
+                raise NoResultsPending
+            # are there still workers to process remaining requests?
+            elif block and not self.workers:
+                raise NoWorkersAvailable
             try:
-                # still results pending?
-                if not self.workRequests:
-                    raise NoResultsPending
-                # are there still workers to process remaining requests?
-                elif block and not self.workers:
-                    raise NoWorkersAvailable
                 # get back next results
-                request, result = self.resultsQueue.get(block=block)
-                # and hand them to the callback, if any
-                if request.callback:
+                request, result = self._results_queue.get(block=block)
+                # has an exception occured?
+                if request.exception and request.exc_callback:
+                    request.exc_callback(request, result)
+                # hand results to callback, if any
+                if request.callback and not \
+                       (request.exception and request.exc_callback):
                     request.callback(request, result)
                 del self.workRequests[request.requestID]
@@ -189,5 +321,4 @@
     def wait(self):
         """Wait for results, blocking until all have arrived."""
-
         while 1:
             try:
@@ -196,35 +327,8 @@
                 break
 
-def makeRequests(callable, args_list, callback=None):
-    """Convenience function for building several work requests for the same
-    callable with different arguments for each call.
-
-    args_list contains the parameters for each invocation of callable.
-    Each item in 'argslist' should be either a 2-item tuple of the list of
-    positional arguments and a dictionary of keyword arguments or a single,
-    non-tuple argument.
-
-    callback is called when the results arrive in the result queue.
-    """
-
-    requests = []
-    
-    for item in args_list:
-        
-        if item is None:
-            """no arguments"""
-            requests.append(
-              WorkRequest(callable, None, None, callback=callback))
-            
-        elif item == isinstance(item, tuple):
-            """arguments and keywords"""
-            requests.append(
-              WorkRequest(callable, item[0], item[1], callback=callback))
-        else:
-            """only keywords"""
-            requests.append(
-              WorkRequest(callable, [item], None, callback=callback))
-    return requests
-
+
+################
+# USAGE EXAMPLE
+################
 
 if __name__ == '__main__':
@@ -235,16 +339,44 @@
     def do_something(data):
         time.sleep(random.randint(1,5))
-        return round(random.random() * data, 5)
+        result = round(random.random() * data, 5)
+        # just to show off, we throw an exception once in a while
+        if result > 5:
+            raise RuntimeError("Something extraordinary happened!")
+        return result
 
     # this will be called each time a result is available
     def print_result(request, result):
-        print "Result: %s from request #%s" % (result, request.requestID)
+        print "**** Result from request #%s: %r" % (request.requestID, result)
+
+    # this will be called when an exception occurs within a thread
+    # this example exception handler does little more than the default handler
+    def handle_exception(request, exc_info):
+        if not isinstance(exc_info, tuple):
+            # Something is seriously wrong...
+            print request
+            print exc_info
+            raise SystemExit
+        print "**** Exception occured in request #%s: %s" % \
+          (request.requestID, exc_info)
 
     # assemble the arguments for each job to a list...
     data = [random.randint(1,10) for i in range(20)]
     # ... and build a WorkRequest object for each item in data
-    requests = makeRequests(do_something, data, print_result)
-
-    # we create a pool of 10 worker threads
+    requests = makeRequests(do_something, data, print_result, handle_exception)
+    # to use the default exception handler, uncomment next line and comment out
+    # the preceding one.
+    #requests = makeRequests(do_something, data, print_result)
+
+    # or the other form of args_lists accepted by makeRequests: ((,), {})
+    data = [((random.randint(1,10),), {}) for i in range(20)]
+    requests.extend(
+        makeRequests(do_something, data, print_result, handle_exception)
+        #makeRequests(do_something, data, print_result)
+        # to use the default exception handler, uncomment next line and comment
+        # out the preceding one.
+    )
+
+    # we create a pool of 3 worker threads
+    print "Creating thread pool with 3 worker threads."
     main = ThreadPool(3)
 
@@ -257,18 +389,30 @@
 
     # ...and wait for the results to arrive in the result queue
-    # wait() will return when results for all work requests have arrived
+    # by using ThreadPool.wait(). This would block until results for
+    # all work requests have arrived:
     # main.wait()
 
-    # alternatively poll for results while doing something else:
+    # instead we can poll for results while doing something else:
     i = 0
-    while 1:
+    while True:
         try:
+            time.sleep(0.5)
             main.poll()
-            print "Main thread working..."
-            time.sleep(0.5)
+            print "Main thread working...",
+            print "(active worker threads: %i)" % (threading.activeCount()-1, )
             if i == 10:
-                print "Adding 3 more worker threads..."
+                print "**** Adding 3 more worker threads..."
                 main.createWorkers(3)
+            if i == 20:
+                print "**** Dismissing 2 worker threads..."
+                main.dismissWorkers(2)
             i += 1
-        except (KeyboardInterrupt, NoResultsPending):
+        except KeyboardInterrupt:
+            print "**** Interrupted!"
             break
+        except NoResultsPending:
+            print "**** No pending results."
+            break
+    if main.dismissedWorkers:
+        print "Joining all dismissed worker threads..."
+        main.joinAllDismissedWorkers()
Index: tbroadcast/HEAD/scripts/tbroadcast
===================================================================
--- tbroadcast/HEAD/scripts/tbroadcast	(revision 511)
+++ tbroadcast/HEAD/scripts/tbroadcast	(revision 517)
@@ -3,33 +3,42 @@
 # -- Author: V. Garonne
 # -- Mail: garonne@lal.in2p3.fr
-# -- Date: 08/25/2006 
+# -- Date: 08/25/2006
 # -- Name: tbroadcast
 # -- Description: main program
 #----------------------------------#
 
-import os
 import sys
-import string
+import time
 
 def usage():
-    print 'Usage : > tbroadcast [global options] [<command>]'
-    print '# command :'
-    print '#   <command>: command to execute'
-    print '# global options :'
-    print '#   -f=<file>             : Input file'
-    print '#   -help                 : Print help'
-    print '#   -local                : Reach packages only within the current project'
-    print '#   -global               : Reach packages in all CMTPATH/CMTPROJECTPATH items'    
-    print '#   -ignore_cycles        : Suppress automatically the cycles'
-    print '#   -make=<file>          : Generate a recursive Make, [see: http://www.tip.net.au/~millerp/rmch/recu-make-cons-harm.html]'
-    print '#   -nb=<num_worker>      : Change the total number of threads[default is 20]'
-    print '#   -no_keep_going        : Exit after the first exit code > 1 found and return it in the shell'
-    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 '#   -perf=<file>          : Store for each package the time for executing the command in the <file> file'
-    print '#   -print                : Print dependencies for each package' 
-    print '#   -version              : version of tbroadcast'        
-    print '#   -silent               : Disable print'     
-    print '#   -test                 : Simulate execution'
+    print """
+Usage: tbroadcast [global options] [<command>]
+#
+# <command> is executed in <package>/cmt
+#
+# global options :
+#   -help                 : Print help
+#   -local                : Reach packages only within the current project
+#                         : if not specified reach packages in all CMTPATH/CMTPROJECTPATH items
+#   -ignore[_cycles]      : Suppress automatically the cycles
+#   -sort                 : Compile packages in order of use count, most significant first
+#   -nb=<num_worker>      : Change the total number of threads[default is 20]
+#   -output=<location>    : Output directory to store output files with the form <package>_output.log
+#   -error=<location>     : Output directory to store error output with the form <package>_error.log
+#   -perf=<file>          : Store for each package the time for executing the command in the <file> file
+#   -make=<file>          : Generate a recursive Make, [see: http://www.tip.net.au/~millerp/rmch/recu-make-cons-harm.html]
+#   -print                : Print dependencies for each package and exit
+#   -version              : Print version of tbroadcast and exit
+#   -test                 : Simulate execution and exit
+#   -                     : is accepted and does nothing
+#
+# Example:
+#   tbroadcast -local -ignore -nb=4 'make -j6'
+"""
+
+# Unused options
+#   -f=<file>             : Input file (option for debug only)
+#   -no_keep_going        : Exit after the first exit code > 1 found and return it in the shell
+#   -silent               : Disable print
 
 if __name__ == '__main__':
@@ -38,6 +47,7 @@
     cur_version = sys.version_info
 
-    if not (cur_version[0] > req_version[0] or (cur_version[0] == req_version[0]  and cur_version[1] >= req_version[1])): 
-        raise "must use python 2.5 or greater"
+    if not (cur_version[0] > req_version[0] or (cur_version[0] == req_version[0] and cur_version[1] >= req_version[1])):
+        print "tbroadcast: must use python 2.5 or greater"
+        sys.exit(-1)
 
     from tbroadcast import Scheduler
@@ -46,12 +56,12 @@
     num_worker    = 20
     command       = ''
-    version       = 'v2.0.4'
+    version       = 'v2.0.7'
     test          = False
-    check         = False
     print_graph   = False
     local         = False
     ignore_cycles = False
     silent        = False
-    perf          = False     
+    perf          = False
+    sort          = False
     output        = None
     error         = None
@@ -60,62 +70,73 @@
     makefile      = 'Makefile'
     keep_going    = True
-    
+
     if len(sys.argv) == 1:
-        test = True
-    else:        
-        for arg in sys.argv[1:len(sys.argv)]:            
+        usage()
+        sys.exit(-1)
+    else:
+        for arg in sys.argv[1:len(sys.argv)]:
+#            print "Argument is",arg
              if arg[0]=='-':
-                 option = string.split(arg,'=')[0]
+                 option = arg.split('=')[0]
                  if option == '-version':
                      print version
                      sys.exit(-1)
-                 if option == '-nb':
-                    num_worker = int (string.split(arg,'=')[1]) 
-                 if option == '-f':
-                    file = string.split(arg,'=')[1]
-                 if option == '-perf':
-                    perf = string.split(arg,'=')[1]
-                 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':             
+                 elif option == '-nb':
+                     num_worker = int (arg.split('=')[1])
+                 elif option == '-f':
+                     file = arg.split('=')[1]
+                 elif option == '-perf':
+                     perf = arg.split('=')[1]
+                 elif option == '-output':
+                     output = arg.split('=')[1]
+                 elif option == '-error':
+                     error = arg.split('=')[1]
+                 elif option == '-local':
+                     local = True
+                 elif option == '-sort':
+                     sort = True
+                 elif option[:7] == '-ignore':
                      ignore_cycles = True
-                 if option == '-silent':             
-                     silent = True        
-                 if option == '-no_keep_going':             
+                 elif option == '-silent':
+                     silent = True
+                 elif option == '-no_keep_going':
                      keep_going = False
-                              
-                 if option == '-help':    
-                      usage()
-                      sys.exit(-1)      
-                                  
-                 if option == '-test':
-                         test = True   
-                 elif option == '-check':
-                         check = True   
+                 elif option == '-help':
+                     usage()
+                     sys.exit(-1)
+                 elif option == '-test':
+                     test = True
                  elif option == '-print':
-                         print_graph = True   
-                 elif option == '-make':        
+                     print_graph = True
+                 elif option == '-make':
                      make     = True
-                     makefile = string.split(arg,'=')[1]
+                     makefile = arg.split('=')[1]
+                 elif option == '-':
+                     pass
+                 else:
+                     print 'tbroadcast: bad option "%s", use -help for help' % option
+                     sys.exit(-1)
              else:
-                  command = arg
+                 command = arg
 
-    master = Scheduler (num_workers=num_worker, file=file, ignore_cycles=ignore_cycles, 
-                        local=local, output=output, error=error, silent=silent, perf=perf, 
-                        keep_going=keep_going)
+#   print "End of arguments. Command to execute", command
+
+    if not (command or test or print_graph):
+        print 'tbroadcast: no command specified'
+        sys.exit(-1)
+
+    master = Scheduler (num_workers=num_worker, file=file, ignore_cycles=ignore_cycles,
+                        local=local, output=output, error=error, silent=silent, perf=perf,
+                        keep_going=keep_going, sort=sort)
     if test:
         master.simulate_execution()
-    elif check:
-        master.check_execution (package=master.get_current_package())    
     elif print_graph:
-        master.print_dependencies ()
+        master.print_dependencies()
     elif make:
-        master.generate_make (makefile, command)    
+        master.generate_make (makefile, command)
     else:
+        print 'tbroadcast: start of job at', time.strftime('%d-%b-%Y %T')
         master.execute_all (command)
-    #sys.exit(-1);
+        print 'tbroadcast: end of job at', time.strftime('%d-%b-%Y %T')
+    #sys.exit(-1)
 #--------- EoF --------#
