Python: Issue stopping thread with keyboard interrupts via atexit

Question

As per implementation of StoppableThreads in Python - How can I implement a 'stoppable' thread?, I tried to adapt the same to handle Keyboard interrupts via atexit. Refer implementation below:

import threading, atexit


class ThreadStoppable(threading.Thread):
    def __init__(self, *args, **kwargs):
        threading.Thread.__init__(self, *args, **kwargs)
        self._stop_event = threading.Event()
        
        ######## AT_EXIT HANDLING
        atexit.register(self._stop_event.set)

    def stop(self):
        """
        stop handler - handles raising of stop event
                    any required pre and post associated logic could be extended
                    by overriding definition in child class for more handling if required
        Returns:
        None
        """
        print('stopped Yahooooooooooooooooooooooooooooooooooooooooooooooooooo')
        self._stop_event.set()

    def run(self):
        try:
            if self._target:
                self._target(*self._args, **self._kwargs, stop_event=self._stop_event)
        finally:
            # Avoid a refcycle if the thread is running a function with
            # an argument that has a member that points to the thread.
            del self._target, self._args, self._kwargs

    def is_stopped(self):
        """
        checks whether thread is topped
                    any required pre and post associated logic could be extended
                    by overriding definition in child class for more handling if required
        Returns:
        <bool> True if thread is stopped else False
        """
        return self._stop_event.is_set()


def timed_output(name, delay, run_event_, **kwargs):
    i = 0
    while run_event_.is_set() and not kwargs.get('stop_event').is_set():
        i += 1
        time.sleep(delay)
        print(name, ": New Message!")
        print('i is', i)
        if kwargs.get('max', 10) < i:
            print('breaking')
            break


def func_test(stop=False):
    print('Func Test with stop' if stop else 'Func Test without stop')
    run_event = threading.Event()
    run_event.set()
    d1 = 1
    t1 = ThreadStoppable(target=timed_output, args=("bob", d1, run_event))
    t1.start()
    if stop:
        t1.stop()
    print('joining')
    t1.join()
    print('joined')


if __name__ == '__main__':
    func_test(stop=True)
    func_test()

The stop is triggerred when t1.stop() is called but atexit.register(self._stop_event.set) does not trigger t1.stop()

The thread continues executing despite ctrl+c. Following is the output:

    C:\>python Thread_Stopper.py
Func Test with stop
stopped Yahooooooooooooooooooooooooooooooooooooooooooooooooooo
joining
bob : New Message!
i is 1
joined
Func Test without stop
joining
bob : New Message!
i is 1
bob : New Message!
i is 2
bob : New Message!
i is 3
bob : New Message!
i is 4
bob : New Message!
i is 5
bob : New Message!
i is 6
bob : New Message!
i is 7
bob : New Message!
i is 8
bob : New Message!
i is 9
bob : New Message!
i is 10
bob : New Message!
i is 11
breaking
Traceback (most recent call last):
  File "C:\Thread_Stopper.py", line 71, in <module>
    func_test()
  File "C:\Thread_Stopper.py", line 65, in func_test
  File "C:\Users\admin\AppData\Local\Programs\Python\Python39\lib\threading.py", line 1053, in join
    self._wait_for_tstate_lock()
  File "C:\Users\admin\AppData\Local\Programs\Python\Python39\lib\threading.py", line 1069, in _wait_for_tstate_lock
    elif lock.acquire(block, timeout):
KeyboardInterru

pt

Python: Issue stopping thread with keyboard interrupts via atexit

0 Answers0