← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/cethread.py

Exclude more files from dumb-rsync upload

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)