Merge pull request #80 from zopefoundation/msgpack

Msgpack

.travis.yml

@@ -13,7 +13,10 @@ matrix:
           env: ZEO_MTACCEPTOR=1
         - os: linux
           python: 3.5
-          env: ZEO_MTACCEPTOR=1
+          env: ZEO_MSGPACK=1 ZEO_MTACCEPTOR=1
+        - os: linux
+          python: 2.7
+          env: ZEO_MSGPACK=1
         - os: linux
           python: 2.7
           env: ZEO4_SERVER=1

CHANGES.rst

 Changelog
 =========
 
+- Added support for serializing ZEO messages using `msgpack
+  <http://msgpack.org/index.html>`_ rather than pickle.  This helps
+  pave the way to supporting `byteserver
+  <https://github.com/jimfulton/byteserver>`_, but it also allows ZEO
+  servers to support Python 2 or 3 clients (but not both at the same
+  time) and may provide a small performance improvement.
+
 5.0.4 (2016-11-18)
 ------------------
 

README.rst

@@ -289,6 +289,17 @@ client-conflict-resolution
         Flag indicating that clients should perform conflict
         resolution. This option defaults to false.
 
+msgpack
+        Use `msgpack <http://msgpack.org/index.html>`_ to serialize
+        and de-serialize ZEO protocol messages.
+
+        An advantage of using msgpack for ZEO communication is that
+        it's a tiny bit faster and a ZEO server can support Python 2
+        or Python 3 clients (but not both).
+
+        msgpack can also be enabled by setting the ``ZEO_MSGPACK``
+        environment to a non-empty string.
+
 Server SSL configuration
 ~~~~~~~~~~~~~~~~~~~~~~~~
 

setup.py

@@ -36,7 +36,7 @@ install_requires = [
     'zope.interface',
     ]
 
-tests_require = ['zope.testing', 'manuel', 'random2', 'mock']
+tests_require = ['zope.testing', 'manuel', 'random2', 'mock', 'msgpack-python']
 
 if sys.version_info[:2] < (3, ):
     install_requires.extend(('futures', 'trollius'))
