imap_unordered() treo lên nếu iterable ném một lỗi

Question

Hãy xem xét các tập tin sample.py chứa đoạn mã sau:

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" 

Khi tôi thực hiện nó, tôi nhận được kết quả như sau:

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! 

Sau đó, ứng dụng bị treo. Làm thế nào tôi có thể xử lý tình huống mà không bị treo cổ?

Answer 1

Vì septi đã đề cập đến sự thụt lề của bạn là (vẫn) sai. Thụt lề tuyên bố lợi nhuận để tôi nằm trong phạm vi của nó. Tôi không hoàn toàn chắc chắn về những gì thực sự xảy ra trong luật của mình, nhưng năng suất một biến đó là ra khỏi phạm vi không có vẻ như là một ý tưởng tốt:

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" 

Xử lý ngoại lệ trong máy phát điện của bạn, bạn có thể sử dụng một wrapper chẳng hạn như cái này:

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

Tôi không biết bạn sẽ xảy ra. Tại sao chức năng loại map phải biết phải làm gì trong trường hợp ngoại lệ. Điều hợp lý duy nhất nó có thể làm là không xử lý ngoại lệ. Nếu bạn cần xử lý các ngoại lệ thì bạn cần báo cho trình tạo của mình cách xử lý ngoại lệ. Vấn đề không phải là imap_unordered đã mất vị trí của nó, nhưng thực tế là máy phát điện đã mất vị trí của nó. Các máy phát điện sẽ không biết nơi để tiếp tục khi nó tiếp theo yêu cầu cho một mục.