CODEWITHSUNDEEP
pythonwindowsmultithreadingwindows-7python-3.x
Navin
Navin

Reputation: 4107

Catch Keyboard Interrupt in program that is waiting on an Event

The following program hangs the terminal such that it ignores Ctrl+C. This is rather annoying since I have to restart the terminal every time one of the threads hang.

Is there any way to catch the KeyboardInterrupt while waiting on an event?

import threading
def main():
    finished_event = threading.Event()
    startThread(finished_event)
    finished_event.wait()#I want to stop the program here
    print('done!')
def startThread(evt):
    """Start a thread that will trigger evt when it is done"""
    #evt.set()
if __name__ == '__main__':
    main()

Upvotes: 7

Views: 12399

Answers (4)

Jeronimo
Jeronimo

Reputation: 2387

Based on @Pete's answer, but with subclassing and using the actual Event.wait method, just with smaller timeouts to allow handling of KeyboardInterrupts and such in between:

class InterruptableEvent(threading.Event):
    def wait(self, timeout=None):
        wait = super().wait  # get once, use often
        if timeout is None:            
            while not wait(0.01):  pass
        else:
            wait(timeout)

Upvotes: 1

jfs
jfs

Reputation: 414179

Update: On the current Python 3 finished_event.wait() works on my Ubuntu machine (starting with Python 3.2). You don't need to specify the timeout parameter, to interrupt it using Ctrl+C. You need to pass the timeout parameter on CPython 2.

Here's a complete code example:

#!/usr/bin/env python3
import threading

def f(event):
    while True:
        pass
    # never reached, otherwise event.set() would be here

event = threading.Event()
threading.Thread(target=f, args=[event], daemon=True).start()
try:
    print('Press Ctrl+C to exit')
    event.wait()
except KeyboardInterrupt:
    print('got Ctrl+C')

There could be bugs related to Ctrl+C. Test whether it works in your environment.

Old polling answer:

You could try to allow the interpreter to run the main thread:

while not finished_event.wait(.1): # timeout in seconds
    pass

If you just want to wait until the child thread is done:

while thread.is_alive():
    thread.join(.1)

Upvotes: 4

Pieter van den Ham
Pieter van den Ham

Reputation: 4484

You could also patch the Event.wait() function in the following manner:

def InterruptableEvent():
    e = threading.Event()

    def patched_wait():
        while not e.is_set():
            e._wait(3)

    e._wait = e.wait
    e.wait = patched_wait
    return e


>>> event = InterruptableEvent()
>>> try:
...     event.wait()
... except KeyboardInterrupt:
...     print "Received KeyboardInterrupt"
... 
^CReceived KeyboardInterrupt

This works because wait() with a timeout argument will raise a KeyboardInterrupt.

Upvotes: 1

GingerNinja23
GingerNinja23

Reputation: 162

If you want to avoid polling, you can use the pause() function of the signal module instead of finished_event.wait(). signal.pause() is a blocking function and gets unblocked when a signal is received by the process. In this case, when ^C is pressed, SIGINT signal unblocks the function. Note that the function does not work on Windows according to the documentation. I've tried it on Linux and it worked for me.

I came across this solution in this SO thread.

Upvotes: 5

Related Questions