1
"""Thread module emulating a subset of Java's threading model."""
8
del _sys.modules[__name__]
13
from functools import wraps
14
from time import time as _time, sleep as _sleep
15
from traceback import format_exc as _format_exc
16
from collections import deque
18
# Note regarding PEP 8 compliant aliases
19
# This threading model was originally inspired by Java, and inherited
20
# the convention of camelCase function and method names from that
21
# language. While those names are not in any imminent danger of being
22
# deprecated, starting with Python 2.6, the module now provides a
23
# PEP 8 compliant alias for any such method name.
24
# Using the new PEP 8 compliant names also facilitates substitution
25
# with the multiprocessing module, which doesn't provide the old
26
# Java inspired names.
29
# Rename some stuff so "from threading import *" is safe
30
__all__ = ['activeCount', 'active_count', 'Condition', 'currentThread',
31
'current_thread', 'enumerate', 'Event',
32
'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',
33
'Timer', 'setprofile', 'settrace', 'local', 'stack_size']
35
_start_new_thread = thread.start_new_thread
36
_allocate_lock = thread.allocate_lock
37
_get_ident = thread.get_ident
38
ThreadError = thread.error
42
# sys.exc_clear is used to work around the fact that except blocks
43
# don't fully clear the exception until 3.0.
44
warnings.filterwarnings('ignore', category=DeprecationWarning,
45
module='threading', message='sys.exc_clear')
47
# Debug support (adapted from ihooks.py).
48
# All the major classes here derive from _Verbose. We force that to
49
# be a new-style class so that all the major classes here are new-style.
50
# This helps debugging (type(instance) is more revealing for instances
51
# of new-style classes).
57
class _Verbose(object):
59
def __init__(self, verbose=None):
62
self.__verbose = verbose
64
def _note(self, format, *args):
66
format = format % args
67
format = "%s: %s\n" % (
68
current_thread().name, format)
69
_sys.stderr.write(format)
72
# Disable this when using "python -O"
73
class _Verbose(object):
74
def __init__(self, verbose=None):
76
def _note(self, *args):
79
# Support for profile and trace hooks
92
# Synchronization classes
96
def RLock(*args, **kwargs):
97
return _RLock(*args, **kwargs)
99
class _RLock(_Verbose):
101
def __init__(self, verbose=None):
102
_Verbose.__init__(self, verbose)
103
self.__block = _allocate_lock()
109
return "<%s(%s, %d)>" % (
110
self.__class__.__name__,
111
owner and owner.name,
114
def acquire(self, blocking=1):
115
me = current_thread()
116
if self.__owner is me:
117
self.__count = self.__count + 1
119
self._note("%s.acquire(%s): recursive success", self, blocking)
121
rc = self.__block.acquire(blocking)
126
self._note("%s.acquire(%s): initial success", self, blocking)
129
self._note("%s.acquire(%s): failure", self, blocking)
135
if self.__owner is not current_thread():
136
raise RuntimeError("cannot release un-aquired lock")
137
self.__count = count = self.__count - 1
140
self.__block.release()
142
self._note("%s.release(): final release", self)
145
self._note("%s.release(): non-final release", self)
147
def __exit__(self, t, v, tb):
150
# Internal methods used by condition variables
152
def _acquire_restore(self, count_owner):
153
count, owner = count_owner
154
self.__block.acquire()
158
self._note("%s._acquire_restore()", self)
160
def _release_save(self):
162
self._note("%s._release_save()", self)
167
self.__block.release()
168
return (count, owner)
171
return self.__owner is current_thread()
174
def Condition(*args, **kwargs):
175
return _Condition(*args, **kwargs)
177
class _Condition(_Verbose):
179
def __init__(self, lock=None, verbose=None):
180
_Verbose.__init__(self, verbose)
184
# Export the lock's acquire() and release() methods
185
self.acquire = lock.acquire
186
self.release = lock.release
187
# If the lock defines _release_save() and/or _acquire_restore(),
188
# these override the default implementations (which just call
189
# release() and acquire() on the lock). Ditto for _is_owned().
191
self._release_save = lock._release_save
192
except AttributeError:
195
self._acquire_restore = lock._acquire_restore
196
except AttributeError:
199
self._is_owned = lock._is_owned
200
except AttributeError:
205
return self.__lock.__enter__()
207
def __exit__(self, *args):
208
return self.__lock.__exit__(*args)
211
return "<Condition(%s, %d)>" % (self.__lock, len(self.__waiters))
213
def _release_save(self):
214
self.__lock.release() # No state to save
216
def _acquire_restore(self, x):
217
self.__lock.acquire() # Ignore saved state
220
# Return True if lock is owned by current_thread.
221
# This method is called only if __lock doesn't have _is_owned().
222
if self.__lock.acquire(0):
223
self.__lock.release()
228
def wait(self, timeout=None):
229
if not self._is_owned():
230
raise RuntimeError("cannot wait on un-aquired lock")
231
waiter = _allocate_lock()
233
self.__waiters.append(waiter)
234
saved_state = self._release_save()
235
try: # restore state no matter what (e.g., KeyboardInterrupt)
239
self._note("%s.wait(): got it", self)
241
# Balancing act: We can't afford a pure busy loop, so we
242
# have to sleep; but if we sleep the whole timeout time,
243
# we'll be unresponsive. The scheme here sleeps very
244
# little at first, longer as time goes on, but never longer
245
# than 20 times per second (or the timeout time remaining).
246
endtime = _time() + timeout
247
delay = 0.0005 # 500 us -> initial delay of 1 ms
249
gotit = waiter.acquire(0)
252
remaining = endtime - _time()
255
delay = min(delay * 2, remaining, .05)
259
self._note("%s.wait(%s): timed out", self, timeout)
261
self.__waiters.remove(waiter)
266
self._note("%s.wait(%s): got it", self, timeout)
268
self._acquire_restore(saved_state)
270
def notify(self, n=1):
271
if not self._is_owned():
272
raise RuntimeError("cannot notify on un-aquired lock")
273
__waiters = self.__waiters
274
waiters = __waiters[:n]
277
self._note("%s.notify(): no waiters", self)
279
self._note("%s.notify(): notifying %d waiter%s", self, n,
281
for waiter in waiters:
284
__waiters.remove(waiter)
289
self.notify(len(self.__waiters))
291
notify_all = notifyAll
294
def Semaphore(*args, **kwargs):
295
return _Semaphore(*args, **kwargs)
297
class _Semaphore(_Verbose):
299
# After Tim Peters' semaphore class, but not quite the same (no maximum)
301
def __init__(self, value=1, verbose=None):
303
raise ValueError("semaphore initial value must be >= 0")
304
_Verbose.__init__(self, verbose)
305
self.__cond = Condition(Lock())
308
def acquire(self, blocking=1):
310
self.__cond.acquire()
311
while self.__value == 0:
315
self._note("%s.acquire(%s): blocked waiting, value=%s",
316
self, blocking, self.__value)
319
self.__value = self.__value - 1
321
self._note("%s.acquire: success, value=%s",
324
self.__cond.release()
330
self.__cond.acquire()
331
self.__value = self.__value + 1
333
self._note("%s.release: success, value=%s",
336
self.__cond.release()
338
def __exit__(self, t, v, tb):
342
def BoundedSemaphore(*args, **kwargs):
343
return _BoundedSemaphore(*args, **kwargs)
345
class _BoundedSemaphore(_Semaphore):
346
"""Semaphore that checks that # releases is <= # acquires"""
347
def __init__(self, value=1, verbose=None):
348
_Semaphore.__init__(self, value, verbose)
349
self._initial_value = value
352
if self._Semaphore__value >= self._initial_value:
353
raise ValueError, "Semaphore released too many times"
354
return _Semaphore.release(self)
357
def Event(*args, **kwargs):
358
return _Event(*args, **kwargs)
360
class _Event(_Verbose):
362
# After Tim Peters' event class (without is_posted())
364
def __init__(self, verbose=None):
365
_Verbose.__init__(self, verbose)
366
self.__cond = Condition(Lock())
375
self.__cond.acquire()
378
self.__cond.notify_all()
380
self.__cond.release()
383
self.__cond.acquire()
387
self.__cond.release()
389
def wait(self, timeout=None):
390
self.__cond.acquire()
393
self.__cond.wait(timeout)
395
self.__cond.release()
397
# Helper to generate new thread names
399
def _newname(template="Thread-%d"):
401
_counter = _counter + 1
402
return template % _counter
404
# Active thread administration
405
_active_limbo_lock = _allocate_lock()
406
_active = {} # maps thread id to Thread object
410
# Main class for threads
412
class Thread(_Verbose):
414
__initialized = False
415
# Need to store a reference to sys.exc_info for printing
416
# out exceptions when a thread tries to use a global var. during interp.
417
# shutdown and thus raises an exception about trying to perform some
418
# operation on/with a NoneType
419
__exc_info = _sys.exc_info
420
# Keep sys.exc_clear too to clear the exception just before
421
# allowing .join() to return.
422
__exc_clear = _sys.exc_clear
424
def __init__(self, group=None, target=None, name=None,
425
args=(), kwargs=None, verbose=None):
426
assert group is None, "group argument must be None for now"
427
_Verbose.__init__(self, verbose)
430
self.__target = target
431
self.__name = str(name or _newname())
433
self.__kwargs = kwargs
434
self.__daemonic = self._set_daemon()
436
self.__started = Event()
437
self.__stopped = False
438
self.__block = Condition(Lock())
439
self.__initialized = True
440
# sys.stderr is not stored in the class like
441
# sys.exc_info since it can be changed between instances
442
self.__stderr = _sys.stderr
444
def _set_daemon(self):
445
# Overridden in _MainThread and _DummyThread
446
return current_thread().daemon
449
assert self.__initialized, "Thread.__init__() was not called"
451
if self.__started.is_set():
457
if self.__ident is not None:
458
status += " %s" % self.__ident
459
return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
462
if not self.__initialized:
463
raise RuntimeError("thread.__init__() not called")
464
if self.__started.is_set():
465
raise RuntimeError("thread already started")
467
self._note("%s.start(): starting thread", self)
468
_active_limbo_lock.acquire()
470
_active_limbo_lock.release()
471
_start_new_thread(self.__bootstrap, ())
472
self.__started.wait()
477
self.__target(*self.__args, **self.__kwargs)
479
# Avoid a refcycle if the thread is running a function with
480
# an argument that has a member that points to the thread.
481
del self.__target, self.__args, self.__kwargs
483
def __bootstrap(self):
484
# Wrapper around the real bootstrap code that ignores
485
# exceptions during interpreter cleanup. Those typically
486
# happen when a daemon thread wakes up at an unfortunate
487
# moment, finds the world around it destroyed, and raises some
488
# random exception *** while trying to report the exception in
489
# __bootstrap_inner() below ***. Those random exceptions
490
# don't help anybody, and they confuse users, so we suppress
491
# them. We suppress them only when it appears that the world
492
# indeed has already been destroyed, so that exceptions in
493
# __bootstrap_inner() during normal business hours are properly
494
# reported. Also, we only suppress them for daemonic threads;
495
# if a non-daemonic encounters this, something else is wrong.
497
self.__bootstrap_inner()
499
if self.__daemonic and _sys is None:
503
def __bootstrap_inner(self):
505
self.__ident = _get_ident()
507
_active_limbo_lock.acquire()
508
_active[self.__ident] = self
510
_active_limbo_lock.release()
512
self._note("%s.__bootstrap(): thread started", self)
515
self._note("%s.__bootstrap(): registering trace hook", self)
516
_sys.settrace(_trace_hook)
518
self._note("%s.__bootstrap(): registering profile hook", self)
519
_sys.setprofile(_profile_hook)
525
self._note("%s.__bootstrap(): raised SystemExit", self)
528
self._note("%s.__bootstrap(): unhandled exception", self)
529
# If sys.stderr is no more (most likely from interpreter
530
# shutdown) use self.__stderr. Otherwise still use sys (as in
531
# _sys) in case sys.stderr was redefined since the creation of
534
_sys.stderr.write("Exception in thread %s:\n%s\n" %
535
(self.name, _format_exc()))
537
# Do the best job possible w/o a huge amt. of code to
538
# approximate a traceback (code ideas from
540
exc_type, exc_value, exc_tb = self.__exc_info()
542
print>>self.__stderr, (
543
"Exception in thread " + self.name +
544
" (most likely raised during interpreter shutdown):")
545
print>>self.__stderr, (
546
"Traceback (most recent call last):")
548
print>>self.__stderr, (
549
' File "%s", line %s, in %s' %
550
(exc_tb.tb_frame.f_code.co_filename,
552
exc_tb.tb_frame.f_code.co_name))
553
exc_tb = exc_tb.tb_next
554
print>>self.__stderr, ("%s: %s" % (exc_type, exc_value))
555
# Make sure that exc_tb gets deleted since it is a memory
556
# hog; deleting everything else is just for thoroughness
558
del exc_type, exc_value, exc_tb
561
self._note("%s.__bootstrap(): normal return", self)
564
# test_threading.test_no_refcycle_through_target when
565
# the exception keeps the target alive past when we
566
# assert that it's dead.
569
with _active_limbo_lock:
572
# We don't call self.__delete() because it also
573
# grabs _active_limbo_lock.
574
del _active[_get_ident()]
579
self.__block.acquire()
580
self.__stopped = True
581
self.__block.notify_all()
582
self.__block.release()
585
"Remove current thread from the dict of currently running threads."
587
# Notes about running with dummy_thread:
589
# Must take care to not raise an exception if dummy_thread is being
590
# used (and thus this module is being used as an instance of
591
# dummy_threading). dummy_thread.get_ident() always returns -1 since
592
# there is only one thread if dummy_thread is being used. Thus
593
# len(_active) is always <= 1 here, and any Thread instance created
594
# overwrites the (if any) thread currently registered in _active.
596
# An instance of _MainThread is always created by 'threading'. This
597
# gets overwritten the instant an instance of Thread is created; both
598
# threads return -1 from dummy_thread.get_ident() and thus have the
599
# same key in the dict. So when the _MainThread instance created by
600
# 'threading' tries to clean itself up when atexit calls this method
601
# it gets a KeyError if another Thread instance was created.
603
# This all means that KeyError from trying to delete something from
604
# _active if dummy_threading is being used is a red herring. But
605
# since it isn't if dummy_threading is *not* being used then don't
606
# hide the exception.
609
with _active_limbo_lock:
610
del _active[_get_ident()]
611
# There must not be any python code between the previous line
612
# and after the lock is released. Otherwise a tracing function
613
# could try to acquire the lock again in the same thread, (in
614
# current_thread()), and would block.
616
if 'dummy_threading' not in _sys.modules:
619
def join(self, timeout=None):
620
if not self.__initialized:
621
raise RuntimeError("Thread.__init__() not called")
622
if not self.__started.is_set():
623
raise RuntimeError("cannot join thread before it is started")
624
if self is current_thread():
625
raise RuntimeError("cannot join current thread")
628
if not self.__stopped:
629
self._note("%s.join(): waiting until thread stops", self)
630
self.__block.acquire()
633
while not self.__stopped:
636
self._note("%s.join(): thread stopped", self)
638
deadline = _time() + timeout
639
while not self.__stopped:
640
delay = deadline - _time()
643
self._note("%s.join(): timed out", self)
645
self.__block.wait(delay)
648
self._note("%s.join(): thread stopped", self)
650
self.__block.release()
654
assert self.__initialized, "Thread.__init__() not called"
658
def name(self, name):
659
assert self.__initialized, "Thread.__init__() not called"
660
self.__name = str(name)
664
assert self.__initialized, "Thread.__init__() not called"
668
assert self.__initialized, "Thread.__init__() not called"
669
return self.__started.is_set() and not self.__stopped
675
assert self.__initialized, "Thread.__init__() not called"
676
return self.__daemonic
679
def daemon(self, daemonic):
680
if not self.__initialized:
681
raise RuntimeError("Thread.__init__() not called")
682
if self.__started.is_set():
683
raise RuntimeError("cannot set daemon status of active thread");
684
self.__daemonic = daemonic
689
def setDaemon(self, daemonic):
690
self.daemon = daemonic
695
def setName(self, name):
698
# The timer class was contributed by Itamar Shtull-Trauring
700
def Timer(*args, **kwargs):
701
return _Timer(*args, **kwargs)
703
class _Timer(Thread):
704
"""Call a function after a specified number of seconds:
706
t = Timer(30.0, f, args=[], kwargs={})
708
t.cancel() # stop the timer's action if it's still waiting
711
def __init__(self, interval, function, args=[], kwargs={}):
712
Thread.__init__(self)
713
self.interval = interval
714
self.function = function
717
self.finished = Event()
720
"""Stop the timer if it hasn't finished yet"""
724
self.finished.wait(self.interval)
725
if not self.finished.is_set():
726
self.function(*self.args, **self.kwargs)
729
# Special thread class to represent the main thread
730
# This is garbage collected through an exit handler
732
class _MainThread(Thread):
735
Thread.__init__(self, name="MainThread")
736
self._Thread__started.set()
737
_active_limbo_lock.acquire()
738
_active[_get_ident()] = self
739
_active_limbo_lock.release()
741
def _set_daemon(self):
746
t = _pickSomeNonDaemonThread()
749
self._note("%s: waiting for other threads", self)
752
t = _pickSomeNonDaemonThread()
754
self._note("%s: exiting", self)
755
self._Thread__delete()
757
def _pickSomeNonDaemonThread():
758
for t in enumerate():
759
if not t.daemon and t.is_alive():
764
# Dummy thread class to represent threads not started here.
765
# These aren't garbage collected when they die, nor can they be waited for.
766
# If they invoke anything in threading.py that calls current_thread(), they
767
# leave an entry in the _active dict forever after.
768
# Their purpose is to return *something* from current_thread().
769
# They are marked as daemon threads so we won't wait for them
770
# when we exit (conform previous semantics).
772
class _DummyThread(Thread):
775
Thread.__init__(self, name=_newname("Dummy-%d"))
777
# Thread.__block consumes an OS-level locking primitive, which
778
# can never be used by a _DummyThread. Since a _DummyThread
779
# instance is immortal, that's bad, so release this resource.
780
del self._Thread__block
782
self._Thread__started.set()
783
_active_limbo_lock.acquire()
784
_active[_get_ident()] = self
785
_active_limbo_lock.release()
787
def _set_daemon(self):
790
def join(self, timeout=None):
791
assert False, "cannot join a dummy thread"
794
# Global API functions
798
return _active[_get_ident()]
800
##print "current_thread(): no current thread for", _get_ident()
801
return _DummyThread()
803
current_thread = currentThread
806
_active_limbo_lock.acquire()
807
count = len(_active) + len(_limbo)
808
_active_limbo_lock.release()
811
active_count = activeCount
814
_active_limbo_lock.acquire()
815
active = _active.values() + _limbo.values()
816
_active_limbo_lock.release()
819
from thread import stack_size
821
# Create the main thread object,
822
# and make it available for the interpreter
823
# (Py_Main) as threading._shutdown.
825
_shutdown = _MainThread()._exitfunc
827
# get thread-local implementation, either from the thread
828
# module, or from the python fallback
831
from thread import _local as local
833
from _threading_local import local
837
# This function is called by Python/ceval.c:PyEval_ReInitThreads which
838
# is called from PyOS_AfterFork. Here we cleanup threading module state
839
# that should not exist after a fork.
841
# Reset _active_limbo_lock, in case we forked while the lock was held
842
# by another (non-forked) thread. http://bugs.python.org/issue874900
843
global _active_limbo_lock
844
_active_limbo_lock = _allocate_lock()
846
# fork() only copied the current thread; clear references to others.
848
current = current_thread()
849
with _active_limbo_lock:
850
for thread in _active.itervalues():
851
if thread is current:
852
# There is only one active thread. We reset the ident to
853
# its new value since it can have changed.
855
thread._Thread__ident = ident
856
new_active[ident] = thread
858
# All the others are already stopped.
859
# We don't call _Thread__stop() because it tries to acquire
860
# thread._Thread__block which could also have been held while
862
thread._Thread__stopped = True
866
_active.update(new_active)
867
assert len(_active) == 1
874
class BoundedQueue(_Verbose):
876
def __init__(self, limit):
877
_Verbose.__init__(self)
879
self.rc = Condition(self.mon)
880
self.wc = Condition(self.mon)
886
while len(self.queue) >= self.limit:
887
self._note("put(%s): queue full", item)
889
self.queue.append(item)
890
self._note("put(%s): appended, length now %d",
891
item, len(self.queue))
897
while not self.queue:
898
self._note("get(): queue empty")
900
item = self.queue.popleft()
901
self._note("get(): got %s, %d left", item, len(self.queue))
906
class ProducerThread(Thread):
908
def __init__(self, queue, quota):
909
Thread.__init__(self, name="Producer")
914
from random import random
916
while counter < self.quota:
917
counter = counter + 1
918
self.queue.put("%s.%d" % (self.name, counter))
919
_sleep(random() * 0.00001)
922
class ConsumerThread(Thread):
924
def __init__(self, queue, count):
925
Thread.__init__(self, name="Consumer")
930
while self.count > 0:
931
item = self.queue.get()
933
self.count = self.count - 1
942
t = ProducerThread(Q, NI)
943
t.name = ("Producer-%d" % (i+1))
945
C = ConsumerThread(Q, NI*NP)
954
if __name__ == '__main__':
added
removed
Lib/threading.py
