並行処理：Aの処理が早く終わればBの処理を止める．Bの処理が早く終わればAの処理を止める

Question

write_txt() ：test.txtに不定期的に1,2,3を書き込む関数
check_file.over5s() : 5秒間test.txtが更新されなければ"5秒"と返すメソッド

write_txt()とcheck_file.over5s()を同時に動かすことでtest.txtが5秒間更新されなければ処理を停止するプログラムを書きたいと考えています．

下のコードは5秒間test.txtが更新されなければ"5秒"と出力され，逆に5秒以内にtest.txtが更新され続ければ"書き込み完了"と期待どおり出力されます．

しかし，"5秒"と出力された後もwrite_txt()は動き続けるため，ファイルが更新されてしまいます．
2つの処理を同時に行い，一方の処理が早く終了すれば，もう一方の処理を強制終了させるためにはどのようにすれば良いでしょうか?

from concurrent.futures import ThreadPoolExecutor
import concurrent
import random,time,pathlib


def write_txt(text):
    for i in range(10):
        with open("test.txt","a") as f:
            f.write(f"{i}\n")
            sleep_time = random.random()*10
        time.sleep(sleep_time)
    return text

class check_file():
    def __init__(self):
        self.file = pathlib.Path('test.txt')
        self.update_time = self.file.stat().st_mtime
    
    def over5s(self,text):
        while True:
            time.sleep(5)
            if self.update_time != self.file.stat().st_mtime:
                self.update_time = self.file.stat().st_mtime
            else:               
                break                
        return text
            
with open("test.txt","w") as f:
    pass
with ThreadPoolExecutor(max_workers=2) as executor:
    file_checker = check_file()
    results = []
    results.append(executor.submit(file_checker.over5s, "5秒"))
    results.append(executor.submit(write_txt, "書き込み完了"))
    concurrent.futures.wait(results, return_when=concurrent.futures.FIRST_COMPLETED)
    for f in concurrent.futures.as_completed(results):
        f_success = f.result()
        print(f_success)
        

concurrent.futures.wait(results, return_when=concurrent.futures.FIRST_COMPLETED)でうまくいくかなと思ったのですが，うまくいきませんでした．
もう一方の処理を停止するコードも書かなきゃだめそうな事はわかるのですが，ドキュメントを調べてもよくやり方が分かりませんでした．

Answer 1

すでに回答が付いていますが、以前回答したやつも少しは役に立つと思うので、貼っておきます。

https://teratail.com/questions/ei3ctb6rr2thsy

Comments

[kt19906]

回答ありがとうございます．
私の説明不足で申し訳ないのですが，実際にはwrite_txtのような処理をライブラリを使って1行のコードで行っており．添付して下さったURLのようにWhile文で逐一変数を確認することできません．
そのため強制的に終了させる必要があると思うのですが，そのような方法はありますか？

ライブラリを使ってあるデータの解析を行なっており，解析の進捗状況はtxtファイルに書き出されます．データは複数あるのでfor文を使って解析を行なっていますが，データによっては解析に時間がかかり，数時間経ってもファイルが更新されないことがあります．そのようなデータは無視し，次のループに移りたいと考えています．

[TakaiY]

ライブラリの関数をそのままスレッド/プロセスで動かすのでなく、そのライブラリを使って計算する関数なりを作って動かせばいいでしょう。
そうすれば、その関数の中で状態は確認できます。

「数時間経ってもファイルが更新されないことがあります」というような場合に、強制的にプロセスを止めたりすれば、メモリが解放されなかったり、リソースを掴んだままだったりと、不都合が置きる可能性もありますが、そのあたりは大丈夫ですか？

[kt19906]

メモリなどに関する知識があまりないのですが，大量にデータを流す予定なので，その間メモリが解放されなかったらよくないと思うので関数を作って動かしてみたいと思います．

その場合は並行処理せずともファイル更新するまでの処理を関数化しその関数にタイムアウトを設ければよさそうですね

[TakaiY]

と、書きましたが、その呼び出したライブラリ内で時間がかかっている場合は手が出ませんね。

基本的にthreadを強制的に終了させる方法は無いので、どうしてもというなら、process にして killするのがよさそうですね。 (やったことはありません)

[kt19906]

ライブラリソースコードを見たところ，while文で逐次textファイルを更新していたので，while文の中身を関数化してtimeout_decoratorというライブラリでその関数のタイムアウトの時間を設定し解決しました．

ありがとうございました．

Answer 2

threading.Event を使ってみてはいかがでしょうか。
（処理の流れがわかりやすいように、print を追加しています）

from concurrent.futures import ThreadPoolExecutor
import concurrent
import random,time,pathlib
from threading import Event

def write_txt(text,event):
    for i in range(10):
        with open("test.txt","a") as f:
            f.write(f"{i}\n")
            print(f"write_txt: ファイルに{i}を書き込みました。")
        sleep_time = random.random()*10
        print(f"write_txt: {sleep_time}秒待ちます。")
        if event.wait(sleep_time):
            print(f"write_txt: 処理終了の通知を受けました。終了します。")
            break
    return text

class check_file():
    def __init__(self, event):
        self.file = pathlib.Path('test.txt')
        self.update_time = self.file.stat().st_mtime
        self.event = event
    
    def over5s(self,text):
        while True:
            print("check_file: 5秒待ちます。")
            time.sleep(5)
            print("check_file: 5秒経過。ファイルの更新時間を判定します。")
            if self.update_time != self.file.stat().st_mtime:
                print("check_file: 5秒以内にファイルが更新されていました。update_time を更新して、処理を継続します。")
                self.update_time = self.file.stat().st_mtime
            else:
                print("check_file: 5秒以内にファイルが更新されていませんでした。write_fileに処理終了を通知します。")
                self.event.set()           
                break                
        return text
            
with open("test.txt","w") as f:
    pass
with ThreadPoolExecutor(max_workers=2) as executor:
    event = Event()
    file_checker = check_file(event)
    results = []
    results.append(executor.submit(file_checker.over5s, "5秒"))
    results.append(executor.submit(write_txt, "書き込み完了", event))

    for f in concurrent.futures.as_completed(results):
        f_success = f.result()
        print(f_success)
        

補足

event.wait(timeout)

timeout が与えられている場合、
timeout 秒経過するか、他のスレッドに set されるまで、スレッドをブロックします。
timeout 秒経過前に 他のスレッドに set された場合、True を返します。
setされる前にtimeout秒経過した場合、ブロックされていたスレッドを再開し、Falseを返します。

Comments

[kt19906]

回答ありがとうございます．
私の説明不足で申し訳ないのですが，TakaiYさんの方にも返信したように，write_txtの部分を実際にはライブラリを使って1行のコードで行っており．if event.wait(sleep_time)のように途中でeventを確認する作業を追加することができません．
このような場合でも停止する方法はありますか？
せっかく丁寧にコードを書いてくださったのに申し訳ありません!