Multithreaded Python programs often ignore the SIGINT generated by a Keyboard Interrupt, especially if the thread that gets the signal is waiting or sleeping. This module provides a workaround by forking a child process that executes the rest of the program while the parent process waits for signals and kills the child process.
| Python |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | import threading, time, os, signal, sys, operator
class MyThread(threading.Thread):
"""this is a wrapper for threading.Thread that improves
the syntax for creating and starting threads.
"""
def __init__(self, target, *args):
threading.Thread.__init__(self, target=target, args=args)
self.start()
class Watcher:
"""this class solves two problems with multithreaded
programs in Python, (1) a signal might be delivered
to any thread (which is just a malfeature) and (2) if
the thread that gets the signal is waiting, the signal
is ignored (which is a bug).
The watcher is a concurrent process (not thread) that
waits for a signal and the process that contains the
threads. See Appendix A of The Little Book of Semaphores.
http://greenteapress.com/semaphores/
I have only tested this on Linux. I would expect it to
work on the Macintosh and not work on Windows.
"""
def __init__(self):
""" Creates a child thread, which returns. The parent
thread waits for a KeyboardInterrupt and then kills
the child thread.
"""
self.child = os.fork()
if self.child == 0:
return
else:
self.watch()
def watch(self):
try:
os.wait()
except KeyboardInterrupt:
# I put the capital B in KeyBoardInterrupt so I can
# tell when the Watcher gets the SIGINT
print 'KeyBoardInterrupt'
self.kill()
sys.exit()
def kill(self):
try:
os.kill(self.child, signal.SIGKILL)
except OSError: pass
def counter(xs, delay=1):
"""print the elements of xs, waiting delay seconds in between"""
for x in xs:
print x
time.sleep(delay)
def main(script, flag='with'):
"""This example runs two threads that print a sequence, sleeping
one second between each. If you run it with no command-line args,
or with the argument 'with', you should be able it interrupt it
with Control-C.
If you run it with the command-line argument 'without', and press
Control-C, you will probably get a traceback from the main thread,
but the child thread will run to completion, and then print a
traceback, no matter how many times you try to interrupt.
"""
if flag == 'with':
Watcher()
elif flag != 'without':
print 'unrecognized flag: ' + flag
sys.exit()
t = range(1, 10)
# create a child thread that runs counter
MyThread(counter, t)
# run counter in the parent thread
counter(t)
if __name__ == '__main__':
main(*sys.argv)
|
Discussion
The comments in the code explain the details. This workaround is also presented in Appendix A of The Little Book of Semaphores which is available for download from http://greenteapress.com/semaphores
This bug has been discussed here:
http://groups.google.ca/group/comp.lang.python/msg/30205fd38b590685
and here:
http://mail.python.org/pipermail/python-bugs-list/2005-March/028189.html
I have only tested this on Linux. I would expect it to work on the Macintosh and not work on Windows.


Comments
The script seems to work just fine on MacOS X. Here is an example
/Jean Brouwers
PS) This is the ActiveState Python 2.4.3 build 11 for MacOS X PPC running on MacOS X 10.3.9.
Sign in to comment