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 StringIO import StringIO as _StringIO
  12 from time import time as _time, sleep as _sleep
  13 from traceback import print_exc as _print_exc
  14
  15 # Rename some stuff so "from threading import *" is safe
  16 __all__ = ['activeCount', 'Condition', 'currentThread', 'enumerate', 'Event',
  17            'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread', 'Timer']
  18
  19 _start_new_thread = thread.start_new_thread
  20 _allocate_lock = thread.allocate_lock
  21 _get_ident = thread.get_ident
  22 ThreadError = thread.error
  23 del thread
  24
  25
  26 # Debug support (adapted from ihooks.py)
  27
  28 _VERBOSE = 0 # XXX Bool or int?
  29
  30 if __debug__:
  31
  32     class _Verbose:
  33
  34         def __init__(self, verbose=None):
  35             if verbose is None:
  36                 verbose = _VERBOSE
  37             self.__verbose = verbose
  38
  39         def _note(self, format, *args):
  40             if self.__verbose:
  41                 format = format % args
  42                 format = "%s: %s\n" % (
  43                     currentThread().getName(), format)
  44                 _sys.stderr.write(format)
  45
  46 else:
  47     # Disable this when using "python -O"
  48     class _Verbose:
  49         def __init__(self, verbose=None):
  50             pass
  51         def _note(self, *args):
  52             pass
  53
  54
  55 # Synchronization classes
  56
  57 Lock = _allocate_lock
  58
  59 def RLock(*args, **kwargs):
  60     return apply(_RLock, args, kwargs)
  61
  62 class _RLock(_Verbose):
  63
  64     def __init__(self, verbose=None):
  65         _Verbose.__init__(self, verbose)
  66         self.__block = _allocate_lock()
  67         self.__owner = None
  68         self.__count = 0
  69
  70     def __repr__(self):
  71         return "<%s(%s, %d)>" % (
  72                 self.__class__.__name__,
  73                 self.__owner and self.__owner.getName(),
  74                 self.__count)
  75
  76     def acquire(self, blocking=1):
  77         me = currentThread()
  78         if self.__owner is me:
  79             self.__count = self.__count + 1
  80             if __debug__:
  81                 self._note("%s.acquire(%s): recursive success", self, blocking)
  82             return 1
  83         rc = self.__block.acquire(blocking)
  84         if rc:
  85             self.__owner = me
  86             self.__count = 1
  87             if __debug__:
  88                 self._note("%s.acquire(%s): initial succes", self, blocking)
  89         else:
  90             if __debug__:
  91                 self._note("%s.acquire(%s): failure", self, blocking)
  92         return rc
  93
  94     def release(self):
  95         me = currentThread()
  96         assert self.__owner is me, "release() of un-acquire()d lock"
  97         self.__count = count = self.__count - 1
  98         if not count:
  99             self.__owner = None
 100             self.__block.release()
 101             if __debug__:
 102                 self._note("%s.release(): final release", self)
 103         else:
 104             if __debug__:
 105                 self._note("%s.release(): non-final release", self)
 106
 107     # Internal methods used by condition variables
 108
 109     def _acquire_restore(self, (count, owner)):
 110         self.__block.acquire()
 111         self.__count = count
 112         self.__owner = owner
 113         if __debug__:
 114             self._note("%s._acquire_restore()", self)
 115
 116     def _release_save(self):
 117         if __debug__:
 118             self._note("%s._release_save()", self)
 119         count = self.__count
 120         self.__count = 0
 121         owner = self.__owner
 122         self.__owner = None
 123         self.__block.release()
 124         return (count, owner)
 125
 126     def _is_owned(self):
 127         return self.__owner is currentThread()
 128
 129
 130 def Condition(*args, **kwargs):
 131     return apply(_Condition, args, kwargs)
 132
 133 class _Condition(_Verbose):
 134
 135     def __init__(self, lock=None, verbose=None):
 136         _Verbose.__init__(self, verbose)
 137         if lock is None:
 138             lock = RLock()
 139         self.__lock = lock
 140         # Export the lock's acquire() and release() methods
 141         self.acquire = lock.acquire
 142         self.release = lock.release
 143         # If the lock defines _release_save() and/or _acquire_restore(),
 144         # these override the default implementations (which just call
 145         # release() and acquire() on the lock).  Ditto for _is_owned().
 146         try:
 147             self._release_save = lock._release_save
 148         except AttributeError:
 149             pass
 150         try:
 151             self._acquire_restore = lock._acquire_restore
 152         except AttributeError:
 153             pass
 154         try:
 155             self._is_owned = lock._is_owned
 156         except AttributeError:
 157             pass
 158         self.__waiters = []
 159
 160     def __repr__(self):
 161         return "<Condition(%s, %d)>" % (self.__lock, len(self.__waiters))
 162
 163     def _release_save(self):
 164         self.__lock.release()           # No state to save
 165
 166     def _acquire_restore(self, x):
 167         self.__lock.acquire()           # Ignore saved state
 168
 169     def _is_owned(self):
 170         # Return True if lock is owned by currentThread.
 171         # This method is called only if __lock doesn't have _is_owned().
 172         if self.__lock.acquire(0):
 173             self.__lock.release()
 174             return False
 175         else:
 176             return True
 177
 178     def wait(self, timeout=None):
 179         currentThread() # for side-effect
 180         assert self._is_owned(), "wait() of un-acquire()d lock"
 181         waiter = _allocate_lock()
 182         waiter.acquire()
 183         self.__waiters.append(waiter)
 184         saved_state = self._release_save()
 185         try:    # restore state no matter what (e.g., KeyboardInterrupt)
 186             if timeout is None:
 187                 waiter.acquire()
 188                 if __debug__:
 189                     self._note("%s.wait(): got it", self)
 190             else:
 191                 # Balancing act:  We can't afford a pure busy loop, so we
 192                 # have to sleep; but if we sleep the whole timeout time,
 193                 # we'll be unresponsive.  The scheme here sleeps very
 194                 # little at first, longer as time goes on, but never longer
 195                 # than 20 times per second (or the timeout time remaining).
 196                 endtime = _time() + timeout
 197                 delay = 0.0005 # 500 us -> initial delay of 1 ms
 198                 while True:
 199                     gotit = waiter.acquire(0)
 200                     if gotit:
 201                         break
 202                     remaining = endtime - _time()
 203                     if remaining <= 0:
 204                         break
 205                     delay = min(delay * 2, remaining, .05)
 206                     _sleep(delay)
 207                 if not gotit:
 208                     if __debug__:
 209                         self._note("%s.wait(%s): timed out", self, timeout)
 210                     try:
 211                         self.__waiters.remove(waiter)
 212                     except ValueError:
 213                         pass
 214                 else:
 215                     if __debug__:
 216                         self._note("%s.wait(%s): got it", self, timeout)
 217         finally:
 218             self._acquire_restore(saved_state)
 219
 220     def notify(self, n=1):
 221         currentThread() # for side-effect
 222         assert self._is_owned(), "notify() of un-acquire()d lock"
 223         __waiters = self.__waiters
 224         waiters = __waiters[:n]
 225         if not waiters:
 226             if __debug__:
 227                 self._note("%s.notify(): no waiters", self)
 228             return
 229         self._note("%s.notify(): notifying %d waiter%s", self, n,
 230                    n!=1 and "s" or "")
 231         for waiter in waiters:
 232             waiter.release()
 233             try:
 234                 __waiters.remove(waiter)
 235             except ValueError:
 236                 pass
 237
 238     def notifyAll(self):
 239         self.notify(len(self.__waiters))
 240
 241
 242 def Semaphore(*args, **kwargs):
 243     return apply(_Semaphore, args, kwargs)
 244
 245 class _Semaphore(_Verbose):
 246
 247     # After Tim Peters' semaphore class, but not quite the same (no maximum)
 248
 249     def __init__(self, value=1, verbose=None):
 250         assert value >= 0, "Semaphore initial value must be >= 0"
 251         _Verbose.__init__(self, verbose)
 252         self.__cond = Condition(Lock())
 253         self.__value = value
 254
 255     def acquire(self, blocking=1):
 256         rc = False
 257         self.__cond.acquire()
 258         while self.__value == 0:
 259             if not blocking:
 260                 break
 261             if __debug__:
 262                 self._note("%s.acquire(%s): blocked waiting, value=%s",
 263                            self, blocking, self.__value)
 264             self.__cond.wait()
 265         else:
 266             self.__value = self.__value - 1
 267             if __debug__:
 268                 self._note("%s.acquire: success, value=%s",
 269                            self, self.__value)
 270             rc = True
 271         self.__cond.release()
 272         return rc
 273
 274     def release(self):
 275         self.__cond.acquire()
 276         self.__value = self.__value + 1
 277         if __debug__:
 278             self._note("%s.release: success, value=%s",
 279                        self, self.__value)
 280         self.__cond.notify()
 281         self.__cond.release()
 282
 283
 284 def BoundedSemaphore(*args, **kwargs):
 285     return apply(_BoundedSemaphore, args, kwargs)
 286
 287 class _BoundedSemaphore(_Semaphore):
 288     """Semaphore that checks that # releases is <= # acquires"""
 289     def __init__(self, value=1, verbose=None):
 290         _Semaphore.__init__(self, value, verbose)
 291         self._initial_value = value
 292
 293     def release(self):
 294         if self._Semaphore__value >= self._initial_value:
 295             raise ValueError, "Semaphore released too many times"
 296         return _Semaphore.release(self)
 297
 298
 299 def Event(*args, **kwargs):
 300     return apply(_Event, args, kwargs)
 301
 302 class _Event(_Verbose):
 303
 304     # After Tim Peters' event class (without is_posted())
 305
 306     def __init__(self, verbose=None):
 307         _Verbose.__init__(self, verbose)
 308         self.__cond = Condition(Lock())
 309         self.__flag = False
 310
 311     def isSet(self):
 312         return self.__flag
 313
 314     def set(self):
 315         self.__cond.acquire()
 316         try:
 317             self.__flag = True
 318             self.__cond.notifyAll()
 319         finally:
 320             self.__cond.release()
 321
 322     def clear(self):
 323         self.__cond.acquire()
 324         try:
 325             self.__flag = False
 326         finally:
 327             self.__cond.release()
 328
 329     def wait(self, timeout=None):
 330         self.__cond.acquire()
 331         try:
 332             if not self.__flag:
 333                 self.__cond.wait(timeout)
 334         finally:
 335             self.__cond.release()
 336
 337 # Helper to generate new thread names
 338 _counter = 0
 339 def _newname(template="Thread-%d"):
 340     global _counter
 341     _counter = _counter + 1
 342     return template % _counter
 343
 344 # Active thread administration
 345 _active_limbo_lock = _allocate_lock()
 346 _active = {}
 347 _limbo = {}
 348
 349
 350 # Main class for threads
 351
 352 class Thread(_Verbose):
 353
 354     __initialized = False
 355
 356     def __init__(self, group=None, target=None, name=None,
 357                  args=(), kwargs={}, verbose=None):
 358         assert group is None, "group argument must be None for now"
 359         _Verbose.__init__(self, verbose)
 360         self.__target = target
 361         self.__name = str(name or _newname())
 362         self.__args = args
 363         self.__kwargs = kwargs
 364         self.__daemonic = self._set_daemon()
 365         self.__started = False
 366         self.__stopped = False
 367         self.__block = Condition(Lock())
 368         self.__initialized = True
 369
 370     def _set_daemon(self):
 371         # Overridden in _MainThread and _DummyThread
 372         return currentThread().isDaemon()
 373
 374     def __repr__(self):
 375         assert self.__initialized, "Thread.__init__() was not called"
 376         status = "initial"
 377         if self.__started:
 378             status = "started"
 379         if self.__stopped:
 380             status = "stopped"
 381         if self.__daemonic:
 382             status = status + " daemon"
 383         return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
 384
 385     def start(self):
 386         assert self.__initialized, "Thread.__init__() not called"
 387         assert not self.__started, "thread already started"
 388         if __debug__:
 389             self._note("%s.start(): starting thread", self)
 390         _active_limbo_lock.acquire()
 391         _limbo[self] = self
 392         _active_limbo_lock.release()
 393         _start_new_thread(self.__bootstrap, ())
 394         self.__started = True
 395         _sleep(0.000001)    # 1 usec, to let the thread run (Solaris hack)
 396
 397     def run(self):
 398         if self.__target:
 399             apply(self.__target, self.__args, self.__kwargs)
 400
 401     def __bootstrap(self):
 402         try:
 403             self.__started = True
 404             _active_limbo_lock.acquire()
 405             _active[_get_ident()] = self
 406             del _limbo[self]
 407             _active_limbo_lock.release()
 408             if __debug__:
 409                 self._note("%s.__bootstrap(): thread started", self)
 410             try:
 411                 self.run()
 412             except SystemExit:
 413                 if __debug__:
 414                     self._note("%s.__bootstrap(): raised SystemExit", self)
 415             except:
 416                 if __debug__:
 417                     self._note("%s.__bootstrap(): unhandled exception", self)
 418                 s = _StringIO()
 419                 _print_exc(file=s)
 420                 _sys.stderr.write("Exception in thread %s:\n%s\n" %
 421                                  (self.getName(), s.getvalue()))
 422             else:
 423                 if __debug__:
 424                     self._note("%s.__bootstrap(): normal return", self)
 425         finally:
 426             self.__stop()
 427             try:
 428                 self.__delete()
 429             except:
 430                 pass
 431
 432     def __stop(self):
 433         self.__block.acquire()
 434         self.__stopped = True
 435         self.__block.notifyAll()
 436         self.__block.release()
 437
 438     def __delete(self):
 439         _active_limbo_lock.acquire()
 440         del _active[_get_ident()]
 441         _active_limbo_lock.release()
 442
 443     def join(self, timeout=None):
 444         assert self.__initialized, "Thread.__init__() not called"
 445         assert self.__started, "cannot join thread before it is started"
 446         assert self is not currentThread(), "cannot join current thread"
 447         if __debug__:
 448             if not self.__stopped:
 449                 self._note("%s.join(): waiting until thread stops", self)
 450         self.__block.acquire()
 451         if timeout is None:
 452             while not self.__stopped:
 453                 self.__block.wait()
 454             if __debug__:
 455                 self._note("%s.join(): thread stopped", self)
 456         else:
 457             deadline = _time() + timeout
 458             while not self.__stopped:
 459                 delay = deadline - _time()
 460                 if delay <= 0:
 461                     if __debug__:
 462                         self._note("%s.join(): timed out", self)
 463                     break
 464                 self.__block.wait(delay)
 465             else:
 466                 if __debug__:
 467                     self._note("%s.join(): thread stopped", self)
 468         self.__block.release()
 469
 470     def getName(self):
 471         assert self.__initialized, "Thread.__init__() not called"
 472         return self.__name
 473
 474     def setName(self, name):
 475         assert self.__initialized, "Thread.__init__() not called"
 476         self.__name = str(name)
 477
 478     def isAlive(self):
 479         assert self.__initialized, "Thread.__init__() not called"
 480         return self.__started and not self.__stopped
 481
 482     def isDaemon(self):
 483         assert self.__initialized, "Thread.__init__() not called"
 484         return self.__daemonic
 485
 486     def setDaemon(self, daemonic):
 487         assert self.__initialized, "Thread.__init__() not called"
 488         assert not self.__started, "cannot set daemon status of active thread"
 489         self.__daemonic = daemonic
 490
 491 # The timer class was contributed by Itamar Shtull-Trauring
 492
 493 def Timer(*args, **kwargs):
 494     return _Timer(*args, **kwargs)
 495
 496 class _Timer(Thread):
 497     """Call a function after a specified number of seconds:
 498
 499     t = Timer(30.0, f, args=[], kwargs={})
 500     t.start()
 501     t.cancel() # stop the timer's action if it's still waiting
 502     """
 503
 504     def __init__(self, interval, function, args=[], kwargs={}):
 505         Thread.__init__(self)
 506         self.interval = interval
 507         self.function = function
 508         self.args = args
 509         self.kwargs = kwargs
 510         self.finished = Event()
 511
 512     def cancel(self):
 513         """Stop the timer if it hasn't finished yet"""
 514         self.finished.set()
 515
 516     def run(self):
 517         self.finished.wait(self.interval)
 518         if not self.finished.isSet():
 519             self.function(*self.args, **self.kwargs)
 520         self.finished.set()
 521
 522 # Special thread class to represent the main thread
 523 # This is garbage collected through an exit handler
 524
 525 class _MainThread(Thread):
 526
 527     def __init__(self):
 528         Thread.__init__(self, name="MainThread")
 529         self._Thread__started = True
 530         _active_limbo_lock.acquire()
 531         _active[_get_ident()] = self
 532         _active_limbo_lock.release()
 533         import atexit
 534         atexit.register(self.__exitfunc)
 535
 536     def _set_daemon(self):
 537         return False
 538
 539     def __exitfunc(self):
 540         self._Thread__stop()
 541         t = _pickSomeNonDaemonThread()
 542         if t:
 543             if __debug__:
 544                 self._note("%s: waiting for other threads", self)
 545         while t:
 546             t.join()
 547             t = _pickSomeNonDaemonThread()
 548         if __debug__:
 549             self._note("%s: exiting", self)
 550         self._Thread__delete()
 551
 552 def _pickSomeNonDaemonThread():
 553     for t in enumerate():
 554         if not t.isDaemon() and t.isAlive():
 555             return t
 556     return None
 557
 558
 559 # Dummy thread class to represent threads not started here.
 560 # These aren't garbage collected when they die,
 561 # nor can they be waited for.
 562 # Their purpose is to return *something* from currentThread().
 563 # They are marked as daemon threads so we won't wait for them
 564 # when we exit (conform previous semantics).
 565
 566 class _DummyThread(Thread):
 567
 568     def __init__(self):
 569         Thread.__init__(self, name=_newname("Dummy-%d"))
 570         self._Thread__started = True
 571         _active_limbo_lock.acquire()
 572         _active[_get_ident()] = self
 573         _active_limbo_lock.release()
 574
 575     def _set_daemon(self):
 576         return True
 577
 578     def join(self, timeout=None):
 579         assert False, "cannot join a dummy thread"
 580
 581
 582 # Global API functions
 583
 584 def currentThread():
 585     try:
 586         return _active[_get_ident()]
 587     except KeyError:
 588         ##print "currentThread(): no current thread for", _get_ident()
 589         return _DummyThread()
 590
 591 def activeCount():
 592     _active_limbo_lock.acquire()
 593     count = len(_active) + len(_limbo)
 594     _active_limbo_lock.release()
 595     return count
 596
 597 def enumerate():
 598     _active_limbo_lock.acquire()
 599     active = _active.values() + _limbo.values()
 600     _active_limbo_lock.release()
 601     return active
 602
 603
 604 # Create the main thread object
 605
 606 _MainThread()
 607
 608
 609 # Self-test code
 610
 611 def _test():
 612
 613     class BoundedQueue(_Verbose):
 614
 615         def __init__(self, limit):
 616             _Verbose.__init__(self)
 617             self.mon = RLock()
 618             self.rc = Condition(self.mon)
 619             self.wc = Condition(self.mon)
 620             self.limit = limit
 621             self.queue = []
 622
 623         def put(self, item):
 624             self.mon.acquire()
 625             while len(self.queue) >= self.limit:
 626                 self._note("put(%s): queue full", item)
 627                 self.wc.wait()
 628             self.queue.append(item)
 629             self._note("put(%s): appended, length now %d",
 630                        item, len(self.queue))
 631             self.rc.notify()
 632             self.mon.release()
 633
 634         def get(self):
 635             self.mon.acquire()
 636             while not self.queue:
 637                 self._note("get(): queue empty")
 638                 self.rc.wait()
 639             item = self.queue.pop(0)
 640             self._note("get(): got %s, %d left", item, len(self.queue))
 641             self.wc.notify()
 642             self.mon.release()
 643             return item
 644
 645     class ProducerThread(Thread):
 646
 647         def __init__(self, queue, quota):
 648             Thread.__init__(self, name="Producer")
 649             self.queue = queue
 650             self.quota = quota
 651
 652         def run(self):
 653             from random import random
 654             counter = 0
 655             while counter < self.quota:
 656                 counter = counter + 1
 657                 self.queue.put("%s.%d" % (self.getName(), counter))
 658                 _sleep(random() * 0.00001)
 659
 660
 661     class ConsumerThread(Thread):
 662
 663         def __init__(self, queue, count):
 664             Thread.__init__(self, name="Consumer")
 665             self.queue = queue
 666             self.count = count
 667
 668         def run(self):
 669             while self.count > 0:
 670                 item = self.queue.get()
 671                 print item
 672                 self.count = self.count - 1
 673
 674     NP = 3
 675     QL = 4
 676     NI = 5
 677
 678     Q = BoundedQueue(QL)
 679     P = []
 680     for i in range(NP):
 681         t = ProducerThread(Q, NI)
 682         t.setName("Producer-%d" % (i+1))
 683         P.append(t)
 684     C = ConsumerThread(Q, NI*NP)
 685     for t in P:
 686         t.start()
 687         _sleep(0.000001)
 688     C.start()
 689     for t in P:
 690         t.join()
 691     C.join()
 692
 693 if __name__ == '__main__':
 694     _test()