imap_unordered() zawiesza się, jeśli iterable zgłasza błąd

Question

rozważyć plik sample.py zawiera następujący kod:

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" 

Kiedy go wykonać, pojawia się następujący komunikat:

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! 

Po tym, aplikacja się zawiesza. Jak mogę sobie poradzić z sytuacją bez zawieszania się?

Answer 1

Jak wspomniano w septi, twoje wcięcie jest (nadal) złe. Wcięcie rachunku zysków i strat tak, aby mieściło się w jego zasięgu. Nie jestem do końca pewien, co rzeczywiście dzieje się w kodzie, ale otrzymując zmienną, która wykracza poza zakres nie wydaje się dobrym pomysłem:

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" 

do obsługi wyjątków w generatorze, można użyć otoki taki jak ten:

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

Nie wiem, czy się spodziewasz. Dlaczego funkcja typu map wie, co zrobić w przypadku wyjątku. Jedyną rozsądną rzeczą, którą może zrobić, to nie obsługiwać wyjątku. Jeśli potrzebujesz obsługi wyjątków, musisz powiedzieć generatorowi, jak sobie z tym poradzić. Problem nie polega na tym, że imap_unordered stracił swoje miejsce, ale fakt, że generator utracił swoje miejsce. Generator nie będzie wiedział, gdzie powrócić, gdy pojawi się następny monit o inny przedmiot.