Skip to main content

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()