【Python基礎】並列処理concurrent.futures ThreadPoolExecutorで他の処理の終了を待つ方法

2023年4月30日2023年5月2日

URLをコピーしました！

concurrent.futures ThreadPoolExecutor

前回、ast.literal_eval（）で文字列を解釈、適切な型に変換する方法を解説しました。

終了を待たない例

まずは他の処理の終了を待たない例です。

from concurrent.futures import ThreadPoolExecutor
import time

def number5():
    for i in range(5):
        print(f'number5: {i}')
        time.sleep(0.5)
        
def number10():
    for i in range(10):
        print(f'number10: {i}')
        time.sleep(0.5)
        
def main():
    with ThreadPoolExecutor(max_workers=2) as executor:
        executor.submit(number5)
        executor.submit(number10)
        print('Executor finish')

    print('END')
    
if __name__ == '__main__':
    main()

実行結果
number5: 0
number10: 0
Executor finish
number5: 1
number10: 1
number5: 2
number10: 2
number5: 3number10: 3

number10: 4number5: 4

number10: 5
number10: 6
number10: 7
number10: 8
number10: 9
END

CPUの利用状況によって実行結果が多少変わると思いますがそこは気にせずにいてください。

このプログラムではまずnumber5関数とnumber10関数とmain関数の3つがあります。

そしてmain関数ではnumber5関数とnumber10関数をconcurrent.futuresのThreadPoolExecutorで並列処理しています。

ちなみにnumber5関数とnumber10関数ではそれぞれ５回と10回繰り返して、その繰り返し回数を表示し、0.5秒待つようになっています。

このため実行結果としては最初はnumber5関数の出力とnumber10関数の出力が入り乱れて表示されています。

そしてnumber5関数の方が繰り返し回数が少ないため先に終わり、number10関数の処理が後まで続いています。

ここで注目すべきはnumber5関数とnumber10関数の並列処理の後で実行されている「print(‘Executor finish’)」の処理です。

number5関数とnumber10関数が1度ずつ処理された後、すぐに「print(‘Executor finish’)」が処理されています。

つまりこの並列処理ではnumber5関数の処理もnumber10関数の処理も待たずに次の処理が進められてしまっているわけです。

それではこの「print(‘Executor finish’)」の処理をnumber5関数とnumber10関数の処理を待ってから行うにはどうしたらいいでしょうか。

解決策１：並列処理の外側に書く

先ほどの例のようにwith構文を使っているのなら、単純な解決策としては並列処理のwith構文の外側に後の処理を書くことです。

from concurrent.futures import ThreadPoolExecutor
import time

def number5():
    for i in range(5):
        print(f'number5: {i}')
        time.sleep(0.5)
        
def number10():
    for i in range(10):
        print(f'number10: {i}')
        time.sleep(0.5)
        
def main():
    with ThreadPoolExecutor(max_workers=2) as executor:
        executor.submit(number5)
        executor.submit(number10)
        
    print('Executor finish')
    print('END')
    
if __name__ == '__main__':
    main()

実行結果
number5: 0number10: 0

number10: 1number5: 1

number10: 2number5: 2

number5: 3number10: 3

number10: 4number5: 4

number10: 5
number10: 6
number10: 7
number10: 8
number10: 9
Executor finish
END

こうすれば「print(‘Executor finish’)」の処理はThreadPoolExecutorの処理外になるので、先ほどのように並列処理と混じることはありません。

ですが、ThreadPoolExecutorはこのようにも書くことができます。

from concurrent.futures import ThreadPoolExecutor
import time

def number5():
    for i in range(5):
        print(f'number5: {i}')
        time.sleep(0.5)
        
def number10():
    for i in range(10):
        print(f'number10: {i}')
        time.sleep(0.5)
        
def main():
    executor = ThreadPoolExecutor(max_workers=2)
    executor.submit(number5)
    executor.submit(number10)
        
    print('Executor finish')
    print('END')
    
if __name__ == '__main__':
    main()

実行結果
number5: 0number10: 0

Executor finish
END
number10: 1number5: 1

number5: 2number10: 2

number10: 3number5: 3

number5: 4number10: 4

number10: 5
number10: 6
number10: 7
number10: 8
number10: 9

この場合は処理を待つというコマンドが必要になります。

解決策２：shutdown(wait=True)を使う

並列処理を実行した後に「executor.shutdown(wait=True)」を実行するとその場で全ての並列処理が終わるのを待ってくれます。

つまり先ほどのwith構文を使わない例ならこんな感じです。

from concurrent.futures import ThreadPoolExecutor
import time

def number5():
    for i in range(5):
        print(f'number5: {i}')
        time.sleep(0.5)
        
def number10():
    for i in range(10):
        print(f'number10: {i}')
        time.sleep(0.5)
        
def main():
    executor = ThreadPoolExecutor(max_workers=2)
    executor.submit(number5)
    executor.submit(number10)
    
    executor.shutdown(wait=True)
        
    print('Executor finish')
    print('END')
    
if __name__ == '__main__':
    main()

実行結果
number5: 0
number10: 0
number5: 1
number10: 1
number5: 2number10: 2

number10: 3number5: 3

number5: 4number10: 4

number10: 5
number10: 6
number10: 7
number10: 8
number10: 9
Executor finish
END

「executor.shutdown(wait=True)」で全ての並列処理が終わるのを待っているので、「Executor finish」と「END」は並列処理が終わった後に処理されています。

もちろんこの「executor.shutdown(wait=True)」はwith構文でも使うことができます。

from concurrent.futures import ThreadPoolExecutor
import time

def number5():
    for i in range(5):
        print(f'number5: {i}')
        time.sleep(0.5)
        
def number10():
    for i in range(10):
        print(f'number10: {i}')
        time.sleep(0.5)
        
def main():
    with ThreadPoolExecutor(max_workers=2) as executor:
        executor.submit(number5)
        executor.submit(number10)
        executor.shutdown(wait=True)
        print('Executor finish')

    print('END')
    
if __name__ == '__main__':
    main()

実行結果
number5: 0
number10: 0
number10: 1
number5: 1
number10: 2number5: 2

number5: 3number10: 3

number10: 4number5: 4

number10: 5
number10: 6
number10: 7
number10: 8
number10: 9
Executor finish
END

個人的にはwith構文を使うのが好きなので、最後の例をよく使いますが、そこら辺は個人の好みでいいんじゃないかなと思います。

ということで並列処理concurrent.futuresのThreadPoolExecutorで複数の処理を実行した際、他の処理を待つ方法でした。

次回はnumpyの三角関数sin、cos、tanをいじっていきます。