Created
June 19, 2017 03:03
-
-
Save rongyi/a3b94e0629b810de7b1c24e069abb8e6 to your computer and use it in GitHub Desktop.
This file contains hidden or 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/env python | |
# -*- coding: utf-8 -*- | |
'''For running command line executables with a timeout''' | |
import subprocess | |
import threading | |
import shlex | |
import sys | |
import traceback | |
class TimedProc(object): | |
''' | |
Create a TimedProc object, calls subprocess. | |
''' | |
def __init__(self, cmd): | |
self.command = cmd | |
cmdargs = shlex.split(cmd) | |
self.process = subprocess.Popen(cmdargs, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | |
self.stdout = "" | |
self.stderr = "" | |
def wait(self, timeout=None): | |
''' | |
wait for subprocess to terminate and return subprocess' return code. | |
If timeout is reached, throw TimedProcTimeoutError | |
''' | |
def receive(): | |
(self.stdout, self.stderr) = self.process.communicate() | |
if timeout: | |
if not isinstance(timeout, (int, float)): | |
raise Exception('Error: timeout must be a number') | |
rt = threading.Thread(target=receive) | |
rt.start() | |
rt.join(timeout) | |
if rt.isAlive(): | |
# Subprocess cleanup (best effort) | |
self.process.kill() | |
def terminate(): | |
if rt.isAlive(): | |
self.process.terminate() | |
# quit after 1 seconds | |
threading.Timer(1, terminate).start() | |
raise Exception( | |
'%s : Timed out after %s seconds' % ( | |
self.command, | |
str(timeout), | |
) | |
) | |
else: | |
receive() | |
return self.process.returncode | |
def run_cmd(cmd, timeout=5): | |
if cmd is None: | |
return (-1, '', 'command should not be empty') | |
tci = TimedProc(cmd) | |
try: | |
ret = tci.wait(timeout) | |
if ret == 0: | |
return (0, tci.stdout, '') | |
else: | |
sys.stderr.write(tci.stderr) | |
return (-1, '', tci.stderr) | |
except Exception, e: | |
return (-2, '', traceback.format_exc(e)) | |
if __name__ == '__main__': | |
rc, stdout, stderr = run_cmd('sleep 7') | |
if rc == 0: | |
print(stdout) | |
else: | |
print(stderr) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment