作成日: 2023.04.27

Python で Thread を使っているプログラムを Ctrl + C で終了させる

普通、Pythonのスクリプトは何もしなくてもCtrl+Cとかで死にますが、Thread を立ち上げて呼び出し元がjoinで待機している時とかだと終了しません。

理由は呼び出し元がjoinで固まっていて、シグナルを受け取らない（正確には受け取っているけどjoinが終わるまでその処理を実行しない）ので、終了できないのです。

いくつか解決方法がありますが、joinをやめてthreadをdaemon化する方法を紹介します。

とりあえずCtrl + C で強制終了してみる

そもそもスレッドを使っていないプログラムの実行中にCtrl+Cしたらどうなるか見てみましょう。

別にスレッド立ち上げて join してなければ、普通にCtrl+C でプログラムは終了します。

例えば以下のコード(main.py)を、

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import time

def main():
    while True:
        print("main", flush=True)
        time.sleep(1)

if __name__ == "__main__":
    main()

実行して Ctrl + c すると、

PS >python .\main.py
main
main
main
Traceback (most recent call last):
  File "main.py", line 12, in <module>
    main()
  File "main.py", line 9, in main
    time.sleep(1)
KeyboardInterrupt

といった感じで終了します。

Thread 使わない普通のプログラムならこんなもんです。

ただ、例外メッセージがビヨッと出てちょっと良くない感じですね。本当に強制終了、という感じで。

SIGINT シグナルを拾って終了する

ブツ切りじゃなく、ちゃんとCtrl + c が押されたのを確認して終了したいですね。

まず１つには、SIGINT (Ctrl + c が押されたときに飛んでくるシグナル）を受信した時に事前に登録していた関数をコールバックさせて処理をする方法があります。

先ほどのソースの上のほうに、

import signal
import sys

def signal_handler(signum, frame):
    print("SIGINT")
    sys.exit()

signal.signal(signal.SIGINT, signal_handler)

を追記してください。完全なコードは以下です。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import signal
import sys
import time

def signal_handler(signum, frame):
    print("SIGINT")
    sys.exit()

signal.signal(signal.SIGINT, signal_handler)

def main():
    while True:
        print("main", flush=True)
        time.sleep(1)

if __name__ == "__main__":
    main()

実行してCtrl＋Cをしてみてください。

PS >python .\main.py
main
main
main
SIGINT

Ctrl + c が押されたときに飛んできたSIGINTが受信したタイミングで事前登録しておいたsignal_handler関数がコールバックされて exit() で明示的に終了できました。

余談ですが、ここで exit()しない という選択もあります。
どういうことかというと、実行中にCtrl + cが押されても終了させたく「ない」場合に、受け取ったシグナルを「無視」することで終了できなくさせることができる、ということです。

try except KeyboardInterrupt で 終了する

シグナルを直接拾ってコールバック関数で対応するのは簡単ですが、なんかコードがCっぽいというか粗野な感じですね。

ので、 try except する方法も述べます。
main()内で、

try:

    処理

except KeyboardInterrupt:
    exit()

とする感じです。

ソースを書き換えると（さっきのimport signalとかはもういらないので削除しましょう）

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import time

def main():
    try:
        while True:
            print("main", flush=True)
            time.sleep(1)
    except KeyboardInterrupt:
        print("except KeyboardInterrupt")
        sys.exit()
if __name__ == "__main__":
    main()

な感じで、実行してCtrl+cすると、

PS > python .\main.py
main
main
main
except KeyboardInterrupt

と、期待通りです。

まぁ、Threadを立ち上げてjoinしていなければCtrl+Cで殺したりシグナル受信したりは容易、ということの確認作業でした。

ThreadのコードでCtrl+Cを失敗してみる

では、とりあえず失敗してみましょう。

以下のコードを実行すると、

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import time
import threading

def th_func():
    while True:
        print("thread", flush=True)
        time.sleep(1)

def main():
    try:
        th = threading.Thread(target=th_func)
        th.start()
        th.join()
    except KeyboardInterrupt:
        print("except KeyboardInterrupt")
        sys.exit()

if __name__ == "__main__":
    main()

PS > python .\main.py
thread
thread
thread
...

Ctrl+cしても止めることができません。

...困りましたね。

とりあえず、プロセスを検索して直接killしてしまいましょう。

以下のコマンドをPowerShellで叩き(ここではmain.pyというプログラムだったとしています) 、

PS >WMIC path win32_process get Caption,Processid,Commandline /format:csv | sls "main.py" |  % {$_.Line.Split(",")[3]}
32888

実施中のPythonのプログラムのプロセスIDが32888と分かったので、

PS >Taskkill /PID 32888 /f
成功: PID 32888 のプロセスは強制終了されました。

で殺しましょう。

さてどうしましょうか？

joinを使わない

結局、joinを使ったせいで問題になってるわけですね。
なので、joinを使わないようにしましょう。

is_alive()でスレッドが生きているかどうかをwhileの無限ループで確認することでjoinの代替とします。
これでmain側にも制御が回ってくるので、Ctrl+cイベントがちゃんと拾えるようになるわけです。

ではソースコードのth.joinのところを以下のように変えて、

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import time
import threading

def th_func():
    while True:
        print("thread", flush=True)
        time.sleep(1)

def main():
    try:
        th = threading.Thread(target=th_func)
        th.start()
        while th.is_alive() :
            print("main", flush=True)
            time.sleep(1)
    except KeyboardInterrupt:
        print("except KeyboardInterrupt")
        sys.exit()

if __name__ == "__main__":
    main()

実行して、Ctrl+cしてみましょう。

PS > python .\main.py
thread
main
thread
main
thread
main
except KeyboardInterrupt
thread
thread
thread
thread
thread
thread
...

あれ？mainは消えましたがthreadが続いてしまいます。

うまくいきませんでしたね（笑）

そうなんです。mainスレッドは死にましたが、立ち上げたスレッドが生き続けてしまうんですね。

ではどうしたらよいでしょうか？

いろいろ方法はあるのですが、シンプルなのはスレッドのデーモン化です。

threadをdaemon化するとメインスレッドが死んだときに一緒に死んでくれるようになるんですね。ある意味悪魔です(笑）

と、いうことで、以下の th.daemon = True を追加した最終ソースで確認してみましょう。
これはth.start()より先に書かないといけないので注意しましょう。
なお、ググるとよく出てくる古い記事の、 th.setDaemon(True) という指定のしかたは現在は deprecated なので注意しましょう。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import time
import threading

def th_func():
    while True:
        print("thread", flush=True)
        time.sleep(1)

def main():
    try:
        th = threading.Thread(target=th_func)
        th.daemon= True
        th.start()
        while th.is_alive() :
            print("main", flush=True)
            time.sleep(1)
    except KeyboardInterrupt:
        print("except KeyboardInterrupt")
        sys.exit()

if __name__ == "__main__":
    main()

実行して、Ctrl+Cすると、

PS > python .\main.py
thread
main
thread
main
thread
main
except KeyboardInterrupt

ばっちりです！

いやぁ大変でしたね。ただ単にCtrl+Cでプログラム停めたいだけだったのに。

マルチプロセスで対応する

main以外のプロセスをos.forkして追加し、そっちでCtrl+Cを受け取って終了処理をする、という方法もあります。
ただ、これは、Windows だと os.fork 機能がないのでできません。

Windowsでもマルチプロセスで対応したいのであれば multiprocessing を使えば可能です。
が、なんかだんだん面倒になってきましたし、別に上記の方法で困らないので上記の方法にします。

以上。