Changed close to use the asyncore thread to try to avoid a race and

breaking the client loop.

src/ZEO/ClientStorage.py

@@ -442,11 +442,20 @@ class ClientStorage(object):
             logger.info("%s Waiting for cache verification to finish",
                         self.__name__)
 
-    def close(self):
-        """Storage API: finalize the storage, releasing external resources."""
+    def close(self, kill=False):
+        "Storage API: finalize the storage, releasing external resources."
         if self._rpc_mgr is not None:
             self._rpc_mgr.close()
             self._rpc_mgr = None
+
+        if (self._connection is not None) and not kill:
+            event = threading.Event()
+            self._connection.trigger.pull_trigger(lambda: self._close(event))
+            event.wait(9)
+        else:
+            self._close()
+
+    def _close(self, event=None):
         if self._connection is not None:
             self._connection.register_object(None) # Don't call me!
             self._connection.close()
@@ -462,6 +471,9 @@ class ClientStorage(object):
         if self._check_blob_size_thread is not None:
             self._check_blob_size_thread.join()
 
+        if event is not None:
+            event.set()
+
     _check_blob_size_thread = None
     def _check_blob_size(self, bytes=None):
         if self._blob_cache_size is None:

src/ZEO/tests/testZEO.py

@@ -951,6 +951,8 @@ def tpc_finish_error():
     ...         pass
     ...     def close(self):
     ...         print 'connection closed'
+    ...     trigger = property(lambda self: self)
+    ...     pull_trigger = lambda self, func: func()
 
     >>> class ConnectionManager:
     ...     def __init__(self, addr, client, tmin, tmax):
@@ -1228,6 +1230,35 @@ def runzeo_without_configfile():
     testing exit immediately
     """
 
+def close_client_storage_w_invalidations():
+    r"""
+Invalidations could cause errors when closing client storages,
+
+    >>> addr, _ = start_server()
+    >>> writing = threading.Event()
+    >>> def mad_write_thread():
+    ...     global writing
+    ...     conn = ZEO.connection(addr)
+    ...     writing.set()
+    ...     while writing.isSet():
+    ...         conn.root.x = 1
+    ...         transaction.commit()
+    ...     conn.close()
+
+    >>> thread = threading.Thread(target=mad_write_thread)
+    >>> thread.setDaemon(True)
+    >>> thread.start()
+    >>> writing.wait()
+    >>> time.sleep(.01)
+    >>> for i in range(10):
+    ...     conn = ZEO.connection(addr)
+    ...     _ = conn._storage.load('\0'*8)
+    ...     conn.close()
+
+    >>> writing.clear()
+    >>> thread.join(1)
+    """
+
 slow_test_classes = [
     BlobAdaptedFileStorageTests, BlobWritableCacheTests,
     DemoStorageTests, FileStorageTests, MappingStorageTests,