2

時間のかかる処理(例えばネットワーク通信処理など)を並行実行するコードを作成しました。
これに加え、時間のかかる処理の内部で例外が発生した場合は、
未実行の処理を実行させないようにしたいのですが、どのようにすればよいでしょうか?
(並行で実行中のメソッドはそのまま実行されるのは仕方ないと考えています。)

また応用として、時間のかかる処理で10回以上例外が発生したら、以降の処理は実行しないようにしたい(キャンセルしたい)のですが、可能でしょうか。
キャンセルする条件が成り立ったら、全てのfutureオブジェクトに対し、cancelメソッドを実行すればできそうなのですが、
ループとなるためスマートでなく、他によい方法があるのではないのかと思い質問しました。

import time
from concurrent.futures import ThreadPoolExecutor

def func(value: int) -> str:
    # この関数内部で例外が発生したら、他のスレッドも停止状態にしたい。
    time.sleep(1)
    return f'Value:[{value}]'

with ThreadPoolExecutor(max_workers=2) as executor:
    future_list = executor.map(func, range(0, 10), timeout=5)

for value in future_list:  # 実行結果を表示
    print(value)

以下、サンプルの実行結果です。

Value:[0]
Value:[1]
Value:[2]
Value:[3]
Value:[4]
Value:[5]
Value:[6]
Value:[7]
Value:[8]
Value:[9]
3
  • こんな記事を参考に、フラグやカウンタとなる変数を用意して制御するとか? 並列処理を含むプログラムを強制終了させたい
    – kunif
    Commented 2022年10月14日 16:00
  • 「時間のかかる処理(例えばWebスクレイピングなど)を並行実行するコード」については、相手先サーバーに負担をかける行為は辞めるべきかと思います。実際に実行してませんよね?例として挙げるのも不適切かと思います。
    – merino
    Commented 2022年10月14日 22:55
  • 例えばHTTP通信など … について。それらがほぼ同じサイトで占められるなら, 別々ではなく 同じコネクションで HTTP/1.1 HTTP/2 あるいはそれ以降を使ったほうが大抵高速でサーバーにとっても望ましい。URLがほぼ別々のサイトであるのなら別コネクションになるが だからといってスレッド使うまでもなく普通に(順番に)アクセスしたら並列アクセスになります
    – oriri
    Commented 2022年10月15日 5:21

2 件の回答 2

1

他の方のコメントのような、元々の処理がこれを適用して良い物かどうかは横に置いておいて、私のコメントで紹介した以下記事を応用して、こんな風に出来ると考えられます。
並列処理を含むプログラムを強制終了させたい

import random  #### 例外発生シミュレーション用に追加
import time
from concurrent.futures import ThreadPoolExecutor

#### 例外発生シミュレーション用に例外クラス追加
class OriginalError(Exception):
    pass

#### 例外発生カウンタ 0初期化
excepcount = 0

def func(value: int) -> str:
    #### 処理開始前に例外発生回数が限度をオーバーしているかチェック
    global excepcount
    if excepcount >= 3: #### ここで回数を調節
        return f'Value:[{value}] excepcount >= 3 check'
    
    # この関数内部で例外が発生したら、他のスレッドも停止状態にしたい。
    try:
        time.sleep(1)
        #### 例外発生のシミュレーション処理
        num = random.randint(0, 100)
        if num < 25:
            raise OriginalError("OriginalErrorです。")
    except Exception as e:
        excepcount += 1  #### 例外発生回数を加算
        return f'Value:[{value}] exception occurred' ####このスレッドは例外終了
    
    return f'Value:[{value}]'

with ThreadPoolExecutor(max_workers=2) as executor:
    future_list = executor.map(func, range(0, 15), timeout=5)  ####スレッド数増加

for value in future_list:  # 実行結果を表示
    print(value)
1
  • ありがとうございます!期待通りの結果を得られました。例外発生回数をカウントし、キャンセル条件成立したときに、処理打ち切りできること理解しました。
    – Sasaki
    Commented 2022年10月15日 15:02
0

回答出てるけど一応


concurrent.futuresExecutor.map() には以下のように記されてます

The returned iterator raises a concurrent.futures.TimeoutError if __next__() is called and the result isn’t available after timeout seconds from the original call to Executor.map(). timeout can be an int or a float. If timeout is not specified or None, there is no limit to the wait time.
If a func call raises an exception, then that exception will be raised when its value is retrieved from the iterator.

  • Executor.map() が返すのは iterator でタイムアウト時は …
  • func の呼び出しが例外を送出した場合、その例外はイテレータから値を受け取る時に送出されます

なので, value == 1 のときに例外が発生しても, その次までは実行されてしまうかも

時間のかかる処理で10回以上例外が発生したら

こちらは別回答が参考になるでしょう


また with の使い方で, 例えば with open() as fp: のインデント内で処理を記述する必要があり, 外側で fp.write() としても書き込まれない (エラー or 無効)

ThreadPoolExecutor の場合も, スレッド実行中に何らかのアクション起こすなら そのインデント内に記述しないと意味はないでしょう。(インデント外では既に終了してる)

上記 Executor.map() のことも踏まえると以下の通り

import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError

def func(value: int, ch: str) -> str:
    # この関数内部で例外が発生したら、他のスレッドも停止状態にしたい。
    print(ch*3, value)
    time.sleep(1)
    if value == 1:
        raise Exception(f'なんらか {value=}')
    return f'Value:[{value}]'

with ThreadPoolExecutor(max_workers=2) as executor:
    it = executor.map(func, range(0, 10), list('ABCDEFGHIJ'), timeout=5)
    try:
        future_list = list(it)
    except TimeoutError:
        print('タイムアウト')
    except Exception as e:
        print(f'generated an exception: {e}')
    else:
        print('正常終了')
        for value in future_list:  # 実行結果を表示
            print(value)

# (結果)
AAA 0
BBB 1
CCC 2
DDD 3
generated an exception: なんらか value=1

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。