You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2023/02/16 06:50:12 UTC
[ignite-3] branch main updated: IGNITE-18613 .NET: Balance requests across connections (#1677)
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new f456cd8314 IGNITE-18613 .NET: Balance requests across connections (#1677)
f456cd8314 is described below
commit f456cd8314ee34fc1b10da17e85fc6df764f2dbf
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Thu Feb 16 08:50:05 2023 +0200
IGNITE-18613 .NET: Balance requests across connections (#1677)
When .NET client is connected to more than one server node, send requests to different connections in a round-robin fashion instead of using the same default connection every time.
---
.../Compute/ComputeClusterAwarenessTests.cs | 9 +-
.../Apache.Ignite.Tests/PartitionAwarenessTests.cs | 34 +++-----
.../Apache.Ignite.Tests/RequestBalancingTests.cs | 64 +++++++++++++++
.../Apache.Ignite/Internal/ClientFailoverSocket.cs | 96 +++++++++++-----------
.../Apache.Ignite/Internal/Table/RecordView.cs | 8 +-
5 files changed, 136 insertions(+), 75 deletions(-)
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/Compute/ComputeClusterAwarenessTests.cs b/modules/platforms/dotnet/Apache.Ignite.Tests/Compute/ComputeClusterAwarenessTests.cs
index bb94f710ac..652dc89ae3 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/Compute/ComputeClusterAwarenessTests.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/Compute/ComputeClusterAwarenessTests.cs
@@ -17,6 +17,7 @@
namespace Apache.Ignite.Tests.Compute
{
+ using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Internal.Proto;
@@ -95,14 +96,16 @@ namespace Apache.Ignite.Tests.Compute
// ReSharper disable once AccessToDisposedClosure
TestUtils.WaitForCondition(() => client.GetConnections().Count == 2, 5000);
+ var nodeNames = new HashSet<string>();
+
for (int i = 0; i < 100; i++)
{
var node = i % 2 == 0 ? server1.Node : server2.Node;
-
var res = await client.Compute.ExecuteAsync<string>(nodes: new[] { node }, jobClassName: string.Empty);
-
- Assert.AreEqual(node.Name, res);
+ nodeNames.Add(res);
}
+
+ CollectionAssert.AreEquivalent(new[] { "s1", "s2" }, nodeNames);
}
}
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/PartitionAwarenessTests.cs b/modules/platforms/dotnet/Apache.Ignite.Tests/PartitionAwarenessTests.cs
index b5d14cd7d6..0d92d217ac 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/PartitionAwarenessTests.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/PartitionAwarenessTests.cs
@@ -68,15 +68,10 @@ public class PartitionAwarenessTests
{
using var client = await GetClient();
var recordView = (await client.Tables.GetTableAsync(FakeServer.ExistingTableName))!.GetRecordView<int>();
- var (defaultServer, _) = GetServerPair();
// Warm up.
await recordView.UpsertAsync(null, 1);
- Assert.AreEqual(
- new[] { ClientOp.TableGet, ClientOp.SchemasGet, ClientOp.PartitionAssignmentGet },
- defaultServer.ClientOps.Take(3));
-
// Check.
await AssertOpOnNode(async () => await recordView.UpsertAsync(null, 1), ClientOp.TupleUpsert, _server2, _server1);
await AssertOpOnNode(async () => await recordView.UpsertAsync(null, 3), ClientOp.TupleUpsert, _server1, _server2);
@@ -85,22 +80,20 @@ public class PartitionAwarenessTests
}
[Test]
- public async Task TestPutWithTxUsesDefaultNode()
+ public async Task TestPutWithTxUsesTxNode()
{
using var client = await GetClient();
var recordView = (await client.Tables.GetTableAsync(FakeServer.ExistingTableName))!.GetRecordView<int>();
- var tx = await client.Transactions.BeginAsync();
- var (defaultServer, secondaryServer) = GetServerPair();
- // Second server.
- await recordView.UpsertAsync(tx, 1);
- await recordView.UpsertAsync(tx, 3);
+ var tx = await client.Transactions.BeginAsync();
+ var txServer = new[] { _server1, _server2 }.Single(x => x.ClientOps.Contains(ClientOp.TxBegin));
- Assert.AreEqual(
- new[] { ClientOp.TableGet, ClientOp.TxBegin, ClientOp.SchemasGet, ClientOp.TupleUpsert, ClientOp.TupleUpsert },
- defaultServer.ClientOps);
+ for (int i = 0; i < 10; i++)
+ {
+ await recordView.UpsertAsync(tx, i);
+ }
- CollectionAssert.IsEmpty(secondaryServer.ClientOps);
+ Assert.AreEqual(Enumerable.Repeat(ClientOp.TupleUpsert, 10), txServer.ClientOps.TakeLast(10));
}
[Test]
@@ -108,7 +101,6 @@ public class PartitionAwarenessTests
{
using var client = await GetClient();
var recordView = (await client.Tables.GetTableAsync(FakeServer.ExistingTableName))!.GetRecordView<int>();
- var (defaultServer, _) = GetServerPair();
// Check default assignment.
await recordView.UpsertAsync(null, 1);
@@ -123,7 +115,9 @@ public class PartitionAwarenessTests
}
// First request on default node receives update flag.
- await AssertOpOnNode(() => client.Tables.GetTablesAsync(), ClientOp.TablesGet, defaultServer);
+ // Make two requests because balancing uses round-robin node.
+ await client.Tables.GetTablesAsync();
+ await client.Tables.GetTablesAsync();
// Second request loads and uses new assignment.
await AssertOpOnNode(() => recordView.UpsertAsync(null, 1), ClientOp.TupleUpsert, _server1, allowExtraOps: true);
@@ -378,12 +372,6 @@ public class PartitionAwarenessTests
return await IgniteClient.StartAsync(cfg);
}
- private (FakeServer Default, FakeServer Secondary) GetServerPair()
- {
- // Any server can be primary due to round-robin balancing in ClientFailoverSocket.
- return _server1.ClientOps.Count > 0 ? (_server1, _server2) : (_server2, _server1);
- }
-
// ReSharper disable NotAccessedPositionalProperty.Local
private record CompositeKey(string IdStr, Guid IdGuid);
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/RequestBalancingTests.cs b/modules/platforms/dotnet/Apache.Ignite.Tests/RequestBalancingTests.cs
new file mode 100644
index 0000000000..7b092f33da
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/RequestBalancingTests.cs
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Tests;
+
+using System.Threading.Tasks;
+using Internal.Proto;
+using NUnit.Framework;
+
+/// <summary>
+/// Tests request balancing across active connections.
+/// When client is connected to multiple nodes, requests will use all available connections in a round-robin fashion
+/// (unless an operation is tied to a specific connection, like a transaction or a cursor).
+/// </summary>
+public class RequestBalancingTests
+{
+ [Test]
+ public async Task TestRequestsAreRoundRobinBalanced()
+ {
+ using var server1 = new FakeServer(nodeName: "s1");
+ using var server2 = new FakeServer(nodeName: "s2");
+ using var server3 = new FakeServer(nodeName: "s3");
+
+ var clientCfg = new IgniteClientConfiguration
+ {
+ Endpoints = { server1.Endpoint, server2.Endpoint, server3.Endpoint }
+ };
+
+ using var client = await IgniteClient.StartAsync(clientCfg);
+
+ // ReSharper disable once AccessToDisposedClosure
+ TestUtils.WaitForCondition(() => client.GetConnections().Count == 3, 5000);
+
+ for (var i = 0; i < 10; i++)
+ {
+ await client.Tables.GetTablesAsync();
+ await client.Tables.GetTableAsync(FakeServer.ExistingTableName);
+ }
+
+ // All servers get their share of requests.
+ foreach (var server in new[] { server1, server2, server3 })
+ {
+ CollectionAssert.Contains(server.ClientOps, ClientOp.TableGet, server.Node.Name);
+ CollectionAssert.Contains(server.ClientOps, ClientOp.TablesGet, server.Node.Name);
+ }
+
+ // Total number of requests across all servers.
+ Assert.AreEqual(20, server1.ClientOps.Count + server2.ClientOps.Count + server3.ClientOps.Count);
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientFailoverSocket.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientFailoverSocket.cs
index 86cde54da6..409b62e75b 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientFailoverSocket.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientFailoverSocket.cs
@@ -38,7 +38,7 @@ namespace Apache.Ignite.Internal
internal sealed class ClientFailoverSocket : IDisposable
{
/** Current global endpoint index for Round-robin. */
- private static long _endPointIndex;
+ private static long _globalEndPointIndex;
/** Logger. */
private readonly IIgniteLogger? _logger;
@@ -72,6 +72,9 @@ namespace Apache.Ignite.Internal
/** Cluster id from the first handshake. */
private Guid? _clusterId;
+ /** Local index for round-robin balancing within this FailoverSocket. */
+ private long _endPointIndex = Interlocked.Increment(ref _globalEndPointIndex);
+
/// <summary>
/// Initializes a new instance of the <see cref="ClientFailoverSocket"/> class.
/// </summary>
@@ -114,40 +117,12 @@ namespace Apache.Ignite.Internal
// Because this call is not awaited, execution of the current method continues before the call is completed.
// Secondary connections are established in the background.
+ // TODO IGNITE-18808 Do this periodically.
_ = socket.ConnectAllSockets();
return socket;
}
- /// <summary>
- /// Performs an in-out operation.
- /// </summary>
- /// <param name="clientOp">Client op code.</param>
- /// <param name="tx">Transaction.</param>
- /// <param name="request">Request data.</param>
- /// <param name="preferredNode">Preferred node.</param>
- /// <returns>Response data.</returns>
- public async Task<PooledBuffer> DoOutInOpAsync(
- ClientOp clientOp,
- Transaction? tx,
- PooledArrayBuffer? request = null,
- PreferredNode preferredNode = default)
- {
- if (tx == null)
- {
- // Use failover socket with reconnect and retry behavior.
- return await DoOutInOpAsync(clientOp, request, preferredNode).ConfigureAwait(false);
- }
-
- if (tx.FailoverSocket != this)
- {
- throw new IgniteClientException(ErrorGroups.Client.Connection, "Specified transaction belongs to a different IgniteClient instance.");
- }
-
- // Use tx-specific socket without retry and failover.
- return await tx.Socket.DoOutInOpAsync(clientOp, request).ConfigureAwait(false);
- }
-
/// <summary>
/// Performs an in-out operation.
/// </summary>
@@ -242,6 +217,16 @@ namespace Apache.Ignite.Internal
}
}
+ /// <summary>
+ /// Gets active connections.
+ /// </summary>
+ /// <returns>Active connections.</returns>
+ public IEnumerable<ConnectionContext> GetConnections() =>
+ _endpoints
+ .Select(e => e.Socket?.ConnectionContext)
+ .Where(ctx => ctx != null)
+ .ToList()!;
+
/// <summary>
/// Gets the primary socket. Reconnects if necessary.
/// </summary>
@@ -251,7 +236,7 @@ namespace Apache.Ignite.Internal
"Microsoft.Design",
"CA1031:DoNotCatchGeneralExceptionTypes",
Justification = "Any connection exception should be handled.")]
- public async ValueTask<ClientSocket> GetSocketAsync(PreferredNode preferredNode = default)
+ private async ValueTask<ClientSocket> GetSocketAsync(PreferredNode preferredNode = default)
{
await _socketLock.WaitAsync().ConfigureAwait(false);
@@ -259,6 +244,7 @@ namespace Apache.Ignite.Internal
{
ThrowIfDisposed();
+ // 1. Preferred node connection.
if (preferredNode != default)
{
var key = preferredNode.Id ?? preferredNode.Name;
@@ -277,6 +263,13 @@ namespace Apache.Ignite.Internal
}
}
+ // 2. Round-robin connection.
+ if (GetNextSocketWithoutReconnect() is { } nextSocket)
+ {
+ return nextSocket;
+ }
+
+ // 3. Default connection.
if (_socket == null || _socket.IsDisposed)
{
if (_socket?.IsDisposed == true)
@@ -295,16 +288,6 @@ namespace Apache.Ignite.Internal
}
}
- /// <summary>
- /// Gets active connections.
- /// </summary>
- /// <returns>Active connections.</returns>
- public IEnumerable<ConnectionContext> GetConnections() =>
- _endpoints
- .Select(e => e.Socket?.ConnectionContext)
- .Where(ctx => ctx != null)
- .ToList()!;
-
[SuppressMessage(
"Microsoft.Design",
"CA1031:DoNotCatchGeneralExceptionTypes",
@@ -331,7 +314,7 @@ namespace Apache.Ignite.Internal
continue;
}
- tasks.Add(ConnectAsync(endpoint));
+ tasks.Add(ConnectAsync(endpoint).AsTask());
}
await Task.WhenAll(tasks).ConfigureAwait(false);
@@ -358,7 +341,7 @@ namespace Apache.Ignite.Internal
}
/// <summary>
- /// Gets next connected socket, or connects a new one.
+ /// Gets the next connected socket, or connects a new one.
/// </summary>
[SuppressMessage("Maintainability", "CA1508:Avoid dead conditional code", Justification = "False positive")]
private async ValueTask<ClientSocket> GetNextSocketAsync()
@@ -392,10 +375,31 @@ namespace Apache.Ignite.Internal
"Failed to establish Ignite thin client connection, examine inner exceptions for details.", errors!);
}
+ /// <summary>
+ /// Gets the next connected socket, without establishing new connections.
+ /// </summary>
+ private ClientSocket? GetNextSocketWithoutReconnect()
+ {
+ var startIdx = (int) Interlocked.Increment(ref _endPointIndex);
+
+ for (var i = 0; i < _endpoints.Count; i++)
+ {
+ var idx = (startIdx + i) % _endpoints.Count;
+ var endPoint = _endpoints[idx];
+
+ if (endPoint.Socket is { IsDisposed: false })
+ {
+ return endPoint.Socket;
+ }
+ }
+
+ return null;
+ }
+
/// <summary>
/// Connects to the given endpoint.
/// </summary>
- private async Task<ClientSocket> ConnectAsync(SocketEndpoint endpoint)
+ private async ValueTask<ClientSocket> ConnectAsync(SocketEndpoint endpoint)
{
if (endpoint.Socket?.IsDisposed == false)
{
@@ -470,10 +474,8 @@ namespace Apache.Ignite.Internal
{
try
{
- IPAddress? ip;
-
// GetHostEntry accepts IPs, but TryParse is a more efficient shortcut.
- return IPAddress.TryParse(host, out ip) ? new[] {ip} : Dns.GetHostEntry(host).AddressList;
+ return IPAddress.TryParse(host, out var ip) ? new[] { ip } : Dns.GetHostEntry(host).AddressList;
}
catch (SocketException e)
{
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/Table/RecordView.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/Table/RecordView.cs
index dfedfc782a..aaf2aa7901 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/Table/RecordView.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/Table/RecordView.cs
@@ -396,8 +396,12 @@ namespace Apache.Ignite.Internal.Table
ClientOp clientOp,
Transaction? tx,
PooledArrayBuffer? request = null,
- PreferredNode preferredNode = default) =>
- await _table.Socket.DoOutInOpAsync(clientOp, tx, request, preferredNode).ConfigureAwait(false);
+ PreferredNode preferredNode = default)
+ {
+ var (buf, _) = await _table.Socket.DoOutInOpAndGetSocketAsync(clientOp, tx, request, preferredNode).ConfigureAwait(false);
+
+ return buf;
+ }
private async Task<PooledBuffer> DoRecordOutOpAsync(
ClientOp op,