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