python-sclapp 0.5.3-3 / usr / share / pyshared / sclapp / signals.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.

import os, signal
from copy import copy

try:
    set
except NameError:
    from sets import Set as set

try:
    reversed
except NameError:
    from sclapp.legacy_support import reversed

from sclapp.exceptions import ExitSignalError, SignalError
from sclapp.error_output import printWarning, printDebug

def _getAllSignals_iter():
    for name in dir(signal):
        if name.startswith('SIG') \
          and (name != 'SIG_IGN') \
          and (name != 'SIG_DFL'):
            signum = getattr(signal, name)
            yield signum

def _getRealSignals_iter(names):
    for name in names:
        if hasattr(signal, name):
            signum = getattr(signal, name)
            yield signum

ALL_SIGNALS = tuple(list(_getAllSignals_iter()))

_STD_EXIT_SIGNALS = ( )
_STD_NOTIFY_SIGNALS = ( )
_STD_DEFAULT_SIGNALS = (
  'SIGHUP', 'SIGINT', 'SIGQUIT', 'SIGILL',
  'SIGABRT', 'SIGFPE', 'SIGPIPE', 'SIGTERM',
  'SIGBUS', 'SIGPROF', 'SIGSYS', 'SIGTRAP',
  'SIGXCPU', 'SIGXFSZ',
)
_STD_IGNORE_SIGNALS = ( 'SIGUSR1', 'SIGUSR2', 'SIGALRM', )

STD_EXIT_SIGNALS = tuple(list(_getRealSignals_iter(_STD_EXIT_SIGNALS)))
STD_NOTIFY_SIGNALS = tuple(list(_getRealSignals_iter(_STD_NOTIFY_SIGNALS)))
STD_DEFAULT_SIGNALS = tuple(list(_getRealSignals_iter(_STD_DEFAULT_SIGNALS)))
STD_IGNORE_SIGNALS = tuple(list(_getRealSignals_iter(_STD_IGNORE_SIGNALS)))

_announce_signals = True
_announce_signals_stack = [ ]

# store raised SignalError here
_exit_signal_exception = None

# signal current being handled
_handling_signal = None

# for keeping track of signal handlers; old handlers are pushed here, and
# popped back into place later
_signal_mapping_stack = [ ]

def _addToMapping(d, signums, handler_factory):
    if signums is not None:
        for signum in signums:
            assert (signum not in d), 'Signal %u mapped more than once' % signum
            d[signum] = handler_factory(signum)

def _pushSignalMapping(
  exit_signals = None,
  notify_signals = None,
  default_signals = None,
  ignore_signals = None,
  announce_signals = True,
  custom_handlers = None
):
    global _signal_mapping_stack
    d = { }

    _addToMapping(d, custom_handlers, lambda signum: custom_handlers[signum])
    _addToMapping(d, exit_signals, lambda signum: _exitSignalHandler)
    _addToMapping(d, notify_signals, lambda signum: _notifySignalHandler)
    _addToMapping(d, default_signals, lambda signum: signal.SIG_DFL)
    _addToMapping(d, ignore_signals, lambda signum: signal.SIG_IGN)

    for signum in d.keys():
        try:
            signal.signal(signum, d[signum])
        except RuntimeError, e:
            if e[0] == 22:
                printDebug(
                  'Setting signal handler for signum %u: Invalid argument' % \
                    signum)
                d[signum] = signal.getsignal(signum)
            else:
                raise

    _signal_mapping_stack.append(d)
    global _announce_signals, _announce_signals_stack
    _announce_signals_stack.append(_announce_signals)
    _announce_signals = announce_signals

def _popSignalMapping():
    global _signal_mapping_stack
    d = _signal_mapping_stack.pop()
    for signum in d:
        _restoreSignalHandler(signum)
    global _announce_signals, _announce_signals_stack
    _announce_signals = _announce_signals_stack.pop()
    return d

def _pushCurrentSignalMapping():
    global _signal_mapping_stack
    d = { }
    for signum in ALL_SIGNALS:
        d[signum] = signal.getsignal(signum)
    _signal_mapping_stack.append(d)

def _restoreSignalHandler(signum):
    for mapping in reversed(_signal_mapping_stack):
        if signum in mapping:
            signal.signal(signum, mapping[signum])

def _getSignalsByHandler_iter(handler):
    for signum in ALL_SIGNALS:
        if signal.getsignal(signum) == handler:
            yield signum

def _getSignalHandlers_iter():
    for signum in ALL_SIGNALS:
        yield signal.getsignal(signum)

# stores list of caught signal numbers
_caught_signals = set([ ])

def _ignoreExitSignals():
    for signum in getExitSignals():
        signal.signal(signum, _ignoreExitSignalHandler)

def _ignoreExitSignalHandler(signum, frame):
    handler = signal.signal(signum, signal.SIG_IGN)
    global _caught_signals, _exit_signal_exception, _handling_signal
    if _handling_signal is None:
        _handling_signal = signum
    else:
        _handleLameSignal(signum, handler)
        return
    _signalPrint('caught signal %i (ignored)' % signum)
    _caught_signals.add(signum)
    _handling_signal = None
    signal.signal(signum, handler)

def _exitSignalHandler(signum, frame):
    signal.signal(signum, signal.SIG_IGN)
    global _handling_signal, _exit_signal_exception
    if _exit_signal_exception is not None:
        _handleLameSignal(signum, _ignoreExitSignalHandler)
        return
    _exit_signal_exception = ExitSignalError(signum)
    _handling_signal = signum
    _signalPrint('caught exit signal %i' % signum)
    _caught_signals.add(signum)
    _handling_signal = None
    _ignoreExitSignals()
    raise _exit_signal_exception

