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