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 a385120aca IGNITE-20900 .NET: Handle replica assignment timestamp
(#2900)
a385120aca is described below
commit a385120aca9b4c3062da8feea5c1ae37fcc8cf24
Author: Pavel Tupitsyn <[email protected]>
AuthorDate: Wed Nov 29 14:00:12 2023 +0200
IGNITE-20900 .NET: Handle replica assignment timestamp (#2900)
---
.../Apache.Ignite.Tests/ClientSocketTests.cs | 2 +-
.../dotnet/Apache.Ignite.Tests/FakeServer.cs | 13 ++-----
.../dotnet/Apache.Ignite.Tests/MetricsTests.cs | 2 +-
.../Apache.Ignite.Tests/PartitionAwarenessTests.cs | 45 +++++++++++++++++++---
.../Apache.Ignite/Internal/ClientFailoverSocket.cs | 29 +++++++-------
.../dotnet/Apache.Ignite/Internal/ClientSocket.cs | 3 +-
.../Internal/IClientSocketEventListener.cs | 4 +-
.../dotnet/Apache.Ignite/Internal/Table/Table.cs | 37 ++++++++++--------
8 files changed, 83 insertions(+), 52 deletions(-)
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/ClientSocketTests.cs
b/modules/platforms/dotnet/Apache.Ignite.Tests/ClientSocketTests.cs
index 64d23585d5..58ed2ec36d 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/ClientSocketTests.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/ClientSocketTests.cs
@@ -88,7 +88,7 @@ namespace Apache.Ignite.Tests
private class NoOpListener : IClientSocketEventListener
{
- public void OnAssignmentChanged(ClientSocket clientSocket)
+ public void OnAssignmentChanged(long timestamp)
{
// No-op.
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
b/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
index 3de2bdba55..8dc99ec868 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
@@ -95,7 +95,7 @@ namespace Apache.Ignite.Tests
public string[] PartitionAssignment { get; set; }
- public bool PartitionAssignmentChanged { get; set; }
+ public long PartitionAssignmentTimestamp { get; set; }
public TimeSpan HandshakeDelay { get; set; }
@@ -348,15 +348,8 @@ namespace Apache.Ignite.Tests
writer.Write(0); // Message type.
writer.Write(requestId);
- if (PartitionAssignmentChanged)
- {
- writer.Write((int)ResponseFlags.PartitionAssignmentChanged);
- writer.Write(DateTime.UtcNow.Ticks);
- }
- else
- {
- writer.Write(0);
- }
+ writer.Write((int)ResponseFlags.PartitionAssignmentChanged);
+ writer.Write(PartitionAssignmentTimestamp);
writer.Write(ObservableTimestamp); // Observable timestamp.
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/MetricsTests.cs
b/modules/platforms/dotnet/Apache.Ignite.Tests/MetricsTests.cs
index 0eacdd5aa8..f2dd0fffec 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/MetricsTests.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/MetricsTests.cs
@@ -87,7 +87,7 @@ public class MetricsTests
await client.Tables.GetTablesAsync();
AssertMetric("bytes-sent", 17);
- AssertMetric("bytes-received", 73);
+ AssertMetric("bytes-received", 74);
}
[Test]
diff --git
a/modules/platforms/dotnet/Apache.Ignite.Tests/PartitionAwarenessTests.cs
b/modules/platforms/dotnet/Apache.Ignite.Tests/PartitionAwarenessTests.cs
index 4cc45ffde5..980bf7c9c0 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/PartitionAwarenessTests.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/PartitionAwarenessTests.cs
@@ -54,8 +54,13 @@ public class PartitionAwarenessTests
_server2 = new FakeServer(nodeName: "srv2");
var assignment = new[] { _server1.Node.Name, _server2.Node.Name };
- _server1.PartitionAssignment = assignment;
- _server2.PartitionAssignment = assignment;
+ var assignmentTimestamp = DateTime.UtcNow.AddDays(-1).Ticks; // Old
assignment.
+
+ foreach (var server in new[] { _server1, _server2 })
+ {
+ server.PartitionAssignment = assignment;
+ server.PartitionAssignmentTimestamp = assignmentTimestamp;
+ }
}
[TearDown]
@@ -325,6 +330,34 @@ public class PartitionAwarenessTests
expectedNode);
}
+ [Test]
+ public async Task TestOldAssignmentIsIgnored()
+ {
+ using var client = await GetClient();
+ var recordView = (await
client.Tables.GetTableAsync(FakeServer.ExistingTableName))!.GetRecordView<int>();
+
+ // Check default assignment.
+ await recordView.UpsertAsync(null, 1);
+ await AssertOpOnNode(() => recordView.UpsertAsync(null, 1),
ClientOp.TupleUpsert, _server2);
+
+ // One server has old assignment
+ _server1.PartitionAssignment =
_server1.PartitionAssignment.Reverse().ToArray();
+ _server1.PartitionAssignmentTimestamp -= 1000;
+
+ // Multiple requests to receive timestamp from all servers.
+ for (int i = 0; i < 10; i++)
+ {
+ await client.Tables.GetTablesAsync();
+ }
+
+ // Check that assignment has not changed - update with old timestamp
was ignored.
+ _server1.ClearOps();
+ _server2.ClearOps();
+
+ await recordView.UpsertAsync(null, 1);
+ await AssertOpOnNode(() => recordView.UpsertAsync(null, 1),
ClientOp.TupleUpsert, _server2);
+ }
+
private static async Task AssertOpOnNode(
Func<Task> action,
ClientOp op,
@@ -362,16 +395,16 @@ public class PartitionAwarenessTests
await AssertOpOnNode(() => func(recordView), op, _server2);
// Update assignment.
+ var assignmentTimestamp = DateTime.UtcNow.Ticks;
+
foreach (var server in new[] { _server1, _server2 })
{
server.ClearOps();
server.PartitionAssignment =
server.PartitionAssignment.Reverse().ToArray();
- server.PartitionAssignmentChanged = true;
+ server.PartitionAssignmentTimestamp = assignmentTimestamp;
}
- // First request on default node receives update flag.
- // Make two requests because balancing uses round-robin node.
- await client.Tables.GetTablesAsync();
+ // First request receives update flag.
await client.Tables.GetTablesAsync();
// Second request loads and uses new assignment.
diff --git
a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientFailoverSocket.cs
b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientFailoverSocket.cs
index 4650b2d150..2ee5c4b2e9 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientFailoverSocket.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientFailoverSocket.cs
@@ -59,15 +59,12 @@ namespace Apache.Ignite.Internal
Justification = "WaitHandle is not used in SemaphoreSlim, no need
to dispose.")]
private readonly SemaphoreSlim _socketLock = new(1);
- /** Last connected socket. Used to track partition assignment updates.
*/
- private volatile ClientSocket? _lastConnectedSocket;
-
/** Disposed flag. */
private volatile bool _disposed;
/** Local topology assignment version. Instead of using event handlers
to notify all tables about assignment change,
* the table will compare its version with channel version to detect
an update. */
- private int _assignmentVersion;
+ private long _assignmentTimestamp;
/** Cluster id from the first handshake. */
private Guid? _clusterId;
@@ -104,9 +101,9 @@ namespace Apache.Ignite.Internal
public IgniteClientConfiguration Configuration { get; }
/// <summary>
- /// Gets the partition assignment version.
+ /// Gets the partition assignment timestamp.
/// </summary>
- public int PartitionAssignmentVersion =>
Interlocked.CompareExchange(ref _assignmentVersion, -1, -1);
+ public long PartitionAssignmentTimestamp => Interlocked.Read(ref
_assignmentTimestamp);
/// <summary>
/// Gets the observable timestamp.
@@ -255,15 +252,20 @@ namespace Apache.Ignite.Internal
}
/// <inheritdoc/>
- void IClientSocketEventListener.OnAssignmentChanged(ClientSocket
clientSocket)
+ void IClientSocketEventListener.OnAssignmentChanged(long timestamp)
{
- // NOTE: Multiple channels will send the same update to us,
resulting in multiple cache invalidations.
- // This could be solved with a cluster-wide AssignmentVersion, but
we don't have that.
- // So we only react to updates from the last known good channel.
When no user-initiated operations are performed on that
- // channel, heartbeat messages will trigger updates.
- if (clientSocket == _lastConnectedSocket)
+ while (true)
{
- Interlocked.Increment(ref _assignmentVersion);
+ var oldTimestamp = Interlocked.Read(ref _assignmentTimestamp);
+ if (oldTimestamp >= timestamp)
+ {
+ return;
+ }
+
+ if (Interlocked.CompareExchange(ref _assignmentTimestamp,
value: timestamp, comparand: oldTimestamp) == oldTimestamp)
+ {
+ return;
+ }
}
}
@@ -491,7 +493,6 @@ namespace Apache.Ignite.Internal
endpoint.Socket = socket;
_endpointsByName[socket.ConnectionContext.ClusterNode.Name] =
endpoint;
- _lastConnectedSocket = socket;
return socket;
}
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.cs
b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.cs
index 9a119c7aa5..2d211b0476 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.cs
@@ -726,8 +726,7 @@ namespace Apache.Ignite.Internal
_logger.LogPartitionAssignmentChangeNotificationInfo(ConnectionContext.ClusterNode.Address,
timestamp);
- // TODO IGNITE-20900: Propagate assignment timestamp.
- _listener.OnAssignmentChanged(this);
+ _listener.OnAssignmentChanged(timestamp);
}
var observableTimestamp = reader.ReadInt64();
diff --git
a/modules/platforms/dotnet/Apache.Ignite/Internal/IClientSocketEventListener.cs
b/modules/platforms/dotnet/Apache.Ignite/Internal/IClientSocketEventListener.cs
index 469afbfd2d..3dbbd4bb95 100644
---
a/modules/platforms/dotnet/Apache.Ignite/Internal/IClientSocketEventListener.cs
+++
b/modules/platforms/dotnet/Apache.Ignite/Internal/IClientSocketEventListener.cs
@@ -25,8 +25,8 @@ internal interface IClientSocketEventListener
/// <summary>
/// Called when partition assignment changes.
/// </summary>
- /// <param name="clientSocket">Source socket.</param>
- void OnAssignmentChanged(ClientSocket clientSocket);
+ /// <param name="timestamp">Timestamp.</param>
+ void OnAssignmentChanged(long timestamp);
/// <summary>
/// Called when observable timestamp changes.
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/Table/Table.cs
b/modules/platforms/dotnet/Apache.Ignite/Internal/Table/Table.cs
index 80e5e97502..9e1fc51842 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/Table/Table.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/Table/Table.cs
@@ -74,7 +74,7 @@ namespace Apache.Ignite.Internal.Table
private volatile int _latestSchemaVersion = SchemaVersionUnknown;
/** */
- private volatile int _partitionAssignmentVersion = -1;
+ private long _partitionAssignmentTimestamp = -1;
/** */
private volatile string?[]? _partitionAssignment;
@@ -215,11 +215,11 @@ namespace Apache.Ignite.Internal.Table
/// <returns>Partition assignment.</returns>
internal async ValueTask<string?[]?> GetPartitionAssignmentAsync()
{
- var socketVer = _socket.PartitionAssignmentVersion;
+ var latestKnownTimestamp = _socket.PartitionAssignmentTimestamp;
var assignment = _partitionAssignment;
// Async double-checked locking. Assignment changes rarely, so we
avoid the lock if possible.
- if (_partitionAssignmentVersion == socketVer && assignment != null)
+ if (Interlocked.Read(ref _partitionAssignmentTimestamp) >=
latestKnownTimestamp && assignment != null)
{
return assignment;
}
@@ -228,20 +228,27 @@ namespace Apache.Ignite.Internal.Table
try
{
- socketVer = _socket.PartitionAssignmentVersion;
+ latestKnownTimestamp = _socket.PartitionAssignmentTimestamp;
assignment = _partitionAssignment;
- if (_partitionAssignmentVersion == socketVer && assignment !=
null)
+ if (Interlocked.Read(ref _partitionAssignmentTimestamp) >=
latestKnownTimestamp && assignment != null)
{
return assignment;
}
- assignment = await
LoadPartitionAssignmentAsync().ConfigureAwait(false);
+ var res = await
LoadPartitionAssignmentAsync(latestKnownTimestamp).ConfigureAwait(false);
+ if (res == null)
+ {
+ // Assignment for the given timestamp is not available yet
(some nodes can lag behind).
+ return null;
+ }
- _partitionAssignment = assignment;
- _partitionAssignmentVersion = socketVer;
+ Debug.Assert(res.Value.Timestamp >= latestKnownTimestamp,
"res.Value.Timestamp >= socketTimestamp");
- return assignment;
+ _partitionAssignment = res.Value.Assignment;
+ Interlocked.Exchange(ref _partitionAssignmentTimestamp,
res.Value.Timestamp);
+
+ return res.Value.Assignment;
}
finally
{
@@ -389,7 +396,7 @@ namespace Apache.Ignite.Internal.Table
/// Loads the partition assignment.
/// </summary>
/// <returns>Partition assignment.</returns>
- private async Task<string?[]?> LoadPartitionAssignmentAsync()
+ private async Task<(string?[]? Assignment, long Timestamp)?>
LoadPartitionAssignmentAsync(long timestamp)
{
using var writer = ProtoCommon.GetMessageWriter();
Write(writer.MessageWriter);
@@ -400,10 +407,10 @@ namespace Apache.Ignite.Internal.Table
void Write(MsgPackWriter w)
{
w.Write(Id);
- w.Write(0); // TODO IGNITE-20900: Send timestamp.
+ w.Write(timestamp);
}
- string?[]? Read()
+ (string?[] Assignment, long Timestamp)? Read()
{
var r = resBuf.GetReader();
var count = r.ReadInt32();
@@ -413,9 +420,7 @@ namespace Apache.Ignite.Internal.Table
return null;
}
- // TODO IGNITE-20900: Handle timestamp.
- _ = r.ReadInt64();
-
+ var resTimestamp = r.ReadInt64();
var res = new string?[count];
for (int i = 0; i < count; i++)
@@ -423,7 +428,7 @@ namespace Apache.Ignite.Internal.Table
res[i] = r.ReadStringNullable();
}
- return res;
+ return (res, resTimestamp);
}
}
}
