git.gag.com Git - debian/gnuradio/blob - gnuradio-core/src/python/gnuradio/gr/gr_threading_23.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.3
   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 StringIO import StringIO as _StringIO
  16 from time import time as _time, sleep as _sleep
  17 from traceback import print_exc as _print_exc
  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']
  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         currentThread() # for side-effect
 201         assert self._is_owned(), "wait() of un-acquire()d lock"
 202         waiter = _allocate_lock()
 203         waiter.acquire()
 204         self.__waiters.append(waiter)
 205         saved_state = self._release_save()
 206         try:    # restore state no matter what (e.g., KeyboardInterrupt)
 207             if timeout is None:
 208                 waiter.acquire()
 209                 if __debug__:
 210                     self._note("%s.wait(): got it", self)
 211             else:
 212                 # Balancing act:  We can't afford a pure busy loop, so we
 213                 # have to sleep; but if we sleep the whole timeout time,
 214                 # we'll be unresponsive.  The scheme here sleeps very
 215                 # little at first, longer as time goes on, but never longer
 216                 # than 20 times per second (or the timeout time remaining).
 217                 endtime = _time() + timeout
 218                 delay = 0.0005 # 500 us -> initial delay of 1 ms
 219                 while True:
 220                     gotit = waiter.acquire(0)
 221                     if gotit:
 222                         break
 223                     remaining = endtime - _time()
 224                     if remaining <= 0:
 225                         break
 226                     delay = min(delay * 2, remaining, .05)
 227                     _sleep(delay)
 228                 if not gotit:
 229                     if __debug__:
 230                         self._note("%s.wait(%s): timed out", self, timeout)
 231                     try:
 232                         self.__waiters.remove(waiter)
 233                     except ValueError:
 234                         pass
 235                 else:
 236                     if __debug__:
 237                         self._note("%s.wait(%s): got it", self, timeout)
 238         finally:
 239             self._acquire_restore(saved_state)
 240
 241     def notify(self, n=1):
 242         currentThread() # for side-effect
 243         assert self._is_owned(), "notify() of un-acquire()d lock"
 244         __waiters = self.__waiters
 245         waiters = __waiters[:n]
 246         if not waiters:
 247             if __debug__:
 248                 self._note("%s.notify(): no waiters", self)
 249             return
 250         self._note("%s.notify(): notifying %d waiter%s", self, n,
 251                    n!=1 and "s" or "")
 252         for waiter in waiters:
 253             waiter.release()
 254             try:
 255                 __waiters.remove(waiter)
 256             except ValueError:
 257                 pass
 258
 259     def notifyAll(self):
 260         self.notify(len(self.__waiters))
 261
 262
 263 def Semaphore(*args, **kwargs):
 264     return _Semaphore(*args, **kwargs)
 265
 266 class _Semaphore(_Verbose):
 267
 268     # After Tim Peters' semaphore class, but not quite the same (no maximum)
 269
 270     def __init__(self, value=1, verbose=None):
 271         assert value >= 0, "Semaphore initial value must be >= 0"
 272         _Verbose.__init__(self, verbose)
 273         self.__cond = Condition(Lock())
 274         self.__value = value
 275
 276     def acquire(self, blocking=1):
 277         rc = False
 278         self.__cond.acquire()
 279         while self.__value == 0:
 280             if not blocking:
 281                 break
 282             if __debug__:
 283                 self._note("%s.acquire(%s): blocked waiting, value=%s",
 284                            self, blocking, self.__value)
 285             self.__cond.wait()
 286         else:
 287             self.__value = self.__value - 1
 288             if __debug__:
 289                 self._note("%s.acquire: success, value=%s",
 290                            self, self.__value)
 291             rc = True
 292         self.__cond.release()
 293         return rc
 294
 295     def release(self):
 296         self.__cond.acquire()
 297         self.__value = self.__value + 1
 298         if __debug__:
 299             self._note("%s.release: success, value=%s",
 300                        self, self.__value)
 301         self.__cond.notify()
 302         self.__cond.release()
 303
 304
 305 def BoundedSemaphore(*args, **kwargs):
 306     return _BoundedSemaphore(*args, **kwargs)
 307
 308 class _BoundedSemaphore(_Semaphore):
 309     """Semaphore that checks that # releases is <= # acquires"""
 310     def __init__(self, value=1, verbose=None):
 311         _Semaphore.__init__(self, value, verbose)
 312         self._initial_value = value
 313
 314     def release(self):
 315         if self._Semaphore__value >= self._initial_value:
 316             raise ValueError, "Semaphore released too many times"
 317         return _Semaphore.release(self)
 318
 319
 320 def Event(*args, **kwargs):
 321     return _Event(*args, **kwargs)
 322
 323 class _Event(_Verbose):
 324
 325     # After Tim Peters' event class (without is_posted())
 326
 327     def __init__(self, verbose=None):
 328         _Verbose.__init__(self, verbose)
 329         self.__cond = Condition(Lock())
 330         self.__flag = False
 331
 332     def isSet(self):
 333         return self.__flag
 334
 335     def set(self):
 336         self.__cond.acquire()
 337         try:
 338             self.__flag = True
 339             self.__cond.notifyAll()
 340         finally:
 341             self.__cond.release()
 342
 343     def clear(self):
 344         self.__cond.acquire()
 345         try:
 346             self.__flag = False
 347         finally:
 348             self.__cond.release()
 349
 350     def wait(self, timeout=None):
 351         self.__cond.acquire()
 352         try:
 353             if not self.__flag:
 354                 self.__cond.wait(timeout)
 355         finally:
 356             self.__cond.release()
 357
 358 # Helper to generate new thread names
 359 _counter = 0
 360 def _newname(template="Thread-%d"):
 361     global _counter
 362     _counter = _counter + 1
 363     return template % _counter
 364
 365 # Active thread administration
 366 _active_limbo_lock = _allocate_lock()
 367 _active = {}
 368 _limbo = {}
 369
 370
 371 # Main class for threads
 372
 373 class Thread(_Verbose):
 374
 375     __initialized = False
 376
 377     def __init__(self, group=None, target=None, name=None,
 378                  args=(), kwargs={}, verbose=None):
 379         assert group is None, "group argument must be None for now"
 380         _Verbose.__init__(self, verbose)
 381         self.__target = target
 382         self.__name = str(name or _newname())
 383         self.__args = args
 384         self.__kwargs = kwargs
 385         self.__daemonic = self._set_daemon()
 386         self.__started = False
 387         self.__stopped = False
 388         self.__block = Condition(Lock())
 389         self.__initialized = True
 390
 391     def _set_daemon(self):
 392         # Overridden in _MainThread and _DummyThread
 393         return currentThread().isDaemon()
 394
 395     def __repr__(self):
 396         assert self.__initialized, "Thread.__init__() was not called"
 397         status = "initial"
 398         if self.__started:
 399             status = "started"
 400         if self.__stopped:
 401             status = "stopped"
 402         if self.__daemonic:
 403             status = status + " daemon"
 404         return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
 405
 406     def start(self):
 407         assert self.__initialized, "Thread.__init__() not called"
 408         assert not self.__started, "thread already started"
 409         if __debug__:
 410             self._note("%s.start(): starting thread", self)
 411         _active_limbo_lock.acquire()
 412         _limbo[self] = self
 413         _active_limbo_lock.release()
 414         _start_new_thread(self.__bootstrap, ())
 415         self.__started = True
 416         _sleep(0.000001)    # 1 usec, to let the thread run (Solaris hack)
 417
 418     def run(self):
 419         if self.__target:
 420             self.__target(*self.__args, **self.__kwargs)
 421
 422     def __bootstrap(self):
 423         try:
 424             self.__started = True
 425             _active_limbo_lock.acquire()
 426             _active[_get_ident()] = self
 427             del _limbo[self]
 428             _active_limbo_lock.release()
 429             if __debug__:
 430                 self._note("%s.__bootstrap(): thread started", self)
 431
 432             if _trace_hook:
 433                 self._note("%s.__bootstrap(): registering trace hook", self)
 434                 _sys.settrace(_trace_hook)
 435             if _profile_hook:
 436                 self._note("%s.__bootstrap(): registering profile hook", self)
 437                 _sys.setprofile(_profile_hook)
 438
 439             try:
 440                 self.run()
 441             except SystemExit:
 442                 if __debug__:
 443                     self._note("%s.__bootstrap(): raised SystemExit", self)
 444             except:
 445                 if __debug__:
 446                     self._note("%s.__bootstrap(): unhandled exception", self)
 447                 s = _StringIO()
 448                 _print_exc(file=s)
 449                 _sys.stderr.write("Exception in thread %s:\n%s\n" %
 450                                  (self.getName(), s.getvalue()))
 451             else:
 452                 if __debug__:
 453                     self._note("%s.__bootstrap(): normal return", self)
 454         finally:
 455             self.__stop()
 456             try:
 457                 self.__delete()
 458             except:
 459                 pass
 460
 461     def __stop(self):
 462         self.__block.acquire()
 463         self.__stopped = True
 464         self.__block.notifyAll()
 465         self.__block.release()
 466
 467     def __delete(self):
 468         _active_limbo_lock.acquire()
 469         del _active[_get_ident()]
 470         _active_limbo_lock.release()
 471
 472     def join(self, timeout=None):
 473         assert self.__initialized, "Thread.__init__() not called"
 474         assert self.__started, "cannot join thread before it is started"
 475         assert self is not currentThread(), "cannot join current thread"
 476         if __debug__:
 477             if not self.__stopped:
 478                 self._note("%s.join(): waiting until thread stops", self)
 479         self.__block.acquire()
 480         try:
 481             if timeout is None:
 482                 while not self.__stopped:
 483                     self.__block.wait()
 484                 if __debug__:
 485                     self._note("%s.join(): thread stopped", self)
 486             else:
 487                 deadline = _time() + timeout
 488                 while not self.__stopped:
 489                     delay = deadline - _time()
 490                     if delay <= 0:
 491                         if __debug__:
 492                             self._note("%s.join(): timed out", self)
 493                         break
 494                     self.__block.wait(delay)
 495                 else:
 496                     if __debug__:
 497                         self._note("%s.join(): thread stopped", self)
 498         finally:
 499             self.__block.release()
 500
 501     def getName(self):
 502         assert self.__initialized, "Thread.__init__() not called"
 503         return self.__name
 504
 505     def setName(self, name):
 506         assert self.__initialized, "Thread.__init__() not called"
 507         self.__name = str(name)
 508
 509     def isAlive(self):
 510         assert self.__initialized, "Thread.__init__() not called"
 511         return self.__started and not self.__stopped
 512
 513     def isDaemon(self):
 514         assert self.__initialized, "Thread.__init__() not called"
 515         return self.__daemonic
 516
 517     def setDaemon(self, daemonic):
 518         assert self.__initialized, "Thread.__init__() not called"
 519         assert not self.__started, "cannot set daemon status of active thread"
 520         self.__daemonic = daemonic
 521
 522 # The timer class was contributed by Itamar Shtull-Trauring
 523
 524 def Timer(*args, **kwargs):
 525     return _Timer(*args, **kwargs)
 526
 527 class _Timer(Thread):
 528     """Call a function after a specified number of seconds:
 529
 530     t = Timer(30.0, f, args=[], kwargs={})
 531     t.start()
 532     t.cancel() # stop the timer's action if it's still waiting
 533     """
 534
 535     def __init__(self, interval, function, args=[], kwargs={}):
 536         Thread.__init__(self)
 537         self.interval = interval
 538         self.function = function
 539         self.args = args
 540         self.kwargs = kwargs
 541         self.finished = Event()
 542
 543     def cancel(self):
 544         """Stop the timer if it hasn't finished yet"""
 545         self.finished.set()
 546
 547     def run(self):
 548         self.finished.wait(self.interval)
 549         if not self.finished.isSet():
 550             self.function(*self.args, **self.kwargs)
 551         self.finished.set()
 552
 553 # Special thread class to represent the main thread
 554 # This is garbage collected through an exit handler
 555
 556 class _MainThread(Thread):
 557
 558     def __init__(self):
 559         Thread.__init__(self, name="MainThread")
 560         self._Thread__started = True
 561         _active_limbo_lock.acquire()
 562         _active[_get_ident()] = self
 563         _active_limbo_lock.release()
 564         import atexit
 565         atexit.register(self.__exitfunc)
 566
 567     def _set_daemon(self):
 568         return False
 569
 570     def __exitfunc(self):
 571         self._Thread__stop()
 572         t = _pickSomeNonDaemonThread()
 573         if t:
 574             if __debug__:
 575                 self._note("%s: waiting for other threads", self)
 576         while t:
 577             t.join()
 578             t = _pickSomeNonDaemonThread()
 579         if __debug__:
 580             self._note("%s: exiting", self)
 581         self._Thread__delete()
 582
 583 def _pickSomeNonDaemonThread():
 584     for t in enumerate():
 585         if not t.isDaemon() and t.isAlive():
 586             return t
 587     return None
 588
 589
 590 # Dummy thread class to represent threads not started here.
 591 # These aren't garbage collected when they die,
 592 # nor can they be waited for.
 593 # Their purpose is to return *something* from currentThread().
 594 # They are marked as daemon threads so we won't wait for them
 595 # when we exit (conform previous semantics).
 596
 597 class _DummyThread(Thread):
 598
 599     def __init__(self):
 600         Thread.__init__(self, name=_newname("Dummy-%d"))
 601         self._Thread__started = True
 602         _active_limbo_lock.acquire()
 603         _active[_get_ident()] = self
 604         _active_limbo_lock.release()
 605
 606     def _set_daemon(self):
 607         return True
 608
 609     def join(self, timeout=None):
 610         assert False, "cannot join a dummy thread"
 611
 612
 613 # Global API functions
 614
 615 def currentThread():
 616     try:
 617         return _active[_get_ident()]
 618     except KeyError:
 619         ##print "currentThread(): no current thread for", _get_ident()
 620         return _DummyThread()
 621
 622 def activeCount():
 623     _active_limbo_lock.acquire()
 624     count = len(_active) + len(_limbo)
 625     _active_limbo_lock.release()
 626     return count
 627
 628 def enumerate():
 629     _active_limbo_lock.acquire()
 630     active = _active.values() + _limbo.values()
 631     _active_limbo_lock.release()
 632     return active
 633
 634 # Create the main thread object
 635
 636 _MainThread()
 637
 638
 639 # Self-test code
 640
 641 def _test():
 642
 643     class BoundedQueue(_Verbose):
 644
 645         def __init__(self, limit):
 646             _Verbose.__init__(self)
 647             self.mon = RLock()
 648             self.rc = Condition(self.mon)
 649             self.wc = Condition(self.mon)
 650             self.limit = limit
 651             self.queue = []
 652
 653         def put(self, item):
 654             self.mon.acquire()
 655             while len(self.queue) >= self.limit:
 656                 self._note("put(%s): queue full", item)
 657                 self.wc.wait()
 658             self.queue.append(item)
 659             self._note("put(%s): appended, length now %d",
 660                        item, len(self.queue))
 661             self.rc.notify()
 662             self.mon.release()
 663
 664         def get(self):
 665             self.mon.acquire()
 666             while not self.queue:
 667                 self._note("get(): queue empty")
 668                 self.rc.wait()
 669             item = self.queue.pop(0)
 670             self._note("get(): got %s, %d left", item, len(self.queue))
 671             self.wc.notify()
 672             self.mon.release()
 673             return item
 674
 675     class ProducerThread(Thread):
 676
 677         def __init__(self, queue, quota):
 678             Thread.__init__(self, name="Producer")
 679             self.queue = queue
 680             self.quota = quota
 681
 682         def run(self):
 683             from random import random
 684             counter = 0
 685             while counter < self.quota:
 686                 counter = counter + 1
 687                 self.queue.put("%s.%d" % (self.getName(), counter))
 688                 _sleep(random() * 0.00001)
 689
 690
 691     class ConsumerThread(Thread):
 692
 693         def __init__(self, queue, count):
 694             Thread.__init__(self, name="Consumer")
 695             self.queue = queue
 696             self.count = count
 697
 698         def run(self):
 699             while self.count > 0:
 700                 item = self.queue.get()
 701                 print item
 702                 self.count = self.count - 1
 703
 704     NP = 3
 705     QL = 4
 706     NI = 5
 707
 708     Q = BoundedQueue(QL)
 709     P = []
 710     for i in range(NP):
 711         t = ProducerThread(Q, NI)
 712         t.setName("Producer-%d" % (i+1))
 713         P.append(t)
 714     C = ConsumerThread(Q, NI*NP)
 715     for t in P:
 716         t.start()
 717         _sleep(0.000001)
 718     C.start()
 719     for t in P:
 720         t.join()
 721     C.join()
 722
 723 if __name__ == '__main__':
 724     _test()