thread-safe messenger

direct/src/showbase/Messenger.py

@@ -5,6 +5,7 @@ __all__ = ['Messenger']
 
 from PythonUtil import *
 from direct.directnotify import DirectNotifyGlobal
+from direct.stdpy.threading import Lock
 import types
 
 class Messenger:
@@ -34,6 +35,14 @@ class Messenger:
         self._messengerIdGen = 0
         self._id2object = {}
 
+        # A mapping of taskChain -> eventList, used for sending events
+        # across task chains (and therefore across threads).
+        self._eventQueuesByTaskChain = {}
+
+        # This protects the data structures within this object from
+        # multithreaded access.
+        self.lock = Lock()
+
         if __debug__:
             self.__isWatching=0
             self.__watching={}
@@ -58,7 +67,7 @@ class Messenger:
 
     def _storeObject(self, object):
         # store reference-counted reference to object in case we need to
-        # retrieve it later
+        # retrieve it later.  assumes lock is held.
         id = self._getMessengerId(object)
         if id not in self._id2object:
             self._id2object[id] = [1, object]
@@ -69,6 +78,7 @@ class Messenger:
         return self._id2object[id][1]
 
     def _releaseObject(self, object):
+        # assumes lock is held.
         id = self._getMessengerId(object)
         if id in self._id2object:
             record = self._id2object[id]
@@ -97,6 +107,8 @@ class Messenger:
             "method not callable in accept (ignoring): %s %s"%
             (method, extraArgs))
 
+        self.lock.acquire()
+        try:
             acceptorDict = self.__callbacks.setdefault(event, {})
 
             id = self._getMessengerId(object)
@@ -121,6 +133,8 @@ class Messenger:
             eventDict = self.__objectEvents.setdefault(id, {})
             eventDict.setdefault(event, None)
             self._storeObject(object)
+        finally:
+            self.lock.release()
 
     def ignore(self, event, object):
         """ ignore(self, string, DirectObject)
@@ -130,6 +144,8 @@ class Messenger:
         if Messenger.notify.getDebug():
             Messenger.notify.debug(`object` + '\n now ignoring: ' + `event`)
 
+        self.lock.acquire()
+        try:
             id = self._getMessengerId(object)
 
             # Find the dictionary of all the objects accepting this event
@@ -150,6 +166,8 @@ class Messenger:
                     del self.__objectEvents[id]
 
             self._releaseObject(object)
+        finally:
+            self.lock.release()
 
     def ignoreAll(self, object):
         """
@@ -158,6 +176,9 @@ class Messenger:
         """
         if Messenger.notify.getDebug():
             Messenger.notify.debug(`object` + '\n now ignoring all events')
+
+        self.lock.acquire()
+        try:
             id = self._getMessengerId(object)
             # Get the list of events this object is listening to
             eventDict = self.__objectEvents.get(id)
@@ -175,11 +196,15 @@ class Messenger:
                 del self.__objectEvents[id]
             if id in self._id2object:
                 del self._id2object[id]
+        finally:
+            self.lock.release()
 
     def getAllAccepting(self, object):
         """
         Returns the list of all events accepted by the indicated object.
         """
+        self.lock.acquire()
+        try:
             id = self._getMessengerId(object)
 
             # Get the list of events this object is listening to
@@ -187,11 +212,15 @@ class Messenger:
             if eventDict:
                 return eventDict.keys()
             return []
+        finally:
+            self.lock.release()
 
     def isAccepting(self, event, object):
         """ isAccepting(self, string, DirectOject)
         Is this object accepting this event?
         """
+        self.lock.acquire()
+        try:
             acceptorDict = self.__callbacks.get(event)
             id = self._getMessengerId(object)
             if acceptorDict and acceptorDict.has_key(id):
@@ -200,6 +229,8 @@ class Messenger:
             # If we looked in both dictionaries and made it here
             # that object must not be accepting that event.
             return 0
+        finally:
+            self.lock.release()
 
     def whoAccepts(self, event):
         """
@@ -213,17 +244,28 @@ class Messenger:
         """
         return (not self.isAccepting(event, object))
 
-    def send(self, event, sentArgs=[]):
+    def send(self, event, sentArgs=[], taskChain = None):
         """
+        Send this event, optionally passing in arguments
+
         event is usually a string.
         sentArgs is a list of any data that you want passed along to the
             handlers listening to this event.
 
-        Send this event, optionally passing in arguments
+        If taskChain is not None, it is the name of the task chain
+        which should receive the event.  If taskChain is None, the
+        event is handled immediately.  Setting a non-None taskChain
+        will defer the event (possibly till next frame or even later)
+        and create a new, temporary task within the named taskChain,
+        but this is the only way to send an event across threads.
         """
         if Messenger.notify.getDebug() and not self.quieting.get(event):
             assert Messenger.notify.debug(
-                'sent event: ' + event + ' sentArgs: ' + `sentArgs`)
+                'sent event: %s sentArgs = %s, taskChain = %s' % (
+                event, sentArgs, taskChain))
+
+        self.lock.acquire()
+        try:
             if __debug__:
                 foundWatch=0
                 if self.__isWatching:
@@ -237,6 +279,24 @@ class Messenger:
                     if foundWatch:
                         print "Messenger: \"%s\" was sent, but no function in Python listened."%(event,)
                 return
+
+            if taskChain:
+                # Queue the event onto the indicated task chain.
+                taskMgr.add(self.__lockAndDispatch, name = 'Messenger-%s-%s' % (event, taskChain), extraArgs = [acceptorDict, event, sentArgs, foundWatch], taskChain = taskChain)
+            else:
+                # Handle the event immediately.
+                self.__dispatch(acceptorDict, event, sentArgs, foundWatch)
+        finally:
+            self.lock.release()
+
+    def __lockAndDispatch(self, acceptorDict, event, sentArgs, foundWatch):
+        self.lock.acquire()
+        try:
+            self.__dispatch(acceptorDict, event, sentArgs, foundWatch)
+        finally:
+            self.lock.release()
+
+    def __dispatch(self, acceptorDict, event, sentArgs, foundWatch):
         for id in acceptorDict.keys():
             # We have to make this apparently redundant check, because
             # it is possible that one object removes its own hooks
@@ -283,14 +343,24 @@ class Messenger:
                 # method itself might call accept() or acceptOnce()
                 # again.
                 assert callable(method)
+
+                # Release the lock temporarily while we call the method.
+                self.lock.release()
+                try:
                     method (*(extraArgs + sentArgs))
+                finally:
+                    self.lock.acquire()
 
     def clear(self):
         """
         Start fresh with a clear dict
         """
+        self.lock.acquire()
+        try:
             self.__callbacks.clear()
             self.__objectEvents.clear()
+        finally:
+            self.lock.release()
 
     def isEmpty(self):
         return (len(self.__callbacks) == 0)