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