CODEWITHSUNDEEP
pythonmultithreadingsignals
Alejo Dev
Alejo Dev

Reputation: 2576

Stop Python script with ctrl + c

I have a script which uses threads, but it is unable to catch Ctrl + C.

Here it is the sample code to reproduce this error:

import threading
import time
import signal

class DummyThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)
        self._running = True
        signal.signal(signal.SIGINT, self.stop)
        signal.signal(signal.SIGTERM, self.stop)

    def stop(self, signum=None, frame=None):
        self._running = False

    def run(self):
        while self._running:
            time.sleep(1)
            print("Running")

if __name__ == "__main__":
    try:
        t = DummyThread()
        t.start()
        while True:
            print("Main thread running")
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("This never gets printed")
        t.stop()
    finally:
        print("Exit")

When I run python3 script.py it starts running, but it does not catch ctrl+c. I have googled it but I have not found a solution. I have to kill the script with SIGTERM, but I want DummyThread to stop gracefully.

Upvotes: 1

Views: 4827

Answers (3)

mkrieger1
mkrieger1

Reputation: 23114

class DummyThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)
        self._running = True
        signal.signal(signal.SIGINT, self.stop)
        signal.signal(signal.SIGTERM, self.stop)

The program actually does not work as expected because of those last two lines and would work without them.

The reason is that, if you press Ctrl-C, the SIGINT signal is handled by the signal handler that is set up by signal.signal and self.stop is called. So the thread should actually stop.

But in the main thread, the while True loop is still running. Since the signal has already been handled, there will be no KeyboardInterrupt exception raised by the Python runtime. Therefore you never get to the except part.

if __name__ == "__main__":
    try:
        t = DummyThread()
        t.start()
        while True:            # you are stuck in this loop
            print("Main thread running")
            time.sleep(0.5)
    except KeyboardInterrupt:  # this never happens
        print("This never gets printed")
        t.stop()

Only one signal handler should be set up to call the stop method. So there are two options to solve the problem:

  1. Handle the signal implicitly by catching the KeyboardInterrupt exception. This is achieved by simply removing the two signal.signal(...) lines.

  2. Set up an explicit signal handler (as you did by using signal.signal in DummyThread.__init__), but remove the while True: loop from the main thread and do not try to handle KeyboardInterrupt. Instead, just wait for the DummyThread to finish on its own by using its join method:

    if __name__ == "__main__":
    t = DummyThread()
    t.start()
    t.join()
    print("Exit")

Upvotes: 3

Artiom Kozyrev
Artiom Kozyrev

Reputation: 3826

The main point is that you can't work with signals in any other Thread except the Main Thread. The Main Thread is the only one which can receive signals and handle them. I can offer the following solution, it is based on Event sync primitive.

According to Python documantation:

Signals and threads Python signal handlers are always executed in the main Python thread, even if the signal was received in another thread. This means that signals can’t be used as a means of inter-thread communication. You can use the synchronization primitives from the threading module instead.

Besides, only the main thread is allowed to set a new signal handler.

from threading import Thread, Event
import time


class DummyThread(Thread):

    def __init__(self, event: Event):
        Thread.__init__(self)
        self.stop_event = event

    def run(self):
        # we are monitoring the event in the Main Thread
        while not self.stop_event.is_set():  
            time.sleep(1)
            print("Running")
        # only Main Thread can make the point reachable
        print("I am done !")  


if __name__ == "__main__":
    try:
        e = Event() 
        t = DummyThread(e)
        t.start()
        while True:
            print("Main thread running")
            time.sleep(0.5)
    except KeyboardInterrupt:
        e.set()
    finally:
        print("Exit")

Another possible choice is to use daemon Thread for such tasks like in your code example (when you just printing smth in the screen every second, rather than e.g. close database connection or some similar task). If main thread is stoped the daemon Thread will stop too.

Upvotes: 2

ajinzrathod
ajinzrathod

Reputation: 940

As shown in your code, you used KeyboardInterrupt to call stop() function. See how Listener does the same task and stops the execution which was unable to catch from Ctrl + C. You dont have to kill the script with SIGTERM anymore

import threading
import time
import signal
import os
from pynput.keyboard import Key, Listener

class DummyThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self._running = True
        signal.signal(signal.SIGINT, self.stop)
        signal.signal(signal.SIGTERM, self.stop)

    def stop(self, signum=None, frame=None):
        self._running = False
        print ("Bye Bye . .")
        os._exit(1)

    def run(self):
        while self._running:
            time.sleep(1)
            print("Running")


if __name__ == "__main__":
    t = DummyThread()
    def func2():
        try:
            t.start()
            while True:
                print("Main thread running")
                time.sleep(0.5)
        except KeyboardInterrupt:
            print("No need for this")
            t.stop()
        finally:
            print("Exit")

    def func1():
        with Listener(on_press = t.stop) as listener :
            listener.join()

    threading.Thread(target=func1).start()
    threading.Thread(target=func2).start()

Upvotes: 0

Related Questions