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 2019/03/07 07:36:57 UTC
[tinkerpop] branch TINKERPOP-2135 updated: TINKERPOP-2135 Use COW
collection for connections
This is an automated email from the ASF dual-hosted git repository.
florianhockmann pushed a commit to branch TINKERPOP-2135
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/TINKERPOP-2135 by this push:
new 0615c85 TINKERPOP-2135 Use COW collection for connections
0615c85 is described below
commit 0615c85dd4d189904e73df8303dcc589d9db0a26
Author: Florian Hockmann <fh...@florian-hockmann.de>
AuthorDate: Wed Mar 6 21:38:20 2019 +0100
TINKERPOP-2135 Use COW collection for connections
A custom copy-on-write (COW) collection is used to store the
connections.
Instead of iterating over all connections to find the least used one,
a round-robin scheduling is used now. The first connection is used that
has not reached its limit.
---
.../Driver/ConcurrentConnectionCollection.cs | 51 -------------
.../src/Gremlin.Net/Driver/ConnectionPool.cs | 57 +++++++--------
.../Driver/CopyOnWriteConnectionCollection.cs | 85 ++++++++++++++++++++++
3 files changed, 113 insertions(+), 80 deletions(-)
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ConcurrentConnectionCollection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ConcurrentConnectionCollection.cs
deleted file mode 100644
index c68e0e9..0000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ConcurrentConnectionCollection.cs
+++ /dev/null
@@ -1,51 +0,0 @@
-#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.Collections;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
-
-namespace Gremlin.Net.Driver
-{
- internal class ConcurrentConnectionCollection : IEnumerable<Connection>
- {
- private const int DefaultValue = 0;
-
- // We use a ConcurrentDictionary because we need a concurrent collection with a TryRemove() method.
- // Only the keys are used here, the values just contain a dummy DefaultValue.
- private readonly ConcurrentDictionary<Connection, int> _connections =
- new ConcurrentDictionary<Connection, int>();
-
- public int Count => _connections.Count;
-
- public bool IsEmpty => _connections.IsEmpty;
-
- public void Add(Connection connection) => _connections.TryAdd(connection, DefaultValue);
-
- public bool TryRemove(Connection connection) => _connections.TryRemove(connection, out _);
-
- public IEnumerator<Connection> GetEnumerator() => _connections.Keys.GetEnumerator();
-
- IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs
index 130d6fd..d291c79 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs
@@ -22,7 +22,6 @@
#endregion
using System;
-using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
@@ -33,10 +32,13 @@ namespace Gremlin.Net.Driver
{
internal class ConnectionPool : IDisposable
{
+ private const int ConnectionIndexOverflowLimit = int.MaxValue - 1000000;
+
private readonly ConnectionFactory _connectionFactory;
- private readonly ConcurrentConnectionCollection _connections = new ConcurrentConnectionCollection();
+ private readonly CopyOnWriteConnectionCollection _connections = new CopyOnWriteConnectionCollection();
private readonly int _poolSize;
private readonly int _maxInProcessPerConnection;
+ private int _connectionIndex;
private int _poolState;
private const int PoolIdle = 0;
private const int PoolPopulationInProgress = 1;
@@ -59,7 +61,7 @@ namespace Gremlin.Net.Driver
private async Task EnsurePoolIsPopulatedAsync()
{
- // The pool could have been empty because of connection problems. So, we need to populate it again.
+ // The pool could have been (partially) empty because of connection problems. So, we need to populate it again.
while (true)
{
if (NrConnections >= _poolSize) break;
@@ -87,10 +89,7 @@ namespace Gremlin.Net.Driver
}
var createdConnections = await Task.WhenAll(connectionCreationTasks).ConfigureAwait(false);
- foreach (var c in createdConnections)
- {
- _connections.Add(c);
- }
+ _connections.AddRange(createdConnections);
}
private async Task<Connection> CreateNewConnectionAsync()
@@ -102,39 +101,40 @@ namespace Gremlin.Net.Driver
private Connection GetConnectionFromPool()
{
- while (true)
- {
- var connection = SelectLeastUsedConnection();
- if (connection == null)
- throw new ServerUnavailableException();
- if (connection.NrRequestsInFlight >= _maxInProcessPerConnection)
- throw new ConnectionPoolBusyException(_poolSize, _maxInProcessPerConnection);
+ var connections = _connections.Snapshot;
+ if (connections.Length == 0) throw new ServerUnavailableException();
+
+ if (GetAvailableConnection(connections, out var connection))
return connection;
- }
+ throw new ConnectionPoolBusyException(_poolSize, _maxInProcessPerConnection);
}
- private Connection SelectLeastUsedConnection()
+ private bool GetAvailableConnection(Connection[] connections, out Connection connection)
{
- if (_connections.IsEmpty) return null;
- var nrMinInFlightConnections = int.MaxValue;
- Connection leastBusy = null;
-
- foreach (var connection in _connections)
+ var index = Interlocked.Increment(ref _connectionIndex);
+ ProtectIndexFromOverflowing(index);
+
+ connection = null;
+ for (var i = 0; i < connections.Length; i++)
{
- var nrInFlight = connection.NrRequestsInFlight;
- if (nrInFlight >= nrMinInFlightConnections) continue;
+ connection = connections[(index + i) % connections.Length];
+ if (connection.NrRequestsInFlight >= _maxInProcessPerConnection) continue;
if (!connection.IsOpen)
{
RemoveConnectionFromPool(connection);
continue;
}
- nrMinInFlightConnections = nrInFlight;
- leastBusy = connection;
+ return true;
}
+ return false;
+ }
- return leastBusy;
+ private void ProtectIndexFromOverflowing(int currentIndex)
+ {
+ if (currentIndex > ConnectionIndexOverflowLimit)
+ Interlocked.Exchange(ref _connectionIndex, 0);
}
-
+
private void RemoveConnectionFromPool(Connection connection)
{
if (_connections.TryRemove(connection))
@@ -159,9 +159,8 @@ namespace Gremlin.Net.Driver
private async Task CloseAndRemoveAllConnectionsAsync()
{
- foreach (var connection in _connections)
+ foreach (var connection in _connections.RemoveAndGetAll())
{
- _connections.TryRemove(connection);
await connection.CloseAsync().ConfigureAwait(false);
DefinitelyDestroyConnection(connection);
}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/CopyOnWriteConnectionCollection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/CopyOnWriteConnectionCollection.cs
new file mode 100644
index 0000000..8e774da
--- /dev/null
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/CopyOnWriteConnectionCollection.cs
@@ -0,0 +1,85 @@
+#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
+{
+ internal class CopyOnWriteConnectionCollection
+ {
+ private static readonly Connection[] EmptyConnections = new Connection[0];
+
+ private volatile Connection[] _connections = EmptyConnections;
+ private readonly object _writeLock = new object();
+
+ public int Count => _connections.Length;
+
+ public void AddRange(Connection[] connections)
+ {
+ if (connections == null || connections.Length == 0)
+ return;
+ lock (_writeLock)
+ {
+ var newArray = new Connection[_connections.Length + connections.Length];
+ _connections.CopyTo(newArray, 0);
+ Array.Copy(connections, 0, newArray, _connections.Length, connections.Length);
+ _connections = newArray;
+ }
+ }
+
+ public bool TryRemove(Connection connection)
+ {
+ lock (_writeLock)
+ {
+ var index = Array.IndexOf(_connections, connection);
+ if (index < 0) return false;
+ if (_connections.Length == 1 && index == 0)
+ {
+ _connections = EmptyConnections;
+ return true;
+ }
+
+ var newArray = new Connection[_connections.Length - 1];
+
+ if (index != 0)
+ Array.Copy(_connections, 0, newArray, 0, index);
+ if (index < _connections.Length - 1)
+ Array.Copy(_connections, index + 1, newArray, index, _connections.Length - index - 1);
+ _connections = newArray;
+ return true;
+ }
+ }
+
+ public Connection[] RemoveAndGetAll()
+ {
+ lock (_writeLock)
+ {
+ var old = _connections;
+ _connections = EmptyConnections;
+ return old;
+ }
+ }
+
+ public Connection[] Snapshot => _connections;
+ }
+}
\ No newline at end of file