python-bzrlib 2.5.0-2ubuntu2 / usr / share / pyshared / bzrlib / cleanup.py

# Copyright (C) 2009, 2010 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Helpers for managing cleanup functions and the errors they might raise.

The usual way to run cleanup code in Python is::

    try:
        do_something()
    finally:
        cleanup_something()

However if both `do_something` and `cleanup_something` raise an exception
Python will forget the original exception and propagate the one from
cleanup_something.  Unfortunately, this is almost always much less useful than
the original exception.

If you want to be certain that the first, and only the first, error is raised,
then use::

    operation = OperationWithCleanups(do_something)
    operation.add_cleanup(cleanup_something)
    operation.run_simple()

This is more inconvenient (because you need to make every try block a
function), but will ensure that the first error encountered is the one raised,
while also ensuring all cleanups are run.  See OperationWithCleanups for more
details.
"""

from __future__ import absolute_import

from collections import deque
import sys
from bzrlib import (
    debug,
    trace,
    )

def _log_cleanup_error(exc):
    trace.mutter('Cleanup failed:')
    trace.log_exception_quietly()
    if 'cleanup' in debug.debug_flags:
        trace.warning('bzr: warning: Cleanup failed: %s', exc)


def _run_cleanup(func, *args, **kwargs):
    """Run func(*args, **kwargs), logging but not propagating any error it
    raises.

    :returns: True if func raised no errors, else False.
    """
    try:
        func(*args, **kwargs)
    except KeyboardInterrupt:
        raise
    except Exception, exc:
        _log_cleanup_error(exc)
        return False
    return True


def _run_cleanups(funcs):
    """Run a series of cleanup functions."""
    for func, args, kwargs in funcs:
        _run_cleanup(func, *args, **kwargs)


class ObjectWithCleanups(object):
    """A mixin for objects that hold a cleanup list.

    Subclass or client code can call add_cleanup and then later `cleanup_now`.
    """
    def __init__(self):
        self.cleanups = deque()

    def add_cleanup(self, cleanup_func, *args, **kwargs):
        """Add a cleanup to run.

        Cleanups may be added at any time.  
        Cleanups will be executed in LIFO order.
        """
        self.cleanups.appendleft((cleanup_func, args, kwargs))

    def cleanup_now(self):
        _run_cleanups(self.cleanups)
        self.cleanups.clear()


class OperationWithCleanups(ObjectWithCleanups):
    """A way to run some code with a dynamic cleanup list.

    This provides a way to add cleanups while the function-with-cleanups is
    running.

    Typical use::

        operation = OperationWithCleanups(some_func)
        operation.run(args...)

    where `some_func` is::

        def some_func(operation, args, ...):
            do_something()
            operation.add_cleanup(something)
            # etc

    Note that the first argument passed to `some_func` will be the
    OperationWithCleanups object.  To invoke `some_func` without that, use
    `run_simple` instead of `run`.
    """

    def __init__(self, func):
        super(OperationWithCleanups, self).__init__()
        self.func = func

    def run(self, *args, **kwargs):
        return _do_with_cleanups(
            self.cleanups, self.func, self, *args, **kwargs)

    def run_simple(self, *args, **kwargs):
        return _do_with_cleanups(
            self.cleanups, self.func, *args, **kwargs)


def _do_with_cleanups(cleanup_funcs, func, *args, **kwargs):
    """Run `func`, then call all the cleanup_funcs.

    All the cleanup_funcs are guaranteed to be run.  The first exception raised
    by func or any of the cleanup_funcs is the one that will be propagted by
    this function (subsequent errors are caught and logged).

    Conceptually similar to::

        try:
            return func(*args, **kwargs)
        finally:
            for cleanup, cargs, ckwargs in cleanup_funcs:
                cleanup(*cargs, **ckwargs)

    It avoids several problems with using try/finally directly:
     * an exception from func will not be obscured by a subsequent exception
       from a cleanup.
     * an exception from a cleanup will not prevent other cleanups from
       running (but the first exception encountered is still the one
       propagated).

    Unike `_run_cleanup`, `_do_with_cleanups` can propagate an exception from a
    cleanup, but only if there is no exception from func.
    """
    # As correct as Python 2.4 allows.
    try:
        result = func(*args, **kwargs)
    except:
        # We have an exception from func already, so suppress cleanup errors.
        _run_cleanups(cleanup_funcs)
        raise
    else:
        # No exception from func, so allow the first exception from
        # cleanup_funcs to propagate if one occurs (but only after running all
        # of them).
        exc_info = None
        for cleanup, c_args, c_kwargs in cleanup_funcs:
            # XXX: Hmm, if KeyboardInterrupt arrives at exactly this line, we
            # won't run all cleanups... perhaps we should temporarily install a
            # SIGINT handler?
            if exc_info is None:
                try:
                    cleanup(*c_args, **c_kwargs)
                except:
                    # This is the first cleanup to fail, so remember its
                    # details.
                    exc_info = sys.exc_info()
            else:
                # We already have an exception to propagate, so log any errors
                # but don't propagate them.
                _run_cleanup(cleanup, *c_args, **kwargs)
        if exc_info is not None:
            try:
                raise exc_info[0], exc_info[1], exc_info[2]
            finally:
                del exc_info
        # No error, so we can return the result
        return result