# A parallelized "find(1)" using the thread module. # This demonstrates the use of a work queue and worker threads.# It really does do more stats/sec when using multiple threads,# although the improvement is only about 20-30 percent.# (That was 8 years ago.  In 2002, on Linux, I can't measure# a speedup. :-( ) # I'm too lazy to write a command line parser for the full find(1)# command line syntax, so the predicate it searches for is wired-in,# see function selector() below.  (It currently searches for files with# world write permission.) # Usage: parfind.py [-w nworkers] [directory] ...# Default nworkers is 4  import sysimport getoptimport stringimport timeimport osfrom stat import *import thread  # Work queue class.  Usage:#   wq = WorkQ()#   wq.addwork(func, (arg1, arg2, ...)) # one or more calls#   wq.run(nworkers)# The work is done when wq.run() completes.# The function calls executed by the workers may add more work.# Don't use keyboard interrupts! class WorkQ:     # Invariants:     # - busy and work are only modified when mutex is locked    # - len(work) is the number of jobs ready to be taken    # - busy is the number of jobs being done    # - todo is locked iff there is no work and somebody is busy     def __init__(self):        self.mutex = thread.allocate()        self.todo = thread.allocate()        self.todo.acquire()        self.work = []        self.busy = 0     def addwork(self, func, args):        job = (func, args)        self.mutex.acquire()        self.work.append(job)        self.mutex.release()        if len(self.work) == 1:            self.todo.release()     def _getwork(self):        self.todo.acquire()        self.mutex.acquire()        if self.busy == 0 and len(self.work) == 0:            self.mutex.release()            self.todo.release()            return None        job = self.work[0]        del self.work[0]        self.busy = self.busy + 1        self.mutex.release()        if len(self.work) > 0:            self.todo.release()        return job     def _donework(self):        self.mutex.acquire()        self.busy = self.busy - 1        if self.busy == 0 and len(self.work) == 0:            self.todo.release()        self.mutex.release()     def _worker(self):        time.sleep(0.00001)     # Let other threads run        while 1:            job = self._getwork()            if not job:                break            func, args = job            apply(func, args)            self._donework()     def run(self, nworkers):        if not self.work:            return # Nothing to do        for i in range(nworkers-1):            thread.start_new(self._worker, ())        self._worker()        self.todo.acquire()  # Main program def main():    nworkers = 4    opts, args = getopt.getopt(sys.argv[1:], '-w:')    for opt, arg in opts:        if opt == '-w':            nworkers = string.atoi(arg)    if not args:        args = [os.curdir]     wq = WorkQ()    for dir in args:        wq.addwork(find, (dir, selector, wq))     t1 = time.time()    wq.run(nworkers)    t2 = time.time()     sys.stderr.write('Total time %r sec.\n' % (t2-t1))  # The predicate -- defines what files we look for.# Feel free to change this to suit your purpose def selector(dir, name, fullname, stat):    # Look for world writable files that are not symlinks    return (stat[ST_MODE] & 0002) != 0 and not S_ISLNK(stat[ST_MODE])  # The find procedure -- calls wq.addwork() for subdirectories def find(dir, pred, wq):    try:        names = os.listdir(dir)    except os.error, msg:        print repr(dir), ':', msg        return    for name in names:        if name not in (os.curdir, os.pardir):            fullname = os.path.join(dir, name)            try:                stat = os.lstat(fullname)            except os.error, msg:                print repr(fullname), ':', msg                continue            if pred(dir, name, fullname, stat):                print fullname            if S_ISDIR(stat[ST_MODE]):                if not os.path.ismount(fullname):                    wq.addwork(find, (fullname, pred, wq))  # Call the main program main()