""" This module reimplements Python's native threading module using Panda
threading constructs. It's designed as a drop-in replacement for the
threading module for code that works with Panda; it is necessary because
in some compilation models, Panda's threading constructs are
incompatible with the OS-provided threads used by Python's thread
module.
This module implements the threading module with a thin layer over
Panda's threading constructs. As such, the semantics are close to,
but not precisely, the semantics documented for Python's standard
threading module. If you really do require strict adherence to
Python's semantics, see the threading2 module instead.
However, if you don't need such strict adherence to Python's original
semantics, this module is probably a better choice. It is likely to
be slighly faster than the threading2 module (and even slightly faster
than Python's own threading module). It is also better integrated
with Panda's threads, so that Panda's thread debug mechanisms will be
easier to use and understand.
It is permissible to mix-and-match both threading and threading2
within the same application. """
from panda3d import core
from direct.stdpy import thread as _thread
import sys as _sys
import weakref
__all__ = [
'Thread',
'Lock', 'RLock',
'Condition',
'Semaphore', 'BoundedSemaphore',
'Event',
'Timer',
'ThreadError',
'local',
'current_thread',
'main_thread',
'enumerate', 'active_count',
'settrace', 'setprofile', 'stack_size',
'TIMEOUT_MAX',
]
TIMEOUT_MAX = _thread.TIMEOUT_MAX
local = _thread._local
_newname = _thread._newname
ThreadError = _thread.error
[docs]class ThreadBase:
""" A base class for both Thread and ExternalThread in this
module. """
[docs] def __init__(self):
pass
[docs] def getName(self):
return self.name
[docs] def isDaemon(self):
return self.daemon
[docs] def setDaemon(self, daemon):
if self.is_alive():
raise RuntimeError
self.__dict__['daemon'] = daemon
def __setattr__(self, key, value):
if key == 'name':
self.setName(value)
elif key == 'ident':
raise AttributeError
elif key == 'daemon':
self.setDaemon(value)
else:
self.__dict__[key] = value
# Copy these static methods from Panda's Thread object. These are
# useful if you may be running in Panda's SIMPLE_THREADS compilation
# mode.
ThreadBase.forceYield = core.Thread.forceYield
ThreadBase.considerYield = core.Thread.considerYield
[docs]class Thread(ThreadBase):
""" This class provides a wrapper around Panda's PythonThread
object. The wrapper is designed to emulate Python's own
threading.Thread object. """
[docs] def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, daemon=None):
ThreadBase.__init__(self)
assert group is None
self.__target = target
self.__args = args
self.__kwargs = kwargs
if not name:
name = _newname()
current = current_thread()
if daemon is not None:
self.__dict__['daemon'] = daemon
else:
self.__dict__['daemon'] = current.daemon
self.__dict__['name'] = name
def call_run():
# As soon as the thread is done, break the circular reference.
try:
self.run()
finally:
self.__thread = None
_thread._remove_thread_id(self.ident)
self.__thread = core.PythonThread(call_run, None, name, name)
threadId = _thread._add_thread(self.__thread, weakref.proxy(self))
self.__dict__['ident'] = threadId
def __del__(self):
# On interpreter shutdown, the _thread module might have
# already been cleaned up.
if _thread and _thread._remove_thread_id:
_thread._remove_thread_id(self.ident)
[docs] def is_alive(self):
thread = self.__thread
return thread is not None and thread.is_started()
isAlive = is_alive
[docs] def start(self):
thread = self.__thread
if thread is None or thread.is_started():
raise RuntimeError
if not thread.start(core.TPNormal, True):
raise RuntimeError
[docs] def run(self):
if _settrace_func:
_sys.settrace(_settrace_func)
if _setprofile_func:
_sys.setprofile(_setprofile_func)
self.__target(*self.__args, **self.__kwargs)
[docs] def join(self, timeout = None):
# We don't support a timed join here, sorry.
assert timeout is None
thread = self.__thread
if thread is not None:
thread.join()
# Clear the circular reference.
self.__thread = None
_thread._remove_thread_id(self.ident)
[docs] def setName(self, name):
self.__dict__['name'] = name
self.__thread.setName(name)
[docs]class ExternalThread(ThreadBase):
""" Returned for a Thread object that wasn't created by this
interface. """
[docs] def __init__(self, extThread, threadId):
ThreadBase.__init__(self)
self.__thread = extThread
self.__dict__['daemon'] = True
self.__dict__['name'] = self.__thread.getName()
self.__dict__['ident'] = threadId
[docs] def is_alive(self):
return self.__thread.isStarted()
[docs] def isAlive(self):
return self.__thread.isStarted()
[docs] def start(self):
raise RuntimeError
[docs] def run(self):
raise RuntimeError
[docs] def join(self, timeout = None):
raise RuntimeError
[docs] def setDaemon(self, daemon):
raise RuntimeError
[docs]class MainThread(ExternalThread):
""" Returned for the MainThread object. """
[docs] def __init__(self, extThread, threadId):
ExternalThread.__init__(self, extThread, threadId)
self.__dict__['daemon'] = False
[docs]class Lock(core.Mutex):
""" This class provides a wrapper around Panda's Mutex object.
The wrapper is designed to emulate Python's own threading.Lock
object. """
[docs] def __init__(self, name = "PythonLock"):
core.Mutex.__init__(self, name)
[docs] def acquire(self, blocking = True):
if blocking:
core.Mutex.acquire(self)
return True
else:
return core.Mutex.tryAcquire(self)
__enter__ = acquire
def __exit__(self, t, v, tb):
self.release()
[docs]class RLock(core.ReMutex):
""" This class provides a wrapper around Panda's ReMutex object.
The wrapper is designed to emulate Python's own threading.RLock
object. """
[docs] def __init__(self, name = "PythonRLock"):
core.ReMutex.__init__(self, name)
[docs] def acquire(self, blocking = True):
if blocking:
core.ReMutex.acquire(self)
return True
else:
return core.ReMutex.tryAcquire(self)
__enter__ = acquire
def __exit__(self, t, v, tb):
self.release()
[docs]class Condition(core.ConditionVarFull):
""" This class provides a wrapper around Panda's ConditionVarFull
object. The wrapper is designed to emulate Python's own
threading.Condition object. """
[docs] def __init__(self, lock = None):
if not lock:
lock = Lock()
# Panda doesn't support RLock objects used with condition
# variables.
assert isinstance(lock, Lock)
self.__lock = lock
core.ConditionVarFull.__init__(self, self.__lock)
[docs] def acquire(self, *args, **kw):
return self.__lock.acquire(*args, **kw)
[docs] def release(self):
self.__lock.release()
[docs] def wait(self, timeout = None):
if timeout is None:
core.ConditionVarFull.wait(self)
else:
core.ConditionVarFull.wait(self, timeout)
[docs] def notifyAll(self):
core.ConditionVarFull.notifyAll(self)
notify_all = notifyAll
__enter__ = acquire
def __exit__(self, t, v, tb):
self.release()
[docs]class Semaphore(core.Semaphore):
""" This class provides a wrapper around Panda's Semaphore
object. The wrapper is designed to emulate Python's own
threading.Semaphore object. """
[docs] def __init__(self, value = 1):
core.Semaphore.__init__(self, value)
[docs] def acquire(self, blocking = True):
if blocking:
core.Semaphore.acquire(self)
return True
else:
return core.Semaphore.tryAcquire(self)
__enter__ = acquire
def __exit__(self, t, v, tb):
self.release()
[docs]class BoundedSemaphore(Semaphore):
""" This class provides a wrapper around Panda's Semaphore
object. The wrapper is designed to emulate Python's own
threading.BoundedSemaphore object. """
[docs] def __init__(self, value = 1):
self.__max = value
Semaphore.__init__(value)
[docs] def release(self):
if self.getCount() > self.__max:
raise ValueError
Semaphore.release(self)
[docs]class Event:
""" This class is designed to emulate Python's own threading.Event
object. """
[docs] def __init__(self):
self.__lock = core.Mutex("Python Event")
self.__cvar = core.ConditionVarFull(self.__lock)
self.__flag = False
[docs] def is_set(self):
return self.__flag
isSet = is_set
[docs] def set(self):
self.__lock.acquire()
try:
self.__flag = True
self.__cvar.notifyAll()
finally:
self.__lock.release()
[docs] def clear(self):
self.__lock.acquire()
try:
self.__flag = False
finally:
self.__lock.release()
[docs] def wait(self, timeout = None):
self.__lock.acquire()
try:
if timeout is None:
while not self.__flag:
self.__cvar.wait()
else:
clock = core.TrueClock.getGlobalPtr()
expires = clock.getShortTime() + timeout
while not self.__flag:
wait = expires - clock.getShortTime()
if wait < 0:
return
self.__cvar.wait(wait)
finally:
self.__lock.release()
[docs]class Timer(Thread):
"""Call a function after a specified number of seconds:
t = Timer(30.0, f, args=[], kwargs={})
t.start()
t.cancel() # stop the timer's action if it's still waiting
"""
[docs] def __init__(self, interval, function, args=[], kwargs={}):
Thread.__init__(self)
self.interval = interval
self.function = function
self.args = args
self.kwargs = kwargs
self.finished = Event()
[docs] def cancel(self):
"""Stop the timer if it hasn't finished yet"""
self.finished.set()
[docs] def run(self):
self.finished.wait(self.interval)
if not self.finished.isSet():
self.function(*self.args, **self.kwargs)
self.finished.set()
def _create_thread_wrapper(t, threadId):
""" Creates a thread wrapper for the indicated external thread. """
if isinstance(t, core.MainThread):
pyt = MainThread(t, threadId)
else:
pyt = ExternalThread(t, threadId)
return pyt
[docs]def current_thread():
t = core.Thread.getCurrentThread()
return _thread._get_thread_wrapper(t, _create_thread_wrapper)
[docs]def main_thread():
t = core.Thread.getMainThread()
return _thread._get_thread_wrapper(t, _create_thread_wrapper)
currentThread = current_thread
[docs]def enumerate():
tlist = []
_thread._threadsLock.acquire()
try:
for thread, locals, wrapper in list(_thread._threads.values()):
if wrapper and wrapper.is_alive():
tlist.append(wrapper)
return tlist
finally:
_thread._threadsLock.release()
[docs]def active_count():
return len(enumerate())
activeCount = active_count
_settrace_func = None
[docs]def settrace(func):
global _settrace_func
_settrace_func = func
_setprofile_func = None
[docs]def setprofile(func):
global _setprofile_func
_setprofile_func = func
[docs]def stack_size(size = None):
raise ThreadError
if __debug__:
def _test():
from collections import deque
_sleep = core.Thread.sleep
_VERBOSE = False
class _Verbose(object):
def __init__(self, verbose=None):
if verbose is None:
verbose = _VERBOSE
self.__verbose = verbose
def _note(self, format, *args):
if self.__verbose:
format = format % args
format = "%s: %s\n" % (
currentThread().getName(), format)
_sys.stderr.write(format)
class BoundedQueue(_Verbose):
def __init__(self, limit):
_Verbose.__init__(self)
self.mon = Lock(name = "BoundedQueue.mon")
self.rc = Condition(self.mon)
self.wc = Condition(self.mon)
self.limit = limit
self.queue = deque()
def put(self, item):
self.mon.acquire()
while len(self.queue) >= self.limit:
self._note("put(%s): queue full", item)
self.wc.wait()
self.queue.append(item)
self._note("put(%s): appended, length now %d",
item, len(self.queue))
self.rc.notify()
self.mon.release()
def get(self):
self.mon.acquire()
while not self.queue:
self._note("get(): queue empty")
self.rc.wait()
item = self.queue.popleft()
self._note("get(): got %s, %d left", item, len(self.queue))
self.wc.notify()
self.mon.release()
return item
class ProducerThread(Thread):
def __init__(self, queue, quota):
Thread.__init__(self, name="Producer")
self.queue = queue
self.quota = quota
def run(self):
from random import random
counter = 0
while counter < self.quota:
counter = counter + 1
self.queue.put("%s.%d" % (self.getName(), counter))
_sleep(random() * 0.00001)
class ConsumerThread(Thread):
def __init__(self, queue, count):
Thread.__init__(self, name="Consumer")
self.queue = queue
self.count = count
def run(self):
while self.count > 0:
item = self.queue.get()
print(item)
self.count = self.count - 1
NP = 3
QL = 4
NI = 5
Q = BoundedQueue(QL)
P = []
for i in range(NP):
t = ProducerThread(Q, NI)
t.setName("Producer-%d" % (i+1))
P.append(t)
C = ConsumerThread(Q, NI*NP)
for t in P:
t.start()
_sleep(0.000001)
C.start()
for t in P:
t.join()
C.join()
if __name__ == '__main__':
_test()
