Producer-consumer threading problem

[Carl Banks]

On Jun 10, 11:33 pm, George Sakkis <george.sak... at gmail.com> wrote:
> I pasted my current solution athttp://codepad.org/FXF2SWmg. Any
> feedback, especially if it has to do with proving or disproving its
> correctness, will be appreciated.


It seems like you're reinventing the wheel.  The Queue class does all
this, and it's been thorougly battle-tested.

So first of all, can you tell us why the following wouldn't work?  It
might help us understand the issue you're facing (never mind the
produce and consume arguments for now--I'll cover that below).


def iter_consumed(items):
    q = Queue.Queue()
    sentinel = object()
    def produce_all()
        for item in items:
            q.put()
        q.put(sentinel)
    producer = threading.Thread(target=produce_all)
    producer.start()
    try:
        while True:
            item = q.get()
            if item is sentinel:
                return
            yield item
    finally:
        # for robustness, notify producer to wrap things up
        # left as exercise
        producer.join()


If you want to customize the effect of getting and putting, you can
subclass Queue and override the _get and _put methods (however, last
time I checked, the Queue class expects _put to always add an item to
the internal sequence representing the queue--not necessarily to the
top--and _get to always remove an item--not necessarily from the
bottom).

However, even that's only necessary if you want to get items in a
different order than you put them.  If you just want to filter items
as they're produced or consumed, you should simply define
produce_filter and consume_filter, that are called before q.put and
after q.get, respectively.


One issue from your function.  This line:

done_remaining[1] += 1

is not atomic, but your code depends on it being so.  It can get out
of sync if there is a intervening thread switch between the read and
set.  This was discussed on the list a while back.  I posted an atomic
counter object in that thread (it was written in C--no other way) for
which the += is atomic.  Otherwise you have to use a lock.


Carl Banks