AppPkg/Applications/Python/Python-2.7.2/Lib/dummy_thread.py

   1 """Drop-in replacement for the thread module.
   2
   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.
   5
   6 Suggested usage is::
   7
   8     try:
   9         import thread
  10     except ImportError:
  11         import dummy_thread as thread
  12
  13 """
  14 # Exports only things specified by thread documentation;
  15 # skipping obsolete synonyms allocate(), start_new(), exit_thread().
  16 __all__ = ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',
  17            'interrupt_main', 'LockType']
  18
  19 import traceback as _traceback
  20
  21 class error(Exception):
  22     """Dummy implementation of thread.error."""
  23
  24     def __init__(self, *args):
  25         self.args = args
  26
  27 def start_new_thread(function, args, kwargs={}):
  28     """Dummy implementation of thread.start_new_thread().
  29
  30     Compatibility is maintained by making sure that ``args`` is a
  31     tuple and ``kwargs`` is a dictionary.  If an exception is raised
  32     and it is SystemExit (which can be done by thread.exit()) it is
  33     caught and nothing is done; all other exceptions are printed out
  34     by using traceback.print_exc().
  35
  36     If the executed function calls interrupt_main the KeyboardInterrupt will be
  37     raised when the function returns.
  38
  39     """
  40     if type(args) != type(tuple()):
  41         raise TypeError("2nd arg must be a tuple")
  42     if type(kwargs) != type(dict()):
  43         raise TypeError("3rd arg must be a dict")
  44     global _main
  45     _main = False
  46     try:
  47         function(*args, **kwargs)
  48     except SystemExit:
  49         pass
  50     except:
  51         _traceback.print_exc()
  52     _main = True
  53     global _interrupt
  54     if _interrupt:
  55         _interrupt = False
  56         raise KeyboardInterrupt
  57
  58 def exit():
  59     """Dummy implementation of thread.exit()."""
  60     raise SystemExit
  61
  62 def get_ident():
  63     """Dummy implementation of thread.get_ident().
  64
  65     Since this module should only be used when threadmodule is not
  66     available, it is safe to assume that the current process is the
  67     only thread.  Thus a constant can be safely returned.
  68     """
  69     return -1
  70
  71 def allocate_lock():
  72     """Dummy implementation of thread.allocate_lock()."""
  73     return LockType()
  74
  75 def stack_size(size=None):
  76     """Dummy implementation of thread.stack_size()."""
  77     if size is not None:
  78         raise error("setting thread stack size not supported")
  79     return 0
  80
  81 class LockType(object):
  82     """Class implementing dummy implementation of thread.LockType.
  83
  84     Compatibility is maintained by maintaining self.locked_status
  85     which is a boolean that stores the state of the lock.  Pickling of
  86     the lock, though, should not be done since if the thread module is
  87     then used with an unpickled ``lock()`` from here problems could
  88     occur from this class not having atomic methods.
  89
  90     """
  91
  92     def __init__(self):
  93         self.locked_status = False
  94
  95     def acquire(self, waitflag=None):
  96         """Dummy implementation of acquire().
  97
  98         For blocking calls, self.locked_status is automatically set to
  99         True and returned appropriately based on value of
 100         ``waitflag``.  If it is non-blocking, then the value is
 101         actually checked and not set if it is already acquired.  This
 102         is all done so that threading.Condition's assert statements
 103         aren't triggered and throw a little fit.
 104
 105         """
 106         if waitflag is None or waitflag:
 107             self.locked_status = True
 108             return True
 109         else:
 110             if not self.locked_status:
 111                 self.locked_status = True
 112                 return True
 113             else:
 114                 return False
 115
 116     __enter__ = acquire
 117
 118     def __exit__(self, typ, val, tb):
 119         self.release()
 120
 121     def release(self):
 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:
 126             raise error
 127         self.locked_status = False
 128         return True
 129
 130     def locked(self):
 131         return self.locked_status
 132
 133 # Used to signal that interrupt_main was called in a "thread"
 134 _interrupt = False
 135 # True when not executing in a "thread"
 136 _main = True
 137
 138 def interrupt_main():
 139     """Set _interrupt flag to True to have start_new_thread raise
 140     KeyboardInterrupt upon exiting."""
 141     if _main:
 142         raise KeyboardInterrupt
 143     else:
 144         global _interrupt
 145         _interrupt = True