You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by fl...@apache.org on 2018/12/11 15:37:48 UTC
[tinkerpop] branch TINKERPOP-1775 updated (a47464c -> ccf74d6)
This is an automated email from the ASF dual-hosted git repository.
florianhockmann pushed a change to branch TINKERPOP-1775
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git.
discard a47464c Add request pipelining TINKERPOP-1775/1774
add c923564 TINKERPOP-2106 throw TimeoutException instead of TraversalInterruptedException
add 75af342 Merge branch 'tp33'
add f7e107f Update changelog CTR
add f15b25e Merge branch 'tp33'
add a409850 uses the directly field instead of getter method (#1001)
add 06498ea TINKERPOP-2075 Added ReferenceElementStrategy
add 2cb95c9 Merge pull request #1010 from apache/TINKERPOP-2075
add 65ff08c Deprecate StarGraph builder methods that don't follow our standard pattern CTR
add 22d3aa3 Merge branch 'tp33'
add 09b093f Fixed SubgraphStrategy configuration and derser using that config CTR
add fd45a72 Increased the maxContentLength for tests of remoting CTR
add 7968e29 Merge branch 'tp33'
new ccf74d6 Add request pipelining TINKERPOP-1775/1774
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (a47464c)
\
N -- N -- N refs/heads/TINKERPOP-1775 (ccf74d6)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
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 | 4 ++
docs/src/reference/gremlin-applications.asciidoc | 9 ++-
docs/src/upgrade/release-3.4.x.asciidoc | 38 +++++++++++-
.../jsr223/AbstractGremlinScriptEngineFactory.java | 4 +-
.../tinkerpop/gremlin/jsr223/CoreImports.java | 2 +
.../jsr223/DefaultGremlinScriptEngineManager.java | 2 +-
.../gremlin/process/computer/GraphFilter.java | 4 +-
.../bulkloading/IncrementalBulkLoader.java | 4 +-
.../traversal/MemoryTraversalSideEffects.java | 2 +-
.../process/computer/util/ComputerGraph.java | 4 +-
.../tinkerpop/gremlin/process/traversal/P.java | 4 +-
.../traversal/dsl/graph/GraphTraversalSource.java | 4 +-
.../process/traversal/step/branch/RepeatStep.java | 2 +-
.../process/traversal/step/map/GraphStep.java | 2 +-
.../traversal/step/util/ReducingBarrierStep.java | 2 +-
.../strategy/decoration/SubgraphStrategy.java | 8 ++-
.../finalization/ReferenceElementStrategy.java | 70 ++++++++++++++++++++++
.../process/traversal/util/DefaultTraversal.java | 8 +--
.../structure/io/graphml/GraphMLWriterHelper.java | 2 +-
.../graphson/AbstractGraphSONTypeSerializer.java | 2 +-
.../structure/io/graphson/GraphSONMapper.java | 2 +-
.../io/graphson/GraphSONTypeIdResolver.java | 14 ++---
.../gremlin/structure/util/star/StarGraph.java | 21 ++++++-
.../tinkerpop/gremlin/util/function/Lambda.java | 2 +-
.../decoration/ReferenceElementStrategyTest.java | 51 ++++++++++++++++
.../tinkerpop/gremlin/driver/Channelizer.java | 1 -
.../gremlin/groovy/engine/GremlinExecutor.java | 7 ++-
gremlin-server/scripts/empty-sample-secure.groovy | 7 ++-
gremlin-server/scripts/empty-sample.groovy | 7 ++-
gremlin-server/scripts/generate-classic.groovy | 7 ++-
.../scripts/generate-modern-readonly.groovy | 7 ++-
gremlin-server/scripts/generate-modern.groovy | 7 ++-
gremlin-server/scripts/load-sample.groovy | 7 ++-
gremlin-server/scripts/spark.groovy | 8 ++-
.../gremlin/driver/remote/RemoteGraphProvider.java | 3 +-
.../gremlin/server/GremlinDriverIntegrateTest.java | 16 ++---
.../server/GremlinResultSetIntegrateTest.java | 10 ++--
.../gremlin/server/gremlin-server-integration.yaml | 2 +-
.../src/test/scripts/generate-all.groovy | 14 ++---
39 files changed, 299 insertions(+), 71 deletions(-)
create mode 100644 gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/finalization/ReferenceElementStrategy.java
create mode 100644 gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/decoration/ReferenceElementStrategyTest.java
[tinkerpop] 01/01: Add request pipelining TINKERPOP-1775/1774
Posted by fl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
florianhockmann pushed a commit to branch TINKERPOP-1775
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit ccf74d64e76c4400e711201fe2a2e48c49acb42b
Author: Florian Hockmann <fh...@florian-hockmann.de>
AuthorDate: Tue Dec 11 16:36:02 2018 +0100
Add request pipelining TINKERPOP-1775/1774
This allows connections to be reused for different requests in parallel
which should lead to better utilization of connections / lower resource
usage. The ConnectionPool now also has a min and max size.
A new connection is only created if all existing connections reached
their max in-flight limit (which is also a new setting). If additionally
the max size of the ConnectionPool is reached, then an exception is
thrown which makes this a breaking change.
---
CHANGELOG.asciidoc | 1 +
docs/src/reference/gremlin-variants.asciidoc | 16 ++
docs/src/upgrade/release-3.4.x.asciidoc | 12 ++
.../src/Gremlin.Net/Driver/Connection.cs | 166 +++++++++++++--------
.../src/Gremlin.Net/Driver/ConnectionPool.cs | 125 +++++++++++-----
.../Gremlin.Net/Driver/ConnectionPoolSettings.cs | 58 +++++++
.../Exceptions/NoConnectionAvailableException.cs | 56 +++++++
.../src/Gremlin.Net/Driver/GremlinClient.cs | 5 +-
.../Gremlin.Net/Driver/Messages/ResponseStatus.cs | 10 --
.../Driver/SingleMessageResultReceiver.cs | 88 +++++++++++
.../Driver/ConnectionPoolTests.cs | 113 ++++++++++++--
.../Driver/GremlinClientTests.cs | 3 -
.../DriverRemoteConnection/GraphTraversalTests.cs | 3 +-
.../RemoteConnectionFactory.cs | 4 +-
pom.xml | 1 +
15 files changed, 522 insertions(+), 139 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index ef01dab..24a78b0 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -25,6 +25,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
This release also includes changes from <<release-3-3-3, 3.3.3>>.
+* Improved the Gremlin.NET driver: It now uses request pipelining and its `ConnectionPool` has min and max sizes.
* Made `valueMap()` aware of `by` and `with` modulators and deprecated `valueMap(boolean)` overloads.
* Use `Compare.eq` in `Contains` predicates to ensure the same filter behavior for numeric values.
* Added `OptionsStrategy` to allow traversals to take arbitrary traversal-wide configurations.
diff --git a/docs/src/reference/gremlin-variants.asciidoc b/docs/src/reference/gremlin-variants.asciidoc
index 0c5c107..a8c6e10 100644
--- a/docs/src/reference/gremlin-variants.asciidoc
+++ b/docs/src/reference/gremlin-variants.asciidoc
@@ -1008,6 +1008,22 @@ The connection properties for the Gremlin.Net driver can be passed to the `Greml
|password |The password to submit on requests that require authentication. |_none_
|=========================================================
+==== Connection Pool
+
+It is also possible to configure the `ConnectionPool` of the Gremlin.Net driver.
+These configuration options can be set as properties
+on the `ConnectionPoolSettings` instance that can be passed to the `GremlinClient`:
+
+[width="100%",cols="3,10,^2",options="header"]
+|=========================================================
+|Key |Description |Default
+|MinSize |The minimum size of the connection pool. This determines how many connections are initially created. |2
+|MaxSize |The maximum size of the connection pool. |32
+|MaxInProcessPerConnection |The maximum number of in-flight requests that can occur on a connection. |4
+|=========================================================
+
+A `NoConnectionAvailableException` is thrown if the `ConnectionPool` reaches the `MaxSize` limit and all connections have reached the `MaxInProcessPerConnection` limit when a new request comes in.
+
==== Serialization
The Gremlin.Net driver uses by default GraphSON 3.0 but it is also possible to use GraphSON 2.0 which can be necessary
diff --git a/docs/src/upgrade/release-3.4.x.asciidoc b/docs/src/upgrade/release-3.4.x.asciidoc
index 07d9273..ff67d3e 100644
--- a/docs/src/upgrade/release-3.4.x.asciidoc
+++ b/docs/src/upgrade/release-3.4.x.asciidoc
@@ -29,6 +29,18 @@ Please see the link:https://github.com/apache/tinkerpop/blob/3.4.0/CHANGELOG.asc
=== Upgrading for Users
+==== Gremlin.NET Driver Improvements
+
+The Gremlin.NET driver now uses request pipelining. This allows connections to be reused for different requests in parallel which should lead to better utilization of connections.
+The `ConnectionPool` now also has a min and max size. A new connection is only created if all existing connections reached their `MaxInProcessPerConnection` limit. If additionally
+the max size of the `ConnectionPool` is reached, then a `NoConnectionAvailableException` is thrown which makes this a breaking change.
+
+These settings can be set as properties on the `ConnectionPoolSettings` instance that can be passed to the `GremlinClient`.
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-1774[TINKERPOP-1774],
+link:https://issues.apache.org/jira/browse/TINKERPOP-1775[TINKERPOP-1775],
+link:http://tinkerpop.apache.org/docs/3.4.0/reference/#_connection_pool[Reference Documentation]
+
==== Modulation of valueMap()
The `valueMap()` step now supports `by` and `with` modulation, which also led to the deprecation of `valueMap(true)` overloads.
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
index d8e6103..3a1d298 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
@@ -22,17 +22,26 @@
#endregion
using System;
+using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net.WebSockets;
using System.Text;
+using System.Threading;
using System.Threading.Tasks;
+using Gremlin.Net.Driver.Exceptions;
using Gremlin.Net.Driver.Messages;
-using Gremlin.Net.Driver.ResultsAggregation;
using Gremlin.Net.Structure.IO.GraphSON;
using Newtonsoft.Json.Linq;
namespace Gremlin.Net.Driver
{
+ internal interface IResponseHandlerForSingleRequestMessage
+ {
+ void HandleReceived(ResponseMessage<JToken> received);
+ void Finalize(Dictionary<string, object> statusAttributes);
+ void HandleFailure(Exception objException);
+ }
+
internal class Connection : IConnection
{
private readonly GraphSONReader _graphSONReader;
@@ -42,6 +51,13 @@ namespace Gremlin.Net.Driver
private readonly WebSocketConnection _webSocketConnection;
private readonly string _username;
private readonly string _password;
+ private readonly ConcurrentQueue<RequestMessage> _writeQueue = new ConcurrentQueue<RequestMessage>();
+
+ private readonly ConcurrentDictionary<Guid, IResponseHandlerForSingleRequestMessage> _callbackByRequestId =
+ new ConcurrentDictionary<Guid, IResponseHandlerForSingleRequestMessage>();
+ private int _connectionState = 0;
+ private int _writeInProgress = 0;
+ private const int Closed = 1;
public Connection(Uri uri, string username, string password, GraphSONReader graphSONReader,
GraphSONWriter graphSONWriter, string mimeType, Action<ClientWebSocketOptions> webSocketConfiguration)
@@ -55,96 +71,82 @@ namespace Gremlin.Net.Driver
_webSocketConnection = new WebSocketConnection(webSocketConfiguration);
}
- public async Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage)
- {
- await SendAsync(requestMessage).ConfigureAwait(false);
- return await ReceiveAsync<T>().ConfigureAwait(false);
- }
-
public async Task ConnectAsync()
{
await _webSocketConnection.ConnectAsync(_uri).ConfigureAwait(false);
+ ReceiveNext();
}
- public async Task CloseAsync()
- {
- await _webSocketConnection.CloseAsync().ConfigureAwait(false);
- }
+ public int NrRequestsInFlight => _callbackByRequestId.Count;
public bool IsOpen => _webSocketConnection.IsOpen;
- private async Task SendAsync(RequestMessage message)
+ public Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage)
{
- var graphsonMsg = _graphSONWriter.WriteObject(message);
- var serializedMsg = _messageSerializer.SerializeMessage(graphsonMsg);
- await _webSocketConnection.SendMessageAsync(serializedMsg).ConfigureAwait(false);
+ var receiver = new ResponseHandlerForSingleRequestMessage<T>(_graphSONReader);
+ _callbackByRequestId.GetOrAdd(requestMessage.RequestId, receiver);
+ _writeQueue.Enqueue(requestMessage);
+ SendNextMessage();
+ return receiver.Result;
}
- private async Task<ResultSet<T>> ReceiveAsync<T>()
+ private void ReceiveNext()
{
- ResultSet<T> resultSet = null;
- Dictionary<string, object> statusAttributes = null;
- ResponseStatus status;
- IAggregator aggregator = null;
- var isAggregatingSideEffects = false;
- var result = new List<T>();
- do
- {
- var received = await _webSocketConnection.ReceiveMessageAsync().ConfigureAwait(false);
- var receivedMsg = _messageSerializer.DeserializeMessage<ResponseMessage<JToken>>(received);
-
- status = receivedMsg.Status;
- status.ThrowIfStatusIndicatesError();
-
- if (status.Code == ResponseStatusCode.Authenticate)
- {
- await AuthenticateAsync().ConfigureAwait(false);
- }
- else if (status.Code != ResponseStatusCode.NoContent)
+ var state = Interlocked.CompareExchange(ref _connectionState, 0, 0);
+ if (state == Closed) return;
+ _webSocketConnection.ReceiveMessageAsync().ContinueWith(
+ received =>
{
- var receivedData = typeof(T) == typeof(JToken)
- ? new[] { receivedMsg.Result.Data }
- : _graphSONReader.ToObject(receivedMsg.Result.Data);
+ Parse(received.Result);
+ ReceiveNext();
+ },
+ TaskContinuationOptions.ExecuteSynchronously);
+ }
- foreach (var d in receivedData)
- if (receivedMsg.Result.Meta.ContainsKey(Tokens.ArgsSideEffectKey))
- {
- if (aggregator == null)
- aggregator =
- new AggregatorFactory().GetAggregatorFor(
- (string) receivedMsg.Result.Meta[Tokens.ArgsAggregateTo]);
- aggregator.Add(d);
- isAggregatingSideEffects = true;
- }
- else
- {
- result.Add(d);
- }
- }
+ private void Parse(byte[] received)
+ {
+ var receivedMsg = _messageSerializer.DeserializeMessage<ResponseMessage<JToken>>(received);
- if (status.Code == ResponseStatusCode.Success || status.Code == ResponseStatusCode.NoContent)
- {
- statusAttributes = receivedMsg.Status.Attributes;
- }
+ var status = receivedMsg.Status;
+ if (status.Code.IndicatesError())
+ {
+ _callbackByRequestId[receivedMsg.RequestId].HandleFailure(new ResponseException(status.Code,
+ status.Attributes, $"{status.Code}: {status.Message}"));
+ _callbackByRequestId.TryRemove(receivedMsg.RequestId, out _);
+ return;
+ }
- } while (status.Code == ResponseStatusCode.PartialContent || status.Code == ResponseStatusCode.Authenticate);
+ if (status.Code == ResponseStatusCode.Authenticate)
+ {
+ Authenticate(receivedMsg.RequestId);
+ return;
+ }
+ if (status.Code != ResponseStatusCode.NoContent)
+ _callbackByRequestId[receivedMsg.RequestId].HandleReceived(receivedMsg);
- resultSet = new ResultSet<T>(isAggregatingSideEffects ? new List<T> { (T)aggregator.GetAggregatedResult() } : result, statusAttributes);
-
- return resultSet;
+ if (status.Code == ResponseStatusCode.Success || status.Code == ResponseStatusCode.NoContent)
+ {
+ _callbackByRequestId[receivedMsg.RequestId].Finalize(status.Attributes);
+ _callbackByRequestId.TryRemove(receivedMsg.RequestId, out _);
+ }
}
-
- private async Task AuthenticateAsync()
+
+ private void Authenticate(Guid requestId)
{
if (string.IsNullOrEmpty(_username) || string.IsNullOrEmpty(_password))
- throw new InvalidOperationException(
- $"The Gremlin Server requires authentication, but no credentials are specified - username: {_username}, password: {_password}.");
+ {
+ _callbackByRequestId[requestId].HandleFailure(new InvalidOperationException(
+ $"The Gremlin Server requires authentication, but no credentials are specified - username: {_username}, password: {_password}."));
+ _callbackByRequestId.TryRemove(requestId, out _);
+ return;
+ }
var message = RequestMessage.Build(Tokens.OpsAuthentication).Processor(Tokens.ProcessorTraversal)
.AddArgument(Tokens.ArgsSasl, SaslArgument()).Create();
- await SendAsync(message).ConfigureAwait(false);
+ _writeQueue.Enqueue(message);
+ SendNextMessage();
}
private string SaslArgument()
@@ -154,6 +156,38 @@ namespace Gremlin.Net.Driver
return Convert.ToBase64String(authBytes);
}
+ private void SendNextMessage()
+ {
+ if (Interlocked.CompareExchange(ref _writeInProgress, 1, 0) != 0)
+ return;
+ if (_writeQueue.TryDequeue(out var msg))
+ SendMessageAsync(msg).ContinueWith(
+ t =>
+ {
+ if (t.IsFaulted)
+ {
+ _callbackByRequestId[msg.RequestId].HandleFailure(t.Exception);
+ _callbackByRequestId.TryRemove(msg.RequestId, out _);
+ }
+
+ SendNextMessage();
+ }, TaskContinuationOptions.ExecuteSynchronously);
+ Interlocked.CompareExchange(ref _writeInProgress, 0, 1);
+ }
+
+ private async Task SendMessageAsync(RequestMessage message)
+ {
+ var graphsonMsg = _graphSONWriter.WriteObject(message);
+ var serializedMsg = _messageSerializer.SerializeMessage(graphsonMsg);
+ await _webSocketConnection.SendMessageAsync(serializedMsg).ConfigureAwait(false);
+ }
+
+ public async Task CloseAsync()
+ {
+ Interlocked.Exchange(ref _connectionState, Closed);
+ await _webSocketConnection.CloseAsync().ConfigureAwait(false);
+ }
+
#region IDisposable Support
private bool _disposed;
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs
index d9e53f4..69d3453 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs
@@ -23,8 +23,10 @@
using System;
using System.Collections.Concurrent;
-using System.Linq;
+using System.Collections.Generic;
+using System.Threading;
using System.Threading.Tasks;
+using Gremlin.Net.Driver.Exceptions;
using Gremlin.Net.Process;
namespace Gremlin.Net.Driver
@@ -33,91 +35,131 @@ namespace Gremlin.Net.Driver
{
private readonly ConnectionFactory _connectionFactory;
private readonly ConcurrentBag<Connection> _connections = new ConcurrentBag<Connection>();
- private readonly object _connectionsLock = new object();
+ private readonly int _minPoolSize;
+ private readonly int _maxPoolSize;
+ private readonly int _maxInProcessPerConnection;
+ private int _nrConnections;
- public ConnectionPool(ConnectionFactory connectionFactory)
+ public ConnectionPool(ConnectionFactory connectionFactory, ConnectionPoolSettings settings)
{
_connectionFactory = connectionFactory;
+ _minPoolSize = settings.MinSize;
+ _maxPoolSize = settings.MaxSize;
+ _maxInProcessPerConnection = settings.MaxInProcessPerConnection;
+ PopulatePoolAsync().WaitUnwrap();
}
- public int NrConnections { get; private set; }
+ public int NrConnections => Interlocked.CompareExchange(ref _nrConnections, 0, 0);
+
+ private async Task PopulatePoolAsync()
+ {
+ var connectionCreationTasks = new List<Task<Connection>>(_minPoolSize);
+ for (var i = 0; i < _minPoolSize; i++)
+ {
+ connectionCreationTasks.Add(CreateNewConnectionAsync());
+ }
+
+ var createdConnections = await Task.WhenAll(connectionCreationTasks).ConfigureAwait(false);
+ foreach (var c in createdConnections)
+ {
+ _connections.Add(c);
+ }
+
+ Interlocked.CompareExchange(ref _nrConnections, _minPoolSize, 0);
+ }
public async Task<IConnection> GetAvailableConnectionAsync()
{
- if (!TryGetConnectionFromPool(out var connection))
- connection = await CreateNewConnectionAsync().ConfigureAwait(false);
+ if (TryGetConnectionFromPool(out var connection))
+ return ProxiedConnection(connection);
+ connection = await AddConnectionIfUnderMaximumAsync().ConfigureAwait(false);
+ if (connection == null)
+ throw new NoConnectionAvailableException("no connection available!");
+ return ProxiedConnection(connection);
+ }
- return new ProxyConnection(connection, AddConnectionIfOpen);
+ private IConnection ProxiedConnection(Connection connection)
+ {
+ return new ProxyConnection(connection, ReturnConnectionIfOpen);
}
- private bool TryGetConnectionFromPool(out Connection connection)
+ private void ReturnConnectionIfOpen(Connection connection)
+ {
+ if (connection.IsOpen) return;
+ ConsiderUnavailable();
+ DefinitelyDestroyConnection(connection);
+ }
+
+ private async Task<Connection> AddConnectionIfUnderMaximumAsync()
{
while (true)
{
- connection = null;
- lock (_connectionsLock)
- {
- if (_connections.IsEmpty) return false;
- _connections.TryTake(out connection);
- }
+ var nrOpened = Interlocked.CompareExchange(ref _nrConnections, 0, 0);
+ if (nrOpened >= _maxPoolSize) return null;
- if (connection.IsOpen) return true;
- connection.Dispose();
+ if (Interlocked.CompareExchange(ref _nrConnections, nrOpened + 1, nrOpened) == nrOpened)
+ break;
}
+
+ var connection = await CreateNewConnectionAsync().ConfigureAwait(false);
+ _connections.Add(connection);
+ return connection;
}
private async Task<Connection> CreateNewConnectionAsync()
{
- NrConnections++;
var newConnection = _connectionFactory.CreateConnection();
await newConnection.ConnectAsync().ConfigureAwait(false);
return newConnection;
}
- private void AddConnectionIfOpen(Connection connection)
+ private bool TryGetConnectionFromPool(out Connection connection)
{
- if (!connection.IsOpen)
+ while (true)
{
- ConsiderUnavailable();
- connection.Dispose();
- return;
+ connection = SelectLeastUsedConnection();
+ if (connection == null) return false; // _connections is empty
+ if (connection.NrRequestsInFlight >= _maxInProcessPerConnection)
+ return false;
+ if (connection.IsOpen) return true;
+ DefinitelyDestroyConnection(connection);
}
- AddConnection(connection);
}
- private void AddConnection(Connection connection)
+ private Connection SelectLeastUsedConnection()
{
- lock (_connectionsLock)
+ if (_connections.IsEmpty) return null;
+ var nrMinInFlightConnections = int.MaxValue;
+ Connection leastBusy = null;
+ foreach (var connection in _connections)
{
- _connections.Add(connection);
+ var nrInFlight = connection.NrRequestsInFlight;
+ if (nrInFlight >= nrMinInFlightConnections) continue;
+ nrMinInFlightConnections = nrInFlight;
+ leastBusy = connection;
}
- }
- private void ConsiderUnavailable()
- {
- CloseAndRemoveAllConnections();
+ return leastBusy;
}
- private void CloseAndRemoveAllConnections()
+ private void ConsiderUnavailable()
{
- lock (_connectionsLock)
- {
- TeardownAsync().WaitUnwrap();
- RemoveAllConnections();
- }
+ CloseAndRemoveAllConnectionsAsync().WaitUnwrap();
}
- private void RemoveAllConnections()
+ private async Task CloseAndRemoveAllConnectionsAsync()
{
while (_connections.TryTake(out var connection))
{
- connection.Dispose();
+ await connection.CloseAsync().ConfigureAwait(false);
+ DefinitelyDestroyConnection(connection);
}
}
- private async Task TeardownAsync()
+ private void DefinitelyDestroyConnection(Connection connection)
{
- await Task.WhenAll(_connections.Select(c => c.CloseAsync())).ConfigureAwait(false);
+ connection.Dispose();
+ Interlocked.Decrement(ref _nrConnections);
}
#region IDisposable Support
@@ -135,10 +177,11 @@ namespace Gremlin.Net.Driver
if (!_disposed)
{
if (disposing)
- CloseAndRemoveAllConnections();
+ CloseAndRemoveAllConnectionsAsync().WaitUnwrap();
_disposed = true;
}
}
+
#endregion
}
}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPoolSettings.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPoolSettings.cs
new file mode 100644
index 0000000..1eb4dc7
--- /dev/null
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPoolSettings.cs
@@ -0,0 +1,58 @@
+#region License
+
+/*
+ * 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.
+ */
+
+#endregion
+
+using Gremlin.Net.Driver.Exceptions;
+
+namespace Gremlin.Net.Driver
+{
+ /// <summary>
+ /// Holds settings for the <see cref="ConnectionPool"/>.
+ /// </summary>
+ public class ConnectionPoolSettings
+ {
+ private const int DefaultMinPoolSize = 2;
+ private const int DefaultMaxPoolSize = 32;
+ private const int DefaultMaxInProcessPerConnection = 4;
+
+ /// <summary>
+ /// Gets or sets the minimum size of a connection pool.
+ /// </summary>
+ /// <remarks>The default value is 2.</remarks>
+ public int MinSize { get; set; } = DefaultMinPoolSize;
+
+ /// <summary>
+ /// Gets or sets the maximum size of a connection pool.
+ /// </summary>
+ /// <remarks>
+ /// The default value is 32. A <see cref="NoConnectionAvailableException" /> is thrown if this limit is reached
+ /// and all connections have reached the <see cref="MaxInProcessPerConnection" /> limit when a new request comes in.
+ /// </remarks>
+ public int MaxSize { get; set; } = DefaultMaxPoolSize;
+
+ /// <summary>
+ /// Gets the maximum number of in-flight requests that can occur on a connection.
+ /// </summary>
+ /// <remarks>The default value is 4.</remarks>
+ public int MaxInProcessPerConnection { get; set; } = DefaultMaxInProcessPerConnection;
+ }
+}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/NoConnectionAvailableException.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/NoConnectionAvailableException.cs
new file mode 100644
index 0000000..d62436c
--- /dev/null
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/NoConnectionAvailableException.cs
@@ -0,0 +1,56 @@
+#region License
+
+/*
+ * 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.
+ */
+
+#endregion
+
+using System;
+
+namespace Gremlin.Net.Driver.Exceptions
+{
+ /// <summary>
+ /// The exception that is thrown when no connection is available to service a request.
+ /// </summary>
+ public class NoConnectionAvailableException : Exception
+ {
+ /// <summary>
+ /// Initializes a new instance of the <see cref="NoConnectionAvailableException" /> class.
+ /// </summary>
+ public NoConnectionAvailableException()
+ {
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="NoConnectionAvailableException" /> class.
+ /// </summary>
+ public NoConnectionAvailableException(string message)
+ : base(message)
+ {
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="NoConnectionAvailableException" /> class.
+ /// </summary>
+ public NoConnectionAvailableException(string message, Exception inner)
+ : base(message, inner)
+ {
+ }
+ }
+}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClient.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClient.cs
index 5c0213a..2dc44ec 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClient.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClient.cs
@@ -54,19 +54,22 @@ namespace Gremlin.Net.Driver
/// <param name="graphSONReader">A <see cref="GraphSONReader" /> instance to read received GraphSON data.</param>
/// <param name="graphSONWriter">a <see cref="GraphSONWriter" /> instance to write GraphSON data.</param>
/// <param name="mimeType">The GraphSON version mime type, defaults to latest supported by the server.</param>
+ /// <param name="connectionPoolSettings">The <see cref="ConnectionPoolSettings"/> for the connection pool.</param>
/// <param name="webSocketConfiguration">
/// A delegate that will be invoked with the <see cref="ClientWebSocketOptions" />
/// object used to configure WebSocket connections.
/// </param>
public GremlinClient(GremlinServer gremlinServer, GraphSONReader graphSONReader = null,
GraphSONWriter graphSONWriter = null, string mimeType = null,
+ ConnectionPoolSettings connectionPoolSettings = null,
Action<ClientWebSocketOptions> webSocketConfiguration = null)
{
var reader = graphSONReader ?? new GraphSON3Reader();
var writer = graphSONWriter ?? new GraphSON3Writer();
var connectionFactory = new ConnectionFactory(gremlinServer, reader, writer, mimeType ?? DefaultMimeType,
webSocketConfiguration);
- _connectionPool = new ConnectionPool(connectionFactory);
+ _connectionPool =
+ new ConnectionPool(connectionFactory, connectionPoolSettings ?? new ConnectionPoolSettings());
}
/// <summary>
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatus.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatus.cs
index aa0b1b7..cc48a48 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatus.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatus.cs
@@ -22,7 +22,6 @@
#endregion
using System.Collections.Generic;
-using Gremlin.Net.Driver.Exceptions;
using Newtonsoft.Json;
namespace Gremlin.Net.Driver.Messages
@@ -38,13 +37,4 @@ namespace Gremlin.Net.Driver.Messages
[JsonProperty(PropertyName = "message")]
public string Message { get; set; }
}
-
- internal static class ResponseStatusExtensions
- {
- public static void ThrowIfStatusIndicatesError(this ResponseStatus status)
- {
- if (status.Code.IndicatesError())
- throw new ResponseException(status.Code, status.Attributes, $"{status.Code}: {status.Message}");
- }
- }
}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/SingleMessageResultReceiver.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/SingleMessageResultReceiver.cs
new file mode 100644
index 0000000..0592323
--- /dev/null
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/SingleMessageResultReceiver.cs
@@ -0,0 +1,88 @@
+#region License
+
+/*
+ * 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.
+ */
+
+#endregion
+
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Gremlin.Net.Driver.Messages;
+using Gremlin.Net.Driver.ResultsAggregation;
+using Gremlin.Net.Structure.IO.GraphSON;
+using Newtonsoft.Json.Linq;
+
+namespace Gremlin.Net.Driver
+{
+ internal class ResponseHandlerForSingleRequestMessage<T> : IResponseHandlerForSingleRequestMessage
+ {
+ public Task<ResultSet<T>> Result => _tcs.Task;
+
+ private readonly TaskCompletionSource<ResultSet<T>> _tcs =
+ new TaskCompletionSource<ResultSet<T>>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ private readonly GraphSONReader _graphSONReader;
+ private bool _isAggregatingSideEffects;
+ private IAggregator _aggregator;
+ private readonly List<T> _result = new List<T>();
+
+ public ResponseHandlerForSingleRequestMessage(GraphSONReader graphSonReader)
+ {
+ _graphSONReader = graphSonReader;
+ }
+
+ public void HandleReceived(ResponseMessage<JToken> received)
+ {
+ var receivedData = typeof(T) == typeof(JToken)
+ ? new[] {received.Result.Data}
+ : _graphSONReader.ToObject(received.Result.Data);
+ foreach (var d in receivedData)
+ {
+ if (received.Result.Meta.ContainsKey(Tokens.ArgsSideEffectKey))
+ {
+ if (_aggregator == null)
+ _aggregator =
+ new AggregatorFactory().GetAggregatorFor(
+ (string) received.Result.Meta[Tokens.ArgsAggregateTo]);
+ _aggregator.Add(d);
+ _isAggregatingSideEffects = true;
+ }
+ else
+ {
+ _result.Add(d);
+ }
+ }
+ }
+
+ public void Finalize(Dictionary<string, object> statusAttributes)
+ {
+ var resultSet =
+ new ResultSet<T>(
+ _isAggregatingSideEffects ? new List<T> {(T) _aggregator.GetAggregatedResult()} : _result,
+ statusAttributes);
+ _tcs.SetResult(resultSet);
+ }
+
+ public void HandleFailure(Exception objException)
+ {
+ _tcs.TrySetException(objException);
+ }
+ }
+}
\ No newline at end of file
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ConnectionPoolTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ConnectionPoolTests.cs
index 21a2627..b80af00 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ConnectionPoolTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ConnectionPoolTests.cs
@@ -25,6 +25,7 @@ using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Gremlin.Net.Driver;
+using Gremlin.Net.Driver.Exceptions;
using Gremlin.Net.IntegrationTest.Util;
using Xunit;
@@ -39,18 +40,20 @@ namespace Gremlin.Net.IntegrationTest.Driver
private async Task ExecuteMultipleLongRunningRequestsInParallel(IGremlinClient gremlinClient, int nrRequests,
int requestRunningTimeInMs)
{
- var longRunningRequestMsg = _requestMessageProvider.GetSleepMessage(requestRunningTimeInMs);
var tasks = new List<Task>(nrRequests);
for (var i = 0; i < nrRequests; i++)
- tasks.Add(gremlinClient.SubmitAsync(longRunningRequestMsg));
+ {
+ tasks.Add(gremlinClient.SubmitAsync(_requestMessageProvider.GetSleepMessage(requestRunningTimeInMs)));
+ }
+
await Task.WhenAll(tasks);
}
[Fact]
public async Task ShouldReuseConnectionForSequentialRequests()
{
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using (var gremlinClient = new GremlinClient(gremlinServer))
+ const int minConnectionPoolSize = 1;
+ using (var gremlinClient = CreateGremlinClient(minConnectionPoolSize))
{
await gremlinClient.SubmitAsync("");
await gremlinClient.SubmitAsync("");
@@ -60,31 +63,111 @@ namespace Gremlin.Net.IntegrationTest.Driver
}
}
- [Fact]
- public void ShouldOnlyCreateConnectionWhenNecessary()
+ [Theory]
+ [InlineData(0)]
+ [InlineData(1)]
+ [InlineData(8)]
+ public void ShouldStartWithConfiguredNrMinConnections(int minConnectionPoolSize)
{
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using (var gremlinClient = new GremlinClient(gremlinServer))
+ using (var gremlinClient = CreateGremlinClient(minConnectionPoolSize))
{
var nrConnections = gremlinClient.NrConnections;
- Assert.Equal(0, nrConnections);
+ Assert.Equal(minConnectionPoolSize, nrConnections);
}
}
[Fact]
public async Task ShouldExecuteParallelRequestsOnDifferentConnections()
{
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using (var gremlinClient = new GremlinClient(gremlinServer))
+ const int nrBlockingTasks = 1;
+ const int nrFastTasks = 1;
+ using (var gremlinClient = CreateGremlinClient(nrBlockingTasks + nrFastTasks, nrBlockingTasks + nrFastTasks))
+ {
+ const int sleepTime = 200;
+
+ var blockingConnectionTask =
+ ExecuteMultipleLongRunningRequestsInParallel(gremlinClient, nrBlockingTasks, sleepTime);
+ await Task.Delay(50);
+ var fastTask = ExecuteMultipleFastRequestsInParallel(gremlinClient, nrFastTasks);
+
+ var completedTask =
+ await WaitForTaskOrTimeoutAsync(fastTask, TimeSpan.FromMilliseconds(sleepTime - 10));
+ await fastTask;
+ await blockingConnectionTask;
+ Assert.Equal(fastTask, completedTask);
+ }
+ }
+
+ private async Task ExecuteMultipleFastRequestsInParallel(IGremlinClient gremlinClient, int nrRequests)
+ {
+ var tasks = new List<Task>(nrRequests);
+ for (var i = 0; i < nrRequests; i++)
+ {
+ tasks.Add(gremlinClient.SubmitAsync(_requestMessageProvider.GetDummyMessage()));
+ }
+
+ await Task.WhenAll(tasks);
+ }
+
+ [Fact]
+ public async Task ShouldNotCreateMoreThanConfiguredNrMaxConnections()
+ {
+ const int maxConnectionPoolSize = 1;
+ using (var gremlinClient = CreateGremlinClient(maxConnectionPoolSize, maxConnectionPoolSize, 2))
{
- var sleepTime = 50;
- var nrParallelRequests = 5;
+ const int sleepTime = 100;
+
+ await ExecuteMultipleLongRunningRequestsInParallel(gremlinClient, maxConnectionPoolSize * 2, sleepTime);
+
+ Assert.True(gremlinClient.NrConnections <= maxConnectionPoolSize);
+ }
+ }
+
+ [Theory]
+ [InlineData(1)]
+ [InlineData(2)]
+ public async Task ShouldCreateNewConnectionWhenNoConnectionIsAvailable(int nrParallelRequests)
+ {
+ using (var gremlinClient = CreateGremlinClient(minConnectionPoolSize: nrParallelRequests - 1,
+ maxConnectionPoolSize: nrParallelRequests, maxInProcessPerConnection: 1))
+ {
+ const int sleepTime = 100;
await ExecuteMultipleLongRunningRequestsInParallel(gremlinClient, nrParallelRequests, sleepTime);
- var nrConnections = gremlinClient.NrConnections;
- Assert.Equal(nrParallelRequests, nrConnections);
+ Assert.Equal(nrParallelRequests, gremlinClient.NrConnections);
+ }
+ }
+
+ [Fact]
+ public async Task ShouldThrowTimeoutExceptionWhenNoConnectionIsAvailable()
+ {
+ const int nrParallelRequests = 3;
+ using (var gremlinClient = CreateGremlinClient(maxConnectionPoolSize: nrParallelRequests - 1))
+ {
+ const int sleepTime = 100;
+
+ await Assert.ThrowsAsync<NoConnectionAvailableException>(() =>
+ ExecuteMultipleLongRunningRequestsInParallel(gremlinClient, nrParallelRequests, sleepTime));
}
}
+
+ private static GremlinClient CreateGremlinClient(int minConnectionPoolSize = 0, int maxConnectionPoolSize = 8,
+ int maxInProcessPerConnection = 4)
+ {
+ var gremlinServer = new GremlinServer(TestHost, TestPort);
+ return new GremlinClient(gremlinServer,
+ connectionPoolSettings: new ConnectionPoolSettings
+ {
+ MinSize = minConnectionPoolSize,
+ MaxSize = maxConnectionPoolSize,
+ MaxInProcessPerConnection = maxInProcessPerConnection
+ });
+ }
+
+ private static Task<Task> WaitForTaskOrTimeoutAsync(Task task, TimeSpan timeout)
+ {
+ return Task.WhenAny(task, Task.Delay(timeout));
+ }
}
}
\ No newline at end of file
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
index 22fe351..c4ae588 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
@@ -31,9 +31,6 @@ using Gremlin.Net.Driver.Messages;
using Gremlin.Net.IntegrationTest.Util;
using Newtonsoft.Json.Linq;
using Xunit;
-using Gremlin.Net.Structure.IO.GraphSON;
-using Newtonsoft.Json.Linq;
-using Newtonsoft.Json;
namespace Gremlin.Net.IntegrationTest.Driver
{
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs
index a61d4e5..8ba8ebf 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs
@@ -178,14 +178,13 @@ namespace Gremlin.Net.IntegrationTest.Process.Traversal.DriverRemoteConnection
{
// 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
- var graph = new Graph();
var connection = _connectionFactory.CreateRemoteConnection();
var options = new Dictionary<string,object>
{
{"x", "test"},
{"y", true}
};
- var g = graph.Traversal().WithRemote(connection);
+ var g = AnonymousTraversalSource.Traversal().WithRemote(connection);
var b = new Bindings();
var countWithStrategy = g.WithStrategies(new OptionsStrategy(options)).V().Count().Next();
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/RemoteConnectionFactory.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/RemoteConnectionFactory.cs
index 39b7fea..088db59 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/RemoteConnectionFactory.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/RemoteConnectionFactory.cs
@@ -44,7 +44,9 @@ namespace Gremlin.Net.IntegrationTest.Process.Traversal.DriverRemoteConnection
public IRemoteConnection CreateRemoteConnection(string traversalSource)
{
- var c = new DriverRemoteConnectionImpl(new GremlinClient(new GremlinServer(TestHost, TestPort)),
+ var c = new DriverRemoteConnectionImpl(
+ new GremlinClient(new GremlinServer(TestHost, TestPort),
+ connectionPoolSettings: new ConnectionPoolSettings {MinSize = 1}),
traversalSource);
_connections.Add(c);
return c;
diff --git a/pom.xml b/pom.xml
index bd37141..313d32d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -403,6 +403,7 @@ limitations under the License.
<exclude>**/node/node</exclude>
<exclude>**/npm-debug.log</exclude>
<exclude>**/_site/**</exclude>
+ <exclude>**/.idea/**</exclude>
</excludes>
<licenses>
<license implementation="org.apache.rat.analysis.license.ApacheSoftwareLicense20"/>