+++ /dev/null
-"""Drop-in replacement for the thread module.\r
-\r
-Meant to be used as a brain-dead substitute so that threaded code does\r
-not need to be rewritten for when the thread module is not present.\r
-\r
-Suggested usage is::\r
-\r
- try:\r
- import thread\r
- except ImportError:\r
- import dummy_thread as thread\r
-\r
-"""\r
-# Exports only things specified by thread documentation;\r
-# skipping obsolete synonyms allocate(), start_new(), exit_thread().\r
-__all__ = ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',\r
- 'interrupt_main', 'LockType']\r
-\r
-import traceback as _traceback\r
-\r
-class error(Exception):\r
- """Dummy implementation of thread.error."""\r
-\r
- def __init__(self, *args):\r
- self.args = args\r
-\r
-def start_new_thread(function, args, kwargs={}):\r
- """Dummy implementation of thread.start_new_thread().\r
-\r
- Compatibility is maintained by making sure that ``args`` is a\r
- tuple and ``kwargs`` is a dictionary. If an exception is raised\r
- and it is SystemExit (which can be done by thread.exit()) it is\r
- caught and nothing is done; all other exceptions are printed out\r
- by using traceback.print_exc().\r
-\r
- If the executed function calls interrupt_main the KeyboardInterrupt will be\r
- raised when the function returns.\r
-\r
- """\r
- if type(args) != type(tuple()):\r
- raise TypeError("2nd arg must be a tuple")\r
- if type(kwargs) != type(dict()):\r
- raise TypeError("3rd arg must be a dict")\r
- global _main\r
- _main = False\r
- try:\r
- function(*args, **kwargs)\r
- except SystemExit:\r
- pass\r
- except:\r
- _traceback.print_exc()\r
- _main = True\r
- global _interrupt\r
- if _interrupt:\r
- _interrupt = False\r
- raise KeyboardInterrupt\r
-\r
-def exit():\r
- """Dummy implementation of thread.exit()."""\r
- raise SystemExit\r
-\r
-def get_ident():\r
- """Dummy implementation of thread.get_ident().\r
-\r
- Since this module should only be used when threadmodule is not\r
- available, it is safe to assume that the current process is the\r
- only thread. Thus a constant can be safely returned.\r
- """\r
- return -1\r
-\r
-def allocate_lock():\r
- """Dummy implementation of thread.allocate_lock()."""\r
- return LockType()\r
-\r
-def stack_size(size=None):\r
- """Dummy implementation of thread.stack_size()."""\r
- if size is not None:\r
- raise error("setting thread stack size not supported")\r
- return 0\r
-\r
-class LockType(object):\r
- """Class implementing dummy implementation of thread.LockType.\r
-\r
- Compatibility is maintained by maintaining self.locked_status\r
- which is a boolean that stores the state of the lock. Pickling of\r
- the lock, though, should not be done since if the thread module is\r
- then used with an unpickled ``lock()`` from here problems could\r
- occur from this class not having atomic methods.\r
-\r
- """\r
-\r
- def __init__(self):\r
- self.locked_status = False\r
-\r
- def acquire(self, waitflag=None):\r
- """Dummy implementation of acquire().\r
-\r
- For blocking calls, self.locked_status is automatically set to\r
- True and returned appropriately based on value of\r
- ``waitflag``. If it is non-blocking, then the value is\r
- actually checked and not set if it is already acquired. This\r
- is all done so that threading.Condition's assert statements\r
- aren't triggered and throw a little fit.\r
-\r
- """\r
- if waitflag is None or waitflag:\r
- self.locked_status = True\r
- return True\r
- else:\r
- if not self.locked_status:\r
- self.locked_status = True\r
- return True\r
- else:\r
- return False\r
-\r
- __enter__ = acquire\r
-\r
- def __exit__(self, typ, val, tb):\r
- self.release()\r
-\r
- def release(self):\r
- """Release the dummy lock."""\r
- # XXX Perhaps shouldn't actually bother to test? Could lead\r
- # to problems for complex, threaded code.\r
- if not self.locked_status:\r
- raise error\r
- self.locked_status = False\r
- return True\r
-\r
- def locked(self):\r
- return self.locked_status\r
-\r
-# Used to signal that interrupt_main was called in a "thread"\r
-_interrupt = False\r
-# True when not executing in a "thread"\r
-_main = True\r
-\r
-def interrupt_main():\r
- """Set _interrupt flag to True to have start_new_thread raise\r
- KeyboardInterrupt upon exiting."""\r
- if _main:\r
- raise KeyboardInterrupt\r
- else:\r
- global _interrupt\r
- _interrupt = True\r