Initial revision

diff --git a/Demo/threads/find.py b/Demo/threads/find.py
new file mode 100644 (file)
index 0000000..e97bd99
--- /dev/null
+++ b/Demo/threads/find.py
@@ -0,0 +1,153 @@
+# A parallelized "find(1)" using the thread module (SGI only for now).
+
+# 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.
+
+# 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
+# group or world write permission.)
+
+# Usage: parfind.py [-w nworkers] [directory] ...
+# Default nworkers is 4, maximum appears to be 8 (on Irix 4.0.2)
+
+
+import sys
+import getopt
+import string
+import time
+import os
+from 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, job):
+               if not job:
+                       raise TypeError, 'cannot add null job'
+               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):
+               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.millitimer()
+       wq.run(nworkers)
+       t2 = time.millitimer()
+
+       sys.stderr.write('Total time ' + `t2-t1` + ' msec.\n')
+
+
+# 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 group or world writable files
+       return (stat[ST_MODE] & 0022) != 0
+
+
+# The find procedure -- calls wq.addwork() for subdirectories
+
+def find(dir, pred, wq):
+       try:
+               names = os.listdir(dir)
+       except os.error, msg:
+               print `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 `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()