91616


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

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

[code]

    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)
    
[/code]

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

[code]

    Value:[0]
    Value:[1]
    Value:[2]
    Value:[3]
    Value:[4]
    Value:[5]
    Value:[6]
    Value:[7]
    Value:[8]
    Value:[9]
    
[/code]


CC BY-SA 4.0
2022-10-14T14:28:59.210
91604
2022-12-04T12:17:39.850
2022-10-15T07:11:14.540
35267
35267
question
python python3
例外発生時に並行処理をキャンセルさせたい


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

[code]

    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)
    
[/code]


CC BY-SA 4.0
2022-10-15T14:05:30.830
91616
2022-10-15T14:27:32.013
2022-10-15T14:27:32.013
26370
26370
91604
answer


回答出てるけど一応

* * *

`concurrent.futures`の
[`Executor.map()`](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Executor.map)
には以下のように記されてます

> The returned iterator raises a
> [`concurrent.futures.TimeoutError`](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.TimeoutError)
> if `__next__()` is called and the result isn’t available after _timeout_
> seconds from the original call to
> [Executor.map()](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.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()` のことも踏まえると以下の通り

[code]

    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
    
[/code]


CC BY-SA 4.0
2022-10-16T04:36:51.187
91624
2022-12-04T12:17:39.850
2022-12-04T12:17:39.850
43025
43025
91604
answer