RFC: my iterthreader module

[Justin Azoff]

I have this iterthreader module that I've been working on for a while
now.  It is similar to itertools.imap, but it calls each function in
its own thread and uses Queues for moving the data around.  A better
name for it would probably be ithreadmap, but anyway...

The short explanation of it is if you have a loop like
for item in biglist:
    print "The value for %s is %s" % (item, slowfunc(item))
or
for item,val in ((item, slowfunc(item)) for item in biglist):
    print "The value for %s is %s" % (item, val)

you can simply rewrite it as

for item,val in iterthreader.Threader(slowfunc, biglist):
    print "The value for %s is %s" % (item, val)

and it will hopefully run faster.  The usual GIL issues still apply of
course....  You can also subclass it in various ways, but I almost
always just call it in the above manner.

So, can anyone find any obvious problems with it?  I've been meaning to
re-post [1]  it to the python cookbook, but I'd like to hear what
others think first.  I'm not aware of any other module that makes this
particular use of threading this simple.

[1] I _think_ I posted it before, but that may have just been in a
comment

import threading
import Queue

class Threader:
    def __init__(self, func=None, data=None, numthreads=2):
        if not numthreads > 0:
            raise AssertionError("numthreads should be greater than 0")

        if func:
            self.handle_input=func
        if data:
            self.get_input = lambda : data

        self._numthreads=numthreads
        self.threads = []
        self.run()


    def __iter__(self):
        return self

    def next(self):
        still_running, input, output = self.DQ.get()
        if not still_running:
            raise StopIteration
        return input, output

    def get_input(self):
        raise NotImplementedError, "You must implement get_input as a
function that returns an iterable"

    def handle_input(self, input):
        raise NotImplementedError, "You must implement handle_input as
a function that returns anything"

    def _handle_input(self):
        while 1:
            work_todo, input = self.Q.get()
            if not work_todo:
                break
            self.DQ.put((True, input, self.handle_input(input)))

    def cleanup(self):
        """wait for all threads to stop and tell the main iter to
stop"""
        for t in self.threads:
            t.join()
        self.DQ.put((False,None,None))


    def run(self):
        self.Q=Queue.Queue()
        self.DQ=Queue.Queue()
        for x in range(self._numthreads):
            t=threading.Thread(target=self._handle_input)
            t.start()
            self.threads.append(t)

        try :
            for x in self.get_input():
                self.Q.put((True, x))
        except NotImplementedError, e:
            print e
        for x in range(self._numthreads):
            self.Q.put((False, None))

        threading.Thread(target=self.cleanup).start()


-- 
- Justin