ZEO.asyncio: switch to async/await style

CHANGES.rst

@@ -1,6 +1,26 @@
 Changelog
 =========
 
+6.0 (unreleased)
+----------------
+
+- Drop Python 2 support
+
+- Reimplement and streamline the ``asyncio`` part of the ``ClientStorage``
+  implementation:
+   - switch from futures with explicit callbacks to `async/await` style
+   - use standard ``asyncio`` features to implement timeouts
+   - redesign the API of the class implementing the ZEO client protocol
+   - significantly improve source documentation
+
+- Drop credentials support: the corresponding ``ClientStorage.__init__``
+  parameters (i.e. ``credentials``, ``username``, ``password``)
+  are retained but ignored.
+  Note: the ZEO 5 server never supported credentials; the feature
+  has previously been retained for the use case
+  "ZEO 5 client with ZEO 4 server".
+
+
 5.3.1 (unreleased)
 ------------------
 
@@ -21,6 +41,26 @@ Changelog
 
 
 
+5.3.0 (2022-03-24)
+------------------
+
+- Add ``ConflictError`` to the list of unlogged server exceptions
+  (the client/its application should determine whether it wants
+  them logged).
+
+  Prevent ``no current transaction: tpc_abort()`` server log entries.
+  The storage API allows ``tpc_abort`` to be called with an
+  invalid transaction (the call should be ignored in this case)
+  and the server's ``tpc_vote`` relies on this.
+
+  Change the server's log message label for request exceptions
+  from ``Bad request ...`` to ``... raised exception:``,
+  hinting towards a server rather than client problem.
+
+  See `issue 156 <https://github.com/zopefoundation/ZEO/issues/156>`_.
+
+
+
 5.3.0 (2022-03-24)
 ------------------
 
@@ -105,7 +145,7 @@ Changelog
 
 - Fixed to work with some changes made in ZODB 5.4.0.
 
-  Client-side updates are incuded for ZODB 5.4.0 or databases that
+  Client-side updates are included for ZODB 5.4.0 or databases that
   already had ``zodbpickle.binary`` OIDs. See `issue 113
   <https://github.com/zopefoundation/ZEO/issues/113>`_.
 

setup.py

@@ -33,17 +33,13 @@
     'ZODB >= 5.5.1',
     'zope.testing',
     'manuel',
-    'random2',
-    'mock',
     'msgpack < 1',
     'zope.testrunner',
 ]
 
 classifiers = """
 Intended Audience :: Developers
 License :: OSI Approved :: Zope Public License
-Programming Language :: Python :: 2
-Programming Language :: Python :: 2.7
 Programming Language :: Python :: 3
 Programming Language :: Python :: 3.5
 Programming Language :: Python :: 3.6
@@ -158,5 +154,5 @@ def emit(self, record):
       zeo-nagios = ZEO.nagios:main
       """,
       include_package_data=True,
-      python_requires='>=2.7.9,!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*',
+      python_requires='>=3.5.2',
 )

src/ZEO/ClientStorage.py

@@ -108,6 +108,7 @@ def __init__(self, addr, storage='1', cache_size=20 * MB,
         This is typically invoked from a custom_zodb.py file.
 
         All arguments except addr should be keyword arguments.
+
         Arguments:
 
         addr
@@ -118,6 +119,20 @@ def __init__(self, addr, storage='1', cache_size=20 * MB,
             connection.  A hostname may be a DNS name or a dotted IP
             address.  Required.
 
+            All addresses are assumed to serve (essentially)
+            the same (potentially replicated) storage.
+            A connection tries to connect to those addresses;
+            the first successful connection establishment with
+            the called for ("read_only" or "writable") capabilities
+            is selected and used for storage interaction until
+            the connection is lost. In that case, a
+            reconnection is tried.
+            If ``ClientStorage`` calls for the "writable" capability
+            but allows for a "read only" fallback,,
+            a read only connection can be used as a fallback;
+            if a writable connection becomes available later, a
+            switch to this connection is performed.
+
         storage
             The server storage name, defaulting to '1'.  The name must
             match one of the storage names supported by the server(s)
@@ -132,14 +147,12 @@ def __init__(self, addr, storage='1', cache_size=20 * MB,
             address and the server storage name.  This is used to
             construct the response to getName()
 
-        cache
-            A cache object or a name, relative to the current working
-            directory, used to construct persistent cache filenames.
-            Defaults to None, in which case the cache is not
-            persistent.  See ClientCache for more info.
-
         wait_timeout
-            Maximum time to wait for results, including connecting.
+            Maximum time (seconds) to wait for connections,
+            defaulting to 30.
+            Note: the timeout applies only to [re]connect.
+            Normal operations can take arbitrary long. This
+            is important for long running operations, such as ``pack``.
 
         read_only
             A flag indicating whether this should be a
@@ -160,7 +173,7 @@ def __init__(self, addr, storage='1', cache_size=20 * MB,
         shared_blob_dir
             Flag whether the blob_dir is a server-shared filesystem
             that should be used instead of transferring blob data over
-            ZEO protocol.
+            the ZEO protocol.
 
         blob_cache_size
             Maximum size of the ZEO blob cache, in bytes.  If not set, then
@@ -170,22 +183,49 @@ def __init__(self, addr, storage='1', cache_size=20 * MB,
             This option is ignored if shared_blob_dir is true.
 
         blob_cache_size_check
-            ZEO check size as percent of blob_cache_size.  The ZEO
+            Cache check size as percent of blob_cache_size.  The ZEO
             cache size will be checked when this many bytes have been
             loaded into the cache. Defaults to 10% of the blob cache
             size.   This option is ignored if shared_blob_dir is true.
 
         client_label
             A label to include in server log messages for the client.
 
-        Note that the authentication protocol is defined by the server
-        and is detected by the ClientStorage upon connecting (see
-        testConnection() and doAuth() for details).
-
+        cache
+            A cache object or a file path (relative or absolute).
+            Defaults to None, in which case the cache is determined
+            from client and var.
+
+        ssl
+            An ssl client context (i.e. with purpose "ServerAuth")
+            to call for SSL connections.
+
+        ssl_server_hostname
+            The server hostname - used during the SSL authentication check
+
+        client
+        var
+            If cache is None, client determines the cache:
+            if it is None, then a non persisent cache is used;
+            otherwie, client is used together with var (defaults
+            to the current working directory) to construct the
+            file path for the persistent cache file
+
+        wait
+            Wait for server connection, defaulting to true.
+
+        credentials
+        username
+        password
+        realm
+        disconnect_poll
+        min_disconnect_poll
+        max_disconnect_poll
+        drop_cache_rather_verify
+        server_sync
+            ignored; retained (as parameters) for compatibility
         """
 
-        assert not username or password or realm
-
         if isinstance(addr, int):
             addr = ('127.0.0.1', addr)
 
@@ -260,14 +300,11 @@ def __init__(self, addr, storage='1', cache_size=20 * MB,
                 blob_cache_size * blob_cache_size_check // 100)
             self._check_blob_size()
 
-        self.server_sync = server_sync
-
         self._server = _client_factory(
             addr, self, cache, storage,
             ZEO.asyncio.client.Fallback if read_only_fallback else read_only,
             wait_timeout or 30,
             ssl = ssl, ssl_server_hostname=ssl_server_hostname,
-            credentials=credentials,
             )
         self._call = self._server.call
         self._async = self._server.async_
@@ -280,7 +317,7 @@ def __init__(self, addr, storage='1', cache_size=20 * MB,
             try:
                 self._wait()
             except Exception:
-                # No point in keeping the server going of the storage
+                # No point in keeping the server going if the storage
                 # creation fails
                 self._server.close()
                 raise
@@ -368,8 +405,7 @@ def notify_connected(self, conn, info):
         self._connection_generation += 1
 
         if self._client_label:
-            conn.call_async_from_same_thread(
-                'set_client_label', self._client_label)
+            conn.call_async('set_client_label', self._client_label)
 
         self._info.update(info)
 
@@ -386,12 +422,10 @@ def notify_connected(self, conn, info):
                 zope.interface.alsoProvides(self, iface)
 
         if self.protocol_version[1:] >= b'5':
-            self.ping = lambda : self._call('ping')
+            self.ping = lambda : self._call('ping', timeout=0)
         else:
-            self.ping = lambda : self._call('lastTransaction')
+            self.ping = lambda : self._call('lastTransaction', timeout=0)
 
-        if self.server_sync:
-            self.sync = self.ping
 
     def set_server_addr(self, addr):
         # Normalize server address and convert to string
@@ -863,7 +897,10 @@ def tpc_abort(self, txn, timeout=None):
                 # wait a little while in hopes of reconnecting.  If
                 # we're able to reconnect and retry the transaction,
                 # ten it might succeed!
-                self._call('tpc_abort', id(txn), timeout=timeout)
+                if timeout is not None:
+                    self._call('tpc_abort', id(txn), timeout=timeout)
+                else:
+                    self._call('tpc_abort', id(txn))
             except ClientDisconnected:
                 logger.debug("%s ClientDisconnected in tpc_abort() ignored",
                              self.__name__)
@@ -1034,8 +1071,9 @@ def _iterator_gc(self, disconnected=False):
                 return self._iterator_gc(True)
             self._iterator_ids -= iids
 
-    def server_status(self):
-        return self._call('server_status')
+    def server_status(self, **kw):
+        """supported keywords: ``timeout``."""
+        return self._call('server_status', **kw)
 
 class TransactionIterator(object):
 

src/ZEO/StorageServer.py

@@ -37,7 +37,7 @@
 import zope.interface
 import six
 
-from ZEO._compat import Pickler, Unpickler, PY3, BytesIO
+from ZEO._compat import Pickler, Unpickler, BytesIO
 from ZEO.Exceptions import AuthError
 from ZEO.monitor import StorageStats
 from ZEO.asyncio.server import Delay, MTDelay, Result
@@ -623,6 +623,7 @@ def ruok(self):
     def ping(self):
         pass
 
+
 class StorageServerDB(object):
     """Adapter from StorageServerDB to ZODB.interfaces.IStorageWrapper
 
@@ -949,9 +950,8 @@ def server_status(self, storage_id):
         status['timeout-thread-is-alive'] = lock_manager.timeout.is_alive()
         last_transaction = self.storages[storage_id].lastTransaction()
         last_transaction_hex = codecs.encode(last_transaction, 'hex_codec')
-        if PY3:
-            # doctests and maybe clients expect a str, not bytes
-            last_transaction_hex = str(last_transaction_hex, 'ascii')
+        # doctests and maybe clients expect a str, not bytes
+        last_transaction_hex = str(last_transaction_hex, 'ascii')
         status['last-transaction'] = last_transaction_hex
         return status
 

src/ZEO/_compat.py

@@ -16,50 +16,26 @@
 import sys
 import platform
 
-PY3 = sys.version_info[0] >= 3
-PY32 = sys.version_info[:2] == (3, 2)
 PYPY = getattr(platform, 'python_implementation', lambda: None)() == 'PyPy'
 WIN = sys.platform.startswith('win')
 
-if PY3:
-    from zodbpickle.pickle import Pickler, Unpickler as _Unpickler, dump, dumps, loads
-    class Unpickler(_Unpickler):
-        # Py3: Python 3 doesn't allow assignments to find_global,
-        # instead, find_class can be overridden
+from zodbpickle.pickle import Pickler, Unpickler as _Unpickler, dump, dumps, loads
+class Unpickler(_Unpickler):
+    # Python 3 doesn't allow assignments to find_global,
+    # instead, find_class can be overridden
 
-        find_global = None
+    find_global = None
 
-        def find_class(self, modulename, name):
-            if self.find_global is None:
-                return super(Unpickler, self).find_class(modulename, name)
-            return self.find_global(modulename, name)
-else:
-    try:
-        import zodbpickle.fastpickle as cPickle
-    except ImportError:
-        import zodbpickle.pickle as cPickle
-    Pickler = cPickle.Pickler
-    Unpickler = cPickle.Unpickler
-    dump = cPickle.dump
-    dumps = cPickle.dumps
-    loads = cPickle.loads
+    def find_class(self, modulename, name):
+        if self.find_global is None:
+            return super(Unpickler, self).find_class(modulename, name)
+        return self.find_global(modulename, name)
 
 # String and Bytes IO
 from ZODB._compat import BytesIO
 
-if PY3:
-
-    import _thread as thread
-    if PY32:
-        from threading import _get_ident as get_ident
-    else:
-        from threading import get_ident
-
-
-else:
-
-    import thread
-    from thread import get_ident
+import _thread as thread
+from threading import get_ident
 
 try:
     from cStringIO import StringIO