1 """Thread module emulating a subset of Java's threading model."""
3 # This started life as the threading.py module of Python 2.4
4 # It's been patched to fix a problem with join, where a KeyboardInterrupt
5 # caused a lock to be left in the acquired state.
12 del _sys.modules[__name__]
15 from time import time as _time, sleep as _sleep
16 from traceback import format_exc as _format_exc
17 from collections import deque
19 # Rename some stuff so "from threading import *" is safe
20 __all__ = ['activeCount', 'Condition', 'currentThread', 'enumerate', 'Event',
21 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',
22 'Timer', 'setprofile', 'settrace', 'local']
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 # Debug support (adapted from ihooks.py).
32 # All the major classes here derive from _Verbose. We force that to
33 # be a new-style class so that all the major classes here are new-style.
34 # This helps debugging (type(instance) is more revealing for instances
35 # of new-style classes).
41 class _Verbose(object):
43 def __init__(self, verbose=None):
46 self.__verbose = verbose
48 def _note(self, format, *args):
50 format = format % args
51 format = "%s: %s\n" % (
52 currentThread().getName(), format)
53 _sys.stderr.write(format)
56 # Disable this when using "python -O"
57 class _Verbose(object):
58 def __init__(self, verbose=None):
60 def _note(self, *args):
63 # Support for profile and trace hooks
76 # Synchronization classes
80 def RLock(*args, **kwargs):
81 return _RLock(*args, **kwargs)
83 class _RLock(_Verbose):
85 def __init__(self, verbose=None):
86 _Verbose.__init__(self, verbose)
87 self.__block = _allocate_lock()
92 return "<%s(%s, %d)>" % (
93 self.__class__.__name__,
94 self.__owner and self.__owner.getName(),
97 def acquire(self, blocking=1):
99 if self.__owner is me:
100 self.__count = self.__count + 1
102 self._note("%s.acquire(%s): recursive success", self, blocking)
104 rc = self.__block.acquire(blocking)
109 self._note("%s.acquire(%s): initial succes", self, blocking)
112 self._note("%s.acquire(%s): failure", self, blocking)
117 assert self.__owner is me, "release() of un-acquire()d lock"
118 self.__count = count = self.__count - 1
121 self.__block.release()
123 self._note("%s.release(): final release", self)
126 self._note("%s.release(): non-final release", self)
128 # Internal methods used by condition variables
130 def _acquire_restore(self, (count, owner)):
131 self.__block.acquire()
135 self._note("%s._acquire_restore()", self)
137 def _release_save(self):
139 self._note("%s._release_save()", self)
144 self.__block.release()
145 return (count, owner)
148 return self.__owner is currentThread()
151 def Condition(*args, **kwargs):
152 return _Condition(*args, **kwargs)
154 class _Condition(_Verbose):
156 def __init__(self, lock=None, verbose=None):
157 _Verbose.__init__(self, verbose)
161 # Export the lock's acquire() and release() methods
162 self.acquire = lock.acquire
163 self.release = lock.release
164 # If the lock defines _release_save() and/or _acquire_restore(),
165 # these override the default implementations (which just call
166 # release() and acquire() on the lock). Ditto for _is_owned().
168 self._release_save = lock._release_save
169 except AttributeError:
172 self._acquire_restore = lock._acquire_restore
173 except AttributeError:
176 self._is_owned = lock._is_owned
177 except AttributeError:
182 return "<Condition(%s, %d)>" % (self.__lock, len(self.__waiters))
184 def _release_save(self):
185 self.__lock.release() # No state to save
187 def _acquire_restore(self, x):
188 self.__lock.acquire() # Ignore saved state
191 # Return True if lock is owned by currentThread.
192 # This method is called only if __lock doesn't have _is_owned().
193 if self.__lock.acquire(0):
194 self.__lock.release()
199 def wait(self, timeout=None):
200 assert self._is_owned(), "wait() of un-acquire()d lock"
201 waiter = _allocate_lock()
203 self.__waiters.append(waiter)
204 saved_state = self._release_save()
205 try: # restore state no matter what (e.g., KeyboardInterrupt)
209 self._note("%s.wait(): got it", self)
211 # Balancing act: We can't afford a pure busy loop, so we
212 # have to sleep; but if we sleep the whole timeout time,
213 # we'll be unresponsive. The scheme here sleeps very
214 # little at first, longer as time goes on, but never longer
215 # than 20 times per second (or the timeout time remaining).
216 endtime = _time() + timeout
217 delay = 0.0005 # 500 us -> initial delay of 1 ms
219 gotit = waiter.acquire(0)
222 remaining = endtime - _time()
225 delay = min(delay * 2, remaining, .05)
229 self._note("%s.wait(%s): timed out", self, timeout)
231 self.__waiters.remove(waiter)
236 self._note("%s.wait(%s): got it", self, timeout)
238 self._acquire_restore(saved_state)
240 def notify(self, n=1):
241 assert self._is_owned(), "notify() of un-acquire()d lock"
242 __waiters = self.__waiters
243 waiters = __waiters[:n]
246 self._note("%s.notify(): no waiters", self)
248 self._note("%s.notify(): notifying %d waiter%s", self, n,
250 for waiter in waiters:
253 __waiters.remove(waiter)
258 self.notify(len(self.__waiters))
261 def Semaphore(*args, **kwargs):
262 return _Semaphore(*args, **kwargs)
264 class _Semaphore(_Verbose):
266 # After Tim Peters' semaphore class, but not quite the same (no maximum)
268 def __init__(self, value=1, verbose=None):
269 assert value >= 0, "Semaphore initial value must be >= 0"
270 _Verbose.__init__(self, verbose)
271 self.__cond = Condition(Lock())
274 def acquire(self, blocking=1):
276 self.__cond.acquire()
277 while self.__value == 0:
281 self._note("%s.acquire(%s): blocked waiting, value=%s",
282 self, blocking, self.__value)
285 self.__value = self.__value - 1
287 self._note("%s.acquire: success, value=%s",
290 self.__cond.release()
294 self.__cond.acquire()
295 self.__value = self.__value + 1
297 self._note("%s.release: success, value=%s",
300 self.__cond.release()
303 def BoundedSemaphore(*args, **kwargs):
304 return _BoundedSemaphore(*args, **kwargs)
306 class _BoundedSemaphore(_Semaphore):
307 """Semaphore that checks that # releases is <= # acquires"""
308 def __init__(self, value=1, verbose=None):
309 _Semaphore.__init__(self, value, verbose)
310 self._initial_value = value
313 if self._Semaphore__value >= self._initial_value:
314 raise ValueError, "Semaphore released too many times"
315 return _Semaphore.release(self)
318 def Event(*args, **kwargs):
319 return _Event(*args, **kwargs)
321 class _Event(_Verbose):
323 # After Tim Peters' event class (without is_posted())
325 def __init__(self, verbose=None):
326 _Verbose.__init__(self, verbose)
327 self.__cond = Condition(Lock())
334 self.__cond.acquire()
337 self.__cond.notifyAll()
339 self.__cond.release()
342 self.__cond.acquire()
346 self.__cond.release()
348 def wait(self, timeout=None):
349 self.__cond.acquire()
352 self.__cond.wait(timeout)
354 self.__cond.release()
356 # Helper to generate new thread names
358 def _newname(template="Thread-%d"):
360 _counter = _counter + 1
361 return template % _counter
363 # Active thread administration
364 _active_limbo_lock = _allocate_lock()
369 # Main class for threads
371 class Thread(_Verbose):
373 __initialized = False
374 # Need to store a reference to sys.exc_info for printing
375 # out exceptions when a thread tries to use a global var. during interp.
376 # shutdown and thus raises an exception about trying to perform some
377 # operation on/with a NoneType
378 __exc_info = _sys.exc_info
380 def __init__(self, group=None, target=None, name=None,
381 args=(), kwargs={}, verbose=None):
382 assert group is None, "group argument must be None for now"
383 _Verbose.__init__(self, verbose)
384 self.__target = target
385 self.__name = str(name or _newname())
387 self.__kwargs = kwargs
388 self.__daemonic = self._set_daemon()
389 self.__started = False
390 self.__stopped = False
391 self.__block = Condition(Lock())
392 self.__initialized = True
393 # sys.stderr is not stored in the class like
394 # sys.exc_info since it can be changed between instances
395 self.__stderr = _sys.stderr
397 def _set_daemon(self):
398 # Overridden in _MainThread and _DummyThread
399 return currentThread().isDaemon()
402 assert self.__initialized, "Thread.__init__() was not called"
409 status = status + " daemon"
410 return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
413 assert self.__initialized, "Thread.__init__() not called"
414 assert not self.__started, "thread already started"
416 self._note("%s.start(): starting thread", self)
417 _active_limbo_lock.acquire()
419 _active_limbo_lock.release()
420 _start_new_thread(self.__bootstrap, ())
421 self.__started = True
422 _sleep(0.000001) # 1 usec, to let the thread run (Solaris hack)
426 self.__target(*self.__args, **self.__kwargs)
428 def __bootstrap(self):
430 self.__started = True
431 _active_limbo_lock.acquire()
432 _active[_get_ident()] = self
434 _active_limbo_lock.release()
436 self._note("%s.__bootstrap(): thread started", self)
439 self._note("%s.__bootstrap(): registering trace hook", self)
440 _sys.settrace(_trace_hook)
442 self._note("%s.__bootstrap(): registering profile hook", self)
443 _sys.setprofile(_profile_hook)
449 self._note("%s.__bootstrap(): raised SystemExit", self)
452 self._note("%s.__bootstrap(): unhandled exception", self)
453 # If sys.stderr is no more (most likely from interpreter
454 # shutdown) use self.__stderr. Otherwise still use sys (as in
455 # _sys) in case sys.stderr was redefined since the creation of
458 _sys.stderr.write("Exception in thread %s:\n%s\n" %
459 (self.getName(), _format_exc()))
461 # Do the best job possible w/o a huge amt. of code to
462 # approximate a traceback (code ideas from
464 exc_type, exc_value, exc_tb = self.__exc_info()
466 print>>self.__stderr, (
467 "Exception in thread " + self.getName() +
468 " (most likely raised during interpreter shutdown):")
469 print>>self.__stderr, (
470 "Traceback (most recent call last):")
472 print>>self.__stderr, (
473 ' File "%s", line %s, in %s' %
474 (exc_tb.tb_frame.f_code.co_filename,
476 exc_tb.tb_frame.f_code.co_name))
477 exc_tb = exc_tb.tb_next
478 print>>self.__stderr, ("%s: %s" % (exc_type, exc_value))
479 # Make sure that exc_tb gets deleted since it is a memory
480 # hog; deleting everything else is just for thoroughness
482 del exc_type, exc_value, exc_tb
485 self._note("%s.__bootstrap(): normal return", self)
494 self.__block.acquire()
495 self.__stopped = True
496 self.__block.notifyAll()
497 self.__block.release()
500 "Remove current thread from the dict of currently running threads."
502 # Notes about running with dummy_thread:
504 # Must take care to not raise an exception if dummy_thread is being
505 # used (and thus this module is being used as an instance of
506 # dummy_threading). dummy_thread.get_ident() always returns -1 since
507 # there is only one thread if dummy_thread is being used. Thus
508 # len(_active) is always <= 1 here, and any Thread instance created
509 # overwrites the (if any) thread currently registered in _active.
511 # An instance of _MainThread is always created by 'threading'. This
512 # gets overwritten the instant an instance of Thread is created; both
513 # threads return -1 from dummy_thread.get_ident() and thus have the
514 # same key in the dict. So when the _MainThread instance created by
515 # 'threading' tries to clean itself up when atexit calls this method
516 # it gets a KeyError if another Thread instance was created.
518 # This all means that KeyError from trying to delete something from
519 # _active if dummy_threading is being used is a red herring. But
520 # since it isn't if dummy_threading is *not* being used then don't
521 # hide the exception.
523 _active_limbo_lock.acquire()
526 del _active[_get_ident()]
528 if 'dummy_threading' not in _sys.modules:
531 _active_limbo_lock.release()
533 def join(self, timeout=None):
534 assert self.__initialized, "Thread.__init__() not called"
535 assert self.__started, "cannot join thread before it is started"
536 assert self is not currentThread(), "cannot join current thread"
538 if not self.__stopped:
539 self._note("%s.join(): waiting until thread stops", self)
540 self.__block.acquire()
543 while not self.__stopped:
546 self._note("%s.join(): thread stopped", self)
548 deadline = _time() + timeout
549 while not self.__stopped:
550 delay = deadline - _time()
553 self._note("%s.join(): timed out", self)
555 self.__block.wait(delay)
558 self._note("%s.join(): thread stopped", self)
560 self.__block.release()
563 assert self.__initialized, "Thread.__init__() not called"
566 def setName(self, name):
567 assert self.__initialized, "Thread.__init__() not called"
568 self.__name = str(name)
571 assert self.__initialized, "Thread.__init__() not called"
572 return self.__started and not self.__stopped
575 assert self.__initialized, "Thread.__init__() not called"
576 return self.__daemonic
578 def setDaemon(self, daemonic):
579 assert self.__initialized, "Thread.__init__() not called"
580 assert not self.__started, "cannot set daemon status of active thread"
581 self.__daemonic = daemonic
583 # The timer class was contributed by Itamar Shtull-Trauring
585 def Timer(*args, **kwargs):
586 return _Timer(*args, **kwargs)
588 class _Timer(Thread):
589 """Call a function after a specified number of seconds:
591 t = Timer(30.0, f, args=[], kwargs={})
593 t.cancel() # stop the timer's action if it's still waiting
596 def __init__(self, interval, function, args=[], kwargs={}):
597 Thread.__init__(self)
598 self.interval = interval
599 self.function = function
602 self.finished = Event()
605 """Stop the timer if it hasn't finished yet"""
609 self.finished.wait(self.interval)
610 if not self.finished.isSet():
611 self.function(*self.args, **self.kwargs)
614 # Special thread class to represent the main thread
615 # This is garbage collected through an exit handler
617 class _MainThread(Thread):
620 Thread.__init__(self, name="MainThread")
621 self._Thread__started = True
622 _active_limbo_lock.acquire()
623 _active[_get_ident()] = self
624 _active_limbo_lock.release()
626 atexit.register(self.__exitfunc)
628 def _set_daemon(self):
631 def __exitfunc(self):
633 t = _pickSomeNonDaemonThread()
636 self._note("%s: waiting for other threads", self)
639 t = _pickSomeNonDaemonThread()
641 self._note("%s: exiting", self)
642 self._Thread__delete()
644 def _pickSomeNonDaemonThread():
645 for t in enumerate():
646 if not t.isDaemon() and t.isAlive():
651 # Dummy thread class to represent threads not started here.
652 # These aren't garbage collected when they die,
653 # nor can they be waited for.
654 # Their purpose is to return *something* from currentThread().
655 # They are marked as daemon threads so we won't wait for them
656 # when we exit (conform previous semantics).
658 class _DummyThread(Thread):
661 Thread.__init__(self, name=_newname("Dummy-%d"))
662 self._Thread__started = True
663 _active_limbo_lock.acquire()
664 _active[_get_ident()] = self
665 _active_limbo_lock.release()
667 def _set_daemon(self):
670 def join(self, timeout=None):
671 assert False, "cannot join a dummy thread"
674 # Global API functions
678 return _active[_get_ident()]
680 ##print "currentThread(): no current thread for", _get_ident()
681 return _DummyThread()
684 _active_limbo_lock.acquire()
685 count = len(_active) + len(_limbo)
686 _active_limbo_lock.release()
690 _active_limbo_lock.acquire()
691 active = _active.values() + _limbo.values()
692 _active_limbo_lock.release()
695 # Create the main thread object
699 # get thread-local implementation, either from the thread
700 # module, or from the python fallback
703 from thread import _local as local
705 from _threading_local import local
712 class BoundedQueue(_Verbose):
714 def __init__(self, limit):
715 _Verbose.__init__(self)
717 self.rc = Condition(self.mon)
718 self.wc = Condition(self.mon)
724 while len(self.queue) >= self.limit:
725 self._note("put(%s): queue full", item)
727 self.queue.append(item)
728 self._note("put(%s): appended, length now %d",
729 item, len(self.queue))
735 while not self.queue:
736 self._note("get(): queue empty")
738 item = self.queue.popleft()
739 self._note("get(): got %s, %d left", item, len(self.queue))
744 class ProducerThread(Thread):
746 def __init__(self, queue, quota):
747 Thread.__init__(self, name="Producer")
752 from random import random
754 while counter < self.quota:
755 counter = counter + 1
756 self.queue.put("%s.%d" % (self.getName(), counter))
757 _sleep(random() * 0.00001)
760 class ConsumerThread(Thread):
762 def __init__(self, queue, count):
763 Thread.__init__(self, name="Consumer")
768 while self.count > 0:
769 item = self.queue.get()
771 self.count = self.count - 1
780 t = ProducerThread(Q, NI)
781 t.setName("Producer-%d" % (i+1))
783 C = ConsumerThread(Q, NI*NP)
792 if __name__ == '__main__':