Group().imap on Queue deadlock

[thinxer]

# encoding: utf8

import gevent
from gevent.pool import Group
from gevent.queue import Queue


fetches = Queue()
validations = Queue()
results = Queue()

def searcher():
    fetches.put(True)
    validations.get()
    print "SEARCH DONE"
    fetches.put(StopIteration)


def validator():
    for result in Group().imap(lambda _:_, fetches):
        print "VALIDATION PUT"
        validations.put(True)
    print "VALIDATION DONE"
    validations.put(StopIteration)
    results.put(StopIteration)
    return


gevent.spawn(searcher)
gevent.spawn(validator)

results.get()

Output:

VALIDATION PUT
SEARCH DONE
Traceback (most recent call last):
  File "test.py", line 32, in <module>
    results.get()
  File "/Users/thinxer/Library/Python/2.7/lib/python/site-packages/gevent/queue.py", line 200, in get
    result = waiter.get()
  File "/Users/thinxer/Library/Python/2.7/lib/python/site-packages/gevent/hub.py", line 568, in get
    return self.hub.switch()
  File "/Users/thinxer/Library/Python/2.7/lib/python/site-packages/gevent/hub.py", line 331, in switch
    return greenlet.switch(self)
gevent.hub.LoopExit: This operation would block forever

It seems that imap never stopped, though fetches been put StopIteration.

Update:
Using Python 2.7 on Mac OS X Mavericks.
gevent.version is "1.0".

[thinxer]

A Go version (using imap_unordered instead of imap): http://play.golang.org/p/TGKE5Kh1u0 , which works fine.

[thinxer]

Will anyone tell me if this is intended, or a bug of gevent? Thanks in advance!

[fantix]

@thinxer This seems to be a bug in IMap:

    def _run(self):
        try:
            empty = True
            func = self.func
            for item in self.iterable:
                self.count += 1
                g = self.spawn(func, item)
                ....

    def _on_result(self, greenlet):
        self.count -= 1
        if greenlet.successful():
            self.queue.put((greenlet.index, greenlet.value))
        else:
            self.queue.put((greenlet.index, Failure(greenlet.exception)))
        if self.ready() and self.count <= 0:
            self.maxindex += 1
            self.queue.put((self.maxindex, Failure(StopIteration))) 

    def _on_finish(self, _self):
        if not self.successful():
            self.maxindex += 1
            self.queue.put((self.maxindex, Failure(self.exception)))

This piece of code assumes that _on_result is called after _run fully returns. But actually in your test code, self.iterable was fed with a gevent Queue object, which will switch to the hub when iterated, then no StopIteration would ever be fed to self.queue, because _on_result is called while iterating over self.iterable when the greenlet is not "ready" yet.

[thinxer]

Yes it is. One of my coworkers pointed out the same issue. Checking self.count after the for loop in _run solves this bug. But I'm not sure if this change will make other things stop working.

[fantix]

@thinxer I think it should be okay - even replacing the current empty check is fine.