Created
November 27, 2011 19:00
-
-
Save NelsonMinar/1397988 to your computer and use it in GitHub Desktop.
Python spawnv() with timeout
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python | |
"""Simple module to implement spawnv with a timeout to kill the child. | |
Only tested on Unix. Probably won't play well with threads or many | |
child processes. | |
You can do much more subtle things with process management, but this is | |
easy and useful. | |
By Nelson Minar <nelson@monkey.org> 2004-04-11 | |
""" | |
import os, signal, time | |
class Timeout (Exception): pass | |
class Error (Exception): pass | |
def _alarmHandler(signum, frame): | |
"Handle an alarm for timing out a child process" | |
if signum == signal.SIGALRM: | |
raise Timeout() | |
else: | |
raise Error() | |
def spawnvTimeout(path, args, timeout): | |
"""spawnv a process with a timeout (in seconds). | |
If the process exits before the timeout the exit code is returned | |
with the semantics of os.wait(). | |
If the timeout fires before the process exits, a Timeout exception | |
is raised and the process is sent a SIGTERM | |
For example, to launch a 'sleep 5' but time out after 2 seconds: | |
spawnvTimeout('/bin/sleep', ['sleep', '5'], 2) | |
""" | |
# Install a signal handler to implement the timeout | |
oldHandler = signal.signal(signal.SIGALRM, _alarmHandler) | |
try: | |
# Set the timeout | |
signal.alarm(timeout) | |
# Start the process, wait for it to finish, then clear pid flag | |
pid = os.spawnv(os.P_NOWAIT, path, args) | |
exitcode = os.waitpid(pid, 0)[1] | |
pid = None | |
# Return the code | |
return exitcode | |
finally: | |
# Remove our alarm and reinstall the old signal handler | |
signal.alarm(0) | |
signal.signal(signal.SIGALRM, oldHandler) | |
# If the process is still running (pid is cleared above), kill it | |
if pid: | |
os.kill(pid, signal.SIGTERM) | |
# Just a little demo of how this works. Really should be a unit test. | |
if __name__ == '__main__': | |
def demo(sleep, timeout): | |
print "\nSleeping for %d seconds, timing out in %d" % (sleep, timeout) | |
now = time.time() | |
try: | |
exitcode = spawnvTimeout('/bin/sleep', | |
['sleep', str(sleep)], | |
timeout) | |
print "Finished with exit code %d" % exitcode | |
except Timeout: | |
print "Timed out" | |
print "Execution took %.1f seconds" % (time.time() - now) | |
print "Running demo of spawnvTimeout()" | |
demo(10, 1) | |
demo(2, 3) | |
demo(3, 10) | |
demo(2, 1) | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment