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 # Rename some stuff so "from threading import *" is safe
19 __all__
= ['activeCount', 'active_count', 'Condition', 'currentThread',
20 'current_thread', 'enumerate', 'Event',
21 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',
22 'Timer', 'setprofile', 'settrace', 'local', 'stack_size']
24 _start_new_thread
= thread
.start_new_thread
25 _allocate_lock
= thread
.allocate_lock
26 _get_ident
= thread
.get_ident
27 ThreadError
= thread
.error
31 # sys.exc_clear is used to work around the fact that except blocks
32 # don't fully clear the exception until 3.0.
33 warnings
.filterwarnings('ignore', category
=DeprecationWarning,
34 module
='threading', message
='sys.exc_clear')
37 def _old_api(callable, old_name
):
38 if not _sys
.py3kwarning
:
41 def old(*args
, **kwargs
):
42 warnings
.warnpy3k("In 3.x, {0} is renamed to {1}."
43 .format(old_name
, callable.__name
__),
45 return callable(*args
, **kwargs
)
46 old
.__name
__ = old_name
49 # Debug support (adapted from ihooks.py).
50 # All the major classes here derive from _Verbose. We force that to
51 # be a new-style class so that all the major classes here are new-style.
52 # This helps debugging (type(instance) is more revealing for instances
53 # of new-style classes).
59 class _Verbose(object):
61 def __init__(self
, verbose
=None):
64 self
.__verbose
= verbose
66 def _note(self
, format
, *args
):
68 format
= format
% args
69 format
= "%s: %s\n" % (
70 current_thread().get_name(), format
)
71 _sys
.stderr
.write(format
)
74 # Disable this when using "python -O"
75 class _Verbose(object):
76 def __init__(self
, verbose
=None):
78 def _note(self
, *args
):
81 # Support for profile and trace hooks
94 # Synchronization classes
98 def RLock(*args
, **kwargs
):
99 return _RLock(*args
, **kwargs
)
101 class _RLock(_Verbose
):
103 def __init__(self
, verbose
=None):
104 _Verbose
.__init
__(self
, verbose
)
105 self
.__block
= _allocate_lock()
111 return "<%s(%s, %d)>" % (
112 self
.__class
__.__name
__,
113 owner
and owner
.get_name(),
116 def acquire(self
, blocking
=1):
117 me
= current_thread()
118 if self
.__owner
is me
:
119 self
.__count
= self
.__count
+ 1
121 self
._note
("%s.acquire(%s): recursive success", self
, blocking
)
123 rc
= self
.__block
.acquire(blocking
)
128 self
._note
("%s.acquire(%s): initial success", self
, blocking
)
131 self
._note
("%s.acquire(%s): failure", self
, blocking
)
137 if self
.__owner
is not current_thread():
138 raise RuntimeError("cannot release un-aquired lock")
139 self
.__count
= count
= self
.__count
- 1
142 self
.__block
.release()
144 self
._note
("%s.release(): final release", self
)
147 self
._note
("%s.release(): non-final release", self
)
149 def __exit__(self
, t
, v
, tb
):
152 # Internal methods used by condition variables
154 def _acquire_restore(self
, count_owner
):
155 count
, owner
= count_owner
156 self
.__block
.acquire()
160 self
._note
("%s._acquire_restore()", self
)
162 def _release_save(self
):
164 self
._note
("%s._release_save()", self
)
169 self
.__block
.release()
170 return (count
, owner
)
173 return self
.__owner
is current_thread()
176 def Condition(*args
, **kwargs
):
177 return _Condition(*args
, **kwargs
)
179 class _Condition(_Verbose
):
181 def __init__(self
, lock
=None, verbose
=None):
182 _Verbose
.__init
__(self
, verbose
)
186 # Export the lock's acquire() and release() methods
187 self
.acquire
= lock
.acquire
188 self
.release
= lock
.release
189 # If the lock defines _release_save() and/or _acquire_restore(),
190 # these override the default implementations (which just call
191 # release() and acquire() on the lock). Ditto for _is_owned().
193 self
._release
_save
= lock
._release
_save
194 except AttributeError:
197 self
._acquire
_restore
= lock
._acquire
_restore
198 except AttributeError:
201 self
._is
_owned
= lock
._is
_owned
202 except AttributeError:
207 return self
.__lock
.__enter
__()
209 def __exit__(self
, *args
):
210 return self
.__lock
.__exit
__(*args
)
213 return "<Condition(%s, %d)>" % (self
.__lock
, len(self
.__waiters
))
215 def _release_save(self
):
216 self
.__lock
.release() # No state to save
218 def _acquire_restore(self
, x
):
219 self
.__lock
.acquire() # Ignore saved state
222 # Return True if lock is owned by current_thread.
223 # This method is called only if __lock doesn't have _is_owned().
224 if self
.__lock
.acquire(0):
225 self
.__lock
.release()
230 def wait(self
, timeout
=None):
231 if not self
._is
_owned
():
232 raise RuntimeError("cannot wait on un-aquired lock")
233 waiter
= _allocate_lock()
235 self
.__waiters
.append(waiter
)
236 saved_state
= self
._release
_save
()
237 try: # restore state no matter what (e.g., KeyboardInterrupt)
241 self
._note
("%s.wait(): got it", self
)
243 # Balancing act: We can't afford a pure busy loop, so we
244 # have to sleep; but if we sleep the whole timeout time,
245 # we'll be unresponsive. The scheme here sleeps very
246 # little at first, longer as time goes on, but never longer
247 # than 20 times per second (or the timeout time remaining).
248 endtime
= _time() + timeout
249 delay
= 0.0005 # 500 us -> initial delay of 1 ms
251 gotit
= waiter
.acquire(0)
254 remaining
= endtime
- _time()
257 delay
= min(delay
* 2, remaining
, .05)
261 self
._note
("%s.wait(%s): timed out", self
, timeout
)
263 self
.__waiters
.remove(waiter
)
268 self
._note
("%s.wait(%s): got it", self
, timeout
)
270 self
._acquire
_restore
(saved_state
)
272 def notify(self
, n
=1):
273 if not self
._is
_owned
():
274 raise RuntimeError("cannot notify on un-aquired lock")
275 __waiters
= self
.__waiters
276 waiters
= __waiters
[:n
]
279 self
._note
("%s.notify(): no waiters", self
)
281 self
._note
("%s.notify(): notifying %d waiter%s", self
, n
,
283 for waiter
in waiters
:
286 __waiters
.remove(waiter
)
290 def notify_all(self
):
291 self
.notify(len(self
.__waiters
))
293 notifyAll
= _old_api(notify_all
, "notifyAll")
296 def Semaphore(*args
, **kwargs
):
297 return _Semaphore(*args
, **kwargs
)
299 class _Semaphore(_Verbose
):
301 # After Tim Peters' semaphore class, but not quite the same (no maximum)
303 def __init__(self
, value
=1, verbose
=None):
305 raise ValueError("semaphore initial value must be >= 0")
306 _Verbose
.__init
__(self
, verbose
)
307 self
.__cond
= Condition(Lock())
310 def acquire(self
, blocking
=1):
312 self
.__cond
.acquire()
313 while self
.__value
== 0:
317 self
._note
("%s.acquire(%s): blocked waiting, value=%s",
318 self
, blocking
, self
.__value
)
321 self
.__value
= self
.__value
- 1
323 self
._note
("%s.acquire: success, value=%s",
326 self
.__cond
.release()
332 self
.__cond
.acquire()
333 self
.__value
= self
.__value
+ 1
335 self
._note
("%s.release: success, value=%s",
338 self
.__cond
.release()
340 def __exit__(self
, t
, v
, tb
):
344 def BoundedSemaphore(*args
, **kwargs
):
345 return _BoundedSemaphore(*args
, **kwargs
)
347 class _BoundedSemaphore(_Semaphore
):
348 """Semaphore that checks that # releases is <= # acquires"""
349 def __init__(self
, value
=1, verbose
=None):
350 _Semaphore
.__init
__(self
, value
, verbose
)
351 self
._initial
_value
= value
354 if self
._Semaphore
__value
>= self
._initial
_value
:
355 raise ValueError, "Semaphore released too many times"
356 return _Semaphore
.release(self
)
359 def Event(*args
, **kwargs
):
360 return _Event(*args
, **kwargs
)
362 class _Event(_Verbose
):
364 # After Tim Peters' event class (without is_posted())
366 def __init__(self
, verbose
=None):
367 _Verbose
.__init
__(self
, verbose
)
368 self
.__cond
= Condition(Lock())
374 isSet
= _old_api(is_set
, "isSet")
377 self
.__cond
.acquire()
380 self
.__cond
.notify_all()
382 self
.__cond
.release()
385 self
.__cond
.acquire()
389 self
.__cond
.release()
391 def wait(self
, timeout
=None):
392 self
.__cond
.acquire()
395 self
.__cond
.wait(timeout
)
397 self
.__cond
.release()
399 # Helper to generate new thread names
401 def _newname(template
="Thread-%d"):
403 _counter
= _counter
+ 1
404 return template
% _counter
406 # Active thread administration
407 _active_limbo_lock
= _allocate_lock()
408 _active
= {} # maps thread id to Thread object
412 # Main class for threads
414 class Thread(_Verbose
):
416 __initialized
= False
417 # Need to store a reference to sys.exc_info for printing
418 # out exceptions when a thread tries to use a global var. during interp.
419 # shutdown and thus raises an exception about trying to perform some
420 # operation on/with a NoneType
421 __exc_info
= _sys
.exc_info
422 # Keep sys.exc_clear too to clear the exception just before
423 # allowing .join() to return.
424 __exc_clear
= _sys
.exc_clear
426 def __init__(self
, group
=None, target
=None, name
=None,
427 args
=(), kwargs
=None, verbose
=None):
428 assert group
is None, "group argument must be None for now"
429 _Verbose
.__init
__(self
, verbose
)
432 self
.__target
= target
433 self
.__name
= str(name
or _newname())
435 self
.__kwargs
= kwargs
436 self
.__daemonic
= self
._set
_daemon
()
438 self
.__started
= Event()
439 self
.__stopped
= False
440 self
.__block
= Condition(Lock())
441 self
.__initialized
= True
442 # sys.stderr is not stored in the class like
443 # sys.exc_info since it can be changed between instances
444 self
.__stderr
= _sys
.stderr
446 def _set_daemon(self
):
447 # Overridden in _MainThread and _DummyThread
448 return current_thread().is_daemon()
451 assert self
.__initialized
, "Thread.__init__() was not called"
453 if self
.__started
.is_set():
459 if self
.__ident
is not None:
460 status
+= " %s" % self
.__ident
461 return "<%s(%s, %s)>" % (self
.__class
__.__name
__, self
.__name
, status
)
464 if not self
.__initialized
:
465 raise RuntimeError("thread.__init__() not called")
466 if self
.__started
.is_set():
467 raise RuntimeError("thread already started")
469 self
._note
("%s.start(): starting thread", self
)
470 _active_limbo_lock
.acquire()
472 _active_limbo_lock
.release()
473 _start_new_thread(self
.__bootstrap
, ())
474 self
.__started
.wait()
479 self
.__target
(*self
.__args
, **self
.__kwargs
)
481 # Avoid a refcycle if the thread is running a function with
482 # an argument that has a member that points to the thread.
483 del self
.__target
, self
.__args
, self
.__kwargs
485 def __bootstrap(self
):
486 # Wrapper around the real bootstrap code that ignores
487 # exceptions during interpreter cleanup. Those typically
488 # happen when a daemon thread wakes up at an unfortunate
489 # moment, finds the world around it destroyed, and raises some
490 # random exception *** while trying to report the exception in
491 # __bootstrap_inner() below ***. Those random exceptions
492 # don't help anybody, and they confuse users, so we suppress
493 # them. We suppress them only when it appears that the world
494 # indeed has already been destroyed, so that exceptions in
495 # __bootstrap_inner() during normal business hours are properly
496 # reported. Also, we only suppress them for daemonic threads;
497 # if a non-daemonic encounters this, something else is wrong.
499 self
.__bootstrap
_inner
()
501 if self
.__daemonic
and _sys
is None:
505 def __bootstrap_inner(self
):
507 self
.__ident
= _get_ident()
509 _active_limbo_lock
.acquire()
510 _active
[self
.__ident
] = self
512 _active_limbo_lock
.release()
514 self
._note
("%s.__bootstrap(): thread started", self
)
517 self
._note
("%s.__bootstrap(): registering trace hook", self
)
518 _sys
.settrace(_trace_hook
)
520 self
._note
("%s.__bootstrap(): registering profile hook", self
)
521 _sys
.setprofile(_profile_hook
)
527 self
._note
("%s.__bootstrap(): raised SystemExit", self
)
530 self
._note
("%s.__bootstrap(): unhandled exception", self
)
531 # If sys.stderr is no more (most likely from interpreter
532 # shutdown) use self.__stderr. Otherwise still use sys (as in
533 # _sys) in case sys.stderr was redefined since the creation of
536 _sys
.stderr
.write("Exception in thread %s:\n%s\n" %
537 (self
.get_name(), _format_exc()))
539 # Do the best job possible w/o a huge amt. of code to
540 # approximate a traceback (code ideas from
542 exc_type
, exc_value
, exc_tb
= self
.__exc
_info
()
544 print>>self
.__stderr
, (
545 "Exception in thread " + self
.get_name() +
546 " (most likely raised during interpreter shutdown):")
547 print>>self
.__stderr
, (
548 "Traceback (most recent call last):")
550 print>>self
.__stderr
, (
551 ' File "%s", line %s, in %s' %
552 (exc_tb
.tb_frame
.f_code
.co_filename
,
554 exc_tb
.tb_frame
.f_code
.co_name
))
555 exc_tb
= exc_tb
.tb_next
556 print>>self
.__stderr
, ("%s: %s" % (exc_type
, exc_value
))
557 # Make sure that exc_tb gets deleted since it is a memory
558 # hog; deleting everything else is just for thoroughness
560 del exc_type
, exc_value
, exc_tb
563 self
._note
("%s.__bootstrap(): normal return", self
)
566 # test_threading.test_no_refcycle_through_target when
567 # the exception keeps the target alive past when we
568 # assert that it's dead.
571 with _active_limbo_lock
:
574 # We don't call self.__delete() because it also
575 # grabs _active_limbo_lock.
576 del _active
[_get_ident()]
581 self
.__block
.acquire()
582 self
.__stopped
= True
583 self
.__block
.notify_all()
584 self
.__block
.release()
587 "Remove current thread from the dict of currently running threads."
589 # Notes about running with dummy_thread:
591 # Must take care to not raise an exception if dummy_thread is being
592 # used (and thus this module is being used as an instance of
593 # dummy_threading). dummy_thread.get_ident() always returns -1 since
594 # there is only one thread if dummy_thread is being used. Thus
595 # len(_active) is always <= 1 here, and any Thread instance created
596 # overwrites the (if any) thread currently registered in _active.
598 # An instance of _MainThread is always created by 'threading'. This
599 # gets overwritten the instant an instance of Thread is created; both
600 # threads return -1 from dummy_thread.get_ident() and thus have the
601 # same key in the dict. So when the _MainThread instance created by
602 # 'threading' tries to clean itself up when atexit calls this method
603 # it gets a KeyError if another Thread instance was created.
605 # This all means that KeyError from trying to delete something from
606 # _active if dummy_threading is being used is a red herring. But
607 # since it isn't if dummy_threading is *not* being used then don't
608 # hide the exception.
611 with _active_limbo_lock
:
612 del _active
[_get_ident()]
613 # There must not be any python code between the previous line
614 # and after the lock is released. Otherwise a tracing function
615 # could try to acquire the lock again in the same thread, (in
616 # current_thread()), and would block.
618 if 'dummy_threading' not in _sys
.modules
:
621 def join(self
, timeout
=None):
622 if not self
.__initialized
:
623 raise RuntimeError("Thread.__init__() not called")
624 if not self
.__started
.is_set():
625 raise RuntimeError("cannot join thread before it is started")
626 if self
is current_thread():
627 raise RuntimeError("cannot join current thread")
630 if not self
.__stopped
:
631 self
._note
("%s.join(): waiting until thread stops", self
)
632 self
.__block
.acquire()
635 while not self
.__stopped
:
638 self
._note
("%s.join(): thread stopped", self
)
640 deadline
= _time() + timeout
641 while not self
.__stopped
:
642 delay
= deadline
- _time()
645 self
._note
("%s.join(): timed out", self
)
647 self
.__block
.wait(delay
)
650 self
._note
("%s.join(): thread stopped", self
)
652 self
.__block
.release()
655 assert self
.__initialized
, "Thread.__init__() not called"
658 getName
= _old_api(get_name
, "getName")
660 def set_name(self
, name
):
661 assert self
.__initialized
, "Thread.__init__() not called"
662 self
.__name
= str(name
)
664 setName
= _old_api(set_name
, "setName")
667 assert self
.__initialized
, "Thread.__init__() not called"
671 assert self
.__initialized
, "Thread.__init__() not called"
672 return self
.__started
.is_set() and not self
.__stopped
674 isAlive
= _old_api(is_alive
, "isAlive")
677 assert self
.__initialized
, "Thread.__init__() not called"
678 return self
.__daemonic
680 isDaemon
= _old_api(is_daemon
, "isDaemon")
682 def set_daemon(self
, daemonic
):
683 if not self
.__initialized
:
684 raise RuntimeError("Thread.__init__() not called")
685 if self
.__started
.is_set():
686 raise RuntimeError("cannot set daemon status of active thread");
687 self
.__daemonic
= daemonic
689 setDaemon
= _old_api(set_daemon
, "setDaemon")
691 # The timer class was contributed by Itamar Shtull-Trauring
693 def Timer(*args
, **kwargs
):
694 return _Timer(*args
, **kwargs
)
696 class _Timer(Thread
):
697 """Call a function after a specified number of seconds:
699 t = Timer(30.0, f, args=[], kwargs={})
701 t.cancel() # stop the timer's action if it's still waiting
704 def __init__(self
, interval
, function
, args
=[], kwargs
={}):
705 Thread
.__init
__(self
)
706 self
.interval
= interval
707 self
.function
= function
710 self
.finished
= Event()
713 """Stop the timer if it hasn't finished yet"""
717 self
.finished
.wait(self
.interval
)
718 if not self
.finished
.is_set():
719 self
.function(*self
.args
, **self
.kwargs
)
722 # Special thread class to represent the main thread
723 # This is garbage collected through an exit handler
725 class _MainThread(Thread
):
728 Thread
.__init
__(self
, name
="MainThread")
729 self
._Thread
__started
.set()
730 _active_limbo_lock
.acquire()
731 _active
[_get_ident()] = self
732 _active_limbo_lock
.release()
734 def _set_daemon(self
):
739 t
= _pickSomeNonDaemonThread()
742 self
._note
("%s: waiting for other threads", self
)
745 t
= _pickSomeNonDaemonThread()
747 self
._note
("%s: exiting", self
)
748 self
._Thread
__delete
()
750 def _pickSomeNonDaemonThread():
751 for t
in enumerate():
752 if not t
.is_daemon() and t
.is_alive():
757 # Dummy thread class to represent threads not started here.
758 # These aren't garbage collected when they die, nor can they be waited for.
759 # If they invoke anything in threading.py that calls current_thread(), they
760 # leave an entry in the _active dict forever after.
761 # Their purpose is to return *something* from current_thread().
762 # They are marked as daemon threads so we won't wait for them
763 # when we exit (conform previous semantics).
765 class _DummyThread(Thread
):
768 Thread
.__init
__(self
, name
=_newname("Dummy-%d"))
770 # Thread.__block consumes an OS-level locking primitive, which
771 # can never be used by a _DummyThread. Since a _DummyThread
772 # instance is immortal, that's bad, so release this resource.
773 del self
._Thread
__block
775 self
._Thread
__started
.set()
776 _active_limbo_lock
.acquire()
777 _active
[_get_ident()] = self
778 _active_limbo_lock
.release()
780 def _set_daemon(self
):
783 def join(self
, timeout
=None):
784 assert False, "cannot join a dummy thread"
787 # Global API functions
789 def current_thread():
791 return _active
[_get_ident()]
793 ##print "current_thread(): no current thread for", _get_ident()
794 return _DummyThread()
796 currentThread
= _old_api(current_thread
, "currentThread")
799 _active_limbo_lock
.acquire()
800 count
= len(_active
) + len(_limbo
)
801 _active_limbo_lock
.release()
804 activeCount
= _old_api(active_count
, "activeCount")
807 _active_limbo_lock
.acquire()
808 active
= _active
.values() + _limbo
.values()
809 _active_limbo_lock
.release()
812 from thread
import stack_size
814 # Create the main thread object,
815 # and make it available for the interpreter
816 # (Py_Main) as threading._shutdown.
818 _shutdown
= _MainThread()._exitfunc
820 # get thread-local implementation, either from the thread
821 # module, or from the python fallback
824 from thread
import _local
as local
826 from _threading_local
import local
830 # This function is called by Python/ceval.c:PyEval_ReInitThreads which
831 # is called from PyOS_AfterFork. Here we cleanup threading module state
832 # that should not exist after a fork.
834 # Reset _active_limbo_lock, in case we forked while the lock was held
835 # by another (non-forked) thread. http://bugs.python.org/issue874900
836 global _active_limbo_lock
837 _active_limbo_lock
= _allocate_lock()
839 # fork() only copied the current thread; clear references to others.
841 current
= current_thread()
842 with _active_limbo_lock
:
843 for ident
, thread
in _active
.iteritems():
844 if thread
is current
:
845 # There is only one active thread.
846 new_active
[ident
] = thread
848 # All the others are already stopped.
849 # We don't call _Thread__stop() because it tries to acquire
850 # thread._Thread__block which could also have been held while
852 thread
._Thread
__stopped
= True
856 _active
.update(new_active
)
857 assert len(_active
) == 1
864 class BoundedQueue(_Verbose
):
866 def __init__(self
, limit
):
867 _Verbose
.__init
__(self
)
869 self
.rc
= Condition(self
.mon
)
870 self
.wc
= Condition(self
.mon
)
876 while len(self
.queue
) >= self
.limit
:
877 self
._note
("put(%s): queue full", item
)
879 self
.queue
.append(item
)
880 self
._note
("put(%s): appended, length now %d",
881 item
, len(self
.queue
))
887 while not self
.queue
:
888 self
._note
("get(): queue empty")
890 item
= self
.queue
.popleft()
891 self
._note
("get(): got %s, %d left", item
, len(self
.queue
))
896 class ProducerThread(Thread
):
898 def __init__(self
, queue
, quota
):
899 Thread
.__init
__(self
, name
="Producer")
904 from random
import random
906 while counter
< self
.quota
:
907 counter
= counter
+ 1
908 self
.queue
.put("%s.%d" % (self
.get_name(), counter
))
909 _sleep(random() * 0.00001)
912 class ConsumerThread(Thread
):
914 def __init__(self
, queue
, count
):
915 Thread
.__init
__(self
, name
="Consumer")
920 while self
.count
> 0:
921 item
= self
.queue
.get()
923 self
.count
= self
.count
- 1
932 t
= ProducerThread(Q
, NI
)
933 t
.setName("Producer-%d" % (i
+1))
935 C
= ConsumerThread(Q
, NI
*NP
)
944 if __name__
== '__main__':