1 # A parallelized "find(1)" using the thread module.

     2 

     3 # This demonstrates the use of a work queue and worker threads.

     4 # It really does do more stats/sec when using multiple threads,

     5 # although the improvement is only about 20-30 percent.

     6 # (That was 8 years ago.  In 2002, on Linux, I can't measure

     7 # a speedup. :-( )

     8 

     9 # I'm too lazy to write a command line parser for the full find(1)

    10 # command line syntax, so the predicate it searches for is wired-in,

    11 # see function selector() below.  (It currently searches for files with

    12 # world write permission.)

    13 

    14 # Usage: parfind.py [-w nworkers] [directory] ...

    15 # Default nworkers is 4

    16 

    17 

    18 import sys

    19 import getopt

    20 import string

    21 import time

    22 import os

    23 from stat import *

    24 import thread

    25 

    26 

    27 # Work queue class.  Usage:

    28 #   wq = WorkQ()

    29 #   wq.addwork(func, (arg1, arg2, ...)) # one or more calls

    30 #   wq.run(nworkers)

    31 # The work is done when wq.run() completes.

    32 # The function calls executed by the workers may add more work.

    33 # Don't use keyboard interrupts!

    34 

    35 class WorkQ:

    36 

    37     # Invariants:

    38 

    39     # - busy and work are only modified when mutex is locked

    40     # - len(work) is the number of jobs ready to be taken

    41     # - busy is the number of jobs being done

    42     # - todo is locked iff there is no work and somebody is busy

    43 

    44     def __init__(self):

    45         self.mutex = thread.allocate()

    46         self.todo = thread.allocate()

    47         self.todo.acquire()

    48         self.work = []

    49         self.busy = 0

    50 

    51     def addwork(self, func, args):

    52         job = (func, args)

    53         self.mutex.acquire()

    54         self.work.append(job)

    55         self.mutex.release()

    56         if len(self.work) == 1:

    57             self.todo.release()

    58 

    59     def _getwork(self):

    60         self.todo.acquire()

    61         self.mutex.acquire()

    62         if self.busy == 0 and len(self.work) == 0:

    63             self.mutex.release()

    64             self.todo.release()

    65             return None

    66         job = self.work[0]

    67         del self.work[0]

    68         self.busy = self.busy + 1

    69         self.mutex.release()

    70         if len(self.work) > 0:

    71             self.todo.release()

    72         return job

    73 

    74     def _donework(self):

    75         self.mutex.acquire()

    76         self.busy = self.busy - 1

    77         if self.busy == 0 and len(self.work) == 0:

    78             self.todo.release()

    79         self.mutex.release()

    80 

    81     def _worker(self):

    82         time.sleep(0.00001)     # Let other threads run

    83         while 1:

    84             job = self._getwork()

    85             if not job:

    86                 break

    87             func, args = job

    88             apply(func, args)

    89             self._donework()

    90 

    91     def run(self, nworkers):

    92         if not self.work:

    93             return # Nothing to do

    94         for i in range(nworkers-1):

    95             thread.start_new(self._worker, ())

    96         self._worker()

    97         self.todo.acquire()

    98 

    99 

   100 # Main program

   101 

   102 def main():

   103     nworkers = 4

   104     opts, args = getopt.getopt(sys.argv[1:], '-w:')

   105     for opt, arg in opts:

   106         if opt == '-w':

   107             nworkers = string.atoi(arg)

   108     if not args:

   109         args = [os.curdir]

   110 

   111     wq = WorkQ()

   112     for dir in args:

   113         wq.addwork(find, (dir, selector, wq))

   114 

   115     t1 = time.time()

   116     wq.run(nworkers)

   117     t2 = time.time()

   118 

   119     sys.stderr.write('Total time %r sec.\n' % (t2-t1))

   120 

   121 

   122 # The predicate -- defines what files we look for.

   123 # Feel free to change this to suit your purpose

   124 

   125 def selector(dir, name, fullname, stat):

   126     # Look for world writable files that are not symlinks

   127     return (stat[ST_MODE] & 0002) != 0 and not S_ISLNK(stat[ST_MODE])

   128 

   129 

   130 # The find procedure -- calls wq.addwork() for subdirectories

   131 

   132 def find(dir, pred, wq):

   133     try:

   134         names = os.listdir(dir)

   135     except os.error, msg:

   136         print repr(dir), ':', msg

   137         return

   138     for name in names:

   139         if name not in (os.curdir, os.pardir):

   140             fullname = os.path.join(dir, name)

   141             try:

   142                 stat = os.lstat(fullname)

   143             except os.error, msg:

   144                 print repr(fullname), ':', msg

   145                 continue

   146             if pred(dir, name, fullname, stat):

   147                 print fullname

   148             if S_ISDIR(stat[ST_MODE]):

   149                 if not os.path.ismount(fullname):

   150                     wq.addwork(find, (fullname, pred, wq))

   151 

   152 

   153 # Call the main program

   154 

   155 main()