git.gag.com Git - debian/gnuradio/blob - gnuradio-core/src/python/gnuradio/gr/gr_threading_24.py

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