SVN: ZODB/trunk/src/ZEO/ Fixed a threading bug.

Log message for revision 108740:
  Fixed a threading bug.
  

Changed:
  U   ZODB/trunk/src/ZEO/StorageServer.py
  U   ZODB/trunk/src/ZEO/zrpc/connection.py

-=-
Modified: ZODB/trunk/src/ZEO/StorageServer.py
===================================================================
--- ZODB/trunk/src/ZEO/StorageServer.py	2010-02-03 22:09:53 UTC (rev 108739)
+++ ZODB/trunk/src/ZEO/StorageServer.py	2010-02-03 22:09:56 UTC (rev 108740)
@@ -1346,8 +1346,21 @@
         self.rpc.callAsync('endVerify')
 
     def invalidateTransaction(self, tid, args):
-        self.rpc.callAsync('invalidateTransaction', tid, args)
+        # Note that this method is *always* called from a different
+        # thread than self.rpc's async thread. It is the only method
+        # for which this is true and requires special consideration!
 
+        # callAsyncNoSend is important here because:
+        # - callAsyncNoPoll isn't appropriate because
+        #   the network thread may not wake up for a long time,
+        #   delaying invalidations for too long. (This is demonstrateed
+        #   by a test failure.)
+        # - callAsync isn't appropriate because (on the server) it tries
+        #   to write to the socket.  If self.rpc's network thread also
+        #   tries to write at the ame time, we can run into problems
+        #   because handle_write isn't thread safe.
+        self.rpc.callAsyncNoSend('invalidateTransaction', tid, args)
+
     def serialnos(self, arg):
         self.rpc.callAsyncNoPoll('serialnos', arg)
 

Modified: ZODB/trunk/src/ZEO/zrpc/connection.py
===================================================================
--- ZODB/trunk/src/ZEO/zrpc/connection.py	2010-02-03 22:09:53 UTC (rev 108739)
+++ ZODB/trunk/src/ZEO/zrpc/connection.py	2010-02-03 22:09:56 UTC (rev 108740)
@@ -535,6 +535,15 @@
             raise DisconnectedError()
         self.send_call(method, args, 1)
 
+    def callAsyncNoSend(self, method, *args):
+        # Like CallAsync but doesn't poll.  This exists so that we can
+        # send invalidations atomically to all clients without
+        # allowing any client to sneak in a load request.
+        if self.closed:
+            raise DisconnectedError()
+        self.send_call(method, args, 1)
+        self.call_from_thread()
+
     def callAsyncIterator(self, iterator):
         """Queue a sequence of calls using an iterator