• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Drop-in replacement for the thread module.
2
3Meant to be used as a brain-dead substitute so that threaded code does
4not need to be rewritten for when the thread module is not present.
5
6Suggested usage is::
7
8    try:
9        import _thread
10    except ImportError:
11        import _dummy_thread as _thread
12
13"""
14# Exports only things specified by thread documentation;
15# skipping obsolete synonyms allocate(), start_new(), exit_thread().
16__all__ = ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',
17           'interrupt_main', 'LockType']
18
19# A dummy value
20TIMEOUT_MAX = 2**31
21
22# NOTE: this module can be imported early in the extension building process,
23# and so top level imports of other modules should be avoided.  Instead, all
24# imports are done when needed on a function-by-function basis.  Since threads
25# are disabled, the import lock should not be an issue anyway (??).
26
27error = RuntimeError
28
29def start_new_thread(function, args, kwargs={}):
30    """Dummy implementation of _thread.start_new_thread().
31
32    Compatibility is maintained by making sure that ``args`` is a
33    tuple and ``kwargs`` is a dictionary.  If an exception is raised
34    and it is SystemExit (which can be done by _thread.exit()) it is
35    caught and nothing is done; all other exceptions are printed out
36    by using traceback.print_exc().
37
38    If the executed function calls interrupt_main the KeyboardInterrupt will be
39    raised when the function returns.
40
41    """
42    if type(args) != type(tuple()):
43        raise TypeError("2nd arg must be a tuple")
44    if type(kwargs) != type(dict()):
45        raise TypeError("3rd arg must be a dict")
46    global _main
47    _main = False
48    try:
49        function(*args, **kwargs)
50    except SystemExit:
51        pass
52    except:
53        import traceback
54        traceback.print_exc()
55    _main = True
56    global _interrupt
57    if _interrupt:
58        _interrupt = False
59        raise KeyboardInterrupt
60
61def exit():
62    """Dummy implementation of _thread.exit()."""
63    raise SystemExit
64
65def get_ident():
66    """Dummy implementation of _thread.get_ident().
67
68    Since this module should only be used when _threadmodule is not
69    available, it is safe to assume that the current process is the
70    only thread.  Thus a constant can be safely returned.
71    """
72    return 1
73
74def allocate_lock():
75    """Dummy implementation of _thread.allocate_lock()."""
76    return LockType()
77
78def stack_size(size=None):
79    """Dummy implementation of _thread.stack_size()."""
80    if size is not None:
81        raise error("setting thread stack size not supported")
82    return 0
83
84def _set_sentinel():
85    """Dummy implementation of _thread._set_sentinel()."""
86    return LockType()
87
88class LockType(object):
89    """Class implementing dummy implementation of _thread.LockType.
90
91    Compatibility is maintained by maintaining self.locked_status
92    which is a boolean that stores the state of the lock.  Pickling of
93    the lock, though, should not be done since if the _thread module is
94    then used with an unpickled ``lock()`` from here problems could
95    occur from this class not having atomic methods.
96
97    """
98
99    def __init__(self):
100        self.locked_status = False
101
102    def acquire(self, waitflag=None, timeout=-1):
103        """Dummy implementation of acquire().
104
105        For blocking calls, self.locked_status is automatically set to
106        True and returned appropriately based on value of
107        ``waitflag``.  If it is non-blocking, then the value is
108        actually checked and not set if it is already acquired.  This
109        is all done so that threading.Condition's assert statements
110        aren't triggered and throw a little fit.
111
112        """
113        if waitflag is None or waitflag:
114            self.locked_status = True
115            return True
116        else:
117            if not self.locked_status:
118                self.locked_status = True
119                return True
120            else:
121                if timeout > 0:
122                    import time
123                    time.sleep(timeout)
124                return False
125
126    __enter__ = acquire
127
128    def __exit__(self, typ, val, tb):
129        self.release()
130
131    def release(self):
132        """Release the dummy lock."""
133        # XXX Perhaps shouldn't actually bother to test?  Could lead
134        #     to problems for complex, threaded code.
135        if not self.locked_status:
136            raise error
137        self.locked_status = False
138        return True
139
140    def locked(self):
141        return self.locked_status
142
143    def __repr__(self):
144        return "<%s %s.%s object at %s>" % (
145            "locked" if self.locked_status else "unlocked",
146            self.__class__.__module__,
147            self.__class__.__qualname__,
148            hex(id(self))
149        )
150
151# Used to signal that interrupt_main was called in a "thread"
152_interrupt = False
153# True when not executing in a "thread"
154_main = True
155
156def interrupt_main():
157    """Set _interrupt flag to True to have start_new_thread raise
158    KeyboardInterrupt upon exiting."""
159    if _main:
160        raise KeyboardInterrupt
161    else:
162        global _interrupt
163        _interrupt = True
164