← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/cethread.py

(vila) Fix test failures blocking package builds. (Vincent Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/cethread.py
 
1
# Copyright (C) 2011 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
from __future__ import absolute_import
 
18
 
 
19
import sys
 
20
import threading
 
21
 
 
22
 
 
23
class CatchingExceptionThread(threading.Thread):
 
24
    """A thread that keeps track of exceptions.
 
25
 
 
26
    If an exception occurs during the thread execution, it's caught and
 
27
    re-raised when the thread is joined().
 
28
    """
 
29
 
 
30
    def __init__(self, *args, **kwargs):
 
31
        # There are cases where the calling thread must wait, yet, if an
 
32
        # exception occurs, the event should be set so the caller is not
 
33
        # blocked. The main example is a calling thread that want to wait for
 
34
        # the called thread to be in a given state before continuing.
 
35
        try:
 
36
            sync_event = kwargs.pop('sync_event')
 
37
        except KeyError:
 
38
            # If the caller didn't pass a specific event, create our own
 
39
            sync_event = threading.Event()
 
40
        super(CatchingExceptionThread, self).__init__(*args, **kwargs)
 
41
        self.set_sync_event(sync_event)
 
42
        self.exception = None
 
43
        self.ignored_exceptions = None # see set_ignored_exceptions
 
44
        self.lock = threading.Lock()
 
45
 
 
46
    # compatibility thunk for python-2.4 and python-2.5...
 
47
    if sys.version_info < (2, 6):
 
48
        name = property(threading.Thread.getName, threading.Thread.setName)
 
49
 
 
50
    def set_sync_event(self, event):
 
51
        """Set the ``sync_event`` event used to synchronize exception catching.
 
52
 
 
53
        When the thread uses an event to synchronize itself with another thread
 
54
        (setting it when the other thread can wake up from a ``wait`` call),
 
55
        the event must be set after catching an exception or the other thread
 
56
        will hang.
 
57
 
 
58
        Some threads require multiple events and should set the relevant one
 
59
        when appropriate.
 
60
 
 
61
        Note that the event should be initially cleared so the caller can
 
62
        wait() on him and be released when the thread set the event.
 
63
 
 
64
        Also note that the thread can use multiple events, setting them as it
 
65
        progress, while the caller can chose to wait on any of them. What
 
66
        matters is that there is always one event set so that the caller is
 
67
        always released when an exception is caught. Re-using the same event is
 
68
        therefore risky as the thread itself has no idea about which event the
 
69
        caller is waiting on. If the caller has already been released then a
 
70
        cleared event won't guarantee that the caller is still waiting on it.
 
71
        """
 
72
        self.sync_event = event
 
73
 
 
74
    def switch_and_set(self, new):
 
75
        """Switch to a new ``sync_event`` and set the current one.
 
76
 
 
77
        Using this method protects against race conditions while setting a new
 
78
        ``sync_event``.
 
79
 
 
80
        Note that this allows a caller to wait either on the old or the new
 
81
        event depending on whether it wants a fine control on what is happening
 
82
        inside a thread.
 
83
 
 
84
        :param new: The event that will become ``sync_event``
 
85
        """
 
86
        cur = self.sync_event
 
87
        self.lock.acquire()
 
88
        try: # Always release the lock
 
89
            try:
 
90
                self.set_sync_event(new)
 
91
                # From now on, any exception will be synced with the new event
 
92
            except:
 
93
                # Unlucky, we couldn't set the new sync event, try restoring a
 
94
                # safe state
 
95
                self.set_sync_event(cur)
 
96
                raise
 
97
            # Setting the current ``sync_event`` will release callers waiting
 
98
            # on it, note that it will also be set in run() if an exception is
 
99
            # raised
 
100
            cur.set()
 
101
        finally:
 
102
            self.lock.release()
 
103
 
 
104
    def set_ignored_exceptions(self, ignored):
 
105
        """Declare which exceptions will be ignored.
 
106
 
 
107
        :param ignored: Can be either:
 
108
        
 
109
           - None: all exceptions will be raised,
 
110
           - an exception class: the instances of this class will be ignored,
 
111
           - a tuple of exception classes: the instances of any class of the
 
112
             list will be ignored,
 
113
           - a callable: that will be passed the exception object
 
114
             and should return True if the exception should be ignored
 
115
        """
 
116
        if ignored is None:
 
117
            self.ignored_exceptions = None
 
118
        elif isinstance(ignored, (Exception, tuple)):
 
119
            self.ignored_exceptions = lambda e: isinstance(e, ignored)
 
120
        else:
 
121
            self.ignored_exceptions = ignored
 
122
 
 
123
    def run(self):
 
124
        """Overrides Thread.run to capture any exception."""
 
125
        self.sync_event.clear()
 
126
        try:
 
127
            try:
 
128
                super(CatchingExceptionThread, self).run()
 
129
            except:
 
130
                self.exception = sys.exc_info()
 
131
        finally:
 
132
            # Make sure the calling thread is released
 
133
            self.sync_event.set()
 
134
 
 
135
 
 
136
    def join(self, timeout=None):
 
137
        """Overrides Thread.join to raise any exception caught.
 
138
 
 
139
        Calling join(timeout=0) will raise the caught exception or return None
 
140
        if the thread is still alive.
 
141
        """
 
142
        super(CatchingExceptionThread, self).join(timeout)
 
143
        if self.exception is not None:
 
144
            exc_class, exc_value, exc_tb = self.exception
 
145
            self.exception = None # The exception should be raised only once
 
146
            if (self.ignored_exceptions is None
 
147
                or not self.ignored_exceptions(exc_value)):
 
148
                # Raise non ignored exceptions
 
149
                raise exc_class, exc_value, exc_tb
 
150
 
 
151
    def pending_exception(self):
 
152
        """Raise the caught exception.
 
153
 
 
154
        This does nothing if no exception occurred.
 
155
        """
 
156
        self.join(timeout=0)