TINKERPOP-2546 Gremlin-Python Client Transport Layer to use AIOHTTP

CHANGELOG.asciidoc

@@ -25,6 +25,13 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
 
 This release also includes changes from <<release-3-4-3, 3.4.3>>.
 
+* Changed transport implementation to use AIOHTTP instead of Tornado for gremlin-python.
+* Added max_content_length and unit test for it in gremlin-python.
+* Removed compression_option support for transport in gremlin-python.
+* Fixed event loop issues and added unit test for it in gremlin-python.
+* Fixed DriverRemoteConnection multithreading issues and added unit test for it in gremlin-python.
+* Fixed heartbeat timeout issues and tested with local server manually for gremlin-python.
+* Fixed build errors emitted for gremlin-python (asyncio task destroyed but is pending error).
 * Added `gremlin-language` module.
 * Allowed the possibility for the propagation of `null` as a `Traverser` in Gremlin.
 * Added a fully shaded version of `gremlin-driver`.

docs/src/reference/gremlin-variants.asciidoc

@@ -844,24 +844,27 @@ can be passed to the `Client` or `DriverRemoteConnection` instance as keyword ar
 |password |The password to submit on requests that require authentication. |""
 |pool_size |The number of connections used by the pool. |4
 |protocol_factory |A callable that returns an instance of `AbstractBaseProtocol`. |`gremlin_python.driver.protocol.GremlinServerWSProtocol`
