1 """Drop-in replacement for the thread module.
3 Meant to be used as a brain-dead substitute so that threaded code does
4 not need to be rewritten for when the thread module is not present.
11 import dummy_thread as thread
14 __author__
= "Brett Cannon"
15 __email__
= "brett@python.org"
17 # Exports only things specified by thread documentation
18 # (skipping obsolete synonyms allocate(), start_new(), exit_thread())
19 __all__
= ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',
20 'interrupt_main', 'LockType']
22 import traceback
as _traceback
24 class error(Exception):
25 """Dummy implementation of thread.error."""
27 def __init__(self
, *args
):
30 def start_new_thread(function
, args
, kwargs
={}):
31 """Dummy implementation of thread.start_new_thread().
33 Compatibility is maintained by making sure that ``args`` is a
34 tuple and ``kwargs`` is a dictionary. If an exception is raised
35 and it is SystemExit (which can be done by thread.exit()) it is
36 caught and nothing is done; all other exceptions are printed out
37 by using traceback.print_exc().
39 If the executed function calls interrupt_main the KeyboardInterrupt will be
40 raised when the function returns.
43 if type(args
) != type(tuple()):
44 raise TypeError("2nd arg must be a tuple")
45 if type(kwargs
) != type(dict()):
46 raise TypeError("3rd arg must be a dict")
50 function(*args
, **kwargs
)
54 _traceback
.print_exc()
59 raise KeyboardInterrupt
62 """Dummy implementation of thread.exit()."""
66 """Dummy implementation of thread.get_ident().
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.
75 """Dummy implementation of thread.allocate_lock()."""
78 class LockType(object):
79 """Class implementing dummy implementation of thread.LockType.
81 Compatibility is maintained by maintaining self.locked_status
82 which is a boolean that stores the state of the lock. Pickling of
83 the lock, though, should not be done since if the thread module is
84 then used with an unpickled ``lock()`` from here problems could
85 occur from this class not having atomic methods.
90 self
.locked_status
= False
92 def acquire(self
, waitflag
=None):
93 """Dummy implementation of acquire().
95 For blocking calls, self.locked_status is automatically set to
96 True and returned appropriately based on value of
97 ``waitflag``. If it is non-blocking, then the value is
98 actually checked and not set if it is already acquired. This
99 is all done so that threading.Condition's assert statements
100 aren't triggered and throw a little fit.
104 self
.locked_status
= True
107 if not self
.locked_status
:
108 self
.locked_status
= True
113 self
.locked_status
= True
118 def __exit__(self
, typ
, val
, tb
):
122 """Release the dummy lock."""
123 # XXX Perhaps shouldn't actually bother to test? Could lead
124 # to problems for complex, threaded code.
125 if not self
.locked_status
:
127 self
.locked_status
= False
131 return self
.locked_status
133 # Used to signal that interrupt_main was called in a "thread"
135 # True when not executing in a "thread"
138 def interrupt_main():
139 """Set _interrupt flag to True to have start_new_thread raise
140 KeyboardInterrupt upon exiting."""
142 raise KeyboardInterrupt