ptupitsyn commented on code in PR #11107:
URL: https://github.com/apache/ignite/pull/11107#discussion_r1431139996
##########
modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Services/ServicesClient.cs:
##########
@@ -202,7 +216,123 @@ public IServicesClient WithServerKeepBinary()
: ctx.Reader;
return reader.ReadObject<object>();
- });
+ },
+ GetServiceTopology(serviceName));
+ }
+
+ /// <summary>
+ /// Provides actual known service topology or empty list if: service
topology is not enabled, not supported or
+ /// not received yet.
+ /// </summary>
+ private IList<Guid> GetServiceTopology(string serviceName)
+ {
+ if (_servicesTopologies == null ||
!_ignite.Socket.GetSocket().Features.HasFeature(ClientBitmaskFeature.ServiceTopology))
+ return null;
+
+ return _servicesTopologies.GetOrAdd(serviceName, s => new
ServiceTopology(serviceName, this)).GetAndUpdate();
+ }
+
+ /// <summary>
+ /// Keeps topology of certain service and its update progress.
+ /// </summary>
+ private class ServiceTopology
+ {
+ /** Service name. */
+ private readonly string _svcName;
+
+ /** Ignite services. */
+ private readonly ServicesClient _svcClient;
+
+ /** Flag of topology update progress. */
+ private int _updateInProgress;
+
+ /** Time of the last received topology. */
+ private long _lastUpdateRequestTime;
+
+ /** UUIDs of the nodes with at least one service instance. */
+ private volatile IList<Guid> _nodes;
+
+ /** Last cluster topology version when current service topology
was actual. */
+ private long _lastAffTop;
+
+ /** */
+ internal ServiceTopology(string name, ServicesClient svcClient)
+ {
+ _svcName = name;
+ _svcClient = svcClient;
+ }
+
+ /// <summary>
+ /// Asynchronously updates the service topology.
+ /// </summary>
+ private async void UpdateTopologyAsync()
+ {
+ if (Interlocked.Exchange(ref _updateInProgress, 1) == 1)
+ return;
+
+ var socket = _svcClient._ignite.Socket;
+
+ var topVer = socket.GetTopologyVersion();
+
+ var log = _svcClient._ignite.GetConfiguration().Logger;
+
+ var groupNodes = _svcClient._clusterGroup?.GetNodes();
+
+ _nodes = await
socket.DoOutInOpAsync(ClientOp.ServiceGetTopology,
+ ctx => ctx.Writer.WriteString(_svcName),
+ ctx =>
+ {
+ var cnt = ctx.Reader.ReadInt();
+
+ IList<Guid> res = new List<Guid>(cnt);
+
+ for (var i = 0; i < cnt; ++i)
+ res.Add(BinaryUtils.ReadGuid(ctx.Reader.Stream));
+
+ Interlocked.Exchange(ref _lastUpdateRequestTime,
DateTime.Now.Ticks);
+ Interlocked.Exchange(ref _lastAffTop, topVer);
+ Interlocked.Exchange(ref _updateInProgress, 0);
+
+ var filteredTopology = FilterTopology(res,
groupNodes?.Select(n => n.Id).ToList());
+
+ log.Debug("Topology of service '" + _svcName + "' has
been updated. The " +
+ "service instance nodes: " + string.Join(",
", res.Select(gid=>gid.ToString())) +
+ ". Effective topology with the cluster group
is: " +
+ string.Join(", ",
filteredTopology.Select(gid=>gid.ToString())) + '.');
Review Comment:
At this point we haven't yet updated `_nodes`. Let's move those lines to the
end of `UpdateTopologyAsync` method, since we have `await` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
