Hồ bơi thread Python xử lý các ngoại lệ

Question

Tôi đã xem xét việc triển khai thực hiện một mẫu hồ bơi chủ đề python đơn giản và thực sự không thể tìm thấy bất kỳ thứ gì phù hợp với nhu cầu của tôi. Tôi đang sử dụng python 2,7 và tất cả các mô-đun tôi đã tìm thấy hoặc là không làm việc, hoặc không xử lý ngoại lệ trong công nhân đúng cách. Tôi đã tự hỏi nếu ai đó biết về một thư viện có thể cung cấp các loại chức năng tôi đang tìm kiếm. Giúp đánh giá cao.

Đa

nỗ lực đầu tiên của tôi là với built-in multiprocessing module, nhưng như thế này không sử dụng đề nhưng subprocesses thay vì chúng tôi chạy vào vấn đề mà các đối tượng không thể được ngâm. Không đi đây.

from multiprocessing import Pool 

class Sample(object): 
    def compute_fib(self, n): 
     phi = (1 + 5**0.5)/2 
     self.fib = int(round((phi**n - (1-phi)**n)/5**0.5)) 

samples = [Sample() for i in range(8)] 
pool = Pool(processes=8) 
for s in samples: pool.apply_async(s.compute_fib, [20]) 
pool.join() 
for s in samples: print s.fib 

# PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup __builtin__.instancemethod failed 

Futures

Vì vậy, tôi thấy có một cổng sau của một số tính năng đồng thời mát mẻ của python 3.2 here. Điều này có vẻ hoàn hảo và dễ sử dụng. Vấn đề là khi bạn nhận được một ngoại lệ trong một trong những người lao động, bạn chỉ nhận được loại ngoại lệ như "ZeroDivisionError" nhưng không có traceback và do đó không có dấu hiệu của dòng nào gây ra ngoại lệ. Mã không thể gỡ lỗi được. Không đi.

from concurrent import futures 

class Sample(object): 
    def compute_fib(self, n): 
     phi = (1 + 5**0.5)/2 
     1/0 
     self.fib = int(round((phi**n - (1-phi)**n)/5**0.5)) 

samples = [Sample() for i in range(8)] 
pool = futures.ThreadPoolExecutor(max_workers=8) 
threads = [pool.submit(s.compute_fib, 20) for s in samples] 
futures.wait(threads, return_when=futures.FIRST_EXCEPTION) 
for t in threads: t.result() 
for s in samples: print s.fib 


# futures-2.1.3-py2.7.egg/concurrent/futures/_base.pyc in __get_result(self) 
# 354  def __get_result(self): 
# 355   if self._exception: 
#--> 356    raise self._exception 
# 357   else: 
# 358    return self._result 
# 
# ZeroDivisionError: integer division or modulo by zero 

WorkerPool

Tôi tìm thấy một thi khác của mô hình này here. Lần này khi một ngoại lệ xảy ra nó được in, nhưng sau đó trình thông dịch tương tác ipython của tôi bị bỏ lại trong trạng thái treo và cần phải bị giết từ một trình bao khác. Không đi.

import workerpool 

class Sample(object): 
    def compute_fib(self, n): 
     phi = (1 + 5**0.5)/2 
     1/0 
     self.fib = int(round((phi**n - (1-phi)**n)/5**0.5)) 

samples = [Sample() for i in range(8)] 
pool = workerpool.WorkerPool(size=8) 
for s in samples: pool.map(s.compute_fib, [20]) 
pool.wait() 
for s in samples: print s.fib 

# ZeroDivisionError: integer division or modulo by zero 
# ^C^C^C^C^C^C^C^C^D^D 
# $ kill 1783 

ThreadPool

Tuy nhiên, một thực hiện khác here. Lần này khi một ngoại lệ xảy ra, nó được in đến stderr nhưng kịch bản không bị gián đoạn và thay vào đó tiếp tục thực thi, điều này làm mất mục đích của ngoại lệ và có thể làm cho mọi thứ không an toàn. Vẫn không thể sử dụng được.

import threadpool 

