A RetroSearch Logo

Home - News ( United States | United Kingdom | Italy | Germany ) - Football scores

Search Query:

Showing content from http://mail.python.org/pipermail/python-checkins/2002-December/032247.html below:

[Python-checkins] python/dist/src/Lib dummy_thread.py,NONE,1.1 dummy_threading.py,NONE,1.1

[Python-checkins] python/dist/src/Lib dummy_thread.py,NONE,1.1 dummy_threading.py,NONE,1.1gvanrossum@users.sourceforge.net gvanrossum@users.sourceforge.net
Mon, 30 Dec 2002 14:30:24 -0800
Update of /cvsroot/python/python/dist/src/Lib
In directory sc8-pr-cvs1:/tmp/cvs-serv13682/Lib

Added Files:
	dummy_thread.py dummy_threading.py 
Log Message:
Brett Cannon's dummy_thread and dummy_threading modules (SF patch
622537), with some nitpicking editorial changes.


--- NEW FILE: dummy_thread.py ---
"""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:
        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',
           'LockType']

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().

    """
    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")
    try:
        function(*args, **kwargs)
    except SystemExit:
        pass
    except:
        _traceback.print_exc()

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.
    
    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
    
    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
            return None 
        elif not waitflag:
            if not self.locked_status:
                self.locked_status = True
                return True
            else:
                return False
        else:
            self.locked_status = True
            return True 

    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

--- NEW FILE: dummy_threading.py ---
"""Faux ``threading`` version using ``dummy_thread`` instead of ``thread``.

The module ``_dummy_threading`` is added to ``sys.modules`` in order
to not have ``threading`` considered imported.  Had ``threading`` been
directly imported it would have made all subsequent imports succeed
regardless of whether ``thread`` was available which is not desired.

:Author: Brett Cannon
:Contact: brett@python.org

XXX: Try to get rid of ``_dummy_threading``.

"""
from sys import modules as sys_modules

import dummy_thread

# Declaring now so as to not have to nest ``try``s to get proper clean-up.
holding_thread = False
holding_threading = False

try:
    # Could have checked if ``thread`` was not in sys.modules and gone
    # a different route, but decided to mirror technique used with
    # ``threading`` below.
    if 'thread' in sys_modules:
        held_thread = sys_modules['thread']
        holding_thread = True
    # Must have some module named ``thread`` that implements its API
    # in order to initially import ``threading``.
    sys_modules['thread'] = sys_modules['dummy_thread']

    if 'threading' in sys_modules:
        # If ``threading`` is already imported, might as well prevent
        # trying to import it more than needed by saving it if it is
        # already imported before deleting it.
        held_threading = sys_modules['threading']
        holding_threading = True
        del sys_modules['threading']
    import threading
    # Need a copy of the code kept somewhere...
    sys_modules['_dummy_threading'] = sys_modules['threading']
    del sys_modules['threading']
    from _dummy_threading import *
    from _dummy_threading import __all__

finally:
    # Put back ``threading`` if we overwrote earlier
    if holding_threading:
        sys_modules['threading'] = held_threading
        del held_threading
    del holding_threading

    # Put back ``thread`` if we overwrote, else del the entry we made
    if holding_thread:
        sys_modules['thread'] = held_thread
        del held_thread
    else:
        del sys_modules['thread']
    del holding_thread

    del dummy_thread
    del sys_modules





RetroSearch is an open source project built by @garambo | Open a GitHub Issue

Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo

HTML: 3.2 | Encoding: UTF-8 | Version: 0.7.4