Fix duplicate test numbers in extra.decTest
[python.git] / Lib / threading.py
blobcf20152b4d7ab37c542206989406087657910d5b
1 """Thread module emulating a subset of Java's threading model."""
3 import sys as _sys
5 try:
6 import thread
7 except ImportError:
8 del _sys.modules[__name__]
9 raise
11 import warnings
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
39 del thread
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).
53 _VERBOSE = False
55 if __debug__:
57 class _Verbose(object):
59 def __init__(self, verbose=None):
60 if verbose is None:
61 verbose = _VERBOSE
62 self.__verbose = verbose
64 def _note(self, format, *args):
65 if self.__verbose:
66 format = format % args
67 format = "%s: %s\n" % (
68 current_thread().name, format)
69 _sys.stderr.write(format)
71 else:
72 # Disable this when using "python -O"
73 class _Verbose(object):
74 def __init__(self, verbose=None):
75 pass
76 def _note(self, *args):
77 pass
79 # Support for profile and trace hooks
81 _profile_hook = None
82 _trace_hook = None
84 def setprofile(func):
85 global _profile_hook
86 _profile_hook = func
88 def settrace(func):
89 global _trace_hook
90 _trace_hook = func
92 # Synchronization classes
94 Lock = _allocate_lock
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()
104 self.__owner = None
105 self.__count = 0
107 def __repr__(self):
108 owner = self.__owner
109 return "<%s(%s, %d)>" % (
110 self.__class__.__name__,
111 owner and owner.name,
112 self.__count)
114 def acquire(self, blocking=1):
115 me = current_thread()
116 if self.__owner is me:
117 self.__count = self.__count + 1
118 if __debug__:
119 self._note("%s.acquire(%s): recursive success", self, blocking)
120 return 1
121 rc = self.__block.acquire(blocking)
122 if rc:
123 self.__owner = me
124 self.__count = 1
125 if __debug__:
126 self._note("%s.acquire(%s): initial success", self, blocking)
127 else:
128 if __debug__:
129 self._note("%s.acquire(%s): failure", self, blocking)
130 return rc
132 __enter__ = acquire
134 def release(self):
135 if self.__owner is not current_thread():
136 raise RuntimeError("cannot release un-acquired lock")
137 self.__count = count = self.__count - 1
138 if not count:
139 self.__owner = None
140 self.__block.release()
141 if __debug__:
142 self._note("%s.release(): final release", self)
143 else:
144 if __debug__:
145 self._note("%s.release(): non-final release", self)
147 def __exit__(self, t, v, tb):
148 self.release()
150 # Internal methods used by condition variables
152 def _acquire_restore(self, count_owner):
153 count, owner = count_owner
154 self.__block.acquire()
155 self.__count = count
156 self.__owner = owner
157 if __debug__:
158 self._note("%s._acquire_restore()", self)
160 def _release_save(self):
161 if __debug__:
162 self._note("%s._release_save()", self)
163 count = self.__count
164 self.__count = 0
165 owner = self.__owner
166 self.__owner = None
167 self.__block.release()
168 return (count, owner)
170 def _is_owned(self):
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)
181 if lock is None:
182 lock = RLock()
183 self.__lock = lock
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().
190 try:
191 self._release_save = lock._release_save
192 except AttributeError:
193 pass
194 try:
195 self._acquire_restore = lock._acquire_restore
196 except AttributeError:
197 pass
198 try:
199 self._is_owned = lock._is_owned
200 except AttributeError:
201 pass
202 self.__waiters = []
204 def __enter__(self):
205 return self.__lock.__enter__()
207 def __exit__(self, *args):
208 return self.__lock.__exit__(*args)
210 def __repr__(self):
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
219 def _is_owned(self):
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()
224 return False
225 else:
226 return True
228 def wait(self, timeout=None):
229 if not self._is_owned():
230 raise RuntimeError("cannot wait on un-acquired lock")
231 waiter = _allocate_lock()
232 waiter.acquire()
233 self.__waiters.append(waiter)
234 saved_state = self._release_save()
235 try: # restore state no matter what (e.g., KeyboardInterrupt)
236 if timeout is None:
237 waiter.acquire()
238 if __debug__:
239 self._note("%s.wait(): got it", self)
240 else:
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
248 while True:
249 gotit = waiter.acquire(0)
250 if gotit:
251 break
252 remaining = endtime - _time()
253 if remaining <= 0:
254 break
255 delay = min(delay * 2, remaining, .05)
256 _sleep(delay)
257 if not gotit:
258 if __debug__:
259 self._note("%s.wait(%s): timed out", self, timeout)
260 try:
261 self.__waiters.remove(waiter)
262 except ValueError:
263 pass
264 else:
265 if __debug__:
266 self._note("%s.wait(%s): got it", self, timeout)
267 finally:
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-acquired lock")
273 __waiters = self.__waiters
274 waiters = __waiters[:n]
275 if not waiters:
276 if __debug__:
277 self._note("%s.notify(): no waiters", self)
278 return
279 self._note("%s.notify(): notifying %d waiter%s", self, n,
280 n!=1 and "s" or "")
281 for waiter in waiters:
282 waiter.release()
283 try:
284 __waiters.remove(waiter)
285 except ValueError:
286 pass
288 def notifyAll(self):
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):
302 if value < 0:
303 raise ValueError("semaphore initial value must be >= 0")
304 _Verbose.__init__(self, verbose)
305 self.__cond = Condition(Lock())
306 self.__value = value
308 def acquire(self, blocking=1):
309 rc = False
310 self.__cond.acquire()
311 while self.__value == 0:
312 if not blocking:
313 break
314 if __debug__:
315 self._note("%s.acquire(%s): blocked waiting, value=%s",
316 self, blocking, self.__value)
317 self.__cond.wait()
318 else:
319 self.__value = self.__value - 1
320 if __debug__:
321 self._note("%s.acquire: success, value=%s",
322 self, self.__value)
323 rc = True
324 self.__cond.release()
325 return rc
327 __enter__ = acquire
329 def release(self):
330 self.__cond.acquire()
331 self.__value = self.__value + 1
332 if __debug__:
333 self._note("%s.release: success, value=%s",
334 self, self.__value)
335 self.__cond.notify()
336 self.__cond.release()
338 def __exit__(self, t, v, tb):
339 self.release()
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
351 def release(self):
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())
367 self.__flag = False
369 def isSet(self):
370 return self.__flag
372 is_set = isSet
374 def set(self):
375 self.__cond.acquire()
376 try:
377 self.__flag = True
378 self.__cond.notify_all()
379 finally:
380 self.__cond.release()
382 def clear(self):
383 self.__cond.acquire()
384 try:
385 self.__flag = False
386 finally:
387 self.__cond.release()
389 def wait(self, timeout=None):
390 self.__cond.acquire()
391 try:
392 if not self.__flag:
393 self.__cond.wait(timeout)
394 return self.__flag
395 finally:
396 self.__cond.release()
398 # Helper to generate new thread names
399 _counter = 0
400 def _newname(template="Thread-%d"):
401 global _counter
402 _counter = _counter + 1
403 return template % _counter
405 # Active thread administration
406 _active_limbo_lock = _allocate_lock()
407 _active = {} # maps thread id to Thread object
408 _limbo = {}
411 # Main class for threads
413 class Thread(_Verbose):
415 __initialized = False
416 # Need to store a reference to sys.exc_info for printing
417 # out exceptions when a thread tries to use a global var. during interp.
418 # shutdown and thus raises an exception about trying to perform some
419 # operation on/with a NoneType
420 __exc_info = _sys.exc_info
421 # Keep sys.exc_clear too to clear the exception just before
422 # allowing .join() to return.
423 __exc_clear = _sys.exc_clear
425 def __init__(self, group=None, target=None, name=None,
426 args=(), kwargs=None, verbose=None):
427 assert group is None, "group argument must be None for now"
428 _Verbose.__init__(self, verbose)
429 if kwargs is None:
430 kwargs = {}
431 self.__target = target
432 self.__name = str(name or _newname())
433 self.__args = args
434 self.__kwargs = kwargs
435 self.__daemonic = self._set_daemon()
436 self.__ident = None
437 self.__started = Event()
438 self.__stopped = False
439 self.__block = Condition(Lock())
440 self.__initialized = True
441 # sys.stderr is not stored in the class like
442 # sys.exc_info since it can be changed between instances
443 self.__stderr = _sys.stderr
445 def _set_daemon(self):
446 # Overridden in _MainThread and _DummyThread
447 return current_thread().daemon
449 def __repr__(self):
450 assert self.__initialized, "Thread.__init__() was not called"
451 status = "initial"
452 if self.__started.is_set():
453 status = "started"
454 if self.__stopped:
455 status = "stopped"
456 if self.__daemonic:
457 status += " daemon"
458 if self.__ident is not None:
459 status += " %s" % self.__ident
460 return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
462 def start(self):
463 if not self.__initialized:
464 raise RuntimeError("thread.__init__() not called")
465 if self.__started.is_set():
466 raise RuntimeError("thread already started")
467 if __debug__:
468 self._note("%s.start(): starting thread", self)
469 with _active_limbo_lock:
470 _limbo[self] = self
471 _start_new_thread(self.__bootstrap, ())
472 self.__started.wait()
474 def run(self):
475 try:
476 if self.__target:
477 self.__target(*self.__args, **self.__kwargs)
478 finally:
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.
496 try:
497 self.__bootstrap_inner()
498 except:
499 if self.__daemonic and _sys is None:
500 return
501 raise
503 def _set_ident(self):
504 self.__ident = _get_ident()
506 def __bootstrap_inner(self):
507 try:
508 self._set_ident()
509 self.__started.set()
510 with _active_limbo_lock:
511 _active[self.__ident] = self
512 del _limbo[self]
513 if __debug__:
514 self._note("%s.__bootstrap(): thread started", self)
516 if _trace_hook:
517 self._note("%s.__bootstrap(): registering trace hook", self)
518 _sys.settrace(_trace_hook)
519 if _profile_hook:
520 self._note("%s.__bootstrap(): registering profile hook", self)
521 _sys.setprofile(_profile_hook)
523 try:
524 self.run()
525 except SystemExit:
526 if __debug__:
527 self._note("%s.__bootstrap(): raised SystemExit", self)
528 except:
529 if __debug__:
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
534 # self.
535 if _sys:
536 _sys.stderr.write("Exception in thread %s:\n%s\n" %
537 (self.name, _format_exc()))
538 else:
539 # Do the best job possible w/o a huge amt. of code to
540 # approximate a traceback (code ideas from
541 # Lib/traceback.py)
542 exc_type, exc_value, exc_tb = self.__exc_info()
543 try:
544 print>>self.__stderr, (
545 "Exception in thread " + self.name +
546 " (most likely raised during interpreter shutdown):")
547 print>>self.__stderr, (
548 "Traceback (most recent call last):")
549 while exc_tb:
550 print>>self.__stderr, (
551 ' File "%s", line %s, in %s' %
552 (exc_tb.tb_frame.f_code.co_filename,
553 exc_tb.tb_lineno,
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
559 finally:
560 del exc_type, exc_value, exc_tb
561 else:
562 if __debug__:
563 self._note("%s.__bootstrap(): normal return", self)
564 finally:
565 # Prevent a race in
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.
569 self.__exc_clear()
570 finally:
571 with _active_limbo_lock:
572 self.__stop()
573 try:
574 # We don't call self.__delete() because it also
575 # grabs _active_limbo_lock.
576 del _active[_get_ident()]
577 except:
578 pass
580 def __stop(self):
581 self.__block.acquire()
582 self.__stopped = True
583 self.__block.notify_all()
584 self.__block.release()
586 def __delete(self):
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.
610 try:
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.
617 except KeyError:
618 if 'dummy_threading' not in _sys.modules:
619 raise
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")
629 if __debug__:
630 if not self.__stopped:
631 self._note("%s.join(): waiting until thread stops", self)
632 self.__block.acquire()
633 try:
634 if timeout is None:
635 while not self.__stopped:
636 self.__block.wait()
637 if __debug__:
638 self._note("%s.join(): thread stopped", self)
639 else:
640 deadline = _time() + timeout
641 while not self.__stopped:
642 delay = deadline - _time()
643 if delay <= 0:
644 if __debug__:
645 self._note("%s.join(): timed out", self)
646 break
647 self.__block.wait(delay)
648 else:
649 if __debug__:
650 self._note("%s.join(): thread stopped", self)
651 finally:
652 self.__block.release()
654 @property
655 def name(self):
656 assert self.__initialized, "Thread.__init__() not called"
657 return self.__name
659 @name.setter
660 def name(self, name):
661 assert self.__initialized, "Thread.__init__() not called"
662 self.__name = str(name)
664 @property
665 def ident(self):
666 assert self.__initialized, "Thread.__init__() not called"
667 return self.__ident
669 def isAlive(self):
670 assert self.__initialized, "Thread.__init__() not called"
671 return self.__started.is_set() and not self.__stopped
673 is_alive = isAlive
675 @property
676 def daemon(self):
677 assert self.__initialized, "Thread.__init__() not called"
678 return self.__daemonic
680 @daemon.setter
681 def daemon(self, daemonic):
682 if not self.__initialized:
683 raise RuntimeError("Thread.__init__() not called")
684 if self.__started.is_set():
685 raise RuntimeError("cannot set daemon status of active thread");
686 self.__daemonic = daemonic
688 def isDaemon(self):
689 return self.daemon
691 def setDaemon(self, daemonic):
692 self.daemon = daemonic
694 def getName(self):
695 return self.name
697 def setName(self, name):
698 self.name = name
700 # The timer class was contributed by Itamar Shtull-Trauring
702 def Timer(*args, **kwargs):
703 return _Timer(*args, **kwargs)
705 class _Timer(Thread):
706 """Call a function after a specified number of seconds:
708 t = Timer(30.0, f, args=[], kwargs={})
709 t.start()
710 t.cancel() # stop the timer's action if it's still waiting
713 def __init__(self, interval, function, args=[], kwargs={}):
714 Thread.__init__(self)
715 self.interval = interval
716 self.function = function
717 self.args = args
718 self.kwargs = kwargs
719 self.finished = Event()
721 def cancel(self):
722 """Stop the timer if it hasn't finished yet"""
723 self.finished.set()
725 def run(self):
726 self.finished.wait(self.interval)
727 if not self.finished.is_set():
728 self.function(*self.args, **self.kwargs)
729 self.finished.set()
731 # Special thread class to represent the main thread
732 # This is garbage collected through an exit handler
734 class _MainThread(Thread):
736 def __init__(self):
737 Thread.__init__(self, name="MainThread")
738 self._Thread__started.set()
739 self._set_ident()
740 with _active_limbo_lock:
741 _active[_get_ident()] = self
743 def _set_daemon(self):
744 return False
746 def _exitfunc(self):
747 self._Thread__stop()
748 t = _pickSomeNonDaemonThread()
749 if t:
750 if __debug__:
751 self._note("%s: waiting for other threads", self)
752 while t:
753 t.join()
754 t = _pickSomeNonDaemonThread()
755 if __debug__:
756 self._note("%s: exiting", self)
757 self._Thread__delete()
759 def _pickSomeNonDaemonThread():
760 for t in enumerate():
761 if not t.daemon and t.is_alive():
762 return t
763 return None
766 # Dummy thread class to represent threads not started here.
767 # These aren't garbage collected when they die, nor can they be waited for.
768 # If they invoke anything in threading.py that calls current_thread(), they
769 # leave an entry in the _active dict forever after.
770 # Their purpose is to return *something* from current_thread().
771 # They are marked as daemon threads so we won't wait for them
772 # when we exit (conform previous semantics).
774 class _DummyThread(Thread):
776 def __init__(self):
777 Thread.__init__(self, name=_newname("Dummy-%d"))
779 # Thread.__block consumes an OS-level locking primitive, which
780 # can never be used by a _DummyThread. Since a _DummyThread
781 # instance is immortal, that's bad, so release this resource.
782 del self._Thread__block
784 self._Thread__started.set()
785 self._set_ident()
786 with _active_limbo_lock:
787 _active[_get_ident()] = self
789 def _set_daemon(self):
790 return True
792 def join(self, timeout=None):
793 assert False, "cannot join a dummy thread"
796 # Global API functions
798 def currentThread():
799 try:
800 return _active[_get_ident()]
801 except KeyError:
802 ##print "current_thread(): no current thread for", _get_ident()
803 return _DummyThread()
805 current_thread = currentThread
807 def activeCount():
808 with _active_limbo_lock:
809 return len(_active) + len(_limbo)
811 active_count = activeCount
813 def enumerate():
814 with _active_limbo_lock:
815 return _active.values() + _limbo.values()
817 from thread import stack_size
819 # Create the main thread object,
820 # and make it available for the interpreter
821 # (Py_Main) as threading._shutdown.
823 _shutdown = _MainThread()._exitfunc
825 # get thread-local implementation, either from the thread
826 # module, or from the python fallback
828 try:
829 from thread import _local as local
830 except ImportError:
831 from _threading_local import local
834 def _after_fork():
835 # This function is called by Python/ceval.c:PyEval_ReInitThreads which
836 # is called from PyOS_AfterFork. Here we cleanup threading module state
837 # that should not exist after a fork.
839 # Reset _active_limbo_lock, in case we forked while the lock was held
840 # by another (non-forked) thread. http://bugs.python.org/issue874900
841 global _active_limbo_lock
842 _active_limbo_lock = _allocate_lock()
844 # fork() only copied the current thread; clear references to others.
845 new_active = {}
846 current = current_thread()
847 with _active_limbo_lock:
848 for thread in _active.itervalues():
849 if thread is current:
850 # There is only one active thread. We reset the ident to
851 # its new value since it can have changed.
852 ident = _get_ident()
853 thread._Thread__ident = ident
854 new_active[ident] = thread
855 else:
856 # All the others are already stopped.
857 # We don't call _Thread__stop() because it tries to acquire
858 # thread._Thread__block which could also have been held while
859 # we forked.
860 thread._Thread__stopped = True
862 _limbo.clear()
863 _active.clear()
864 _active.update(new_active)
865 assert len(_active) == 1
868 # Self-test code
870 def _test():
872 class BoundedQueue(_Verbose):
874 def __init__(self, limit):
875 _Verbose.__init__(self)
876 self.mon = RLock()
877 self.rc = Condition(self.mon)
878 self.wc = Condition(self.mon)
879 self.limit = limit
880 self.queue = deque()
882 def put(self, item):
883 self.mon.acquire()
884 while len(self.queue) >= self.limit:
885 self._note("put(%s): queue full", item)
886 self.wc.wait()
887 self.queue.append(item)
888 self._note("put(%s): appended, length now %d",
889 item, len(self.queue))
890 self.rc.notify()
891 self.mon.release()
893 def get(self):
894 self.mon.acquire()
895 while not self.queue:
896 self._note("get(): queue empty")
897 self.rc.wait()
898 item = self.queue.popleft()
899 self._note("get(): got %s, %d left", item, len(self.queue))
900 self.wc.notify()
901 self.mon.release()
902 return item
904 class ProducerThread(Thread):
906 def __init__(self, queue, quota):
907 Thread.__init__(self, name="Producer")
908 self.queue = queue
909 self.quota = quota
911 def run(self):
912 from random import random
913 counter = 0
914 while counter < self.quota:
915 counter = counter + 1
916 self.queue.put("%s.%d" % (self.name, counter))
917 _sleep(random() * 0.00001)
920 class ConsumerThread(Thread):
922 def __init__(self, queue, count):
923 Thread.__init__(self, name="Consumer")
924 self.queue = queue
925 self.count = count
927 def run(self):
928 while self.count > 0:
929 item = self.queue.get()
930 print item
931 self.count = self.count - 1
933 NP = 3
934 QL = 4
935 NI = 5
937 Q = BoundedQueue(QL)
938 P = []
939 for i in range(NP):
940 t = ProducerThread(Q, NI)
941 t.name = ("Producer-%d" % (i+1))
942 P.append(t)
943 C = ConsumerThread(Q, NI*NP)
944 for t in P:
945 t.start()
946 _sleep(0.000001)
947 C.start()
948 for t in P:
949 t.join()
950 C.join()
952 if __name__ == '__main__':
953 _test()