"""Thread module emulating a subset of Java's threading model.""" import sys as _sys try:    import threadexcept ImportError:    del _sys.modules[__name__]    raise from time import time as _time, sleep as _sleepfrom traceback import format_exc as _format_excfrom collections import deque # Rename some stuff so "from threading import *" is safe__all__ = ['activeCount', 'Condition', 'currentThread', 'enumerate', 'Event',           'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',           'Timer', 'setprofile', 'settrace', 'local', 'stack_size'] _start_new_thread = thread.start_new_thread_allocate_lock = thread.allocate_lock_get_ident = thread.get_identThreadError = thread.errordel thread  # Debug support (adapted from ihooks.py).# All the major classes here derive from _Verbose.  We force that to# be a new-style class so that all the major classes here are new-style.# This helps debugging (type(instance) is more revealing for instances# of new-style classes). _VERBOSE = False if __debug__:     class _Verbose(object):         def __init__(self, verbose=None):            if verbose is None:                verbose = _VERBOSE            self.__verbose = verbose         def _note(self, format, *args):            if self.__verbose:                format = format % args                format = "%s: %s\n" % (                    currentThread().getName(), format)                _sys.stderr.write(format) else:    # Disable this when using "python -O"    class _Verbose(object):        def __init__(self, verbose=None):            pass        def _note(self, *args):            pass # Support for profile and trace hooks _profile_hook = None_trace_hook = None def setprofile(func):    global _profile_hook    _profile_hook = func def settrace(func):    global _trace_hook    _trace_hook = func # Synchronization classes Lock = _allocate_lock def RLock(*args, **kwargs):    return _RLock(*args, **kwargs) class _RLock(_Verbose):     def __init__(self, verbose=None):        _Verbose.__init__(self, verbose)        self.__block = _allocate_lock()        self.__owner = None        self.__count = 0     def __repr__(self):        return "<%s(%s, %d)>" % (                self.__class__.__name__,                self.__owner and self.__owner.getName(),                self.__count)     def acquire(self, blocking=1):        me = currentThread()        if self.__owner is me:            self.__count = self.__count + 1            if __debug__:                self._note("%s.acquire(%s): recursive success", self, blocking)            return 1        rc = self.__block.acquire(blocking)        if rc:            self.__owner = me            self.__count = 1            if __debug__:                self._note("%s.acquire(%s): initial success", self, blocking)        else:            if __debug__:                self._note("%s.acquire(%s): failure", self, blocking)        return rc     __enter__ = acquire     def release(self):        me = currentThread()        assert self.__owner is me, "release() of un-acquire()d lock"        self.__count = count = self.__count - 1        if not count:            self.__owner = None            self.__block.release()            if __debug__:                self._note("%s.release(): final release", self)        else:            if __debug__:                self._note("%s.release(): non-final release", self)     def __exit__(self, t, v, tb):        self.release()     # Internal methods used by condition variables     def _acquire_restore(self, (count, owner)):        self.__block.acquire()        self.__count = count        self.__owner = owner        if __debug__:            self._note("%s._acquire_restore()", self)     def _release_save(self):        if __debug__:            self._note("%s._release_save()", self)        count = self.__count        self.__count = 0        owner = self.__owner        self.__owner = None        self.__block.release()        return (count, owner)     def _is_owned(self):        return self.__owner is currentThread()  def Condition(*args, **kwargs):    return _Condition(*args, **kwargs) class _Condition(_Verbose):     def __init__(self, lock=None, verbose=None):        _Verbose.__init__(self, verbose)        if lock is None:            lock = RLock()        self.__lock = lock        # Export the lock's acquire() and release() methods        self.acquire = lock.acquire        self.release = lock.release        # If the lock defines _release_save() and/or _acquire_restore(),        # these override the default implementations (which just call        # release() and acquire() on the lock).  Ditto for _is_owned().        try:            self._release_save = lock._release_save        except AttributeError:            pass        try:            self._acquire_restore = lock._acquire_restore        except AttributeError:            pass        try:            self._is_owned = lock._is_owned        except AttributeError:            pass        self.__waiters = []     def __enter__(self):        return self.__lock.__enter__()     def __exit__(self, *args):        return self.__lock.__exit__(*args)     def __repr__(self):        return "<Condition(%s, %d)>" % (self.__lock, len(self.__waiters))     def _release_save(self):        self.__lock.release()           # No state to save     def _acquire_restore(self, x):        self.__lock.acquire()           # Ignore saved state     def _is_owned(self):        # Return True if lock is owned by currentThread.        # This method is called only if __lock doesn't have _is_owned().        if self.__lock.acquire(0):            self.__lock.release()            return False        else:            return True     def wait(self, timeout=None):        assert self._is_owned(), "wait() of un-acquire()d lock"        waiter = _allocate_lock()        waiter.acquire()        self.__waiters.append(waiter)        saved_state = self._release_save()        try:    # restore state no matter what (e.g., KeyboardInterrupt)            if timeout is None:                waiter.acquire()                if __debug__:                    self._note("%s.wait(): got it", self)            else:                # Balancing act:  We can't afford a pure busy loop, so we                # have to sleep; but if we sleep the whole timeout time,                # we'll be unresponsive.  The scheme here sleeps very                # little at first, longer as time goes on, but never longer                # than 20 times per second (or the timeout time remaining).                endtime = _time() + timeout                delay = 0.0005 # 500 us -> initial delay of 1 ms                while True:                    gotit = waiter.acquire(0)                    if gotit:                        break                    remaining = endtime - _time()                    if remaining <= 0:                        break                    delay = min(delay * 2, remaining, .05)                    _sleep(delay)                if not gotit:                    if __debug__:                        self._note("%s.wait(%s): timed out", self, timeout)                    try:                        self.__waiters.remove(waiter)                    except ValueError:                        pass                else:                    if __debug__:                        self._note("%s.wait(%s): got it", self, timeout)        finally:            self._acquire_restore(saved_state)     def notify(self, n=1):        assert self._is_owned(), "notify() of un-acquire()d lock"        __waiters = self.__waiters        waiters = __waiters[:n]        if not waiters:            if __debug__:                self._note("%s.notify(): no waiters", self)            return        self._note("%s.notify(): notifying %d waiter%s", self, n,                   n!=1 and "s" or "")        for waiter in waiters:            waiter.release()            try:                __waiters.remove(waiter)            except ValueError:                pass     def notifyAll(self):        self.notify(len(self.__waiters))  def Semaphore(*args, **kwargs):    return _Semaphore(*args, **kwargs) class _Semaphore(_Verbose):     # After Tim Peters' semaphore class, but not quite the same (no maximum)     def __init__(self, value=1, verbose=None):        assert value >= 0, "Semaphore initial value must be >= 0"        _Verbose.__init__(self, verbose)        self.__cond = Condition(Lock())        self.__value = value     def acquire(self, blocking=1):        rc = False        self.__cond.acquire()        while self.__value == 0:            if not blocking:                break            if __debug__:                self._note("%s.acquire(%s): blocked waiting, value=%s",                           self, blocking, self.__value)            self.__cond.wait()        else:            self.__value = self.__value - 1            if __debug__:                self._note("%s.acquire: success, value=%s",                           self, self.__value)            rc = True        self.__cond.release()        return rc     __enter__ = acquire     def release(self):        self.__cond.acquire()        self.__value = self.__value + 1        if __debug__:            self._note("%s.release: success, value=%s",                       self, self.__value)        self.__cond.notify()        self.__cond.release()     def __exit__(self, t, v, tb):        self.release()  def BoundedSemaphore(*args, **kwargs):    return _BoundedSemaphore(*args, **kwargs) class _BoundedSemaphore(_Semaphore):    """Semaphore that checks that # releases is <= # acquires"""    def __init__(self, value=1, verbose=None):        _Semaphore.__init__(self, value, verbose)        self._initial_value = value     def release(self):        if self._Semaphore__value >= self._initial_value:            raise ValueError, "Semaphore released too many times"        return _Semaphore.release(self)  def Event(*args, **kwargs):    return _Event(*args, **kwargs) class _Event(_Verbose):     # After Tim Peters' event class (without is_posted())     def __init__(self, verbose=None):        _Verbose.__init__(self, verbose)        self.__cond = Condition(Lock())        self.__flag = False     def isSet(self):        return self.__flag     def set(self):        self.__cond.acquire()        try:            self.__flag = True            self.__cond.notifyAll()        finally:            self.__cond.release()     def clear(self):        self.__cond.acquire()        try:            self.__flag = False        finally:            self.__cond.release()     def wait(self, timeout=None):        self.__cond.acquire()        try:            if not self.__flag:                self.__cond.wait(timeout)        finally:            self.__cond.release() # Helper to generate new thread names_counter = 0def _newname(template="Thread-%d"):    global _counter    _counter = _counter + 1    return template % _counter # Active thread administration_active_limbo_lock = _allocate_lock()_active = {}    # maps thread id to Thread object_limbo = {}  # Main class for threads class Thread(_Verbose):     __initialized = False    # Need to store a reference to sys.exc_info for printing    # out exceptions when a thread tries to use a global var. during interp.    # shutdown and thus raises an exception about trying to perform some    # operation on/with a NoneType    __exc_info = _sys.exc_info     def __init__(self, group=None, target=None, name=None,                 args=(), kwargs=None, verbose=None):        assert group is None, "group argument must be None for now"        _Verbose.__init__(self, verbose)        if kwargs is None:            kwargs = {}        self.__target = target        self.__name = str(name or _newname())        self.__args = args        self.__kwargs = kwargs        self.__daemonic = self._set_daemon()        self.__started = False        self.__stopped = False        self.__block = Condition(Lock())        self.__initialized = True        # sys.stderr is not stored in the class like        # sys.exc_info since it can be changed between instances        self.__stderr = _sys.stderr     def _set_daemon(self):        # Overridden in _MainThread and _DummyThread        return currentThread().isDaemon()     def __repr__(self):        assert self.__initialized, "Thread.__init__() was not called"        status = "initial"        if self.__started:            status = "started"        if self.__stopped:            status = "stopped"        if self.__daemonic:            status = status + " daemon"        return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)     def start(self):        assert self.__initialized, "Thread.__init__() not called"        assert not self.__started, "thread already started"        if __debug__:            self._note("%s.start(): starting thread", self)        _active_limbo_lock.acquire()        _limbo[self] = self        _active_limbo_lock.release()        _start_new_thread(self.__bootstrap, ())        self.__started = True        _sleep(0.000001)    # 1 usec, to let the thread run (Solaris hack)     def run(self):        if self.__target:            self.__target(*self.__args, **self.__kwargs)     def __bootstrap(self):        try:            self.__started = True            _active_limbo_lock.acquire()            _active[_get_ident()] = self            del _limbo[self]            _active_limbo_lock.release()            if __debug__:                self._note("%s.__bootstrap(): thread started", self)             if _trace_hook:                self._note("%s.__bootstrap(): registering trace hook", self)                _sys.settrace(_trace_hook)            if _profile_hook:                self._note("%s.__bootstrap(): registering profile hook", self)                _sys.setprofile(_profile_hook)             try:                self.run()            except SystemExit:                if __debug__:                    self._note("%s.__bootstrap(): raised SystemExit", self)            except:                if __debug__:                    self._note("%s.__bootstrap(): unhandled exception", self)                # If sys.stderr is no more (most likely from interpreter                # shutdown) use self.__stderr.  Otherwise still use sys (as in                # _sys) in case sys.stderr was redefined since the creation of                # self.                if _sys:                    _sys.stderr.write("Exception in thread %s:\n%s\n" %                                      (self.getName(), _format_exc()))                else:                    # Do the best job possible w/o a huge amt. of code to                    # approximate a traceback (code ideas from                    # Lib/traceback.py)                    exc_type, exc_value, exc_tb = self.__exc_info()                    try:                        print>>self.__stderr, (                            "Exception in thread " + self.getName() +                            " (most likely raised during interpreter shutdown):")                        print>>self.__stderr, (                            "Traceback (most recent call last):")                        while exc_tb:                            print>>self.__stderr, (                                '  File "%s", line %s, in %s' %                                (exc_tb.tb_frame.f_code.co_filename,                                    exc_tb.tb_lineno,                                    exc_tb.tb_frame.f_code.co_name))                            exc_tb = exc_tb.tb_next                        print>>self.__stderr, ("%s: %s" % (exc_type, exc_value))                    # Make sure that exc_tb gets deleted since it is a memory                    # hog; deleting everything else is just for thoroughness                    finally:                        del exc_type, exc_value, exc_tb            else:                if __debug__:                    self._note("%s.__bootstrap(): normal return", self)        finally:

Lines 1–500 of 816. Load more