nova/nova/process.py

# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# Copyright 2010 FathomDB Inc.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""
Process pool using twisted threading
"""

import logging
import StringIO

from twisted.internet import defer
from twisted.internet import error
from twisted.internet import protocol
from twisted.internet import reactor

from nova import flags
from nova.exception import ProcessExecutionError

FLAGS = flags.FLAGS
flags.DEFINE_integer('process_pool_size', 4,
                     'Number of processes to use in the process pool')


# This is based on _BackRelay from twister.internal.utils, but modified to
#  capture both stdout and stderr, without odd stderr handling, and also to
#  handle stdin
class BackRelayWithInput(protocol.ProcessProtocol):
    """
    Trivial protocol for communicating with a process and turning its output
    into the result of a L{Deferred}.

    @ivar deferred: A L{Deferred} which will be called back with all of stdout
        and all of stderr as well (as a tuple).  C{terminate_on_stderr} is true
        and any bytes are received over stderr, this will fire with an
        L{_ProcessExecutionError} instance and the attribute will be set to
        C{None}.

    @ivar onProcessEnded: If C{terminate_on_stderr} is false and bytes are
        received over stderr, this attribute will refer to a L{Deferred} which
        will be called back when the process ends.  This C{Deferred} is also
        associated with the L{_ProcessExecutionError} which C{deferred} fires
        with earlier in this case so that users can determine when the process
        has actually ended, in addition to knowing when bytes have been
        received via stderr.
    """

    def __init__(self, deferred, cmd, started_deferred=None,
                 terminate_on_stderr=False, check_exit_code=True,
                 process_input=None):
        self.deferred = deferred
        self.cmd = cmd
        self.stdout = StringIO.StringIO()
        self.stderr = StringIO.StringIO()
        self.started_deferred = started_deferred
        self.terminate_on_stderr = terminate_on_stderr
        self.check_exit_code = check_exit_code
        self.process_input = process_input
        self.on_process_ended = None

    def _build_execution_error(self, exit_code=None):
        return ProcessExecutionError(cmd=self.cmd,
                                     exit_code=exit_code,
                                     stdout=self.stdout.getvalue(),
                                     stderr=self.stderr.getvalue())

    def errReceived(self, text):
        self.stderr.write(text)
        if self.terminate_on_stderr and (self.deferred is not None):
            self.on_process_ended = defer.Deferred()
            self.deferred.errback(self._build_execution_error())
            self.deferred = None
            self.transport.loseConnection()

    def outReceived(self, text):
        self.stdout.write(text)

    def processEnded(self, reason):
        if self.deferred is not None:
            stdout, stderr = self.stdout.getvalue(), self.stderr.getvalue()
            exit_code = reason.value.exitCode
            if self.check_exit_code and exit_code != 0:
                self.deferred.errback(self._build_execution_error(exit_code))
            else:
                try:
                    if self.check_exit_code:
                        reason.trap(error.ProcessDone)
                    self.deferred.callback((stdout, stderr))
                except:
                    # NOTE(justinsb): This logic is a little suspicious to me.
                    # If the callback throws an exception, then errback will
                    # be called also. However, this is what the unit tests
                    # test for.
                    exec_error = self._build_execution_error(exit_code)
                    self.deferred.errback(exec_error)
        elif self.on_process_ended is not None:
            self.on_process_ended.errback(reason)

    def connectionMade(self):
        if self.started_deferred:
            self.started_deferred.callback(self)
        if self.process_input:
            self.transport.write(str(self.process_input))
        self.transport.closeStdin()


def get_process_output(executable, args=None, env=None, path=None,
                       process_reactor=None, check_exit_code=True,
                       process_input=None, started_deferred=None,
                       terminate_on_stderr=False):
    if process_reactor is None:
        process_reactor = reactor
    args = args and args or ()
    env = env and env and {}
    deferred = defer.Deferred()
    cmd = executable
    if args:
        cmd = " ".join([cmd] + args)
    logging.debug("Running cmd: %s", cmd)
    process_handler = BackRelayWithInput(
            deferred,
            cmd,
            started_deferred=started_deferred,
            check_exit_code=check_exit_code,
            process_input=process_input,
            terminate_on_stderr=terminate_on_stderr)
    # NOTE(vish): commands come in as unicode, but self.executes needs
    #             strings or process.spawn raises a deprecation warning
    executable = str(executable)
    if not args is None:
        args = [str(x) for x in args]
    process_reactor.spawnProcess(process_handler, executable,
                                 (executable,) + tuple(args), env, path)
    return deferred


class ProcessPool(object):
    """ A simple process pool implementation using Twisted's Process bits.

    This is pretty basic right now, but hopefully the API will be the correct
    one so that it can be optimized later.
    """
    def __init__(self, size=None):
        self.size = size and size or FLAGS.process_pool_size
        self._pool = defer.DeferredSemaphore(self.size)

    def simple_execute(self, cmd, **kw):
        """ Weak emulation of the old utils.execute() function.

        This only exists as a way to quickly move old execute methods to
        this new style of code.

        NOTE(termie): This will break on args with spaces in them.
        """
        parsed = cmd.split(' ')
        executable, args = parsed[0], parsed[1:]
        return self.execute(executable, args, **kw)

    def execute(self, *args, **kw):
        deferred = self._pool.acquire()

        def _associate_process(proto):
            deferred.process = proto.transport
            return proto.transport

        started = defer.Deferred()
        started.addCallback(_associate_process)
        kw.setdefault('started_deferred', started)

        deferred.process = None
        deferred.started = started

        deferred.addCallback(lambda _: get_process_output(*args, **kw))
        deferred.addBoth(self._release)
        return deferred

    def _release(self, retval=None):
        self._pool.release()
        return retval


class SharedPool(object):
    _instance = None

    def __init__(self):
        if SharedPool._instance is None:
            self.__class__._instance = ProcessPool()

    def __getattr__(self, key):
        return getattr(self._instance, key)


def simple_execute(cmd, **kwargs):
    return SharedPool().simple_execute(cmd, **kwargs)