23

私は、単純な python スレッド プール パターンの適切な実装を見回してきましたが、実際に自分のニーズに合ったものを見つけることができません。私はpython 2.7を使用していますが、見つかったすべてのモジュールが機能しないか、ワーカーの例外を適切に処理しません。私が探しているタイプの機能を提供できるライブラリを誰かが知っているかどうか疑問に思っていました。大変助かります。

マルチプロセッシング

私の最初の試みは組み込みmultiprocessingモジュールでしたが、これはスレッドではなくサブプロセスを使用するため、オブジェクトをピクルできないという問題に遭遇しました。いいえ、ここに行きません。

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

先物

Python 3.2 の優れた同時実行機能のいくつかのバック ポートがここにあることがわかります。これは完璧で使いやすいようです。問題は、ワーカーの 1 つで例外が発生した場合、「ZeroDivisionError」などの例外のタイプのみが取得され、トレースバックが取得されないため、どの行で例外が発生したかがわからないことです。コードがデバッグできなくなります。立ち入り禁止。

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

ワーカープール

このパターンの別の実装を見つけましたhere。今回は例外が発生すると出力されますが、ipython 対話型インタープリターはハング状態のままになり、他のシェルから強制終了する必要があります。立ち入り禁止。

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

スレッドプール

さらに別の実装がここにあります。今回は例外が発生すると、スクリプトに出力されますstderrが、スクリプトは中断されずに実行が継続されます。これは、例外の目的に反し、物事を危険にさらす可能性があります。まだ使えません。

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'

- アップデート -

ライブラリに関してはfutures、python 3 の動作は 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出力:

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出力:

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
4

5 に答える 5

5

concurrent.futuresインターフェイスが非常にシンプルなので、私は個人的に使用しています。トレースバックの問題については、それを維持するための回避策を見つけました。この他の質問に対する私の答えをチェックしてください：

concurrent.futures での例外の元の行番号の取得

于 2014-06-28T11:09:07.033 に答える
1

簡単な解決策：自分に最も適した代替手段を使用try-exceptし、ワーカーに独自のブロックを実装します。必要に応じて、ルート呼び出しを囲みます。

これらのライブラリが例外を「誤って」処理するとは言いません。これらはデフォルトの動作をしますが、原始的です。デフォルトが適切でない場合は、これを自分で処理する必要があります。

于 2013-03-12T11:00:52.753 に答える