Lib/threading.py

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