Lib/threading.py

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