python-sclapp 0.5.3-3 / usr / share / pyshared / sclapp / processes.py

# Copyright (c) 2005-2007 Forest Bond.
# This file is part of the sclapp software package.
# 
# sclapp is free software; you can redistribute it and/or modify it under the
# terms of the GNU General Public License version 2 as published by the Free
# Software Foundation.
# 
# A copy of the license has been included in the COPYING file.

'''Classes for easily managing background processes.'''

import os, signal, traceback
from datetime import datetime

from sclapp import error_output
from sclapp.redirection import redirectFds
from sclapp.util import safe_encode

def waitPid(pid, block = True):
    error_output._printSclappDebug(u'waitPid')
    import errno
    if pid is None:
        status = None
        signal = None
    else:
        try:
            if block:
                (os_pid, os_status) = os.waitpid(pid, 0)
            else:
                (os_pid, os_status) = os.waitpid(pid, os.WNOHANG)
        except OSError, e:
            if e.errno == errno.ESRCH:
                error_output._printSclappDebug(u'waitpid (%i): ESRCH' % pid)
                status = None
                signal = None
            if e.errno == errno.ECHILD:
                error_output._printSclappDebug(u'waitpid (%i): ECHILD' % pid)
                status = None
                signal = None
            else:
                raise e
        else:
            if os_pid:
                if os.WIFSIGNALED(os_status):
                    status = None
                    signal = os.WTERMSIG(os_status)
                    error_output._printSclappDebug(u'signaled: %s' % signal)
                elif os.WIFEXITED(os_status):
                    status = os.WEXITSTATUS(os_status)
                    error_output._printSclappDebug(u'exited: %s' % status)
                    signal = None
                else:
                    error_output._printSclappDebug(u'not signaled, not exited')
                    status = None
                    signal = None
            else:
                error_output._printSclappDebug(u'os_pid is %s' % os_pid)
                status = None
                signal = None
    return status, signal

def _runFunction(fn, args, kwargs, stdin = None, stdout = None, stderr = None):
    pid = os.fork()
    if pid:
        return pid
    try:
        os.setsid()
        redirectFds(stdin, stdout, stderr)
        os._exit(fn(*args, **kwargs))
    except Exception:
        traceback.print_exc()
        os._exit(254)

def _runCommand(command, args, stdin = None, stdout = None, stderr = None):
    pid = os.fork()
    if pid:
        return pid
    try:
        os.setsid()
        redirectFds(stdin, stdout, stderr)
        os.execvp(command, args)
    except Exception:
        traceback.print_exc()
        os._exit(254)

def average(l):
    return (sum(l) / float(len(l)))

class Event(object):
    name = None
    timestamp = None

    def __init__(self, name, *args, **kwargs):
        self.name = name
        self.timestamp = datetime.now()
        super(Event, self).__init__(*args, **kwargs)

class EventHistory(list):
    maxlength = None

    def __init__(self, *args, **kwargs):
        try:
            self.maxlength = kwargs['maxlength']
            del kwargs['maxlength']
        except KeyError:
            self.maxlength = 100

    def averagePeriod(self, name, samplesize):
        if samplesize < 2:
            raise ValueError, u'samplesize must be greater than 1'

        samples = [
          event for event in self if (event.name == name) ][-samplesize:]

        if len(samples) < 2:
            return None

        deltas = [ ]
        for i in range(1, len(samples)):
            deltas.append(samples[i].timestamp - samples[i-1].timestamp)

        deltas_seconds = [
          d.seconds + (d.microseconds / 1000000.0) for d in deltas ]

        return average(deltas_seconds)

    def averageFrequency(self, name, samplesize):
        avg_period = self.averagePeriod(name, samplesize)
        if avg_period is None:
            return 0.0
        return (1.0 / avg_period)

    def __setitem__(self, index, value):
        raise TypeError

    def __delitem__(self, index):
        raise TypeError

    def pop(self, value):
        raise TypeError

    def append(self, value):
        super(EventHistory, self).append(value)
        if len(self) > self.maxlength:
            super(EventHistory, self).__delitem__(0)

