AppPkg/Applications/Python/Python-2.7.10: Initial Checkin part 4/5.

[mirror_edk2.git] / AppPkg / Applications / Python / Python-2.7.10 / Lib / dummy_thread.py

diff --git a/AppPkg/Applications/Python/Python-2.7.10/Lib/dummy_thread.py b/AppPkg/Applications/Python/Python-2.7.10/Lib/dummy_thread.py
new file mode 100644 (file)
index 0000000..4b2da7d
--- /dev/null
+++ b/AppPkg/Applications/Python/Python-2.7.10/Lib/dummy_thread.py
@@ -0,0 +1,145 @@
+"""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