Using select on a unix command in lieu of signal

[Jp Calderone]

On 29 Aug 2005 17:57:34 -0700, rh0dium <sklass at pointcircle.com> wrote:
>So here's how I solved this..  It's seems crude - but hey it works.
>select not needed..
>
>    def runCmd( self, cmd, timeout=None ):
>        self.logger.debug("Initializing function %s - %s" %
>(sys._getframe().f_code.co_name,cmd) )
>        command = cmd + "\n"
>
>        child = popen2.Popen3(command)
>        t0 = time.time()
>
>        out = None
>        while time.time() - t0 < timeout:
>            if child.poll() != -1:
>                self.logger.debug("Command %s completed succesfully" %
>cmd )
>                out = child.poll()
>                results = "".join(child.fromchild.readlines())
>                results = results.rstrip()
>                break
>                print "Still waiting..", child.poll(), time.time() -
>t0, t0
>            time.sleep(.5)
>
>        if out == None:
>            self.logger.warning( "Command: %s failed!" % cmd)
>            kill = os.kill(child.pid,9)
>            self.logger.debug( "Killing command %s - Result: %s" %
>(cmd, kill))
>            out = results = None
>
>        else:
>
>            self.logger.debug("Exit: %s Reullts: %s" % (out,results))
>
>        child.tochild.close()
>        child.fromchild.close()
>        return out,results
>
>Comments..
>

Here's how I'd do it...


from twisted.internet import reactor, protocol, error

class PrematureTermination(Exception):
    """Indicates the process exited abnormally, either by receiving an 
    unhandled signal or with a non-zero exit code.
    """

class TimeoutOutputProcessProtocol(protocol.ProcessProtocol):
    timeoutCall = None
    onCompletion = None

    def __init__(self, onCompletion, timeout=None):
        # Take a Deferred which we will use to signal completion (successful 
        # or otherwise), as well as an optional timeout, which is the maximum 
        # number of seconds (may include a fractional part) for which we will 
        # await the process' completion.
        self.onCompletion = onCompletion
        self.timeout = timeout

    def connectionMade(self):
        # The child process has been created.  Set up a buffer for its output,
        # as well as a timer if we were given a timeout.
        self.output = []
        if self.timeout is not None:
            self.timeoutCall = reactor.callLater(
                self.timeout, self._terminate)

    def outReceived(self, data):
        # Record some data from the child process.  This will be called 
        # repeatedly, possibly with a large amount of data, so we use a list 
        # to accumulate the results to avoid quadratic string-concatenation 
        # behavior.  If desired, this method could also extend the timeout: 
        # since it is producing output, the child process is clearly not hung;
        # for some applications it may make sense to give it some leeway in 
        # this case.  If we wanted to do this, we'd add lines to this effect:
        #     if self.timeoutCall is not None:
        #         self.timeoutCall.delay(someNumberOfSeconds)
        self.output.append(data)

    def _terminate(self):
        # Callback set up in connectionMade - if we get here, we've run out of 
        # time.  Error-back the waiting Deferred with a TimeoutError including
        # the output we've received so far (in case the application can still 
        # make use of it somehow) and kill the child process (rather 
        # forcefully - a nicer implementation might want to start with a 
        # gentler signal and set up another timeout to try again with KILL).
        self.timeoutCall = None
        self.onCompletion.errback(error.TimeoutError(''.join(self.output)))
        self.onCompletion = None
        self.output = None
        self.transport.signalProcess('KILL')

    def processEnded(self, reason):
        # Callback indicating the child process has exited.  If the timeout 
        # has not expired and the process exited normally, callback the 
        # waiting Deferred with all our results.  If we did time out, nothing 
        # more needs to be done here since the Deferred has already been 
        # errored-back.  If we exited abnormally, error-back the Deferred in a
        # different way indicating this.
        if self.onCompletion is not None:
            # We didn't time out
            self.timeoutCall.cancel()
            self.timeoutCall = None
            
            if reason.check(error.ProcessTerminated):
                # The child exited abnormally
                self.onCompletion.errback(
                    PrematureTermination(reason, ''.join(self.output)))
            else:
                # Success!  Pass on our output.
                self.onCompletion.callback(''.join(self.output)))

            # Misc. cleanup
            self.onCompletion = None
            self.output = None

def runCmd(executable, args, timeout=None, **kw):
    d = defer.Deferred()
    p = TimeoutOutputProcessProtocol(d, timeout)
    reactor.spawnProcess(p, executable, args, **kw)
    return d

And there you have it.  It's a bit longer, but that's mostly due to the comments.  The runCmd function has a slightly different signature too, since spawnProcess can control a few more things than Popen3, so it makes sense to make those features available (these include setting up the child's environment variables, the UID and GID it will run as, whether or not to allocate a PTY for it, and the working directory it is given).  The return value differs too, of course: it's a Deferred instead of a two-tuple, but it will eventually fire with roughly the same information.

Hope this helps,

Jp