-|transport_factory |A callable that returns an instance of `AbstractBaseTransport`. |`gremlin_python.driver.tornado.transport.TornadoTransport`
+|transport_factory |A callable that returns an instance of `AbstractBaseTransport`. |`gremlin_python.driver.aiohttp.transport.AiohttpTransport`
 |username |The username to submit on requests that require authentication. |""
 |kerberized_service |the first part of the principal name configured for the gremlin service|"""
 |session | A unique string-based identifier (typically a UUID) to enable a <<sessions,session-based connection>>. This is not a valid configuration for `DriverRemoteConnection`. |None
 |=========================================================
 
-Note that the `transport_factory` can allow for additional configuration of the `TornadoTransport`, which exposes
-options to manage `ioloop` timeouts and compression settings:
+Note that the `transport_factory` can allow for additional configuration of the `AiohttpTransport`, which exposes
+options to manage `asyncio` timeouts, heartbeat, max content length, ssl options, and the ability to call the api from
+an event loop:
 
 [source,python]
 ----
 import ssl
 ...
 g = traversal().withRemote(
   DriverRemoteConnection('ws://localhost:8182/gremlin','g',
-                         transport_factory=lambda: TornadoTransport(read_timeout=10,
+                         transport_factory=lambda: AiohttpTransport(read_timeout=10,
                                                                     write_timeout=10,
-                                                                    compression_options={'compression_level':5,'mem_level':5},
+                                                                    heartbeat=1.0,
+                                                                    call_from_event_loop=True
+                                                                    max_content_length=100*1024*1024,
                                                                     ssl_options=ssl.create_default_context(Purpose.CLIENT_AUTH))))
 ----
 
@@ -1146,7 +1149,7 @@ and is designed to work best with a running <<gremlin-server,Gremlin Server>> co
 [source,shell]
 ----
 pip install gremlinpython
-pip install tornado
+pip install aiohttp
 python example.py
 ----
 

docs/src/upgrade/release-3.5.x.asciidoc

@@ -494,14 +494,26 @@ See: link:https://issues.apache.org/jira/browse/TINKERPOP-2517[TINKERPOP-2517]
 
 ==== Python 2.x Support
 
-The gremlinpython module no longer supports Python 2.x. Users must use Python 3 going forward. For the most part, from
+The gremlin-python module no longer supports Python 2.x. Users must use Python 3 going forward. For the most part, from
 a user's perspective, there are no specific API changes to consider as a result of this change. It is also worth
 noting that Jython support has been removed and that `gremlin-python` no longer produces a JVM-based artifact. This
 change means that the `GremlinJythonScriptEngine` no longer exists and there is no way to write native Python lambdas.
 All lambdas should be written using `gremlin-groovy` if they are needed.
 
 See: link:https://issues.apache.org/jira/browse/TINKERPOP-2317[TINKERPOP-2317]
 
+==== Python Transportation Layer Rewrite
+
+With the removal of Python 2.x support the transport layer of gremlin-python has been rewritten to use a library that
+utilizes the asyncio event loop of Python 3. link:https://github.com/aio-libs/aiohttp[AIOHTTP] utilizes Python 3's
+event loop with a minimal HTTP abstraction and is now used for the transport layer. From a user's perspective there is
+not much of a change except there is now a working heartbeat that can be set and there is no longer a
+compression_options config when connecting with a custom transport factory. This change fixed a number of issues that
+were related to the IOLoop of the old link:https://github.com/tornadoweb/tornado[Tornado] transport layer, which has
+been completely removed from the library.
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-2546[TINKERPOP-2546]
+
 ==== Python Kerberos Support
 
 The Python Driver now supports Kerberos based authentication:

gremlin-python/src/main/python/gremlin_python/driver/tornado/__init__.py → gremlin-python/src/main/python/gremlin_python/driver/aiohttp/__init__.py

@@ -6,9 +6,9 @@
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
-# 
+#
 # http://www.apache.org/licenses/LICENSE-2.0
-# 
+#
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py

@@ -0,0 +1,129 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+
+import aiohttp
+import asyncio
+import async_timeout
+
+from gremlin_python.driver.transport import AbstractBaseTransport
+
+__author__ = 'Lyndon Bauto ([email protected])'
+
+
+class AiohttpTransport(AbstractBaseTransport):
+    nest_asyncio_applied = False
+
+    # Default heartbeat of 5.0 seconds.
+    def __init__(self, read_timeout=None, write_timeout=None, ssl_options=None,
+                 heartbeat=5.0, max_content_length=None, call_from_event_loop=None):
+        if call_from_event_loop is not None and call_from_event_loop and not AiohttpTransport.nest_asyncio_applied:
+            """ 
+                The AiohttpTransport implementation uses the asyncio event loop. Because of this, it cannot be called within an
+                event loop without nest_asyncio. If the code is ever refactored so that it can be called within an event loop
+                this import and call can be removed. Without this, applications which use the event loop to call gremlin-python
+                (such as Jupyter) will not work.
+            """
+            import nest_asyncio
+            nest_asyncio.apply()
+            AiohttpTransport.nest_asyncio_applied = True
+
+        # Start event loop and initialize websocket and client to None
+        self._loop = asyncio.new_event_loop()
+        self._websocket = None
+        self._client_session = None
+
+        # Set all inner variables to parameters passed in.
+        self._read_timeout = read_timeout
+        self._write_timeout = write_timeout
+        self._ssl_options = ssl_options
+        self._heartbeat = heartbeat
+        self._max_content_length = max_content_length
+
+    def connect(self, url, headers=None):
+        # Inner function to perform async connect.
+        async def async_connect():
+            # Start client session and use it to create a websocket with all the connection options provided.
+            self._client_session = aiohttp.ClientSession(loop=self._loop)
+            self._websocket = await self._client_session.ws_connect(url,
+                                                                    ssl=self._ssl_options,
+                                                                    headers=headers,
+                                                                    heartbeat=self._heartbeat,
+                                                                    max_msg_size=self._max_content_length)
+
+        # Execute the async connect synchronously.
+        self._loop.run_until_complete(async_connect())
+
+    def write(self, message):
+        # Inner function to perform async write.
+        async def async_write():
+            async with async_timeout.timeout(self._write_timeout):
+                await self._websocket.send_bytes(message)
+
+        # Execute the async write synchronously.
+        self._loop.run_until_complete(async_write())
+
+    def read(self):
+        # Inner function to perform async read.
+        async def async_read():
+            async with async_timeout.timeout(self._read_timeout):
+                return await self._websocket.receive()
+
+        # Execute the async read synchronously.
+        msg = self._loop.run_until_complete(async_read())
+
+        # Need to handle multiple potential message types.
+        if msg.type == aiohttp.WSMsgType.close:
+            # Server is closing connection, shutdown and throw exception.
+            self.close()
+            raise RuntimeError("Connection was closed by server.")
+        elif msg.type == aiohttp.WSMsgType.closed:
+            # Should not be possible since our loop and socket would be closed.
+            raise RuntimeError("Connection was already closed.")
+        elif msg.type == aiohttp.WSMsgType.error:
+            # Error on connection, try to convert message to a string in error.
+            raise RuntimeError("Received error on read: '" + str(msg.data) + "'")
+        elif msg.type == aiohttp.WSMsgType.text:
+            # Convert message to bytes.
+            data = msg.data.strip().encode('utf-8')
+        else:
+            # General handle, return byte data.
+            data = msg.data
+        return data
+
+    def close(self):
+        # Inner function to perform async close.
+        async def async_close():
+            if not self._websocket.closed:
+                await self._websocket.close()
+            if not self._client_session.closed:
+                await self._client_session.close()
+
+        # If the loop is not closed (connection hasn't already been closed)
+        if not self._loop.is_closed():
+            # Execute the async close synchronously.
+            self._loop.run_until_complete(async_close())
+
+            # Close the event loop.
+            self._loop.close()
+
+    @property
+    def closed(self):
+        # Connection is closed if either the websocket or the client session is closed.
+        return self._websocket.closed or self._client_session.closed

gremlin-python/src/main/python/gremlin_python/driver/client.py

@@ -39,7 +39,9 @@ class Client:
     def __init__(self, url, traversal_source, protocol_factory=None,
                  transport_factory=None, pool_size=None, max_workers=None,
                  message_serializer=None, username="", password="",
-                 kerberized_service="", headers=None, session="", max_content_length=None):
+                 kerberized_service="", headers=None, session="",
+                 max_content_length=None, heartbeat=None,
+                 call_from_event_loop=None):
         self._url = url
         self._headers = headers
         self._traversal_source = traversal_source
@@ -54,17 +56,19 @@ def __init__(self, url, traversal_source, protocol_factory=None,
         self._sessionEnabled = (session != "")
         if transport_factory is None:
             try:
-                from gremlin_python.driver.tornado.transport import (
-                    TornadoTransport)
+                from gremlin_python.driver.aiohttp.transport import (
+                    AiohttpTransport)
             except ImportError:
-                raise Exception("Please install Tornado or pass"
+                raise Exception("Please install AIOHTTP or pass "
                                 "custom transport factory")
             else:
-                transport_factory = lambda: TornadoTransport(
-                    max_content_length=max_content_length)
+                def transport_factory():
+                    return AiohttpTransport(heartbeat=heartbeat,
+                                            max_content_length=max_content_length,
+                                            call_from_event_loop=call_from_event_loop)
         self._transport_factory = transport_factory
         if protocol_factory is None:
-            protocol_factory = lambda: protocol.GremlinServerWSProtocol(
+            def protocol_factory(): return protocol.GremlinServerWSProtocol(
                 self._message_serializer,
                 username=self._username,
                 password=self._password,

gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py

@@ -32,7 +32,9 @@ def __init__(self, url, traversal_source, protocol_factory=None,
                  transport_factory=None, pool_size=None, max_workers=None,
                  username="", password="", kerberized_service='',
                  message_serializer=None, graphson_reader=None,
-                 graphson_writer=None, headers=None, max_content_length=None):
+                 graphson_writer=None, headers=None,
+                 max_content_length=None, heartbeat=None,
+                 call_from_event_loop=None):
         if message_serializer is None:
             message_serializer = serializer.GraphSONMessageSerializer(
                 reader=graphson_reader,
@@ -47,7 +49,9 @@ def __init__(self, url, traversal_source, protocol_factory=None,
                                      password=password,
                                      kerberized_service=kerberized_service,
                                      headers=headers,
-                                     max_content_length=max_content_length)
+                                     heartbeat=heartbeat,
+                                     max_content_length=max_content_length,
+                                     call_from_event_loop=call_from_event_loop)
         self._url = self._client._url
         self._traversal_source = self._client._traversal_source
 

gremlin-python/src/main/python/gremlin_python/driver/transport.py

@@ -41,6 +41,7 @@ def read(self):
     def close(self):
         pass
 
-    @abc.abstractproperty
+    @property
+    @abc.abstractmethod
     def closed(self):
         pass

gremlin-python/src/main/python/setup.py

@@ -45,8 +45,9 @@
 version = __version__.version
 
 install_requires = [
+    'nest_asyncio',
+    'aiohttp>=3.7.0,<=3.7.4',
     'aenum>=1.4.5,<3.0.0',
-    'tornado>=6.0',
     'six>=1.10.0,<2.0.0',
     'isodate>=0.6.0,<1.0.0',
     'pyparsing>=2.4.7,<3.0.0'
@@ -59,7 +60,7 @@
     name='gremlinpython',
     version=version,
     packages=['gremlin_python', 'gremlin_python.driver',
-              'gremlin_python.driver.tornado', 'gremlin_python.process',
+              'gremlin_python.driver.aiohttp', 'gremlin_python.process',
               'gremlin_python.structure', 'gremlin_python.structure.io'],
     license='Apache 2',
     url='http://tinkerpop.apache.org',