Skip Interlocked.Exchange and FlushConfig when no update detected (#905)

* skip Interlocked.Exchange and FlushConfig when no update detected on merge

* special case of zero epoch necessitates tracking of explicit slot update

* version-bump-1.0.50

* ensure src is owner

* update migrateslotwalk test

Author: vazois

Version.props

@@ -1,6 +1,6 @@
 <Project>
 	<!-- Versioning property for builds and packages -->
 	<PropertyGroup>
-		<VersionPrefix>1.0.49</VersionPrefix>
+		<VersionPrefix>1.0.50</VersionPrefix>
 	</PropertyGroup>
 </Project>

libs/cluster/Server/ClusterConfig.cs

@@ -943,6 +943,8 @@ private ClusterConfig MergeWorkerInfo(Worker worker)
 
         public ClusterConfig MergeSlotMap(ClusterConfig senderConfig, ILogger logger = null)
         {
+            // Track if update happened to avoid expensive merge and FlushConfig operation when possible
+            var updated = false;
             var senderSlotMap = senderConfig.slotMap;
             var senderWorkerId = GetWorkerIdFromNodeId(senderConfig.LocalNodeId);
 
@@ -979,12 +981,16 @@ public ClusterConfig MergeSlotMap(ClusterConfig senderConfig, ILogger logger = n
                 if (senderConfig.LocalNodeConfigEpoch != 0 && workers[currentOwnerId].ConfigEpoch >= senderConfig.LocalNodeConfigEpoch)
                     continue;
 
+                // Update happened only if workerId or state changed
+                // NOTE: this avoids message flooding when sender epoch equals zero
+                updated = newSlotMap[i]._workerId != senderWorkerId || newSlotMap[i]._state != SlotState.STABLE;
+
                 // Update ownership of node
                 newSlotMap[i]._workerId = senderWorkerId;
                 newSlotMap[i]._state = SlotState.STABLE;
             }
 
-            return new(newSlotMap, workers);
+            return updated ? new(newSlotMap, workers) : this;
         }
 
         /// <summary>

test/Garnet.test.cluster/ClusterMigrateTests.cs

@@ -1808,7 +1808,7 @@ void MigrateKeys()
                     var status = context.clusterTestUtils.SetSlot(_tgt, slot, "IMPORTING", nodeIds[_src], logger: context.logger);
                     while (string.IsNullOrEmpty(status) || !status.Equals("OK"))
                     {
-                        SetSlot(_src);
+                        SetSlot(_src, slot);
                         ClusterTestUtils.BackOff(cancellationToken: cancellationToken, msg: $"{nodeIds[_src]}({nodeEndpoints[_src].Port}) > {slot} > {nodeIds[_tgt]}({nodeEndpoints[_tgt].Port})");
                         status = context.clusterTestUtils.SetSlot(_tgt, slot, "IMPORTING", nodeIds[_src], logger: context.logger);
                     }
@@ -1847,6 +1847,8 @@ void MigrateSlots()
                         var node = config.GetBySlot(slot);
                         if (node != null && node.NodeId.Equals(nodeIds[_src]))
                             break;
+                        // Force set slot to src node
+                        SetSlot(_src, slot);
                         ClusterTestUtils.BackOff(cancellationToken: cancellationToken);
                     }
 
@@ -1878,15 +1880,6 @@ void MigrateSlots()
                         ClusterTestUtils.BackOff(cancellationToken: cancellationToken);
                     }
                 }
-
-                void SetSlot(int nodeIndex)
-                {
-                    for (var i = 0; i < shards; i++)
-                    {
-                        var resp = context.clusterTestUtils.SetSlot(i, slot, "NODE", nodeIds[nodeIndex], logger: context.logger);
-                        ClassicAssert.AreEqual("OK", resp);
-                    }
-                }
             }
 
             ValidateConfig();
@@ -1932,14 +1925,25 @@ void ValidateConfig()
 
                         if (node == null || nodeIds[shards - 1] != node.NodeId)
                         {
+                            // If failed to converge start from the beginning and backOff to give time to converge
                             i = 0;
+                            SetSlot(shards - 1, slot);
                             ClusterTestUtils.BackOff(cancellationToken: cancellationToken);
                             continue;
                         }
                         ClassicAssert.AreEqual(nodeIds[shards - 1], node.NodeId);
                     }
                 }
             }
+
+            void SetSlot(int nodeIndex, int slot)
+            {
+                for (var i = 0; i < shards; i++)
+                {
+                    var resp = context.clusterTestUtils.SetSlot(i, slot, "NODE", nodeIds[nodeIndex], logger: context.logger);
+                    ClassicAssert.AreEqual("OK", resp);
+                }
+            }
         }
     }
 }