dummy_thread.py 4.18 KB
Newer Older
1 2 3 4 5 6
"""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::
Tim Peters's avatar
Tim Peters committed
7

8 9 10 11 12 13 14 15 16 17 18 19
    try:
        import thread
    except ImportError:
        import dummy_thread as thread

"""
__author__ = "Brett Cannon"
__email__ = "brett@python.org"

# 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',
20
           'interrupt_main', 'LockType']
21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38

import traceback as _traceback

class error(Exception):
    """Dummy implementation of thread.error."""

    def __init__(self, *args):
        self.args = args

def start_new_thread(function, args, kwargs={}):
    """Dummy implementation of thread.start_new_thread().

    Compatibility is maintained by making sure that ``args`` is a
    tuple and ``kwargs`` is a dictionary.  If an exception is raised
    and it is SystemExit (which can be done by thread.exit()) it is
    caught and nothing is done; all other exceptions are printed out
    by using traceback.print_exc().

39 40 41
    If the executed function calls interrupt_main the KeyboardInterrupt will be
    raised when the function returns.

42 43 44 45 46
    """
    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")
47 48
    global _main
    _main = False
49 50 51 52 53 54
    try:
        function(*args, **kwargs)
    except SystemExit:
        pass
    except:
        _traceback.print_exc()
55 56
    _main = True
    global _interrupt
57 58 59
    if _interrupt:
        _interrupt = False
        raise KeyboardInterrupt
60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79

def exit():
    """Dummy implementation of thread.exit()."""
    raise SystemExit

def get_ident():
    """Dummy implementation of thread.get_ident().

    Since this module should only be used when threadmodule is not
    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():
    """Dummy implementation of thread.allocate_lock()."""
    return LockType()

class LockType(object):
    """Class implementing dummy implementation of thread.LockType.
Tim Peters's avatar
Tim Peters committed
80

81 82 83 84 85 86 87 88 89 90
    Compatibility is maintained by maintaining self.locked_status
    which is a boolean that stores the state of the lock.  Pickling of
    the lock, though, should not be done since if the thread module is
    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
Tim Peters's avatar
Tim Peters committed
91

92 93 94 95 96 97 98 99 100 101 102 103 104
    def acquire(self, waitflag=None):
        """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:
            self.locked_status = True
Tim Peters's avatar
Tim Peters committed
105
            return None
106 107 108 109 110 111 112 113
        elif not waitflag:
            if not self.locked_status:
                self.locked_status = True
                return True
            else:
                return False
        else:
            self.locked_status = True
Tim Peters's avatar
Tim Peters committed
114
            return True
115 116 117 118 119 120 121 122 123 124 125 126

    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
127

128
# Used to signal that interrupt_main was called in a "thread"
129
_interrupt = False
130 131
# True when not executing in a "thread"
_main = True
132 133 134 135

def interrupt_main():
    """Set _interrupt flag to True to have start_new_thread raise
    KeyboardInterrupt upon exiting."""
136 137 138 139 140
    if _main:
        raise KeyboardInterrupt
    else:
        global _interrupt
        _interrupt = True