def _notifySignalHandler(signum, frame):
    handler = signal.signal(signum, signal.SIG_IGN)
    global _exit_signal_exception, _handling_signal
    if _exit_signal_exception is not None:
        _handleLameSignal(signum, handler)
        return
    if _handling_signal is None:
        _handling_signal = signum
    elif _handling_signal in getNotifySignals():
        _handleLameSignal(signum, handler)
        return
    _signalPrint('caught signal %i' % signum)
    _caught_signals.add(signum)
    _handling_signal = None
    signal.signal(signum, handler)
    raise SignalError, signum

def _handleLameSignal(signum, handler):
    _signalPrint('caught signal %i (ignored)' % signum)
    _caught_signals.add(signum)
    signal.signal(signum, handler)

def _signalPrint(message):
    _signal_print_fn = _getSignalPrintFn()
    if _signal_print_fn is not None:
        from sclapp.protected_output import _protected_output_enabled, \
          _signalOutput
        if _protected_output_enabled:
            _signalOutput()
        _signal_print_fn(message)

def _getSignalPrintFn():
    global _announce_signals
    if _announce_signals:
        return printWarning
    else:
        return None

def _enableSignals(signums):
    for signum in signums:
        if signum in getExitSignals():
            if _exit_signal_exception is not None:
                signal.signal(signum, _ignoreExitSignalHandler)
            else:
                signal.signal(signum, _exitSignalHandler)
        elif signum in getNotifySignals():
            signal.signal(signum, _notifySignalHandler)
        elif signum in getDefaultSignals():
            signal.signal(signum, signal.SIG_DFL)
        elif signum in getIgnoreSignals():
            signal.signal(signum, signal.SIG_IGN)

def getExitSignals():
    '''getExitSignals() -> list

    Returns a list containing all signal numbers that are currently being
    handled by sclapp as exit signals.
    '''
    return set(
      list(_getSignalsByHandler_iter(_exitSignalHandler)) +
      list(_getSignalsByHandler_iter(_ignoreExitSignalHandler))
    )

def getNotifySignals():
    '''getNotifySignals() -> list

    Returns a list containing all signal numbers that are currently being
    handled by sclapp as notify signals.
    '''
    return set(list(_getSignalsByHandler_iter(_notifySignalHandler)))

def getDefaultSignals():
    '''getDefaultSignals() -> list

    Returns a list containing all signal numbers that are currently being
    handled by sclapp as default signals (SIG_DFL).
    '''
    return set(list(_getSignalsByHandler_iter(signal.SIG_DFL)))

def getIgnoreSignals():
    '''getIgnoreSignals() -> list

    Returns a list containing all signal numbers that are currently being
    handled by sclapp as ignore signals (SIG_IGN).
    '''
    return set(list(_getSignalsByHandler_iter(signal.SIG_IGN)))

def getCaughtSignals():
    '''getCaughtSignals() -> list

    Returns a list containing all signal numbers that have been caught by
    sclapp signal handlers.
    '''
    return copy(_caught_signals)

def _parseSignal(sig):
    if str(sig) == sig:
        if hasattr(signal, sig):
            return getattr(signal, sig)
        return None
    return sig

def _parseSignals(sigs):
    return [signum for signum in [_parseSignal(sig) for sig in sigs] if signum]

def enableSignalHandling(exit_signals = None, notify_signals = None,
  default_signals = None, ignore_signals = None, announce_signals = True,
  custom_handlers = None):
    '''enableSignalHandling() -> None

    Enables sclapp signal handling.
    '''
    # Note that we use an empty list if a group is not specified:
    exit_signals = _parseSignals(exit_signals or [ ])
    notify_signals = _parseSignals(notify_signals or [ ])
    default_signals = _parseSignals(default_signals or [ ])
    ignore_signals = _parseSignals(ignore_signals or [ ])

    if os.name == 'posix':
        global _signal_mapping_stack
        if len(_signal_mapping_stack) < 1:
            _pushCurrentSignalMapping()

        specified_signals = []
        if custom_handlers:
            specified_signals = specified_signals + custom_handlers.keys()
        specified_signals = specified_signals + exit_signals
        specified_signals = specified_signals + notify_signals
        specified_signals = specified_signals + default_signals
        specified_signals = specified_signals + ignore_signals

        for signum in STD_EXIT_SIGNALS:
            if signum not in specified_signals:
                exit_signals.append(signum)
        for signum in STD_NOTIFY_SIGNALS:
            if signum not in specified_signals:
                notify_signals.append(signum)
        for signum in STD_DEFAULT_SIGNALS:
            if signum not in specified_signals:
                default_signals.append(signum)
        for signum in STD_IGNORE_SIGNALS:
            if signum not in specified_signals:
                ignore_signals.append(signum)

        _pushSignalMapping(
          exit_signals = exit_signals,
          notify_signals = notify_signals,
          default_signals = default_signals,
          ignore_signals = ignore_signals,
          announce_signals = announce_signals,
          custom_handlers = custom_handlers
        )

def disableSignalHandling():
    '''disableSignalHandling() -> None

    Disables sclapp signal handling.
    '''
    if os.name == 'posix':
        _popSignalMapping()

def signalHandlingEnabled():
    handlers = list(_getSignalHandlers_iter())
    if _exitSignalHandler in handlers:
        return True
    if _notifySignalHandler in handlers:
        return True
    if _ignoreExitSignalHandler in handlers:
        return True
    return False