Using signals to terminate a killable thread.

signal_stop.py

import time
import threading
import signal

class Job(threading.Thread):
    def __init__(self):
        super().__init__()
        '''
        The shutdown_flag is a threading.Event object that
        indicates whether the thread should be terminated.
        '''
        self.shutdown_flag = threading.Event()

    def run(self):
        print(f'Thread #{self.ident} started')
        
        while not self.shutdown_flag.is_set():
            time.sleep(0.5)

        print(f'Thread #{self.ident} stopped')

class ServiceExit(Exception):
    """
    Custom exception which is used to trigger the clean exit
    """
    pass

def service_shutdown(signum, frame):
    print(f'Caught signal {signum}')
    raise ServiceExit

def main():
    signal.signal(signal.SIGTERM, service_shutdown)
    signal.signal(signal.SIGINT, service_shutdown)

    print('Starting main program')

    # Start the job threads
    try:
        j1 = Job()
        j2 = Job()
        j1.start()
        j2.start()

        # Keep the main thread running, otherwise signals are ignored.
        while True:
            time.sleep(0.5)
    except ServiceExit:
        """
        Terminate the running threads.
        Set the shutdown flag on each thread to trigger a clean shutdown of each thread.
        """
        j1.shutdown_flag.set()
        j2.shutdown_flag.set()
        # Wait for threads to close...
        j1.join()
        j2.join()

    print('Exit main program')

if __name__ == "__main__":
    main()