[Zodb-checkins] SVN: ZODB/branches/jim-thready-zeo/src/ZEO/zrpc/connection.py Try giving each client it's own thread, for methods not otherwise

[Jim Fulton]

Log message for revision 103854:
  Try giving each client it's own thread, for methods not otherwise
  handed off to worker threads,
  

Changed:
  U   ZODB/branches/jim-thready-zeo/src/ZEO/zrpc/connection.py

-=-
Modified: ZODB/branches/jim-thready-zeo/src/ZEO/zrpc/connection.py
===================================================================
--- ZODB/branches/jim-thready-zeo/src/ZEO/zrpc/connection.py	2009-09-11 15:55:30 UTC (rev 103853)
+++ ZODB/branches/jim-thready-zeo/src/ZEO/zrpc/connection.py	2009-09-11 19:34:12 UTC (rev 103854)
@@ -14,6 +14,7 @@
 import asyncore
 import atexit
 import errno
+import Queue
 import select
 import sys
 import threading
@@ -780,8 +781,6 @@
             self.log("poll()", level=TRACE)
         self.trigger.pull_trigger()
 
-
-
 class ManagedServerConnection(Connection):
     """Server-side Connection subclass."""
 
@@ -790,9 +789,27 @@
 
     def __init__(self, sock, addr, obj, mgr):
         self.mgr = mgr
+        self.queue = Queue.Queue()
+        self.thread = threading.Thread(target=self.server_thread)
+        self.thread.setDaemon(True)
+        self.thread.start()
         Connection.__init__(self, sock, addr, obj, 'S')
         self.marshal = ServerMarshaller()
 
+    def server_thread(self):
+        while 1:
+            try:
+                req = self.queue.get()
+                if not req:
+                    break
+                Connection.handle_request(self, *req)
+            except:
+                logger.critical('Error in thready job %r', delay,
+                                exc_info=sys.exc_info())
+
+    def handle_request(self, *args):
+        self.queue.put(args)
+
     def handshake(self):
         # Send the server's preferred protocol to the client.
         self.message_output(self.current_protocol)
@@ -802,6 +819,8 @@
         self.obj.notifyConnected(self)
 
     def close(self):
+        self.queue.put(None)
+        self.thread.join(1)
         self.obj.notifyDisconnected()
         Connection.close(self)