Lib/threading.py

   1 # threading.py:
   2 # Proposed new threading module, emulating a subset of Java's threading model
   3
   4 import sys
   5 import time
   6 import thread
   7 import traceback
   8 import StringIO
   9
  10 # Rename some stuff so "from threading import *" is safe
  11
  12 _sys = sys
  13 del sys
  14
  15 _time = time.time
  16 _sleep = time.sleep
  17 del time
  18
  19 _start_new_thread = thread.start_new_thread
  20 _allocate_lock = thread.allocate_lock
  21 _get_ident = thread.get_ident
  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
  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         if self.__lock.acquire(0):
 176             self.__lock.release()
 177             return 0
 178         else:
 179             return 1
 180
 181     def wait(self, timeout=None):
 182         me = currentThread()
 183         assert self._is_owned(), "wait() of un-acquire()d lock"
 184         waiter = _allocate_lock()
 185         waiter.acquire()
 186         self.__waiters.append(waiter)
 187         saved_state = self._release_save()
 188         if timeout is None:
 189             waiter.acquire()
 190             if __debug__:
 191                 self._note("%s.wait(): got it", self)
 192         else:
 193             endtime = _time() + timeout
 194             delay = 0.000001 # 1 usec
 195             while 1:
 196                 gotit = waiter.acquire(0)
 197                 if gotit or _time() >= endtime:
 198                     break
 199                 _sleep(delay)
 200                 if delay < 1.0:
 201                     delay = delay * 2.0
 202             if not gotit:
 203                 if __debug__:
 204                     self._note("%s.wait(%s): timed out", self, timeout)
 205                 try:
 206                     self.__waiters.remove(waiter)
 207                 except ValueError:
 208                     pass
 209             else:
 210                 if __debug__:
 211                     self._note("%s.wait(%s): got it", self, timeout)
 212         self._acquire_restore(saved_state)
 213
 214     def notify(self, n=1):
 215         me = currentThread()
 216         assert self._is_owned(), "notify() of un-acquire()d lock"
 217         __waiters = self.__waiters
 218         waiters = __waiters[:n]
 219         if not waiters:
 220             if __debug__:
 221                 self._note("%s.notify(): no waiters", self)
 222             return
 223         self._note("%s.notify(): notifying %d waiter%s", self, n,
 224                    n!=1 and "s" or "")
 225         for waiter in waiters:
 226             waiter.release()
 227             try:
 228                 __waiters.remove(waiter)
 229             except ValueError:
 230                 pass
 231
 232     def notifyAll(self):
 233         self.notify(len(self.__waiters))
 234
 235
 236 def Semaphore(*args, **kwargs):
 237     return apply(_Semaphore, args, kwargs)
 238
 239 class _Semaphore(_Verbose):
 240
 241     # After Tim Peters' semaphore class, but bnot quite the same (no maximum)
 242
 243     def __init__(self, value=1, verbose=None):
 244         assert value >= 0, "Semaphore initial value must be >= 0"
 245         _Verbose.__init__(self, verbose)
 246         self.__cond = Condition(Lock())
 247         self.__value = value
 248
 249     def acquire(self, blocking=1):
 250         rc = 0
 251         self.__cond.acquire()
 252         while self.__value == 0:
 253             if not blocking:
 254                 break
 255             self.__cond.wait()
 256         else:
 257             self.__value = self.__value - 1
 258             rc = 1
 259         self.__cond.release()
 260         return rc
 261
 262     def release(self):
 263         self.__cond.acquire()
 264         self.__value = self.__value + 1
 265         self.__cond.notify()
 266         self.__cond.release()
 267
 268
 269 def Event(*args, **kwargs):
 270     return apply(_Event, args, kwargs)
 271
 272 class _Event(_Verbose):
 273
 274     # After Tim Peters' event class (without is_posted())
 275
 276     def __init__(self, verbose=None):
 277         _Verbose.__init__(self, verbose)
 278         self.__cond = Condition(Lock())
 279         self.__flag = 0
 280
 281     def isSet(self):
 282         return self.__flag
 283
 284     def set(self):
 285         self.__cond.acquire()
 286         self.__flag = 1
 287         self.__cond.notifyAll()
 288         self.__cond.release()
 289
 290     def clear(self):
 291         self.__cond.acquire()
 292         self.__flag = 0
 293         self.__cond.release()
 294
 295     def wait(self, timeout=None):
 296         self.__cond.acquire()
 297         if not self.__flag:
 298             self.__cond.wait(timeout)
 299         self.__cond.release()
 300
 301
 302 # Helper to generate new thread names
 303 _counter = 0
 304 def _newname(template="Thread-%d"):
 305     global _counter
 306     _counter = _counter + 1
 307     return template % _counter
 308
 309 # Active thread administration
 310 _active_limbo_lock = _allocate_lock()
 311 _active = {}
 312 _limbo = {}
 313
 314
 315 # Main class for threads
 316
 317 class Thread(_Verbose):
 318
 319     __initialized = 0
 320
 321     def __init__(self, group=None, target=None, name=None,
 322                  args=(), kwargs={}, verbose=None):
 323         assert group is None, "group argument must be None for now"
 324         _Verbose.__init__(self, verbose)
 325         self.__target = target
 326         self.__name = str(name or _newname())
 327         self.__args = args
 328         self.__kwargs = kwargs
 329         self.__daemonic = self._set_daemon()
 330         self.__started = 0
 331         self.__stopped = 0
 332         self.__block = Condition(Lock())
 333         self.__initialized = 1
 334
 335     def _set_daemon(self):
 336         # Overridden in _MainThread and _DummyThread
 337         return currentThread().isDaemon()
 338
 339     def __repr__(self):
 340         assert self.__initialized, "Thread.__init__() was not called"
 341         status = "initial"
 342         if self.__started:
 343             status = "started"
 344         if self.__stopped:
 345             status = "stopped"
 346         if self.__daemonic:
 347             status = status + " daemon"
 348         return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
 349
 350     def start(self):
 351         assert self.__initialized, "Thread.__init__() not called"
 352         assert not self.__started, "thread already started"
 353         if __debug__:
 354             self._note("%s.start(): starting thread", self)
 355         _active_limbo_lock.acquire()
 356         _limbo[self] = self
 357         _active_limbo_lock.release()
 358         _start_new_thread(self.__bootstrap, ())
 359         self.__started = 1
 360         _sleep(0.000001)    # 1 usec, to let the thread run (Solaris hack)
 361
 362     def run(self):
 363         if self.__target:
 364             apply(self.__target, self.__args, self.__kwargs)
 365
 366     def __bootstrap(self):
 367         try:
 368             self.__started = 1
 369             _active_limbo_lock.acquire()
 370             _active[_get_ident()] = self
 371             del _limbo[self]
 372             _active_limbo_lock.release()
 373             if __debug__:
 374                 self._note("%s.__bootstrap(): thread started", self)
 375             try:
 376                 self.run()
 377             except SystemExit:
 378                 if __debug__:
 379                     self._note("%s.__bootstrap(): raised SystemExit", self)
 380             except:
 381                 if __debug__:
 382                     self._note("%s.__bootstrap(): unhandled exception", self)
 383                 s = _StringIO()
 384                 _print_exc(file=s)
 385                 _sys.stderr.write("Exception in thread %s:\n%s\n" %
 386                                  (self.getName(), s.getvalue()))
 387             else:
 388                 if __debug__:
 389                     self._note("%s.__bootstrap(): normal return", self)
 390         finally:
 391             self.__stop()
 392             self.__delete()
 393
 394     def __stop(self):
 395         self.__block.acquire()
 396         self.__stopped = 1
 397         self.__block.notifyAll()
 398         self.__block.release()
 399
 400     def __delete(self):
 401         _active_limbo_lock.acquire()
 402         del _active[_get_ident()]
 403         _active_limbo_lock.release()
 404
 405     def join(self, timeout=None):
 406         assert self.__initialized, "Thread.__init__() not called"
 407         assert self.__started, "cannot join thread before it is started"
 408         assert self is not currentThread(), "cannot join current thread"
 409         if __debug__:
 410             if not self.__stopped:
 411                 self._note("%s.join(): waiting until thread stops", self)
 412         self.__block.acquire()
 413         if timeout is None:
 414             while not self.__stopped:
 415                 self.__block.wait()
 416             if __debug__:
 417                 self._note("%s.join(): thread stopped", self)
 418         else:
 419             deadline = _time() + timeout
 420             while not self.__stopped:
 421                 delay = deadline - _time()
 422                 if delay <= 0:
 423                     if __debug__:
 424                         self._note("%s.join(): timed out", self)
 425                     break
 426                 self.__block.wait(delay)
 427             else:
 428                 if __debug__:
 429                     self._note("%s.join(): thread stopped", self)
 430         self.__block.release()
 431
 432     def getName(self):
 433         assert self.__initialized, "Thread.__init__() not called"
 434         return self.__name
 435
 436     def setName(self, name):
 437         assert self.__initialized, "Thread.__init__() not called"
 438         self.__name = str(name)
 439
 440     def isAlive(self):
 441         assert self.__initialized, "Thread.__init__() not called"
 442         return self.__started and not self.__stopped
 443
 444     def isDaemon(self):
 445         assert self.__initialized, "Thread.__init__() not called"
 446         return self.__daemonic
 447
 448     def setDaemon(self, daemonic):
 449         assert self.__initialized, "Thread.__init__() not called"
 450         assert not self.__started, "cannot set daemon status of active thread"
 451         self.__daemonic = daemonic
 452
 453
 454 # Special thread class to represent the main thread
 455 # This is garbage collected through an exit handler
 456
 457 class _MainThread(Thread):
 458
 459     def __init__(self):
 460         Thread.__init__(self, name="MainThread")
 461         self._Thread__started = 1
 462         _active_limbo_lock.acquire()
 463         _active[_get_ident()] = self
 464         _active_limbo_lock.release()
 465         try:
 466             self.__oldexitfunc = _sys.exitfunc
 467         except AttributeError:
 468             self.__oldexitfunc = None
 469         _sys.exitfunc = self.__exitfunc
 470
 471     def _set_daemon(self):
 472         return 0
 473
 474     def __exitfunc(self):
 475         self._Thread__stop()
 476         t = _pickSomeNonDaemonThread()
 477         if t:
 478             if __debug__:
 479                 self._note("%s: waiting for other threads", self)
 480         while t:
 481             t.join()
 482             t = _pickSomeNonDaemonThread()
 483         if self.__oldexitfunc:
 484             if __debug__:
 485                 self._note("%s: calling exit handler", self)
 486             self.__oldexitfunc()
 487         if __debug__:
 488             self._note("%s: exiting", self)
 489         self._Thread__delete()
 490
 491 def _pickSomeNonDaemonThread():
 492     for t in enumerate():
 493         if not t.isDaemon() and t.isAlive():
 494             return t
 495     return None
 496
 497
 498 # Dummy thread class to represent threads not started here.
 499 # These aren't garbage collected when they die,
 500 # nor can they be waited for.
 501 # Their purpose is to return *something* from currentThread().
 502 # They are marked as daemon threads so we won't wait for them
 503 # when we exit (conform previous semantics).
 504
 505 class _DummyThread(Thread):
 506
 507     def __init__(self):
 508         Thread.__init__(self, name=_newname("Dummy-%d"))
 509         self.__Thread_started = 1
 510         _active_limbo_lock.acquire()
 511         _active[_get_ident()] = self
 512         _active_limbo_lock.release()
 513
 514     def _set_daemon(self):
 515         return 1
 516
 517     def join(self):
 518         assert 0, "cannot join a dummy thread"
 519
 520
 521 # Global API functions
 522
 523 def currentThread():
 524     try:
 525         return _active[_get_ident()]
 526     except KeyError:
 527         print "currentThread(): no current thread for", _get_ident()
 528         return _DummyThread()
 529
 530 def activeCount():
 531     _active_limbo_lock.acquire()
 532     count = len(_active) + len(_limbo)
 533     _active_limbo_lock.release()
 534     return count
 535
 536 def enumerate():
 537     _active_limbo_lock.acquire()
 538     active = _active.values() + _limbo.values()
 539     _active_limbo_lock.release()
 540     return active
 541
 542
 543 # Create the main thread object
 544
 545 _MainThread()
 546
 547
 548 # Self-test code
 549
 550 def _test():
 551
 552     import random
 553
 554     class BoundedQueue(_Verbose):
 555
 556         def __init__(self, limit):
 557             _Verbose.__init__(self)
 558             self.mon = RLock()
 559             self.rc = Condition(self.mon)
 560             self.wc = Condition(self.mon)
 561             self.limit = limit
 562             self.queue = []
 563
 564         def put(self, item):
 565             self.mon.acquire()
 566             while len(self.queue) >= self.limit:
 567                 self._note("put(%s): queue full", item)
 568                 self.wc.wait()
 569             self.queue.append(item)
 570             self._note("put(%s): appended, length now %d",
 571                        item, len(self.queue))
 572             self.rc.notify()
 573             self.mon.release()
 574
 575         def get(self):
 576             self.mon.acquire()
 577             while not self.queue:
 578                 self._note("get(): queue empty")
 579                 self.rc.wait()
 580             item = self.queue[0]
 581             del self.queue[0]
 582             self._note("get(): got %s, %d left", item, len(self.queue))
 583             self.wc.notify()
 584             self.mon.release()
 585             return item
 586
 587     class ProducerThread(Thread):
 588
 589         def __init__(self, queue, quota):
 590             Thread.__init__(self, name="Producer")
 591             self.queue = queue
 592             self.quota = quota
 593
 594         def run(self):
 595             from random import random
 596             counter = 0
 597             while counter < self.quota:
 598                 counter = counter + 1
 599                 self.queue.put("%s.%d" % (self.getName(), counter))
 600                 _sleep(random() * 0.00001)
 601
 602
 603     class ConsumerThread(Thread):
 604
 605         def __init__(self, queue, count):
 606             Thread.__init__(self, name="Consumer")
 607             self.queue = queue
 608             self.count = count
 609
 610         def run(self):
 611             while self.count > 0:
 612                 item = self.queue.get()
 613                 print item
 614                 self.count = self.count - 1
 615
 616     import time
 617
 618     NP = 3
 619     QL = 4
 620     NI = 5
 621
 622     Q = BoundedQueue(QL)
 623     P = []
 624     for i in range(NP):
 625         t = ProducerThread(Q, NI)
 626         t.setName("Producer-%d" % (i+1))
 627         P.append(t)
 628     C = ConsumerThread(Q, NI*NP)
 629     for t in P:
 630         t.start()
 631         _sleep(0.000001)
 632     C.start()
 633     for t in P:
 634         t.join()
 635     C.join()
 636
 637 if __name__ == '__main__':
 638     _test()