Extends the Python Popen class (Popen execute a child program in a new process) with simple method decorators to support timeout operations.
"""
Simple Popen extension to simplify timeout dealings.
Extends the Popen class (Popen execute a child program in a new process)
with simple method decorators to support timeout operations.
Example:
# Call the `sleep` cmd for three seconds, while specifying a ExpirablePopen
# timeout of 2 seconds to test:
ExpirablePopen(['/bin/sleep'] + ['3'], stdout=PIPE, timeout=2).communicate()
ERROR:root:Terminating process id 7266 after timeout limit reached (2 secs).
Inspired by Stack Overflow posting: https://stackoverflow.com/a/41222436
"""
from __future__ import print_function
from logging import error
from subprocess import Popen
from threading import Event
from threading import Thread
class ExpirablePopen(Popen):
def __init__(self, *args, **kwargs):
self.timeout = kwargs.pop('timeout', 0)
self.timer = None
self.done = Event()
Popen.__init__(self, *args, **kwargs)
def __tkill(self):
timeout = self.timeout
if not self.done.wait(timeout):
error("Terminating process id {0} after timeout limit reached ({1} secs).".format(self.pid, timeout))
self.kill()
def expirable(func):
def wrapper(self, *args, **kwargs):
# zero timeout means call of parent method
if self.timeout == 0:
return func(self, *args, **kwargs)
# if timer is None, need to start it
if self.timer is None:
self.timer = thr = Thread(target=self.__tkill)
thr.daemon = True
thr.start()
result = func(self, *args, **kwargs)
self.done.set()
return result
return wrapper
wait = expirable(Popen.wait)
communicate = expirable(Popen.communicate)
if __name__ == '__main__':
from subprocess import PIPE
# Example: call the `sleep` cmd for three seconds,
# while specifying a ExpirablePopen timeout of 2 seconds to test:
ExpirablePopen(['/bin/sleep'] + ['3'], stdout=PIPE, timeout=2).communicate()