1 """Thread module emulating a subset of Java's threading model."""
8 del _sys
.modules
[__name__
]
11 from time
import time
as _time
, sleep
as _sleep
12 from traceback
import format_exc
as _format_exc
13 from collections
import deque
15 # Rename some stuff so "from threading import *" is safe
16 __all__
= ['activeCount', 'Condition', 'currentThread', 'enumerate', 'Event',
17 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',
18 'Timer', 'setprofile', 'settrace', 'local']
20 _start_new_thread
= thread
.start_new_thread
21 _allocate_lock
= thread
.allocate_lock
22 _get_ident
= thread
.get_ident
23 ThreadError
= thread
.error
27 # Debug support (adapted from ihooks.py).
28 # All the major classes here derive from _Verbose. We force that to
29 # be a new-style class so that all the major classes here are new-style.
30 # This helps debugging (type(instance) is more revealing for instances
31 # of new-style classes).
37 class _Verbose(object):
39 def __init__(self
, verbose
=None):
42 self
.__verbose
= verbose
44 def _note(self
, format
, *args
):
46 format
= format
% args
47 format
= "%s: %s\n" % (
48 currentThread().getName(), format
)
49 _sys
.stderr
.write(format
)
52 # Disable this when using "python -O"
53 class _Verbose(object):
54 def __init__(self
, verbose
=None):
56 def _note(self
, *args
):
59 # Support for profile and trace hooks
72 # Synchronization classes
76 def RLock(*args
, **kwargs
):
77 return _RLock(*args
, **kwargs
)
79 class _RLock(_Verbose
):
81 def __init__(self
, verbose
=None):
82 _Verbose
.__init
__(self
, verbose
)
83 self
.__block
= _allocate_lock()
88 return "<%s(%s, %d)>" % (
89 self
.__class
__.__name
__,
90 self
.__owner
and self
.__owner
.getName(),
93 def acquire(self
, blocking
=1):
95 if self
.__owner
is me
:
96 self
.__count
= self
.__count
+ 1
98 self
._note
("%s.acquire(%s): recursive success", self
, blocking
)
100 rc
= self
.__block
.acquire(blocking
)
105 self
._note
("%s.acquire(%s): initial success", self
, blocking
)
108 self
._note
("%s.acquire(%s): failure", self
, blocking
)
113 assert self
.__owner
is me
, "release() of un-acquire()d lock"
114 self
.__count
= count
= self
.__count
- 1
117 self
.__block
.release()
119 self
._note
("%s.release(): final release", self
)
122 self
._note
("%s.release(): non-final release", self
)
124 # Internal methods used by condition variables
126 def _acquire_restore(self
, (count
, owner
)):
127 self
.__block
.acquire()
131 self
._note
("%s._acquire_restore()", self
)
133 def _release_save(self
):
135 self
._note
("%s._release_save()", self
)
140 self
.__block
.release()
141 return (count
, owner
)
144 return self
.__owner
is currentThread()
147 def Condition(*args
, **kwargs
):
148 return _Condition(*args
, **kwargs
)
150 class _Condition(_Verbose
):
152 def __init__(self
, lock
=None, verbose
=None):
153 _Verbose
.__init
__(self
, verbose
)
157 # Export the lock's acquire() and release() methods
158 self
.acquire
= lock
.acquire
159 self
.release
= lock
.release
160 # If the lock defines _release_save() and/or _acquire_restore(),
161 # these override the default implementations (which just call
162 # release() and acquire() on the lock). Ditto for _is_owned().
164 self
._release
_save
= lock
._release
_save
165 except AttributeError:
168 self
._acquire
_restore
= lock
._acquire
_restore
169 except AttributeError:
172 self
._is
_owned
= lock
._is
_owned
173 except AttributeError:
178 return "<Condition(%s, %d)>" % (self
.__lock
, len(self
.__waiters
))
180 def _release_save(self
):
181 self
.__lock
.release() # No state to save
183 def _acquire_restore(self
, x
):
184 self
.__lock
.acquire() # Ignore saved state
187 # Return True if lock is owned by currentThread.
188 # This method is called only if __lock doesn't have _is_owned().
189 if self
.__lock
.acquire(0):
190 self
.__lock
.release()
195 def wait(self
, timeout
=None):
196 assert self
._is
_owned
(), "wait() of un-acquire()d lock"
197 waiter
= _allocate_lock()
199 self
.__waiters
.append(waiter
)
200 saved_state
= self
._release
_save
()
201 try: # restore state no matter what (e.g., KeyboardInterrupt)
205 self
._note
("%s.wait(): got it", self
)
207 # Balancing act: We can't afford a pure busy loop, so we
208 # have to sleep; but if we sleep the whole timeout time,
209 # we'll be unresponsive. The scheme here sleeps very
210 # little at first, longer as time goes on, but never longer
211 # than 20 times per second (or the timeout time remaining).
212 endtime
= _time() + timeout
213 delay
= 0.0005 # 500 us -> initial delay of 1 ms
215 gotit
= waiter
.acquire(0)
218 remaining
= endtime
- _time()
221 delay
= min(delay
* 2, remaining
, .05)
225 self
._note
("%s.wait(%s): timed out", self
, timeout
)
227 self
.__waiters
.remove(waiter
)
232 self
._note
("%s.wait(%s): got it", self
, timeout
)
234 self
._acquire
_restore
(saved_state
)
236 def notify(self
, n
=1):
237 assert self
._is
_owned
(), "notify() of un-acquire()d lock"
238 __waiters
= self
.__waiters
239 waiters
= __waiters
[:n
]
242 self
._note
("%s.notify(): no waiters", self
)
244 self
._note
("%s.notify(): notifying %d waiter%s", self
, n
,
246 for waiter
in waiters
:
249 __waiters
.remove(waiter
)
254 self
.notify(len(self
.__waiters
))
257 def Semaphore(*args
, **kwargs
):
258 return _Semaphore(*args
, **kwargs
)
260 class _Semaphore(_Verbose
):
262 # After Tim Peters' semaphore class, but not quite the same (no maximum)
264 def __init__(self
, value
=1, verbose
=None):
265 assert value
>= 0, "Semaphore initial value must be >= 0"
266 _Verbose
.__init
__(self
, verbose
)
267 self
.__cond
= Condition(Lock())
270 def acquire(self
, blocking
=1):
272 self
.__cond
.acquire()
273 while self
.__value
== 0:
277 self
._note
("%s.acquire(%s): blocked waiting, value=%s",
278 self
, blocking
, self
.__value
)
281 self
.__value
= self
.__value
- 1
283 self
._note
("%s.acquire: success, value=%s",
286 self
.__cond
.release()
290 self
.__cond
.acquire()
291 self
.__value
= self
.__value
+ 1
293 self
._note
("%s.release: success, value=%s",
296 self
.__cond
.release()
299 def BoundedSemaphore(*args
, **kwargs
):
300 return _BoundedSemaphore(*args
, **kwargs
)
302 class _BoundedSemaphore(_Semaphore
):
303 """Semaphore that checks that # releases is <= # acquires"""
304 def __init__(self
, value
=1, verbose
=None):
305 _Semaphore
.__init
__(self
, value
, verbose
)
306 self
._initial
_value
= value
309 if self
._Semaphore
__value
>= self
._initial
_value
:
310 raise ValueError, "Semaphore released too many times"
311 return _Semaphore
.release(self
)
314 def Event(*args
, **kwargs
):
315 return _Event(*args
, **kwargs
)
317 class _Event(_Verbose
):
319 # After Tim Peters' event class (without is_posted())
321 def __init__(self
, verbose
=None):
322 _Verbose
.__init
__(self
, verbose
)
323 self
.__cond
= Condition(Lock())
330 self
.__cond
.acquire()
333 self
.__cond
.notifyAll()
335 self
.__cond
.release()
338 self
.__cond
.acquire()
342 self
.__cond
.release()
344 def wait(self
, timeout
=None):
345 self
.__cond
.acquire()
348 self
.__cond
.wait(timeout
)
350 self
.__cond
.release()
352 # Helper to generate new thread names
354 def _newname(template
="Thread-%d"):
356 _counter
= _counter
+ 1
357 return template
% _counter
359 # Active thread administration
360 _active_limbo_lock
= _allocate_lock()
361 _active
= {} # maps thread id to Thread object
365 # Main class for threads
367 class Thread(_Verbose
):
369 __initialized
= False
370 # Need to store a reference to sys.exc_info for printing
371 # out exceptions when a thread tries to use a global var. during interp.
372 # shutdown and thus raises an exception about trying to perform some
373 # operation on/with a NoneType
374 __exc_info
= _sys
.exc_info
376 def __init__(self
, group
=None, target
=None, name
=None,
377 args
=(), kwargs
=None, verbose
=None):
378 assert group
is None, "group argument must be None for now"
379 _Verbose
.__init
__(self
, verbose
)
382 self
.__target
= target
383 self
.__name
= str(name
or _newname())
385 self
.__kwargs
= kwargs
386 self
.__daemonic
= self
._set
_daemon
()
387 self
.__started
= False
388 self
.__stopped
= False
389 self
.__block
= Condition(Lock())
390 self
.__initialized
= True
391 # sys.stderr is not stored in the class like
392 # sys.exc_info since it can be changed between instances
393 self
.__stderr
= _sys
.stderr
395 def _set_daemon(self
):
396 # Overridden in _MainThread and _DummyThread
397 return currentThread().isDaemon()
400 assert self
.__initialized
, "Thread.__init__() was not called"
407 status
= status
+ " daemon"
408 return "<%s(%s, %s)>" % (self
.__class
__.__name
__, self
.__name
, status
)
411 assert self
.__initialized
, "Thread.__init__() not called"
412 assert not self
.__started
, "thread already started"
414 self
._note
("%s.start(): starting thread", self
)
415 _active_limbo_lock
.acquire()
417 _active_limbo_lock
.release()
418 _start_new_thread(self
.__bootstrap
, ())
419 self
.__started
= True
420 _sleep(0.000001) # 1 usec, to let the thread run (Solaris hack)
424 self
.__target
(*self
.__args
, **self
.__kwargs
)
426 def __bootstrap(self
):
428 self
.__started
= True
429 _active_limbo_lock
.acquire()
430 _active
[_get_ident()] = self
432 _active_limbo_lock
.release()
434 self
._note
("%s.__bootstrap(): thread started", self
)
437 self
._note
("%s.__bootstrap(): registering trace hook", self
)
438 _sys
.settrace(_trace_hook
)
440 self
._note
("%s.__bootstrap(): registering profile hook", self
)
441 _sys
.setprofile(_profile_hook
)
447 self
._note
("%s.__bootstrap(): raised SystemExit", self
)
450 self
._note
("%s.__bootstrap(): unhandled exception", self
)
451 # If sys.stderr is no more (most likely from interpreter
452 # shutdown) use self.__stderr. Otherwise still use sys (as in
453 # _sys) in case sys.stderr was redefined since the creation of
456 _sys
.stderr
.write("Exception in thread %s:\n%s\n" %
457 (self
.getName(), _format_exc()))
459 # Do the best job possible w/o a huge amt. of code to
460 # approximate a traceback (code ideas from
462 exc_type
, exc_value
, exc_tb
= self
.__exc
_info
()
464 print>>self
.__stderr
, (
465 "Exception in thread " + self
.getName() +
466 " (most likely raised during interpreter shutdown):")
467 print>>self
.__stderr
, (
468 "Traceback (most recent call last):")
470 print>>self
.__stderr
, (
471 ' File "%s", line %s, in %s' %
472 (exc_tb
.tb_frame
.f_code
.co_filename
,
474 exc_tb
.tb_frame
.f_code
.co_name
))
475 exc_tb
= exc_tb
.tb_next
476 print>>self
.__stderr
, ("%s: %s" % (exc_type
, exc_value
))
477 # Make sure that exc_tb gets deleted since it is a memory
478 # hog; deleting everything else is just for thoroughness
480 del exc_type
, exc_value
, exc_tb
483 self
._note
("%s.__bootstrap(): normal return", self
)
492 self
.__block
.acquire()
493 self
.__stopped
= True
494 self
.__block
.notifyAll()
495 self
.__block
.release()
498 "Remove current thread from the dict of currently running threads."
500 # Notes about running with dummy_thread:
502 # Must take care to not raise an exception if dummy_thread is being
503 # used (and thus this module is being used as an instance of
504 # dummy_threading). dummy_thread.get_ident() always returns -1 since
505 # there is only one thread if dummy_thread is being used. Thus
506 # len(_active) is always <= 1 here, and any Thread instance created
507 # overwrites the (if any) thread currently registered in _active.
509 # An instance of _MainThread is always created by 'threading'. This
510 # gets overwritten the instant an instance of Thread is created; both
511 # threads return -1 from dummy_thread.get_ident() and thus have the
512 # same key in the dict. So when the _MainThread instance created by
513 # 'threading' tries to clean itself up when atexit calls this method
514 # it gets a KeyError if another Thread instance was created.
516 # This all means that KeyError from trying to delete something from
517 # _active if dummy_threading is being used is a red herring. But
518 # since it isn't if dummy_threading is *not* being used then don't
519 # hide the exception.
521 _active_limbo_lock
.acquire()
524 del _active
[_get_ident()]
526 if 'dummy_threading' not in _sys
.modules
:
529 _active_limbo_lock
.release()
531 def join(self
, timeout
=None):
532 assert self
.__initialized
, "Thread.__init__() not called"
533 assert self
.__started
, "cannot join thread before it is started"
534 assert self
is not currentThread(), "cannot join current thread"
536 if not self
.__stopped
:
537 self
._note
("%s.join(): waiting until thread stops", self
)
538 self
.__block
.acquire()
541 while not self
.__stopped
:
544 self
._note
("%s.join(): thread stopped", self
)
546 deadline
= _time() + timeout
547 while not self
.__stopped
:
548 delay
= deadline
- _time()
551 self
._note
("%s.join(): timed out", self
)
553 self
.__block
.wait(delay
)
556 self
._note
("%s.join(): thread stopped", self
)
558 self
.__block
.release()
561 assert self
.__initialized
, "Thread.__init__() not called"
564 def setName(self
, name
):
565 assert self
.__initialized
, "Thread.__init__() not called"
566 self
.__name
= str(name
)
569 assert self
.__initialized
, "Thread.__init__() not called"
570 return self
.__started
and not self
.__stopped
573 assert self
.__initialized
, "Thread.__init__() not called"
574 return self
.__daemonic
576 def setDaemon(self
, daemonic
):
577 assert self
.__initialized
, "Thread.__init__() not called"
578 assert not self
.__started
, "cannot set daemon status of active thread"
579 self
.__daemonic
= daemonic
581 # The timer class was contributed by Itamar Shtull-Trauring
583 def Timer(*args
, **kwargs
):
584 return _Timer(*args
, **kwargs
)
586 class _Timer(Thread
):
587 """Call a function after a specified number of seconds:
589 t = Timer(30.0, f, args=[], kwargs={})
591 t.cancel() # stop the timer's action if it's still waiting
594 def __init__(self
, interval
, function
, args
=[], kwargs
={}):
595 Thread
.__init
__(self
)
596 self
.interval
= interval
597 self
.function
= function
600 self
.finished
= Event()
603 """Stop the timer if it hasn't finished yet"""
607 self
.finished
.wait(self
.interval
)
608 if not self
.finished
.isSet():
609 self
.function(*self
.args
, **self
.kwargs
)
612 # Special thread class to represent the main thread
613 # This is garbage collected through an exit handler
615 class _MainThread(Thread
):
618 Thread
.__init
__(self
, name
="MainThread")
619 self
._Thread
__started
= True
620 _active_limbo_lock
.acquire()
621 _active
[_get_ident()] = self
622 _active_limbo_lock
.release()
624 atexit
.register(self
.__exitfunc
)
626 def _set_daemon(self
):
629 def __exitfunc(self
):
631 t
= _pickSomeNonDaemonThread()
634 self
._note
("%s: waiting for other threads", self
)
637 t
= _pickSomeNonDaemonThread()
639 self
._note
("%s: exiting", self
)
640 self
._Thread
__delete
()
642 def _pickSomeNonDaemonThread():
643 for t
in enumerate():
644 if not t
.isDaemon() and t
.isAlive():
649 # Dummy thread class to represent threads not started here.
650 # These aren't garbage collected when they die, nor can they be waited for.
651 # If they invoke anything in threading.py that calls currentThread(), they
652 # leave an entry in the _active dict forever after.
653 # Their purpose is to return *something* from currentThread().
654 # They are marked as daemon threads so we won't wait for them
655 # when we exit (conform previous semantics).
657 class _DummyThread(Thread
):
660 Thread
.__init
__(self
, name
=_newname("Dummy-%d"))
662 # Thread.__block consumes an OS-level locking primitive, which
663 # can never be used by a _DummyThread. Since a _DummyThread
664 # instance is immortal, that's bad, so release this resource.
665 del self
._Thread
__block
667 self
._Thread
__started
= True
668 _active_limbo_lock
.acquire()
669 _active
[_get_ident()] = self
670 _active_limbo_lock
.release()
672 def _set_daemon(self
):
675 def join(self
, timeout
=None):
676 assert False, "cannot join a dummy thread"
679 # Global API functions
683 return _active
[_get_ident()]
685 ##print "currentThread(): no current thread for", _get_ident()
686 return _DummyThread()
689 _active_limbo_lock
.acquire()
690 count
= len(_active
) + len(_limbo
)
691 _active_limbo_lock
.release()
695 _active_limbo_lock
.acquire()
696 active
= _active
.values() + _limbo
.values()
697 _active_limbo_lock
.release()
700 # Create the main thread object
704 # get thread-local implementation, either from the thread
705 # module, or from the python fallback
708 from thread
import _local
as local
710 from _threading_local
import local
717 class BoundedQueue(_Verbose
):
719 def __init__(self
, limit
):
720 _Verbose
.__init
__(self
)
722 self
.rc
= Condition(self
.mon
)
723 self
.wc
= Condition(self
.mon
)
729 while len(self
.queue
) >= self
.limit
:
730 self
._note
("put(%s): queue full", item
)
732 self
.queue
.append(item
)
733 self
._note
("put(%s): appended, length now %d",
734 item
, len(self
.queue
))
740 while not self
.queue
:
741 self
._note
("get(): queue empty")
743 item
= self
.queue
.popleft()
744 self
._note
("get(): got %s, %d left", item
, len(self
.queue
))
749 class ProducerThread(Thread
):
751 def __init__(self
, queue
, quota
):
752 Thread
.__init
__(self
, name
="Producer")
757 from random
import random
759 while counter
< self
.quota
:
760 counter
= counter
+ 1
761 self
.queue
.put("%s.%d" % (self
.getName(), counter
))
762 _sleep(random() * 0.00001)
765 class ConsumerThread(Thread
):
767 def __init__(self
, queue
, count
):
768 Thread
.__init
__(self
, name
="Consumer")
773 while self
.count
> 0:
774 item
= self
.queue
.get()
776 self
.count
= self
.count
- 1
785 t
= ProducerThread(Q
, NI
)
786 t
.setName("Producer-%d" % (i
+1))
788 C
= ConsumerThread(Q
, NI
*NP
)
797 if __name__
== '__main__':