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 <[email protected]>
Newsgroups gmane.comp.python.zope.zodb.cvs
Message-ID <20090911193413.231EF94186__23654.8688066707$1252697666$gmane$org@cvs.zope.org>
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)
lmpx.com only provides a reader for public news (NNTP) servers. It is not affiliated with the servers or forums shown here and is not responsible for the content of articles, which is written by their respective authors.