Tried to make management of the client loop more robust and added a

test for it.

src/ZEO/tests/testZEO.py

@@ -271,6 +271,43 @@ class HeartbeatTests(ZEO.tests.ConnectionTests.CommonSetupTearDown):
                      > client_timeout_count)
 
 
+class CatastrophicClientLoopFailure(
+    ZEO.tests.ConnectionTests.CommonSetupTearDown):
+    """Test what happens when the client loop falls over
+    """
+
+    def getConfig(self, path, create, read_only):
+        return """<mappingstorage 1/>"""
+
+    def checkCatastrophicClientLoopFailure(self):
+        self._storage = self.openClientStorage()
+
+        class Evil:
+            def writable(self):
+                raise SystemError("I'm evil")
+
+        log = []
+        ZEO.zrpc.connection.client_logger.critical = (
+            lambda m, *a, **kw: log.append((m % a, kw))
+            )
+
+        ZEO.zrpc.connection.client_map[None] = Evil()
+        
+        try:
+            ZEO.zrpc.connection.client_trigger.pull_trigger()
+        except DisconnectedError:
+            pass
+
+        time.sleep(.1)
+        self.failIf(self._storage.is_connected())
+        self.assertEqual(len(ZEO.zrpc.connection.client_map), 1)
+        del ZEO.zrpc.connection.client_logger.critical
+        self.assertEqual(log[0][0], 'The ZEO cient loop failed.')
+        self.assert_('exc_info' in log[0][1])
+        self.assertEqual(log[1][0], "Couldn't close a dispatcher.")
+        self.assert_('exc_info' in log[1][1])
+    
+
 class DemoStorageWrappedAroundClientStorage(DemoStorageWrappedBase):
 
     def getConfig(self):
@@ -307,6 +344,7 @@ test_classes = [OneTimeTests,
                 MappingStorageTests,
                 DemoStorageWrappedAroundClientStorage,
                 HeartbeatTests,
+                CatastrophicClientLoopFailure,
                ]
 
 def test_suite():

src/ZEO/zrpc/connection.py

@@ -34,19 +34,19 @@ ASYNC = 1
 
 ##############################################################################
 # Dedicated Client select loop:
-client_map = {}
-client_trigger = trigger(client_map)
 client_timeout = 30.0
 client_timeout_count = 0 # for testing
+client_map = {}
+client_trigger = trigger(client_map)
+client_logger = logging.getLogger('ZEO.zrpc.client_loop')
 
 def client_loop():
     map = client_map
-    logger = logging.getLogger('ZEO.zrpc.client_loop')
-    logger.addHandler(logging.StreamHandler())
 
     read = asyncore.read
     write = asyncore.write
     _exception = asyncore._exception
+    loop_failures = 0
     
     while map:
         try:
@@ -106,8 +106,17 @@ def client_loop():
                 _exception(obj)
 
         except:
-            logger.exception('poll failure')
-            raise
+            client_logger.critical('The ZEO cient loop failed.',
+                            exc_info=sys.exc_info())
+            for fd, obj in map.items():
+                if obj is client_trigger:
+                    continue
+                try:
+                    obj.mgr.client.close()
+                except:
+                    map.pop(fd, None)
+                    client_logger.critical("Couldn't close a dispatcher.",
+                                           exc_info=sys.exc_info())
 
 client_thread = threading.Thread(target=client_loop)
 client_thread.setDaemon(True)