1
1

More than 1 year has passed since last update.

Python 複数スレッドをEvent制御してるときCtrl+Cが効かなくなった時のテクニック

Last updated at Posted at 2022-03-23

すべてのスレッドでevent.waitしている場合、各スレッドがロックされるためctrl+cが受け取れない。
そのため、event.waitを使わないスレッドを必ず1つ用意しておく。このスレッドでctrl+cを受け取り終了フラグをTrueにして子スレッドの条件に終了フラグを配置していおけばループから抜けることができる。

import time
import threading

e_task1 = threading.Event()
e_task2 = threading.Event()

f_sys_exit = False


# ===============================
# 制御不可能なchildスレッドの処理
# ===============================
def can_not_control_task1():
    while True:
        print(threading.current_thread().name, ' is waitting.')
        e_task1.wait()
        print(threading.current_thread().name, ' end.')


def cant_not_control_task2():
    while True:
        print(threading.current_thread().name, ' is waitting.')
        e_task2.wait()
        print(threading.current_thread().name, ' end.')


# ===============================
# 制御可能なchildスレッドの処理
# ===============================
def can_control_task1():
    while True:
        if f_sys_exit:
            break
        print(threading.current_thread().name, ' is waitting.')
        e_task1.wait()
        print(threading.current_thread().name, ' end.')


def can_control_task2():
    while True:
        if f_sys_exit:
            break
        print(threading.current_thread().name, ' is waitting.')
        e_task2.wait()
        print(threading.current_thread().name, ' end.')


# ===============================
# childスレッドの実行処理
# ===============================
def run_can_not_control_child_thread():
    """コントール不可能"""
    ths = []
    for task in [can_not_control_task1, cant_not_control_task2]:
        th = threading.Thread(target=task, name=task.__name__, daemon=False)
        th.start()
        ths.append(th)
    for th in ths:
        th.join()


def run_can_control_child_thread():
    """コントール可能"""
    ths = []
    for task in [can_control_task1, can_control_task2]:
        th = threading.Thread(target=task, name=task.__name__, daemon=False)
        th.start()
        ths.append(th)
    for th in ths:
        th.join()


# ===============================
# メインスレッドの実行処理
# ===============================
def can_not_control():
    """コントール不可能"""
    global f_sys_exit
    threading.Thread(target=run_can_not_control_child_thread, daemon=True).start()
    while True:
        try:
            time.sleep(1)
            print('parent thread is running.')
        except KeyboardInterrupt:
            print('pressed to Ctrl + C keys.')
            break


def can_control():
    """コントール可能"""
    global f_sys_exit
    threading.Thread(target=run_can_control_child_thread, daemon=True).start()
    while True:
        try:
            time.sleep(1)
            print('parent thread is running.')
        except KeyboardInterrupt:
            print('pressed to Ctrl + C keys.')
            # sys_exitフラグをTrueにしてから 
            f_sys_exit = True
            # すべてのwait中のスレッドをsetする
            set_all_event()
            break


# ===============================
# イベント処理
# ===============================
def set_all_event():
    """すべてのイベントをset"""
    global e_task1, e_task2
    e_task1.set()
    e_task2.set()


if __name__ == '__main__':
    # can_not_control()
    can_control()

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
1
1