Don't pickle errors.

Don't send errors as pickles. Send error names and class dict or arguments.

This is a step toward eliminating instance pickles from the ZEO protocol.

This is much messier than I'd like it to be because of the way both
python and ZEO handle exceptions.

src/ZEO/Exceptions.py

@@ -37,3 +37,7 @@ class AuthError(StorageError):
 class ProtocolError(ClientStorageError):
     """A client contacted a server with an incomparible protocol
     """
+
+class ServerException(ClientStorageError):
+    """
+    """

src/ZEO/asyncio/client.py

@@ -5,7 +5,7 @@
 else:
     import trollius as asyncio
 
-from ZEO.Exceptions import ClientDisconnected
+from ZEO.Exceptions import ClientDisconnected, ServerException
 import concurrent.futures
 import functools
 import logging
@@ -208,15 +208,22 @@ def message_received(self, data):
         msgid, async, name, args = decode(data)
         if name == '.reply':
             future = self.futures.pop(msgid)
-            if (isinstance(args, tuple) and len(args) > 1 and
-                type(args[0]) == self.exception_type_type and
-                issubclass(args[0], Exception)
-                ):
-                if not issubclass(
-                    args[0], (
-                        ZODB.POSException.POSKeyError,
-                        ZODB.POSException.ConflictError,)
-                    ):
+            if (async): # ZEO 5 exception
+                class_, args = args
+                factory = exc_factories.get(class_)
+                if factory:
+                    exc = factory(class_, args)
+                    if not isinstance(exc, unlogged_exceptions):
+                        logger.error("%s from server: %s:%s",
+                                     self.name, class_, args)
+                else:
+                    exc = ServerException(class_, args)
+                future.set_exception(exc)
+            elif (isinstance(args, tuple) and len(args) > 1 and
+                  type(args[0]) == self.exception_type_type and
+                  issubclass(args[0], Exception)
+                  ):
+                if not issubclass(args[0], unlogged_exceptions):
                     logger.error("%s from server: %s.%s:%s",
                                  self.name,
                                  args[0].__module__,
@@ -270,6 +277,52 @@ def heartbeat(self, write=True):
         self.heartbeat_handle = self.loop.call_later(
             self.heartbeat_interval, self.heartbeat)
 
+def create_Exception(class_, args):
+    return exc_classes[class_](*args)
+
+def create_ConflictError(class_, args):
+    exc = exc_classes[class_](
+        message = args['message'],
+        oid     = args['oid'],
+        serials = args['serials'],
+        )
+    exc.class_name = args.get('class_name')
+    return exc
+
+def create_BTreesConflictError(class_, args):
+    return ZODB.POSException.BTreesConflictError(
+        p1 = args['p1'],
+        p2 = args['p2'],
+        p3 = args['p3'],
+        reason = args['reason'],
+        )
+
+def create_MultipleUndoErrors(class_, args):
+    return ZODB.POSException.MultipleUndoErrors(args['_errs'])
+
+exc_classes = {
+    'builtins.KeyError': KeyError,
+    'builtins.TypeError': TypeError,
+    'ZODB.POSException.ConflictError': ZODB.POSException.ConflictError,
+    'ZODB.POSException.POSKeyError': ZODB.POSException.POSKeyError,
+    'ZODB.POSException.ReadConflictError': ZODB.POSException.ReadConflictError,
+    'ZODB.POSException.ReadOnlyError': ZODB.POSException.ReadOnlyError,
+    'ZODB.POSException.StorageTransactionError':
+    ZODB.POSException.StorageTransactionError,
+    }
+exc_factories = {
+    'builtins.KeyError': create_Exception,
+    'builtins.TypeError': create_Exception,
+    'ZODB.POSException.BTreesConflictError': create_BTreesConflictError,
+    'ZODB.POSException.ConflictError': create_ConflictError,
+    'ZODB.POSException.MultipleUndoErrors': create_MultipleUndoErrors,
+    'ZODB.POSException.POSKeyError': create_Exception,
+    'ZODB.POSException.ReadConflictError': create_ConflictError,
+    'ZODB.POSException.ReadOnlyError': create_Exception,
+    'ZODB.POSException.StorageTransactionError': create_Exception,
+    }
+unlogged_exceptions = (ZODB.POSException.POSKeyError,
+                       ZODB.POSException.ConflictError)
 class Client(object):
     """asyncio low-level ZEO client interface
     """

src/ZEO/asyncio/server.py

@@ -111,9 +111,9 @@ def message_received(self, message):
         if not async:
             self.send_reply(message_id, result)
 
-    def send_reply(self, message_id, result, send_error=False):
+    def send_reply(self, message_id, result, send_error=False, flag=0):
         try:
-            result = self.encode(message_id, 0, '.reply', result)
+            result = self.encode(message_id, flag, '.reply', result)
         except Exception:
             if isinstance(result, Delay):
                 result.set_sender(message_id, self)
@@ -134,7 +134,10 @@ def send_reply_threadsafe(self, message_id, result):
     def send_error(self, message_id, exc, send_error=False):
         """Abstracting here so we can make this cleaner in the future
         """
-        self.send_reply(message_id, (exc.__class__, exc), send_error)
+        class_ = exc.__class__
+        class_ = "%s.%s" % (class_.__module__, class_.__name__)
+        args = class_, exc.__dict__ or exc.args
+        self.send_reply(message_id, args, send_error, 2)
 
     def async(self, method, *args):
         self.call_async(method, args)