Keyboard interrupts not honored by Python thread

Question

Short form question:

How do I get a keyboard interrupt to stop a Python thread?

Details:

Consider the following simple program:

import time

def test():
    while(True):
        print(time.asctime(), flush=True)
        time.sleep(3)

if __name__ == '__main__':
    test()

When I run it in a cmd window and type ^C, it terminates the program as expected:

>py test.py
Tue Jul 18 08:34:33 2023
Tue Jul 18 08:34:36 2023
Traceback (most recent call last):
  File "C:\Users\r\Projects\Pro1\git\pro1-pubsub\test.py", line 22, in <module>
    test()
  File "C:\Users\r\Projects\Pro1\git\pro1-pubsub\test.py", line 13, in test
    time.sleep(3)
KeyboardInterrupt
^C

But if the code is running in a thread, ^C has no effect -- it just keeps running and the only way I've found to stop it is to open the Task Manager and terminate the Python process:

import threading
import time

def test():
    threading.Thread(target=listen_thread).start()

def listen_thread():
    print("listen_thread", flush=True)
    while(True):
        print(time.asctime(), flush=True)
        time.sleep(3)

if __name__ == '__main__':
    test()

I also tried setting a signal handler like this, but signal_handler() never gets called:

import signal
import sys
import threading
import time

runnable = True;

def test():
    threading.Thread(target=listen_thread).start()

def listen_thread():
    print("listen_thread starting", flush=True)
    while(runnable):
        print(time.asctime(), flush=True)
        time.sleep(3)
    print("listen_thread stopping", flush=True)

def signal_handler(signal, frame):
    global runnable
    print("setting runnable to False", flush=True)
    runnable = False

if __name__ == '__main__':
    signal.signal(signal.SIGINT, signal_handler)
    test()

The question (redux):

How do I get a keyboard interrupt to stop the thread?

Answer 1

You Should create a shared variable to be accessed by the main and the child thread. The child thread checks the value of the share variable at each iteration of the loop to determine whether it should continue executing. And the variable can be changed to False by the main thread if a keyboard interrupt occurs in the try-except.

import threading
import time

is_running = True

def test():
    thread = threading.Thread(target=listen_thread)
    thread.start()

    # Wait for keyboard interrupt 
    try:
        while thread.is_alive():
            time.sleep(1)
    except KeyboardInterrupt:
        print("Keyboard interrupt detected. Stopping the thread...")
        global is_running
        is_running = False
        thread.join()
        print("Thread stopped.")

def listen_thread():
    print("listen_thread starting")
    while is_running:
        print(time.asctime())
        time.sleep(3)
    print("listen_thread stopping")

if __name__ == '__main__':
    test()