Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/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
You can’t perform that action at this time.