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