ยปCore Development>Code coverage>Lib/_dummy_thread.py

Python code coverage for Lib/_dummy_thread.py

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