Turning f(callback) into a generator

[Peter Otten]

Jimmy Retzlaff wrote:

> Perhaps there should be a second queue for an exception and the
> sentinel. Then the main loop could look something like this (untested):
> 
>     while terminationQueue.empty() or not queue.empty():
>         # The timeout is needed because we may have arrived here after
>         # the last value was placed in the queue but before the sentinel
>         # or an exception was placed in the terminationQueue.
>         yield queue.get(timeout=0.1)
> 
>     terminationValue = terminationQueue.get()
>     if terminationValue is not sentinel:
>          raise terminationValue
> 
> In this case the launcher function would put the sentinel in the
> terminationQueue instead of the data queue. The idea of the loop
> condition above is to yield everything that came through before the
> exception/sentinel, then re-raise any exception if that was the reason
> for terminating. This would hopefully help clarify where the exception
> occurred (i.e., on which element in the iteration).
> 
> The separate queue also eliminates the problem of someone somehow
> finding a way of injecting the sentinel into the data stream. But this
> does create the need for the timeout (which doesn't work in 2.2). I
> guess it's a judgment call as to whether the sentinel should be placed
> in one queue or the other (or both).

I'd rather not go with the above approach. At first glance it looks more
complicated than the original design. Also, I'm not sure if an item in the
terminationQueue could become visible before all items in the data queue.
However, seeing 

raise terminationValue

was useful in that I noticed that you do not nead a special sentinel -
instead, just put a StopIteration exception into the queue.

> Your generalization could make a nice Python Cookbook recipe if you feel
> like putting it together. At least two of us have needed this sort of
> thing.

Since I posted the code, I've found a serious bug. Unless you run the
generator to exhaustion, you end up with a thread where a callback waits
forever to insert a tuple into the queue. So some two-way communication is
needed here to tell the callback "Stop it, we're done", and I've not yet
figured out how to do that. An intermediate fix is a "high" timeout and
simply returning from launcher() on a Queue.Full exception.

When I can advance code quality to prime time, I will probably turn it into
a recipe - good idea :-)

In the convenience department, this tiny class will make usage even more
intuitive:

class CbWrapper:
    def __init__(self, visit, queuesize=1, callbackpos=None):
        self._visit = visit
        self.queuesize = queuesize
        self.callbackpos = callbackpos
    def __call__(self, *args, **kwd):
        if self.callbackpos is not None:
            args = args[:self.callbackpos] + (CALLBACK,) +
args[self.callbackpos:]
        return cbwrapper(self._visit, self.queuesize, *args, **kwd)
    visit = __call__

# make os.path.walk() almost as easy to use as os.walk()
import os
mywalk = CbWrapper(os.path.walk, callbackpos=1)
for arg, dirname, names in mywalk(".", None):
    print "directory \'%s\' contains %d items" % (dirname, len(names))

Perhaps some extension of the class approach will also help with the
dangling thread problem, killing it at least when the instance goes out of
scope.

Peter