Lib/threading.py

   1 """Thread module emulating a subset of Java's threading model."""
   2
   3 import sys as _sys
   4
   5 try:
   6     import thread
   7 except ImportError:
   8     del _sys.modules[__name__]
   9     raise
  10
  11 import warnings
  12
  13 from time import time as _time, sleep as _sleep
  14 from traceback import format_exc as _format_exc
  15 from collections import deque
  16
  17 # Note regarding PEP 8 compliant aliases
  18 #  This threading model was originally inspired by Java, and inherited
  19 # the convention of camelCase function and method names from that
  20 # language. While those names are not in any imminent danger of being
  21 # deprecated, starting with Python 2.6, the module now provides a
  22 # PEP 8 compliant alias for any such method name.
  23 # Using the new PEP 8 compliant names also facilitates substitution
  24 # with the multiprocessing module, which doesn't provide the old
  25 # Java inspired names.
  26
  27
  28 # Rename some stuff so "from threading import *" is safe
  29 __all__ = ['activeCount', 'active_count', 'Condition', 'currentThread',
  30            'current_thread', 'enumerate', 'Event',
  31            'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',
  32            'Timer', 'setprofile', 'settrace', 'local', 'stack_size']
  33
  34 _start_new_thread = thread.start_new_thread
  35 _allocate_lock = thread.allocate_lock
  36 _get_ident = thread.get_ident
  37 ThreadError = thread.error
  38 del thread
  39
  40
  41 # sys.exc_clear is used to work around the fact that except blocks
  42 # don't fully clear the exception until 3.0.
  43 warnings.filterwarnings('ignore', category=DeprecationWarning,
  44                         module='threading', message='sys.exc_clear')
  45
  46 # Debug support (adapted from ihooks.py).
  47 # All the major classes here derive from _Verbose.  We force that to
  48 # be a new-style class so that all the major classes here are new-style.
  49 # This helps debugging (type(instance) is more revealing for instances
  50 # of new-style classes).
  51
  52 _VERBOSE = False
  53
  54 if __debug__:
  55
  56     class _Verbose(object):
  57
  58         def __init__(self, verbose=None):
  59             if verbose is None:
  60                 verbose = _VERBOSE
  61             self.__verbose = verbose
  62
  63         def _note(self, format, *args):
  64             if self.__verbose:
  65                 format = format % args
  66                 format = "%s: %s\n" % (
  67                     current_thread().name, format)
  68                 _sys.stderr.write(format)
  69
  70 else:
  71     # Disable this when using "python -O"
  72     class _Verbose(object):
  73         def __init__(self, verbose=None):
  74             pass
  75         def _note(self, *args):
  76             pass
  77
  78 # Support for profile and trace hooks
  79
  80 _profile_hook = None
  81 _trace_hook = None
  82
  83 def setprofile(func):
  84     global _profile_hook
  85     _profile_hook = func
  86
  87 def settrace(func):
  88     global _trace_hook
  89     _trace_hook = func
  90
  91 # Synchronization classes
  92
  93 Lock = _allocate_lock
  94
  95 def RLock(*args, **kwargs):
  96     return _RLock(*args, **kwargs)
  97
  98 class _RLock(_Verbose):
  99
 100     def __init__(self, verbose=None):
 101         _Verbose.__init__(self, verbose)
 102         self.__block = _allocate_lock()
 103         self.__owner = None
 104         self.__count = 0
 105
 106     def __repr__(self):
 107         owner = self.__owner
 108         try:
 109             owner = _active[owner].name
 110         except KeyError:
 111             pass
 112         return "<%s owner=%r count=%d>" % (
 113                 self.__class__.__name__, owner, self.__count)
 114
 115     def acquire(self, blocking=1):
 116         me = _get_ident()
 117         if self.__owner == me:
 118             self.__count = self.__count + 1
 119             if __debug__:
 120                 self._note("%s.acquire(%s): recursive success", self, blocking)
 121             return 1
 122         rc = self.__block.acquire(blocking)
 123         if rc:
 124             self.__owner = me
 125             self.__count = 1
 126             if __debug__:
 127                 self._note("%s.acquire(%s): initial success", self, blocking)
 128         else:
 129             if __debug__:
 130                 self._note("%s.acquire(%s): failure", self, blocking)
 131         return rc
 132
 133     __enter__ = acquire
 134
 135     def release(self):
 136         if self.__owner != _get_ident():
 137             raise RuntimeError("cannot release un-acquired lock")
 138         self.__count = count = self.__count - 1
 139         if not count:
 140             self.__owner = None
 141             self.__block.release()
 142             if __debug__:
 143                 self._note("%s.release(): final release", self)
 144         else:
 145             if __debug__:
 146                 self._note("%s.release(): non-final release", self)
 147
 148     def __exit__(self, t, v, tb):
 149         self.release()
 150
 151     # Internal methods used by condition variables
 152
 153     def _acquire_restore(self, count_owner):
 154         count, owner = count_owner
 155         self.__block.acquire()
 156         self.__count = count
 157         self.__owner = owner
 158         if __debug__:
 159             self._note("%s._acquire_restore()", self)
 160
 161     def _release_save(self):
 162         if __debug__:
 163             self._note("%s._release_save()", self)
 164         count = self.__count
 165         self.__count = 0
 166         owner = self.__owner
 167         self.__owner = None
 168         self.__block.release()
 169         return (count, owner)
 170
 171     def _is_owned(self):
 172         return self.__owner == _get_ident()
 173
 174
 175 def Condition(*args, **kwargs):
 176     return _Condition(*args, **kwargs)
 177
 178 class _Condition(_Verbose):
 179
 180     def __init__(self, lock=None, verbose=None):
 181         _Verbose.__init__(self, verbose)
 182         if lock is None:
 183             lock = RLock()
 184         self.__lock = lock
 185         # Export the lock's acquire() and release() methods
 186         self.acquire = lock.acquire
 187         self.release = lock.release
 188         # If the lock defines _release_save() and/or _acquire_restore(),
 189         # these override the default implementations (which just call
 190         # release() and acquire() on the lock).  Ditto for _is_owned().
 191         try:
 192             self._release_save = lock._release_save
 193         except AttributeError:
 194             pass
 195         try:
 196             self._acquire_restore = lock._acquire_restore
 197         except AttributeError:
 198             pass
 199         try:
 200             self._is_owned = lock._is_owned
 201         except AttributeError:
 202             pass
 203         self.__waiters = []
 204
 205     def __enter__(self):
 206         return self.__lock.__enter__()
 207
 208     def __exit__(self, *args):
 209         return self.__lock.__exit__(*args)
 210
 211     def __repr__(self):
 212         return "<Condition(%s, %d)>" % (self.__lock, len(self.__waiters))
 213
 214     def _release_save(self):
 215         self.__lock.release()           # No state to save
 216
 217     def _acquire_restore(self, x):
 218         self.__lock.acquire()           # Ignore saved state
 219
 220     def _is_owned(self):
 221         # Return True if lock is owned by current_thread.
 222         # This method is called only if __lock doesn't have _is_owned().
 223         if self.__lock.acquire(0):
 224             self.__lock.release()
 225             return False
 226         else:
 227             return True
 228
 229     def wait(self, timeout=None):
 230         if not self._is_owned():
 231             raise RuntimeError("cannot wait on un-acquired lock")
 232         waiter = _allocate_lock()
 233         waiter.acquire()
 234         self.__waiters.append(waiter)
 235         saved_state = self._release_save()
 236         try:    # restore state no matter what (e.g., KeyboardInterrupt)
 237             if timeout is None:
 238                 waiter.acquire()
 239                 if __debug__:
 240                     self._note("%s.wait(): got it", self)
 241             else:
 242                 # Balancing act:  We can't afford a pure busy loop, so we
 243                 # have to sleep; but if we sleep the whole timeout time,
 244                 # we'll be unresponsive.  The scheme here sleeps very
 245                 # little at first, longer as time goes on, but never longer
 246                 # than 20 times per second (or the timeout time remaining).
 247                 endtime = _time() + timeout
 248                 delay = 0.0005 # 500 us -> initial delay of 1 ms
 249                 while True:
 250                     gotit = waiter.acquire(0)
 251                     if gotit:
 252                         break
 253                     remaining = endtime - _time()
 254                     if remaining <= 0:
 255                         break
 256                     delay = min(delay * 2, remaining, .05)
 257                     _sleep(delay)
 258                 if not gotit:
 259                     if __debug__:
 260                         self._note("%s.wait(%s): timed out", self, timeout)
 261                     try:
 262                         self.__waiters.remove(waiter)
 263                     except ValueError:
 264                         pass
 265                 else:
 266                     if __debug__:
 267                         self._note("%s.wait(%s): got it", self, timeout)
 268         finally:
 269             self._acquire_restore(saved_state)
 270
 271     def notify(self, n=1):
 272         if not self._is_owned():
 273             raise RuntimeError("cannot notify on un-acquired lock")
 274         __waiters = self.__waiters
 275         waiters = __waiters[:n]
 276         if not waiters:
 277             if __debug__:
 278                 self._note("%s.notify(): no waiters", self)
 279             return
 280         self._note("%s.notify(): notifying %d waiter%s", self, n,
 281                    n!=1 and "s" or "")
 282         for waiter in waiters:
 283             waiter.release()
 284             try:
 285                 __waiters.remove(waiter)
 286             except ValueError:
 287                 pass
 288
 289     def notifyAll(self):
 290         self.notify(len(self.__waiters))
 291
 292     notify_all = notifyAll
 293
 294
 295 def Semaphore(*args, **kwargs):
 296     return _Semaphore(*args, **kwargs)
 297
 298 class _Semaphore(_Verbose):
 299
 300     # After Tim Peters' semaphore class, but not quite the same (no maximum)
 301
 302     def __init__(self, value=1, verbose=None):
 303         if value < 0:
 304             raise ValueError("semaphore initial value must be >= 0")
 305         _Verbose.__init__(self, verbose)
 306         self.__cond = Condition(Lock())
 307         self.__value = value
 308
 309     def acquire(self, blocking=1):
 310         rc = False
 311         self.__cond.acquire()
 312         while self.__value == 0:
 313             if not blocking:
 314                 break
 315             if __debug__:
 316                 self._note("%s.acquire(%s): blocked waiting, value=%s",
 317                            self, blocking, self.__value)
 318             self.__cond.wait()
 319         else:
 320             self.__value = self.__value - 1
 321             if __debug__:
 322                 self._note("%s.acquire: success, value=%s",
 323                            self, self.__value)
 324             rc = True
 325         self.__cond.release()
 326         return rc
 327
 328     __enter__ = acquire
 329
 330     def release(self):
 331         self.__cond.acquire()
 332         self.__value = self.__value + 1
 333         if __debug__:
 334             self._note("%s.release: success, value=%s",
 335                        self, self.__value)
 336         self.__cond.notify()
 337         self.__cond.release()
 338
 339     def __exit__(self, t, v, tb):
 340         self.release()
 341
 342
 343 def BoundedSemaphore(*args, **kwargs):
 344     return _BoundedSemaphore(*args, **kwargs)
 345
 346 class _BoundedSemaphore(_Semaphore):
 347     """Semaphore that checks that # releases is <= # acquires"""
 348     def __init__(self, value=1, verbose=None):
 349         _Semaphore.__init__(self, value, verbose)
 350         self._initial_value = value
 351
 352     def release(self):
 353         if self._Semaphore__value >= self._initial_value:
 354             raise ValueError, "Semaphore released too many times"
 355         return _Semaphore.release(self)
 356
 357
 358 def Event(*args, **kwargs):
 359     return _Event(*args, **kwargs)
 360
 361 class _Event(_Verbose):
 362
 363     # After Tim Peters' event class (without is_posted())
 364
 365     def __init__(self, verbose=None):
 366         _Verbose.__init__(self, verbose)
 367         self.__cond = Condition(Lock())
 368         self.__flag = False
 369
 370     def isSet(self):
 371         return self.__flag
 372
 373     is_set = isSet
 374
 375     def set(self):
 376         self.__cond.acquire()
 377         try:
 378             self.__flag = True
 379             self.__cond.notify_all()
 380         finally:
 381             self.__cond.release()
 382
 383     def clear(self):
 384         self.__cond.acquire()
 385         try:
 386             self.__flag = False
 387         finally:
 388             self.__cond.release()
 389
 390     def wait(self, timeout=None):
 391         self.__cond.acquire()
 392         try:
 393             if not self.__flag:
 394                 self.__cond.wait(timeout)
 395             return self.__flag
 396         finally:
 397             self.__cond.release()
 398
 399 # Helper to generate new thread names
 400 _counter = 0
 401 def _newname(template="Thread-%d"):
 402     global _counter
 403     _counter = _counter + 1
 404     return template % _counter
 405
 406 # Active thread administration
 407 _active_limbo_lock = _allocate_lock()
 408 _active = {}    # maps thread id to Thread object
 409 _limbo = {}
 410
 411
 412 # Main class for threads
 413
 414 class Thread(_Verbose):
 415
 416     __initialized = False
 417     # Need to store a reference to sys.exc_info for printing
 418     # out exceptions when a thread tries to use a global var. during interp.
 419     # shutdown and thus raises an exception about trying to perform some
 420     # operation on/with a NoneType
 421     __exc_info = _sys.exc_info
 422     # Keep sys.exc_clear too to clear the exception just before
 423     # allowing .join() to return.
 424     __exc_clear = _sys.exc_clear
 425
 426     def __init__(self, group=None, target=None, name=None,
 427                  args=(), kwargs=None, verbose=None):
 428         assert group is None, "group argument must be None for now"
 429         _Verbose.__init__(self, verbose)
 430         if kwargs is None:
 431             kwargs = {}
 432         self.__target = target
 433         self.__name = str(name or _newname())
 434         self.__args = args
 435         self.__kwargs = kwargs
 436         self.__daemonic = self._set_daemon()
 437         self.__ident = None
 438         self.__started = Event()
 439         self.__stopped = False
 440         self.__block = Condition(Lock())
 441         self.__initialized = True
 442         # sys.stderr is not stored in the class like
 443         # sys.exc_info since it can be changed between instances
 444         self.__stderr = _sys.stderr
 445
 446     def _set_daemon(self):
 447         # Overridden in _MainThread and _DummyThread
 448         return current_thread().daemon
 449
 450     def __repr__(self):
 451         assert self.__initialized, "Thread.__init__() was not called"
 452         status = "initial"
 453         if self.__started.is_set():
 454             status = "started"
 455         if self.__stopped:
 456             status = "stopped"
 457         if self.__daemonic:
 458             status += " daemon"
 459         if self.__ident is not None:
 460             status += " %s" % self.__ident
 461         return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
 462
 463     def start(self):
 464         if not self.__initialized:
 465             raise RuntimeError("thread.__init__() not called")
 466         if self.__started.is_set():
 467             raise RuntimeError("threads can only be started once")
 468         if __debug__:
 469             self._note("%s.start(): starting thread", self)
 470         with _active_limbo_lock:
 471             _limbo[self] = self
 472         try:
 473             _start_new_thread(self.__bootstrap, ())
 474         except Exception:
 475             with _active_limbo_lock:
 476                 del _limbo[self]
 477             raise
 478         self.__started.wait()
 479
 480     def run(self):
 481         try:
 482             if self.__target:
 483                 self.__target(*self.__args, **self.__kwargs)
 484         finally:
 485             # Avoid a refcycle if the thread is running a function with
 486             # an argument that has a member that points to the thread.
 487             del self.__target, self.__args, self.__kwargs
 488
 489     def __bootstrap(self):
 490         # Wrapper around the real bootstrap code that ignores
 491         # exceptions during interpreter cleanup.  Those typically
 492         # happen when a daemon thread wakes up at an unfortunate
 493         # moment, finds the world around it destroyed, and raises some
 494         # random exception *** while trying to report the exception in
 495         # __bootstrap_inner() below ***.  Those random exceptions
 496         # don't help anybody, and they confuse users, so we suppress
 497         # them.  We suppress them only when it appears that the world
 498         # indeed has already been destroyed, so that exceptions in
 499         # __bootstrap_inner() during normal business hours are properly
 500         # reported.  Also, we only suppress them for daemonic threads;
 501         # if a non-daemonic encounters this, something else is wrong.
 502         try:
 503             self.__bootstrap_inner()
 504         except:
 505             if self.__daemonic and _sys is None:
 506                 return
 507             raise
 508
 509     def _set_ident(self):
 510         self.__ident = _get_ident()
 511
 512     def __bootstrap_inner(self):
 513         try:
 514             self._set_ident()
 515             self.__started.set()
 516             with _active_limbo_lock:
 517                 _active[self.__ident] = self
 518                 del _limbo[self]
 519             if __debug__:
 520                 self._note("%s.__bootstrap(): thread started", self)
 521
 522             if _trace_hook:
 523                 self._note("%s.__bootstrap(): registering trace hook", self)
 524                 _sys.settrace(_trace_hook)
 525             if _profile_hook:
 526                 self._note("%s.__bootstrap(): registering profile hook", self)
 527                 _sys.setprofile(_profile_hook)
 528
 529             try:
 530                 self.run()
 531             except SystemExit:
 532                 if __debug__:
 533                     self._note("%s.__bootstrap(): raised SystemExit", self)
 534             except:
 535                 if __debug__:
 536                     self._note("%s.__bootstrap(): unhandled exception", self)
 537                 # If sys.stderr is no more (most likely from interpreter
 538                 # shutdown) use self.__stderr.  Otherwise still use sys (as in
 539                 # _sys) in case sys.stderr was redefined since the creation of
 540                 # self.
 541                 if _sys:
 542                     _sys.stderr.write("Exception in thread %s:\n%s\n" %
 543                                       (self.name, _format_exc()))
 544                 else:
 545                     # Do the best job possible w/o a huge amt. of code to
 546                     # approximate a traceback (code ideas from
 547                     # Lib/traceback.py)
 548                     exc_type, exc_value, exc_tb = self.__exc_info()
 549                     try:
 550                         print>>self.__stderr, (
 551                             "Exception in thread " + self.name +
 552                             " (most likely raised during interpreter shutdown):")
 553                         print>>self.__stderr, (
 554                             "Traceback (most recent call last):")
 555                         while exc_tb:
 556                             print>>self.__stderr, (
 557                                 '  File "%s", line %s, in %s' %
 558                                 (exc_tb.tb_frame.f_code.co_filename,
 559                                     exc_tb.tb_lineno,
 560                                     exc_tb.tb_frame.f_code.co_name))
 561                             exc_tb = exc_tb.tb_next
 562                         print>>self.__stderr, ("%s: %s" % (exc_type, exc_value))
 563                     # Make sure that exc_tb gets deleted since it is a memory
 564                     # hog; deleting everything else is just for thoroughness
 565                     finally:
 566                         del exc_type, exc_value, exc_tb
 567             else:
 568                 if __debug__:
 569                     self._note("%s.__bootstrap(): normal return", self)
 570             finally:
 571                 # Prevent a race in
 572                 # test_threading.test_no_refcycle_through_target when
 573                 # the exception keeps the target alive past when we
 574                 # assert that it's dead.
 575                 self.__exc_clear()
 576         finally:
 577             with _active_limbo_lock:
 578                 self.__stop()
 579                 try:
 580                     # We don't call self.__delete() because it also
 581                     # grabs _active_limbo_lock.
 582                     del _active[_get_ident()]
 583                 except:
 584                     pass
 585
 586     def __stop(self):
 587         self.__block.acquire()
 588         self.__stopped = True
 589         self.__block.notify_all()
 590         self.__block.release()
 591
 592     def __delete(self):
 593         "Remove current thread from the dict of currently running threads."
 594
 595         # Notes about running with dummy_thread:
 596         #
 597         # Must take care to not raise an exception if dummy_thread is being
 598         # used (and thus this module is being used as an instance of
 599         # dummy_threading).  dummy_thread.get_ident() always returns -1 since
 600         # there is only one thread if dummy_thread is being used.  Thus
 601         # len(_active) is always <= 1 here, and any Thread instance created
 602         # overwrites the (if any) thread currently registered in _active.
 603         #
 604         # An instance of _MainThread is always created by 'threading'.  This
 605         # gets overwritten the instant an instance of Thread is created; both
 606         # threads return -1 from dummy_thread.get_ident() and thus have the
 607         # same key in the dict.  So when the _MainThread instance created by
 608         # 'threading' tries to clean itself up when atexit calls this method
 609         # it gets a KeyError if another Thread instance was created.
 610         #
 611         # This all means that KeyError from trying to delete something from
 612         # _active if dummy_threading is being used is a red herring.  But
 613         # since it isn't if dummy_threading is *not* being used then don't
 614         # hide the exception.
 615
 616         try:
 617             with _active_limbo_lock:
 618                 del _active[_get_ident()]
 619                 # There must not be any python code between the previous line
 620                 # and after the lock is released.  Otherwise a tracing function
 621                 # could try to acquire the lock again in the same thread, (in
 622                 # current_thread()), and would block.
 623         except KeyError:
 624             if 'dummy_threading' not in _sys.modules:
 625                 raise
 626
 627     def join(self, timeout=None):
 628         if not self.__initialized:
 629             raise RuntimeError("Thread.__init__() not called")
 630         if not self.__started.is_set():
 631             raise RuntimeError("cannot join thread before it is started")
 632         if self is current_thread():
 633             raise RuntimeError("cannot join current thread")
 634
 635         if __debug__:
 636             if not self.__stopped:
 637                 self._note("%s.join(): waiting until thread stops", self)
 638         self.__block.acquire()
 639         try:
 640             if timeout is None:
 641                 while not self.__stopped:
 642                     self.__block.wait()
 643                 if __debug__:
 644                     self._note("%s.join(): thread stopped", self)
 645             else:
 646                 deadline = _time() + timeout
 647                 while not self.__stopped:
 648                     delay = deadline - _time()
 649                     if delay <= 0:
 650                         if __debug__:
 651                             self._note("%s.join(): timed out", self)
 652                         break
 653                     self.__block.wait(delay)
 654                 else:
 655                     if __debug__:
 656                         self._note("%s.join(): thread stopped", self)
 657         finally:
 658             self.__block.release()
 659
 660     @property
 661     def name(self):
 662         assert self.__initialized, "Thread.__init__() not called"
 663         return self.__name
 664
 665     @name.setter
 666     def name(self, name):
 667         assert self.__initialized, "Thread.__init__() not called"
 668         self.__name = str(name)
 669
 670     @property
 671     def ident(self):
 672         assert self.__initialized, "Thread.__init__() not called"
 673         return self.__ident
 674
 675     def isAlive(self):
 676         assert self.__initialized, "Thread.__init__() not called"
 677         return self.__started.is_set() and not self.__stopped
 678
 679     is_alive = isAlive
 680
 681     @property
 682     def daemon(self):
 683         assert self.__initialized, "Thread.__init__() not called"
 684         return self.__daemonic
 685
 686     @daemon.setter
 687     def daemon(self, daemonic):
 688         if not self.__initialized:
 689             raise RuntimeError("Thread.__init__() not called")
 690         if self.__started.is_set():
 691             raise RuntimeError("cannot set daemon status of active thread");
 692         self.__daemonic = daemonic
 693
 694     def isDaemon(self):
 695         return self.daemon
 696
 697     def setDaemon(self, daemonic):
 698         self.daemon = daemonic
 699
 700     def getName(self):
 701         return self.name
 702
 703     def setName(self, name):
 704         self.name = name
 705
 706 # The timer class was contributed by Itamar Shtull-Trauring
 707
 708 def Timer(*args, **kwargs):
 709     return _Timer(*args, **kwargs)
 710
 711 class _Timer(Thread):
 712     """Call a function after a specified number of seconds:
 713
 714     t = Timer(30.0, f, args=[], kwargs={})
 715     t.start()
 716     t.cancel() # stop the timer's action if it's still waiting
 717     """
 718
 719     def __init__(self, interval, function, args=[], kwargs={}):
 720         Thread.__init__(self)
 721         self.interval = interval
 722         self.function = function
 723         self.args = args
 724         self.kwargs = kwargs
 725         self.finished = Event()
 726
 727     def cancel(self):
 728         """Stop the timer if it hasn't finished yet"""
 729         self.finished.set()
 730
 731     def run(self):
 732         self.finished.wait(self.interval)
 733         if not self.finished.is_set():
 734             self.function(*self.args, **self.kwargs)
 735         self.finished.set()
 736
 737 # Special thread class to represent the main thread
 738 # This is garbage collected through an exit handler
 739
 740 class _MainThread(Thread):
 741
 742     def __init__(self):
 743         Thread.__init__(self, name="MainThread")
 744         self._Thread__started.set()
 745         self._set_ident()
 746         with _active_limbo_lock:
 747             _active[_get_ident()] = self
 748
 749     def _set_daemon(self):
 750         return False
 751
 752     def _exitfunc(self):
 753         self._Thread__stop()
 754         t = _pickSomeNonDaemonThread()
 755         if t:
 756             if __debug__:
 757                 self._note("%s: waiting for other threads", self)
 758         while t:
 759             t.join()
 760             t = _pickSomeNonDaemonThread()
 761         if __debug__:
 762             self._note("%s: exiting", self)
 763         self._Thread__delete()
 764
 765 def _pickSomeNonDaemonThread():
 766     for t in enumerate():
 767         if not t.daemon and t.is_alive():
 768             return t
 769     return None
 770
 771
 772 # Dummy thread class to represent threads not started here.
 773 # These aren't garbage collected when they die, nor can they be waited for.
 774 # If they invoke anything in threading.py that calls current_thread(), they
 775 # leave an entry in the _active dict forever after.
 776 # Their purpose is to return *something* from current_thread().
 777 # They are marked as daemon threads so we won't wait for them
 778 # when we exit (conform previous semantics).
 779
 780 class _DummyThread(Thread):
 781
 782     def __init__(self):
 783         Thread.__init__(self, name=_newname("Dummy-%d"))
 784
 785         # Thread.__block consumes an OS-level locking primitive, which
 786         # can never be used by a _DummyThread.  Since a _DummyThread
 787         # instance is immortal, that's bad, so release this resource.
 788         del self._Thread__block
 789
 790         self._Thread__started.set()
 791         self._set_ident()
 792         with _active_limbo_lock:
 793             _active[_get_ident()] = self
 794
 795     def _set_daemon(self):
 796         return True
 797
 798     def join(self, timeout=None):
 799         assert False, "cannot join a dummy thread"
 800
 801
 802 # Global API functions
 803
 804 def currentThread():
 805     try:
 806         return _active[_get_ident()]
 807     except KeyError:
 808         ##print "current_thread(): no current thread for", _get_ident()
 809         return _DummyThread()
 810
 811 current_thread = currentThread
 812
 813 def activeCount():
 814     with _active_limbo_lock:
 815         return len(_active) + len(_limbo)
 816
 817 active_count = activeCount
 818
 819 def _enumerate():
 820     # Same as enumerate(), but without the lock. Internal use only.
 821     return _active.values() + _limbo.values()
 822
 823 def enumerate():
 824     with _active_limbo_lock:
 825         return _active.values() + _limbo.values()
 826
 827 from thread import stack_size
 828
 829 # Create the main thread object,
 830 # and make it available for the interpreter
 831 # (Py_Main) as threading._shutdown.
 832
 833 _shutdown = _MainThread()._exitfunc
 834
 835 # get thread-local implementation, either from the thread
 836 # module, or from the python fallback
 837
 838 try:
 839     from thread import _local as local
 840 except ImportError:
 841     from _threading_local import local
 842
 843
 844 def _after_fork():
 845     # This function is called by Python/ceval.c:PyEval_ReInitThreads which
 846     # is called from PyOS_AfterFork.  Here we cleanup threading module state
 847     # that should not exist after a fork.
 848
 849     # Reset _active_limbo_lock, in case we forked while the lock was held
 850     # by another (non-forked) thread.  http://bugs.python.org/issue874900
 851     global _active_limbo_lock
 852     _active_limbo_lock = _allocate_lock()
 853
 854     # fork() only copied the current thread; clear references to others.
 855     new_active = {}
 856     current = current_thread()
 857     with _active_limbo_lock:
 858         for thread in _active.itervalues():
 859             if thread is current:
 860                 # There is only one active thread. We reset the ident to
 861                 # its new value since it can have changed.
 862                 ident = _get_ident()
 863                 thread._Thread__ident = ident
 864                 new_active[ident] = thread
 865             else:
 866                 # All the others are already stopped.
 867                 # We don't call _Thread__stop() because it tries to acquire
 868                 # thread._Thread__block which could also have been held while
 869                 # we forked.
 870                 thread._Thread__stopped = True
 871
 872         _limbo.clear()
 873         _active.clear()
 874         _active.update(new_active)
 875         assert len(_active) == 1
 876
 877
 878 # Self-test code
 879
 880 def _test():
 881
 882     class BoundedQueue(_Verbose):
 883
 884         def __init__(self, limit):
 885             _Verbose.__init__(self)
 886             self.mon = RLock()
 887             self.rc = Condition(self.mon)
 888             self.wc = Condition(self.mon)
 889             self.limit = limit
 890             self.queue = deque()
 891
 892         def put(self, item):
 893             self.mon.acquire()
 894             while len(self.queue) >= self.limit:
 895                 self._note("put(%s): queue full", item)
 896                 self.wc.wait()
 897             self.queue.append(item)
 898             self._note("put(%s): appended, length now %d",
 899                        item, len(self.queue))
 900             self.rc.notify()
 901             self.mon.release()
 902
 903         def get(self):
 904             self.mon.acquire()
 905             while not self.queue:
 906                 self._note("get(): queue empty")
 907                 self.rc.wait()
 908             item = self.queue.popleft()
 909             self._note("get(): got %s, %d left", item, len(self.queue))
 910             self.wc.notify()
 911             self.mon.release()
 912             return item
 913
 914     class ProducerThread(Thread):
 915
 916         def __init__(self, queue, quota):
 917             Thread.__init__(self, name="Producer")
 918             self.queue = queue
 919             self.quota = quota
 920
 921         def run(self):
 922             from random import random
 923             counter = 0
 924             while counter < self.quota:
 925                 counter = counter + 1
 926                 self.queue.put("%s.%d" % (self.name, counter))
 927                 _sleep(random() * 0.00001)
 928
 929
 930     class ConsumerThread(Thread):
 931
 932         def __init__(self, queue, count):
 933             Thread.__init__(self, name="Consumer")
 934             self.queue = queue
 935             self.count = count
 936
 937         def run(self):
 938             while self.count > 0:
 939                 item = self.queue.get()
 940                 print item
 941                 self.count = self.count - 1
 942
 943     NP = 3
 944     QL = 4
 945     NI = 5
 946
 947     Q = BoundedQueue(QL)
 948     P = []
 949     for i in range(NP):
 950         t = ProducerThread(Q, NI)
 951         t.name = ("Producer-%d" % (i+1))
 952         P.append(t)
 953     C = ConsumerThread(Q, NI*NP)
 954     for t in P:
 955         t.start()
 956         _sleep(0.000001)
 957     C.start()
 958     for t in P:
 959         t.join()
 960     C.join()
 961
 962 if __name__ == '__main__':
 963     _test()