Using module 'subprocess' with timeout
I don't know much about the low level details; but, given that inpython 2.6 the API offers the ability to wait for threads andterminate processes, what about running the process in a separatethread?
import subprocess, threadingclass Command(object): def __init__(self, cmd): self.cmd = cmd self.process = None def run(self, timeout): def target(): print 'Thread started' self.process = subprocess.Popen(self.cmd, shell=True) self.process.communicate() print 'Thread finished' thread = threading.Thread(target=target) thread.start() thread.join(timeout) if thread.is_alive(): print 'Terminating process' self.process.terminate() thread.join() print self.process.returncodecommand = Command("echo 'Process started'; sleep 2; echo 'Process finished'")command.run(timeout=3)command.run(timeout=1)
The output of this snippet in my machine is:
Thread startedProcess startedProcess finishedThread finished0Thread startedProcess startedTerminating processThread finished-15
where it can be seen that, in the first execution, the processfinished correctly (return code 0), while the in the second one theprocess was terminated (return code -15).
I haven't tested in windows; but, aside from updating the examplecommand, I think it should work since I haven't found in thedocumentation anything that says that thread.join or process.terminateis not supported.
In Python 3.3+:
from subprocess import STDOUT, check_outputoutput = check_output(cmd, stderr=STDOUT, timeout=seconds)
output is a byte string that contains command's merged stdout, stderr data.
CalledProcessError on non-zero exit status as specified in the question's text unlike
shell=True because it is often used unnecessarily. You can always add it back if
cmd indeed requires it. If you add
shell=True i.e., if the child process spawns its own descendants;
check_output() can return much later than the timeout indicates, see Subprocess timeout failure.
The timeout feature is available on Python 2.x via the
subprocess32 backport of the 3.2+ subprocess module.
jcollado's answer can be simplified using the threading.Timer class:
import shlexfrom subprocess import Popen, PIPEfrom threading import Timerdef run(cmd, timeout_sec): proc = Popen(shlex.split(cmd), stdout=PIPE, stderr=PIPE) timer = Timer(timeout_sec, proc.kill) try: timer.start() stdout, stderr = proc.communicate() finally: timer.cancel()# Examples: both take 1 secondrun("sleep 1", 5) # process ends normally at 1 secondrun("sleep 5", 1) # timeout happens at 1 second