class _BackgroundProcess(object):
    _pid = None
    _status = None
    _signal = None
    _stdin = None
    _stdout = None
    _stderr = None
    _history = None

    def __init__(self, stdin = None, stdout = None, stderr = None):
        self._stdin = stdin
        self._stdout = stdout
        self._stderr = stderr
        self._history = EventHistory()

    def run(self):
        '''Runs the process.'''
        self._history.append(Event('run'))

    def getRunFrequency(self, samplesize = 10):
        '''Returns runs/second over the last samplesize runs.'''
        return self._history.averageFrequency('run', samplesize)

    def getRunCount(self):
        return len([ ev for ev in self._history if ev.name == 'run' ])

    def stop(self):
        '''Stops (pauses) the process (usually by sending SIGSTOP).'''
        return self.kill(signal.SIGSTOP)

    def cont(self):
        '''Re-starts a stopped (paused) process (usually by sending SIGCONT).'''
        return self.kill(signal.SIGCONT)

    def kill(self, signum = signal.SIGINT):
        '''Kills the process, or sends an arbitrary signal (specified by the
        signum argument) to the process.  Returns False if the process doesn't
        appear to be running in the first place, True otherwise.
        '''
        if self._pid is None:
            return False
        os.kill(self._pid, signum)
        return True

    def reap(self):
        '''Tries to reap the process.  Returns True if successful, False
        otherwise.
        '''
        return self.wait(False)

    def wait(self, block = True):
        '''Blocks until the process has terminated.  If the block argument is
        False, process will simply be reaped (if it has terminated), and the
        function will return.  Returns True if process is reaped by this call,
        False otherwise.
        '''
        error_output._printSclappDebug(u'wait')
        if self._pid is None:
            error_output._printSclappDebug(u'_pid is None')
            return False
        self._status, self._signal = waitPid(self._pid, block)
        if (self._status is not None) or (self._signal is not None):
            self._pid = None
            return True
        return False

    def isRunning(self):
        '''Returns True if process is running, False otherwise.'''
        self.reap()
        return (self._pid is not None)

    def isStopped(self):
        '''Returns True if process is stopped, False otherwise.'''
        self.reap()
        return self._stopped

    def getExitStatus(self):
        '''Returns the exit status of the process.'''
        self.reap()
        return self._status

    def getExitSignal(self):
        '''Returns the signal that caused the process to terminate.'''
        self.reap()
        return self._signal

    def getPid(self):
        '''Returns the PID (process ID) of the process.'''
        return self._pid

class BackgroundFunction(_BackgroundProcess):
    '''Runs a function in a forked background process.  The status of the
    forked process can be monitored by the caller during function execution.
    '''
    _function = None
    _args = None
    _kwargs = None

    def __init__(
      self,
      function,
      args = (),
      kwargs = {},
      stdin = None,
      stdout = None,
      stderr = None,
    ):
        self._function = function
        self._args = args
        self._kwargs = dict(kwargs)
        super(BackgroundFunction, self).__init__(stdin, stdout, stderr)

    def run(self):
        self._pid = _runFunction(self._function, self._args, self._kwargs,
          self._stdin, self._stdout, self._stderr)
        return super(BackgroundFunction, self).run()

    def __str__(self):
        return '%s: %s (pid=%s)' % (
          self.__class__.__name__, self._function.__name__,
          self._pid
        )

class BackgroundCommand(_BackgroundProcess):
    '''Runs an external command in a forked process.  The status of the forked
    process can be monitored by the caller.
    '''
    _command = None
    _args = None

    def __init__(self, command, args, stdin = None, stdout = None,
      stderr = None):
        '''command will be run with arguments args.  If stdin, stdout, or
        stderr are specified, the standard I/O file descriptors for the
        sub-process will be redirected.  stdin, stdout, and stderr should be
        specified as for redirectFds(), or left unspecified if no I/O
        redirection is desired.
        '''
        self._command = command
        self._args = args
        super(BackgroundCommand, self).__init__(stdin, stdout, stderr)

    def run(self):
        '''Runs the command.'''
        self._pid = _runCommand(self._command, self._args,
          self._stdin, self._stdout, self._stderr)
        return super(BackgroundCommand, self).run()

    def __str__(self):
        return safe_encode(unicode(self))

    def __unicode__(self):
        return u'%s: %s (pid=%s)' % (
          self.__class__.__name__,
          self._command,
          self._pid
        )