openmedialibrary_platform/Darwin/lib/python3.5/_dummy_thread.py

164 lines
5 KiB
Python
Raw Normal View History

2013-10-11 17:28:32 +00:00
"""Drop-in replacement for the thread module.
Meant to be used as a brain-dead substitute so that threaded code does
not need to be rewritten for when the thread module is not present.
Suggested usage is::
try:
2014-09-30 16:15:32 +00:00
import _thread
2013-10-11 17:28:32 +00:00
except ImportError:
2014-09-30 16:15:32 +00:00
import _dummy_thread as _thread
2013-10-11 17:28:32 +00:00
"""
# Exports only things specified by thread documentation;
# skipping obsolete synonyms allocate(), start_new(), exit_thread().
__all__ = ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',
'interrupt_main', 'LockType']
2014-09-30 16:15:32 +00:00
# A dummy value
TIMEOUT_MAX = 2**31
2013-10-11 17:28:32 +00:00
2014-09-30 16:15:32 +00:00
# NOTE: this module can be imported early in the extension building process,
# and so top level imports of other modules should be avoided. Instead, all
# imports are done when needed on a function-by-function basis. Since threads
# are disabled, the import lock should not be an issue anyway (??).
2013-10-11 17:28:32 +00:00
2014-09-30 16:15:32 +00:00
error = RuntimeError
2013-10-11 17:28:32 +00:00
def start_new_thread(function, args, kwargs={}):
2014-09-30 16:15:32 +00:00
"""Dummy implementation of _thread.start_new_thread().
2013-10-11 17:28:32 +00:00
Compatibility is maintained by making sure that ``args`` is a
tuple and ``kwargs`` is a dictionary. If an exception is raised
2014-09-30 16:15:32 +00:00
and it is SystemExit (which can be done by _thread.exit()) it is
2013-10-11 17:28:32 +00:00
caught and nothing is done; all other exceptions are printed out
by using traceback.print_exc().
If the executed function calls interrupt_main the KeyboardInterrupt will be
raised when the function returns.
"""
if type(args) != type(tuple()):
raise TypeError("2nd arg must be a tuple")
if type(kwargs) != type(dict()):
raise TypeError("3rd arg must be a dict")
global _main
_main = False
try:
function(*args, **kwargs)
except SystemExit:
pass
except:
2014-09-30 16:15:32 +00:00
import traceback
traceback.print_exc()
2013-10-11 17:28:32 +00:00
_main = True
global _interrupt
if _interrupt:
_interrupt = False
raise KeyboardInterrupt
def exit():
2014-09-30 16:15:32 +00:00
"""Dummy implementation of _thread.exit()."""
2013-10-11 17:28:32 +00:00
raise SystemExit
def get_ident():
2014-09-30 16:15:32 +00:00
"""Dummy implementation of _thread.get_ident().
2013-10-11 17:28:32 +00:00
2014-09-30 16:15:32 +00:00
Since this module should only be used when _threadmodule is not
2013-10-11 17:28:32 +00:00
available, it is safe to assume that the current process is the
only thread. Thus a constant can be safely returned.
"""
return -1
def allocate_lock():
2014-09-30 16:15:32 +00:00
"""Dummy implementation of _thread.allocate_lock()."""
2013-10-11 17:28:32 +00:00
return LockType()
def stack_size(size=None):
2014-09-30 16:15:32 +00:00
"""Dummy implementation of _thread.stack_size()."""
2013-10-11 17:28:32 +00:00
if size is not None:
raise error("setting thread stack size not supported")
return 0
2014-09-30 16:15:32 +00:00
def _set_sentinel():
"""Dummy implementation of _thread._set_sentinel()."""
return LockType()
2013-10-11 17:28:32 +00:00
class LockType(object):
2014-09-30 16:15:32 +00:00
"""Class implementing dummy implementation of _thread.LockType.
2013-10-11 17:28:32 +00:00
Compatibility is maintained by maintaining self.locked_status
which is a boolean that stores the state of the lock. Pickling of
2014-09-30 16:15:32 +00:00
the lock, though, should not be done since if the _thread module is
2013-10-11 17:28:32 +00:00
then used with an unpickled ``lock()`` from here problems could
occur from this class not having atomic methods.
"""
def __init__(self):
self.locked_status = False
2014-09-30 16:15:32 +00:00
def acquire(self, waitflag=None, timeout=-1):
2013-10-11 17:28:32 +00:00
"""Dummy implementation of acquire().
For blocking calls, self.locked_status is automatically set to
True and returned appropriately based on value of
``waitflag``. If it is non-blocking, then the value is
actually checked and not set if it is already acquired. This
is all done so that threading.Condition's assert statements
aren't triggered and throw a little fit.
"""
if waitflag is None or waitflag:
self.locked_status = True
return True
else:
if not self.locked_status:
self.locked_status = True
return True
else:
2014-09-30 16:15:32 +00:00
if timeout > 0:
import time
time.sleep(timeout)
2013-10-11 17:28:32 +00:00
return False
__enter__ = acquire
def __exit__(self, typ, val, tb):
self.release()
def release(self):
"""Release the dummy lock."""
# XXX Perhaps shouldn't actually bother to test? Could lead
# to problems for complex, threaded code.
if not self.locked_status:
raise error
self.locked_status = False
return True
def locked(self):
return self.locked_status
2015-11-25 01:25:01 +00:00
def __repr__(self):
return "<%s %s.%s object at %s>" % (
"locked" if self.locked_status else "unlocked",
self.__class__.__module__,
self.__class__.__qualname__,
hex(id(self))
)
2013-10-11 17:28:32 +00:00
# Used to signal that interrupt_main was called in a "thread"
_interrupt = False
# True when not executing in a "thread"
_main = True
def interrupt_main():
"""Set _interrupt flag to True to have start_new_thread raise
KeyboardInterrupt upon exiting."""
if _main:
raise KeyboardInterrupt
else:
global _interrupt
_interrupt = True