You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2021/04/09 13:46:39 UTC
[tinkerpop] branch master updated (c123526 -> ee45fd5)
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git.
from c123526 Merge branch '3.4-dev'
add 972562f TINKERPOP-2457 Added max_content_length as Python driver config
add a83ff01 Merge branch 'TINKERPOP-2457' into 3.4-dev
new ee45fd5 Merge branch '3.4-dev'
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGELOG.asciidoc | 1 +
.../src/main/python/gremlin_python/driver/client.py | 7 +++++--
.../python/gremlin_python/driver/driver_remote_connection.py | 5 +++--
.../main/python/gremlin_python/driver/tornado/transport.py | 8 ++++++--
gremlin-python/src/main/python/tests/driver/test_client.py | 11 +++++++++++
5 files changed, 26 insertions(+), 6 deletions(-)
[tinkerpop] 01/01: Merge branch '3.4-dev'
Posted by sp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit ee45fd562b624ed490aa2d77329c1325a5bc1dc6
Merge: c123526 a83ff01
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Fri Apr 9 09:46:26 2021 -0400
Merge branch '3.4-dev'
CHANGELOG.asciidoc | 1 +
.../src/main/python/gremlin_python/driver/client.py | 7 +++++--
.../python/gremlin_python/driver/driver_remote_connection.py | 5 +++--
.../main/python/gremlin_python/driver/tornado/transport.py | 8 ++++++--
gremlin-python/src/main/python/tests/driver/test_client.py | 11 +++++++++++
5 files changed, 26 insertions(+), 6 deletions(-)
diff --cc gremlin-python/src/main/python/gremlin_python/driver/client.py
index 4e2d026,0000000..d5952c7
mode 100644,000000..100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/client.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/client.py
@@@ -1,149 -1,0 +1,152 @@@
+#
+# 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.
+#
+from concurrent.futures import ThreadPoolExecutor
+
+from six.moves import queue
+
+from gremlin_python.driver import connection, protocol, request, serializer
+from gremlin_python.process import traversal
+
+# This is until concurrent.futures backport 3.1.0 release
+try:
+ from multiprocessing import cpu_count
+except ImportError:
+ # some platforms don't have multiprocessing
+ def cpu_count():
+ return None
+
+__author__ = 'David M. Brown (davebshow@gmail.com)'
+
+
+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=""):
++ kerberized_service="", headers=None, session="", max_content_length=None):
+ self._url = url
+ self._headers = headers
+ self._traversal_source = traversal_source
++ if max_content_length is None:
++ max_content_length = 10 * 1024 * 1024
+ if message_serializer is None:
+ message_serializer = serializer.GraphSONSerializersV3d0()
+ self._message_serializer = message_serializer
+ self._username = username
+ self._password = password
+ self._session = session
+ self._sessionEnabled = (session != "")
+ if transport_factory is None:
+ try:
+ from gremlin_python.driver.tornado.transport import (
+ TornadoTransport)
+ except ImportError:
+ raise Exception("Please install Tornado or pass"
+ "custom transport factory")
+ else:
- transport_factory = lambda: TornadoTransport()
++ transport_factory = lambda: TornadoTransport(
++ max_content_length=max_content_length)
+ self._transport_factory = transport_factory
+ if protocol_factory is None:
+ protocol_factory = lambda: protocol.GremlinServerWSProtocol(
+ self._message_serializer,
+ username=self._username,
+ password=self._password,
+ kerberized_service=kerberized_service)
+ self._protocol_factory = protocol_factory
+ if self._sessionEnabled:
+ if pool_size is None:
+ pool_size = 1
+ elif pool_size != 1:
+ raise Exception("PoolSize must be 1 on session mode!")
+ if pool_size is None:
+ pool_size = 4
+ self._pool_size = pool_size
+ # This is until concurrent.futures backport 3.1.0 release
+ if max_workers is None:
+ # If your application is overlapping Gremlin I/O on multiple threads
+ # consider passing kwarg max_workers = (cpu_count() or 1) * 5
+ max_workers = pool_size
+ self._executor = ThreadPoolExecutor(max_workers=max_workers)
+ # Threadsafe queue
+ self._pool = queue.Queue()
+ self._fill_pool()
+
+ @property
+ def available_pool_size(self):
+ return self._pool.qsize()
+
+ @property
+ def executor(self):
+ return self._executor
+
+ @property
+ def traversal_source(self):
+ return self._traversal_source
+
+ def _fill_pool(self):
+ for i in range(self._pool_size):
+ conn = self._get_connection()
+ self._pool.put_nowait(conn)
+
+ def close(self):
+ if self._sessionEnabled:
+ self._close_session()
+ while not self._pool.empty():
+ conn = self._pool.get(True)
+ conn.close()
+ self._executor.shutdown()
+
+ def _close_session(self):
+ message = request.RequestMessage(
+ processor='session', op='close',
+ args={'session': self._session})
+ conn = self._pool.get(True)
+ return conn.write(message).result()
+
+ def _get_connection(self):
+ protocol = self._protocol_factory()
+ return connection.Connection(
+ self._url, self._traversal_source, protocol,
+ self._transport_factory, self._executor, self._pool,
+ headers=self._headers)
+
+ def submit(self, message, bindings=None, request_options=None):
+ return self.submitAsync(message, bindings=bindings, request_options=request_options).result()
+
+ def submitAsync(self, message, bindings=None, request_options=None):
+ if isinstance(message, traversal.Bytecode):
+ message = request.RequestMessage(
+ processor='traversal', op='bytecode',
+ args={'gremlin': message,
+ 'aliases': {'g': self._traversal_source}})
+ elif isinstance(message, str):
+ message = request.RequestMessage(
+ processor='', op='eval',
+ args={'gremlin': message,
+ 'aliases': {'g': self._traversal_source}})
+ if bindings:
+ message.args.update({'bindings': bindings})
+ if self._sessionEnabled:
+ message = message._replace(processor='session')
+ message.args.update({'session': self._session})
+ conn = self._pool.get(True)
+ if request_options:
+ message.args.update(request_options)
+ return conn.write(message)
diff --cc gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
index 516ba4e,0000000..8f002c9
mode 100644,000000..100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
@@@ -1,85 -1,0 +1,86 @@@
+#
+# 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.
+#
+from concurrent.futures import Future
+
+from gremlin_python.driver import client, serializer
+from gremlin_python.driver.remote_connection import (
+ RemoteConnection, RemoteTraversal)
+from gremlin_python.process.strategies import OptionsStrategy
+
+__author__ = 'David M. Brown (davebshow@gmail.com)'
+
+
+class DriverRemoteConnection(RemoteConnection):
+
+ 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):
++ graphson_writer=None, headers=None, max_content_length=None):
+ if message_serializer is None:
+ message_serializer = serializer.GraphSONMessageSerializer(
+ reader=graphson_reader,
+ writer=graphson_writer)
+ self._client = client.Client(url, traversal_source,
+ protocol_factory=protocol_factory,
+ transport_factory=transport_factory,
+ pool_size=pool_size,
+ max_workers=max_workers,
+ message_serializer=message_serializer,
+ username=username,
+ password=password,
+ kerberized_service=kerberized_service,
- headers=headers)
++ headers=headers,
++ max_content_length=max_content_length)
+ self._url = self._client._url
+ self._traversal_source = self._client._traversal_source
+
+ def close(self):
+ self._client.close()
+
+ def submit(self, bytecode):
+ result_set = self._client.submit(bytecode, request_options=self._extract_request_options(bytecode))
+ results = result_set.all().result()
+ return RemoteTraversal(iter(results))
+
+ def submitAsync(self, bytecode):
+ future = Future()
+ future_result_set = self._client.submitAsync(bytecode, request_options=self._extract_request_options(bytecode))
+
+ def cb(f):
+ try:
+ result_set = f.result()
+ results = result_set.all().result()
+ future.set_result(RemoteTraversal(iter(results)))
+ except Exception as e:
+ future.set_exception(e)
+
+ future_result_set.add_done_callback(cb)
+ return future
+
+ @staticmethod
+ def _extract_request_options(bytecode):
+ options_strategy = next((x for x in bytecode.source_instructions
+ if x[0] == "withStrategies" and type(x[1]) is OptionsStrategy), None)
+ request_options = None
+ if options_strategy:
+ allowed_keys = ['evaluationTimeout', 'scriptEvaluationTimeout', 'batchSize', 'requestId', 'userAgent']
+ request_options = {allowed: options_strategy[1].configuration[allowed] for allowed in allowed_keys
+ if allowed in options_strategy[1].configuration}
+ return request_options
diff --cc gremlin-python/src/main/python/tests/driver/test_client.py
index 54784c4,0000000..56e2d8c
mode 100644,000000..100644
--- a/gremlin-python/src/main/python/tests/driver/test_client.py
+++ b/gremlin-python/src/main/python/tests/driver/test_client.py
@@@ -1,339 -1,0 +1,350 @@@
+#
+# 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 threading
+import uuid
+
+from gremlin_python.driver.client import Client
+from gremlin_python.driver.protocol import GremlinServerError
+from gremlin_python.driver.request import RequestMessage
+from gremlin_python.process.graph_traversal import __
+from gremlin_python.process.strategies import OptionsStrategy
+from gremlin_python.structure.graph import Graph
+from gremlin_python.driver.tornado.transport import TornadoTransport
+from tornado.util import TimeoutError
+
+__author__ = 'David M. Brown (davebshow@gmail.com)'
+
+
+def test_connection(connection):
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ results_set = connection.write(message).result()
+ future = results_set.all()
+ results = future.result()
+ assert len(results) == 6
+ assert isinstance(results, list)
+ assert results_set.done.done()
+ assert 'host' in results_set.status_attributes
+
+
++def test_client_message_too_big(client):
++ try:
++ client = Client("http://localhost", 'g', max_content_length=1024)
++ client.submit("1+1").all().result()
++ assert False
++ except Exception:
++ assert True
++ finally:
++ client.close()
++
++
+def test_client_simple_eval(client):
+ assert client.submit('1 + 1').all().result()[0] == 2
+
+
+def test_client_simple_eval_bindings(client):
+ assert client.submit('x + x', {'x': 2}).all().result()[0] == 4
+
+
+def test_client_eval_traversal(client):
+ assert len(client.submit('g.V()').all().result()) == 6
+
+
+def test_client_error(client):
+ try:
+ # should fire an exception
+ client.submit('1/0').all().result()
+ assert False
+ except GremlinServerError as ex:
+ assert 'exceptions' in ex.status_attributes
+ assert 'stackTrace' in ex.status_attributes
+
+
+def test_client_connection_pool_after_error(client):
+ # Overwrite fixture with pool_size=1 client
+ client = Client('ws://localhost:45940/gremlin', 'gmodern', pool_size=1)
+
+ try:
+ # should fire an exception
+ client.submit('1/0').all().result()
+ assert False
+ except GremlinServerError as gse:
+ # expecting the pool size to be 1 again after query returned
+ assert gse.status_code == 597
+ assert client.available_pool_size == 1
+
+
+def test_client_side_timeout_set_for_tornado(client):
+ client = Client('ws://localhost:45940/gremlin', 'gmodern',
+ transport_factory=lambda: TornadoTransport(read_timeout=1, write_timeout=1))
+
+ try:
+ # should fire an exception
+ client.submit('Thread.sleep(2000);1').all().result()
+ assert False
+ except TimeoutError as toerr:
+ assert str(toerr) == "Operation timed out after 1 seconds"
+
+
+def test_client_bytecode(client):
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ result_set = client.submit(message)
+ assert len(result_set.all().result()) == 6
+
+
+def test_client_bytecode_options(client):
+ # smoke test to validate serialization of OptionsStrategy. no way to really validate this from an integration
+ # test perspective because there's no way to access the internals of the strategy via bytecode
+ g = Graph().traversal()
+ t = g.withStrategies(OptionsStrategy(options={"x": "test", "y": True})).V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ result_set = client.submit(message)
+ assert len(result_set.all().result()) == 6
+ ##
+ t = g.with_("x", "test").with_("y", True).V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ result_set = client.submit(message)
+ assert len(result_set.all().result()) == 6
+
+
+def test_iterate_result_set(client):
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ result_set = client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 6
+
+
+def test_client_async(client):
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ future = client.submitAsync(message)
+ result_set = future.result()
+ assert len(result_set.all().result()) == 6
+
+
+def test_connection_share(client):
+ # Overwrite fixture with pool_size=1 client
+ client = Client('ws://localhost:45940/gremlin', 'gmodern', pool_size=1)
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ message2 = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ future = client.submitAsync(message)
+ future2 = client.submitAsync(message2)
+
+ result_set2 = future2.result()
+ assert len(result_set2.all().result()) == 6
+
+ # This future has to finish for the second to yield result - pool_size=1
+ assert future.done()
+ result_set = future.result()
+ assert len(result_set.all().result()) == 6
+
+
+def test_multi_conn_pool(client):
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ message2 = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ client = Client('ws://localhost:45940/gremlin', 'g', pool_size=1)
+ future = client.submitAsync(message)
+ future2 = client.submitAsync(message2)
+
+ result_set2 = future2.result()
+ assert len(result_set2.all().result()) == 6
+
+ # with connection pool `future` may or may not be done here
+ result_set = future.result()
+ assert len(result_set.all().result()) == 6
+
+
+def test_multi_thread_pool(client):
+ g = Graph().traversal()
+ traversals = [g.V(),
+ g.V().count(),
+ g.E(),
+ g.E().count()
+ ]
+ results = [[] for _ in traversals]
+
+ # Use a condition variable to synchronise a group of threads, which should also inject some
+ # non-determinism into the run-time execution order
+ condition = threading.Condition()
+
+ def thread_run(tr, result_list):
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': tr.bytecode, 'aliases': {'g': 'gmodern'}})
+ with condition:
+ condition.wait(5)
+ result_set = client.submit(message)
+ for result in result_set:
+ result_list.append(result)
+
+ threads = []
+ for i in range(len(results)):
+ thread = threading.Thread(target=thread_run,
+ args=(traversals[i], results[i]),
+ name="test_multi_thread_pool_%d" % i)
+ thread.daemon = True
+ threads.append(thread)
+ thread.start()
+ with condition:
+ condition.notify_all()
+
+ for t in threads:
+ t.join(5)
+
+ assert len(results[0][0]) == 6
+ assert results[1][0][0].object == 6
+ assert len(results[2][0]) == 6
+ assert results[3][0][0].object == 6
+
+
+def test_client_bytecode_with_int(client):
+ g = Graph().traversal()
+ t = g.V().has('age', 851401972585122).count()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
+ result_set = client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 1
+
+
+def test_multi_request_in_session(client):
+ # Overwrite fixture with session client
+ session_id = str(uuid.uuid4())
+ client = Client('ws://localhost:45940/gremlin', 'g', session=session_id)
+
+ assert client.submit('x = 1').all().result()[0] == 1
+ assert client.submit('x + 2').all().result()[0] == 3
+
+ client.close()
+
+ # attempt reconnect to session and make sure "x" is no longer a thing
+ client = Client('ws://localhost:45940/gremlin', 'g', session=session_id)
+ try:
+ # should fire an exception
+ client.submit('x').all().result()
+ assert False
+ except Exception:
+ assert True
+
+
+def test_client_pool_in_session(client):
+ # Overwrite fixture with pool_size=2 client
+ try:
+ # should fire an exception
+ client = Client('ws://localhost:45940/gremlin', 'g', session=str(uuid.uuid4()), pool_size=2)
+ assert False
+ except Exception:
+ assert True
+
+
+def test_big_result_set(client):
+ g = Graph().traversal()
+ t = g.inject(1).repeat(__.addV('person').property('name', __.loops())).times(20000).count()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 1
+
+ t = g.V().limit(10)
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 10
+
+ t = g.V().limit(100)
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 100
+
+ t = g.V().limit(1000)
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 1000
+
+ t = g.V().limit(10000)
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 10000
+
+
+def test_big_result_set_secure(authenticated_client):
+ g = Graph().traversal()
+ t = g.inject(1).repeat(__.addV('person').property('name', __.loops())).times(20000).count()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = authenticated_client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 1
+
+ t = g.V().limit(10)
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = authenticated_client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 10
+
+ t = g.V().limit(100)
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = authenticated_client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 100
+
+ t = g.V().limit(1000)
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = authenticated_client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 1000
+
+ t = g.V().limit(10000)
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'g'}})
+ result_set = authenticated_client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 10000