class Sample(object): 
    def compute_fib(self, n): 
     phi = (1 + 5**0.5)/2 
     1/0 
     self.fib = int(round((phi**n - (1-phi)**n)/5**0.5)) 

samples = [Sample() for i in range(8)] 
pool = threadpool.ThreadPool(8) 
requests = [threadpool.makeRequests(s.compute_fib, [20]) for s in samples] 
requests = [y for x in requests for y in x] 
for r in requests: pool.putRequest(r) 
pool.wait() 
for s in samples: print s.fib 

# ZeroDivisionError: integer division or modulo by zero 
# ZeroDivisionError: integer division or modulo by zero 
# ZeroDivisionError: integer division or modulo by zero 
# ZeroDivisionError: integer division or modulo by zero 
# ZeroDivisionError: integer division or modulo by zero 
# ZeroDivisionError: integer division or modulo by zero 
# ZeroDivisionError: integer division or modulo by zero 
# ZeroDivisionError: integer division or modulo by zero 
#---> 17 for s in samples: print s.fib 
# 
#AttributeError: 'Sample' object has no attribute 'fib' 

- Cập nhật -

Dường như liên quan đến thư viện futures, hành vi của python 3 là không giống như python 2.

futures_exceptions.py:

from concurrent.futures import ThreadPoolExecutor, as_completed 

def div_zero(x): 
    return x/0 

with ThreadPoolExecutor(max_workers=4) as executor: 
    futures = executor.map(div_zero, range(4)) 
    for future in as_completed(futures): print(future) 

Python 2.7.6 đầu ra:

Traceback (most recent call last): 
    File "...futures_exceptions.py", line 12, in <module> 
    for future in as_completed(futures): 
    File "...python2.7/site-packages/concurrent/futures/_base.py", line 198, in as_completed 
    with _AcquireFutures(fs): 
    File "...python2.7/site-packages/concurrent/futures/_base.py", line 147, in __init__ 
    self.futures = sorted(futures, key=id) 
    File "...python2.7/site-packages/concurrent/futures/_base.py", line 549, in map 
    yield future.result() 
    File "...python2.7/site-packages/concurrent/futures/_base.py", line 397, in result 
    return self.__get_result() 
    File "...python2.7/site-packages/concurrent/futures/_base.py", line 356, in __get_result 
    raise self._exception 
ZeroDivisionError: integer division or modulo by zero 

Python 3.3.2 đầu ra:

Traceback (most recent call last): 
    File "...futures_exceptions.py", line 11, in <module> 
    for future in as_completed(futures): 
    File "...python3.3/concurrent/futures/_base.py", line 193, in as_completed 
    with _AcquireFutures(fs): 
    File "...python3.3/concurrent/futures/_base.py", line 142, in __init__ 
    self.futures = sorted(futures, key=id) 
    File "...python3.3/concurrent/futures/_base.py", line 546, in result_iterator 
    yield future.result() 
    File "...python3.3/concurrent/futures/_base.py", line 392, in result 
    return self.__get_result() 
    File "...python3.3/concurrent/futures/_base.py", line 351, in __get_result 
    raise self._exception 
    File "...python3.3/concurrent/futures/thread.py", line 54, in run 
    result = self.fn(*self.args, **self.kwargs) 
    File "...futures_exceptions.py", line 7, in div_zero 
    return x/0 
ZeroDivisionError: division by zero 

Answer 1

Cá nhân tôi sử dụng Futures vì giao diện rất đơn giản. Đối với vấn đề truy nguyên, tôi tìm thấy một cách giải quyết để bảo vệ nó. Thanh toán trả lời của tôi cho câu hỏi khác này:

Getting original line number for exception in concurrent.futures

Answer 2

Giải pháp dễ dàng: Sử dụng bất cứ điều gì phù hợp với lựa chọn bạn tốt nhất, và thực hiện try-except khối của riêng bạn trong công nhân của bạn. Quanh cuộc gọi gốc nếu bạn phải.

Tôi sẽ không nói những thư viện này xử lý ngoại lệ "không chính xác". Họ có một hành vi mặc định, tuy nhiên nguyên thủy. Bạn sẽ tự mình xử lý nếu mặc định không phù hợp với bạn.