@@ -128,7 +128,11 @@ setup(name="ZEO",
       classifiers = classifiers,
       test_suite="__main__.alltests", # to support "setup.py test"
       tests_require = tests_require,
-      extras_require = dict(test=tests_require, uvloop=['uvloop >=0.5.1']),
+      extras_require = dict(
+          test=tests_require,
+          uvloop=['uvloop >=0.5.1'],
+          msgpack=['msgpack-python'],
+          ),
       install_requires = install_requires,
       zip_safe = False,
       entry_points = """

src/ZEO/ClientStorage.py

@@ -383,7 +383,7 @@ class ClientStorage(ZODB.ConflictResolution.ConflictResolvingStorage):
                 'interfaces', ()):
                 zope.interface.alsoProvides(self, iface)
 
-        if self.protocol_version >= b'Z5':
+        if self.protocol_version[1:] >= b'5':
             self.ping = lambda : self._call('ping')
         else:
             self.ping = lambda : self._call('lastTransaction')

src/ZEO/StorageServer.py

@@ -228,7 +228,7 @@ class ZEOStorage:
         storage = self.storage
 
         supportsUndo = (getattr(storage, 'supportsUndo', lambda : False)()
-                        and self.connection.protocol_version >= b'Z310')
+                        and self.connection.protocol_version[1:] >= b'310')
 
         # Communicate the backend storage interfaces to the client
         storage_provides = zope.interface.providedBy(storage)
@@ -660,6 +660,7 @@ class StorageServer:
                  ssl=None,
                  client_conflict_resolution=False,
                  Acceptor=Acceptor,
+                 msgpack=False,
                  ):
         """StorageServer constructor.
 
@@ -754,7 +755,7 @@ class StorageServer:
         self.client_conflict_resolution = client_conflict_resolution
 
         if addr is not None:
-            self.acceptor = Acceptor(self, addr, ssl)
+            self.acceptor = Acceptor(self, addr, ssl, msgpack)
             if isinstance(addr, tuple) and addr[0]:
                 self.addr = self.acceptor.addr
             else:

src/ZEO/asyncio/base.py

@@ -10,8 +10,6 @@ import socket
 from struct import unpack
 import sys
 
-from .marshal import encoder
-
 logger = logging.getLogger(__name__)
 
 INET_FAMILIES = socket.AF_INET, socket.AF_INET6
@@ -135,7 +133,6 @@ class Protocol(asyncio.Protocol):
     def first_message_received(self, protocol_version):
         # Handler for first/handshake message, set up in __init__
         del self.message_received # use default handler from here on
-        self.encode = encoder()
         self.finish_connect(protocol_version)
 
     def call_async(self, method, args):

src/ZEO/asyncio/client.py

@@ -13,7 +13,7 @@ import ZEO.interfaces
 
 from . import base
 from .compat import asyncio, new_event_loop
-from .marshal import decode
+from .marshal import encoder, decoder
 
 logger = logging.getLogger(__name__)
 
@@ -63,7 +63,7 @@ class Protocol(base.Protocol):
     # One place where special care was required was in cache setup on
     # connect. See finish connect below.
 
-    protocols = b'Z309', b'Z310', b'Z3101', b'Z4', b'Z5'
+    protocols = b'309', b'310', b'3101', b'4', b'5'
 
     def __init__(self, loop,
                  addr, client, storage_key, read_only, connect_poll=1,
@@ -165,13 +165,17 @@ class Protocol(base.Protocol):
         # lastTid before processing (and possibly missing) subsequent
         # invalidations.
 
-        self.protocol_version = min(protocol_version, self.protocols[-1])
-
-        if self.protocol_version not in self.protocols:
+        version = min(protocol_version[1:], self.protocols[-1])
+        if version not in self.protocols:
             self.client.register_failed(
                 self, ZEO.Exceptions.ProtocolError(protocol_version))
             return
 
+        self.protocol_version = protocol_version[:1] + version
+        self.encode = encoder(protocol_version)
+        self.decode = decoder(protocol_version)
+        self.heartbeat_bytes = self.encode(-1, 0, '.reply', None)
+
         self._write(self.protocol_version)
 
         credentials = (self.credentials,) if self.credentials else ()
@@ -199,7 +203,7 @@ class Protocol(base.Protocol):
 
     exception_type_type = type(Exception)
     def message_received(self, data):
-        msgid, async, name, args = decode(data)
+        msgid, async, name, args = self.decode(data)
         if name == '.reply':
             future = self.futures.pop(msgid)
             if (async): # ZEO 5 exception
@@ -267,7 +271,7 @@ class Protocol(base.Protocol):
 
     def heartbeat(self, write=True):
         if write:
-            self._write(b'(J\xff\xff\xff\xffK\x00U\x06.replyNt.')
+            self._write(self.heartbeat_bytes)
         self.heartbeat_handle = self.loop.call_later(
             self.heartbeat_interval, self.heartbeat)
 

src/ZEO/asyncio/marshal.py

@@ -26,10 +26,21 @@ from ..shortrepr import short_repr
 
 logger = logging.getLogger(__name__)
 
-def encoder():
+def encoder(protocol, server=False):
     """Return a non-thread-safe encoder
     """
 
+    if protocol[:1] == b'M':
+        from msgpack import packb
+        default = server_default if server else None
+        def encode(*args):
+            return packb(
+                args, use_bin_type=True, default=default)
+
+        return encode
+    else:
+        assert protocol[:1] == b'Z'
+
     if PY3 or PYPY:
         f = BytesIO()
         getvalue = f.getvalue
@@ -54,9 +65,20 @@ def encoder():
 
 def encode(*args):
 
-    return encoder()(*args)
+    return encoder(b'Z')(*args)
 
-def decode(msg):
+def decoder(protocol):
+    if protocol[:1] == b'M':
+        from msgpack import unpackb
+        def msgpack_decode(data):
+            """Decodes msg and returns its parts"""
+            return unpackb(data, encoding='utf-8', use_list=False)
+        return msgpack_decode
+    else:
+        assert protocol[:1] == b'Z'
+        return pickle_decode
+
+def pickle_decode(msg):
     """Decodes msg and returns its parts"""
     unpickler = Unpickler(BytesIO(msg))
     unpickler.find_global = find_global
@@ -71,7 +93,14 @@ def decode(msg):
         logger.error("can't decode message: %s" % short_repr(msg))
         raise
 
-def server_decode(msg):
+def server_decoder(protocol):
+    if protocol[:1] == b'M':
+        return decoder(protocol)
+    else:
+        assert protocol[:1] == b'Z'
+        return pickle_server_decode
+
+def pickle_server_decode(msg):
     """Decodes msg and returns its parts"""
     unpickler = Unpickler(BytesIO(msg))
     unpickler.find_global = server_find_global
@@ -87,6 +116,17 @@ def server_decode(msg):
         logger.error("can't decode message: %s" % short_repr(msg))
         raise
 
+def server_default(obj):
+    if isinstance(obj, Exception):
+        return reduce_exception(obj)
+    else:
+        return obj
+
+def reduce_exception(exc):
+    class_ = exc.__class__
+    class_ = "%s.%s" % (class_.__module__, class_.__name__)
+    return class_, exc.__dict__ or exc.args
+
 _globals = globals()
 _silly = ('__doc__',)
 

src/ZEO/asyncio/mtacceptor.py

@@ -76,13 +76,14 @@ class Acceptor(asyncore.dispatcher):
     And creates a separate thread for each.
     """
 
-    def __init__(self, storage_server, addr, ssl):
+    def __init__(self, storage_server, addr, ssl, msgpack):
         self.storage_server = storage_server
         self.addr = addr
         self.__socket_map = {}
         asyncore.dispatcher.__init__(self, map=self.__socket_map)
 
         self.ssl_context = ssl
+        self.msgpack = msgpack
         self._open_socket()
 
     def _open_socket(self):
@@ -165,7 +166,7 @@ class Acceptor(asyncore.dispatcher):
             def run():
                 loop = new_event_loop()
                 zs = self.storage_server.create_client_handler()
-                protocol = ServerProtocol(loop, self.addr, zs)
+                protocol = ServerProtocol(loop, self.addr, zs, self.msgpack)
                 protocol.stop = loop.stop
 
                 if self.ssl_context is None:

src/ZEO/asyncio/server.py

@@ -11,13 +11,13 @@ from ..shortrepr import short_repr
 
 from . import base
 from .compat import asyncio, new_event_loop
-from .marshal import server_decode
+from .marshal import server_decoder, encoder, reduce_exception
 
 class ServerProtocol(base.Protocol):
     """asyncio low-level ZEO server interface
     """
 
-    protocols = (b'Z5', )
+    protocols = (b'5', )
 
     name = 'server protocol'
     methods = set(('register', ))
@@ -26,12 +26,16 @@ class ServerProtocol(base.Protocol):
         ZODB.POSException.POSKeyError,
         )
 
-    def __init__(self, loop, addr, zeo_storage):
+    def __init__(self, loop, addr, zeo_storage, msgpack):
         """Create a server's client interface
         """
         super(ServerProtocol, self).__init__(loop, addr)
         self.zeo_storage = zeo_storage
 
+        self.announce_protocol = (
+          (b'M' if msgpack else b'Z') + best_protocol_version
+        )
+
     closed = False
     def close(self):
         logger.debug("Closing server protocol")
@@ -44,7 +48,7 @@ class ServerProtocol(base.Protocol):
     def connection_made(self, transport):
         self.connected = True
         super(ServerProtocol, self).connection_made(transport)
-        self._write(best_protocol_version)
+        self._write(self.announce_protocol)
 
     def connection_lost(self, exc):
         self.connected = False
@@ -61,10 +65,13 @@ class ServerProtocol(base.Protocol):
             self._write(json.dumps(self.zeo_storage.ruok()).encode("ascii"))
             self.close()
         else:
-            if protocol_version in self.protocols:
+            version = protocol_version[1:]
+            if version in self.protocols:
                 logger.info("received handshake %r" %
                             str(protocol_version.decode('ascii')))
                 self.protocol_version = protocol_version
+                self.encode = encoder(protocol_version, True)
+                self.decode = server_decoder(protocol_version)
                 self.zeo_storage.notify_connected(self)
             else:
                 logger.error("bad handshake %s" % short_repr(protocol_version))
@@ -79,7 +86,7 @@ class ServerProtocol(base.Protocol):
 
     def message_received(self, message):
         try:
-            message_id, async, name, args = server_decode(message)
+            message_id, async, name, args = self.decode(message)
         except Exception:
             logger.exception("Can't deserialize message")
             self.close()
@@ -128,10 +135,7 @@ class ServerProtocol(base.Protocol):
     def send_error(self, message_id, exc, send_error=False):
         """Abstracting here so we can make this cleaner in the future
         """
-        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)
+        self.send_reply(message_id, reduce_exception(exc), send_error, 2)
 
     def async(self, method, *args):
         self.call_async(method, args)
@@ -144,8 +148,8 @@ best_protocol_version = os.environ.get(
     ServerProtocol.protocols[-1].decode('utf-8')).encode('utf-8')
 assert best_protocol_version in ServerProtocol.protocols
 
-def new_connection(loop, addr, socket, zeo_storage):
-    protocol = ServerProtocol(loop, addr, zeo_storage)
+def new_connection(loop, addr, socket, zeo_storage, msgpack):
+    protocol = ServerProtocol(loop, addr, zeo_storage, msgpack)
     cr = loop.create_connection((lambda : protocol), sock=socket)
     asyncio.async(cr, loop=loop)
 
@@ -213,10 +217,11 @@ class MTDelay(Delay):
 
 class Acceptor(object):
 
-    def __init__(self, storage_server, addr, ssl):
+    def __init__(self, storage_server, addr, ssl, msgpack):
         self.storage_server = storage_server
         self.addr = addr
         self.ssl_context = ssl
+        self.msgpack = msgpack
         self.event_loop = loop = new_event_loop()
 
         if isinstance(addr, tuple):
@@ -243,7 +248,8 @@ class Acceptor(object):
         try:
             logger.debug("Accepted connection")
             zs = self.storage_server.create_client_handler()
-            protocol = ServerProtocol(self.event_loop, self.addr, zs)
+            protocol = ServerProtocol(
+                self.event_loop, self.addr, zs, self.msgpack)
         except Exception:
             logger.exception("Failure in protocol factory")
 

src/ZEO/runzeo.py

@@ -100,6 +100,7 @@ class ZEOOptionsMixin:
         self.add("client_conflict_resolution",
                  "zeo.client_conflict_resolution",
                  default=0)
+        self.add("msgpack", "zeo.msgpack", default=0)
         self.add("invalidation_queue_size", "zeo.invalidation_queue_size",
                  default=100)
         self.add("invalidation_age", "zeo.invalidation_age")
@@ -347,6 +348,8 @@ def create_server(storages, options):
         storages,
         read_only = options.read_only,
         client_conflict_resolution=options.client_conflict_resolution,
+        msgpack=(options.msgpack if isinstance(options.msgpack, bool)
+                 else os.environ.get('ZEO_MSGPACK')),
         invalidation_queue_size = options.invalidation_queue_size,
         invalidation_age = options.invalidation_age,
         transaction_timeout = options.transaction_timeout,

src/ZEO/server.xml

@@ -115,6 +115,19 @@
       </description>
     </key>
 
+    <key name="msgpack" datatype="boolean" required="no">
+      <description>
+        Use msgpack to serialize and de-serialize ZEO protocol messages.
+
+        An advantage of using msgpack for ZEO communication is that
+        it's a tiny bit faster and a ZEO server can support Python 2
+        or Python 3 clients (but not both).
+
+        msgpack can also be enabled by setting the ``ZEO_MSGPACK``
+        environment to a non-empty string.
+      </description>
+    </key>
+
   </sectiontype>
 
 </component>

src/ZEO/tests/forker.py

@@ -70,7 +70,7 @@ class ZEOConfig:
 
         for name in (
             'invalidation_queue_size', 'invalidation_age',
-            'transaction_timeout', 'pid_filename',
+            'transaction_timeout', 'pid_filename', 'msgpack',
             'ssl_certificate', 'ssl_key', 'client_conflict_resolution',
             ):
             v = getattr(self, name, None)
@@ -200,6 +200,7 @@ def start_zeo_server(storage_conf=None, zeo_conf=None, port=None, keep=False,
                      path='Data.fs', protocol=None, blob_dir=None,
                      suicide=True, debug=False,
                      threaded=False, start_timeout=33, name=None, log=None,
+                     show_config=False
                      ):
     """Start a ZEO server in a separate process.
 
@@ -231,11 +232,14 @@ def start_zeo_server(storage_conf=None, zeo_conf=None, port=None, keep=False,
             z.__dict__.update(zeo_conf)
         zeo_conf = str(z)
 
+    zeo_conf = str(zeo_conf) + '\n\n' + storage_conf
+    if show_config:
+        print(zeo_conf)
+
     # Store the config info in a temp file.
     tmpfile = tempfile.mktemp(".conf", dir=os.getcwd())
     fp = open(tmpfile, 'w')
-    fp.write(str(zeo_conf) + '\n\n')
-    fp.write(storage_conf)
+    fp.write(zeo_conf)
     fp.close()
 
     if threaded:

src/ZEO/tests/protocols.test

@@ -17,15 +17,15 @@ Let's start a Z4 server
     ... '''
 
     >>> addr, stop = start_server(
-    ...    storage_conf, dict(invalidation_queue_size=5), protocol=b'Z4')
+    ...    storage_conf, dict(invalidation_queue_size=5), protocol=b'4')
 
 A current client should be able to connect to a old server:
 
     >>> import ZEO, ZODB.blob, transaction
     >>> db = ZEO.DB(addr, client='client', blob_dir='blobs')
     >>> wait_connected(db.storage)
-    >>> str(db.storage.protocol_version.decode('ascii'))
-    'Z4'
+    >>> str(db.storage.protocol_version.decode('ascii'))[1:]
+    '4'
 
     >>> conn = db.open()
     >>> conn.root().x = 0

src/ZEO/tests/servertesting.py

@@ -44,7 +44,7 @@ class StorageServer(ZEO.StorageServer.StorageServer):
 def client(server, name='client'):
     zs = ZEO.StorageServer.ZEOStorage(server)
     protocol = ZEO.asyncio.tests.server_protocol(
-        zs, protocol_version=b'Z5', addr='test-addr-%s' % name)
+        False, zs, protocol_version=b'Z5', addr='test-addr-%s' % name)
     zs.notify_connected(protocol)
     zs.register('1', 0)
     return zs

src/ZEO/tests/testZEO.py

@@ -1465,6 +1465,36 @@ def ClientDisconnected_errors_are_TransientErrors():
     True
     """
 
+if not os.environ.get('ZEO4_SERVER'):
+    if os.environ.get('ZEO_MSGPACK'):
+        def test_runzeo_msgpack_support():
+            """
+            >>> import ZEO
+
+            >>> a, s = ZEO.server(threaded=False)
+            >>> conn = ZEO.connection(a)
+            >>> str(conn.db().storage.protocol_version.decode('ascii'))
+            'M5'
+            >>> conn.close(); s()
+            """
+    else:
+        def test_runzeo_msgpack_support():
+            """
+            >>> import ZEO
+
+            >>> a, s = ZEO.server(threaded=False)
+            >>> conn = ZEO.connection(a)
+            >>> str(conn.db().storage.protocol_version.decode('ascii'))
+            'Z5'
+            >>> conn.close(); s()
+
+            >>> a, s = ZEO.server(zeo_conf=dict(msgpack=True), threaded=False)
+            >>> conn = ZEO.connection(a)
+            >>> str(conn.db().storage.protocol_version.decode('ascii'))
+            'M5'
+            >>> conn.close(); s()
+            """
+
 if sys.platform.startswith('win'):
     del runzeo_logrotate_on_sigusr2
     del unix_domain_sockets

src/ZEO/tests/utils.py

@@ -39,7 +39,7 @@ class StorageServer:
     """
 
     def __init__(self, test, storage,
-                 protocol_version=best_protocol_version,
+                 protocol_version=b'Z' + best_protocol_version,
                  **kw):
         self.test = test
         self.storage_server = ZEO.StorageServer.StorageServer(