Imap_unordered () freezes if iterable causes an error

Question

Consider a sample.py file containing the following code:

 from multiprocessing import Pool def sample_worker(x): print "sample_worker processes item", x return x def get_sample_sequence(): for i in xrange(2,30): if i % 10 == 0: raise Exception('That sequence is corrupted!') yield i if __name__ == "__main__": pool = Pool(24) try: for x in pool.imap_unordered(sample_worker, get_sample_sequence()): print "sample_worker returned value", x except: print "Outer exception caught!" pool.close() pool.join() print "done" 

When I completed it, I get the following output:

 Exception in thread Thread-2: Traceback (most recent call last): File "C:\Python27\lib\threading.py", line 810, in __bootstrap_inner self.run() File "C:\Python27\lib\threading.py", line 763, in run self.__target(*self.__args, **self.__kwargs) File "C:\Python27\lib\multiprocessing\pool.py", line 338, in _handle_tasks for i, task in enumerate(taskseq): File "C:\Python27\lib\multiprocessing\pool.py", line 278, in <genexpr> self._taskqueue.put((((result._job, i, func, (x,), {}) File "C:\Users\renat-nasyrov\Desktop\sample.py", line 10, in get_sample_sequence raise Exception('That sequence is corrupted!') Exception: That sequence is corrupted! 

After that, the application freezes. How can I handle a situation without freezes?

Answer 1

As septi said, your indentation is (still) wrong. Reject the yield statement so that I am within its scope. I'm not quite sure what is really going on in your code, but assigning a variable that goes beyond does not seem like a good idea:

 from multiprocessing import Pool def sample_worker(x): print "sample_worker processes item", x return x def get_sample_sequence(): for i in xrange(2,30): if i % 10 == 0: raise Exception('That sequence is corrupted!') yield i # fixed if __name__ == "__main__": pool = Pool(24) try: for x in pool.imap_unordered(sample_worker, get_sample_sequence()): print "sample_worker returned value", x except: print "Outer exception caught!" pool.close() pool.join() print "done" 

To handle exceptions in the generator, you can use a wrapper like this one:

 import logging def robust_generator(): try: for i in get_sample_sequence(): logging.debug("yield "+str(i)) yield i except Exception, e: logging.exception(e) raise StopIteration() 

Answer 2

I do not know what you expect. Why does a function like map know what to do in case of an exception. The only reasonable thing he can do is not handle the exception. If you need to handle exceptions, then you need to tell the generator how to handle the exception. The problem is not that imap_unordered lost its place, but that the generator has lost its place. The generator will not know where to resume when the next time he is asked for another item.