High CPU Utilization from Gossip Fix (#1315)

* background task for flush config

* add comments

* add parameter to control the config flush frequency

* refactor gossip codebase

* cleanup and consolidate gossip merge

* make AddConnection and TryRemoveConnection async

* cleanup code duplication

* add dictionary to GarnetClusterConnection

* fix typo

* Rename cluster config frequency with ms suffix

* update defaults.conf

Author: vazois

libs/cluster/Server/ClusterManager.cs

@@ -3,7 +3,9 @@
 
 using System;
 using System.Threading;
+using System.Threading.Tasks;
 using Garnet.common;
+using Garnet.server;
 using Garnet.server.TLS;
 using Microsoft.Extensions.Logging;
 using Tsavorite.core;
@@ -30,32 +32,42 @@ internal sealed partial class ClusterManager : IDisposable
         /// </summary>
         readonly IGarnetTlsOptions tlsOptions;
 
+        /// <summary>
+        /// Garnet server options
+        /// </summary>
+        readonly GarnetServerOptions serverOptions;
+
         /// <summary>
         /// ClusterProvider
         /// </summary>
         public readonly ClusterProvider clusterProvider;
 
+        /// <summary>
+        /// Flush count used to indicate a pending flush operation.
+        /// </summary>
+        int flushCount = 0;
+
         /// <summary>
         /// Constructor
         /// </summary>
-        public unsafe ClusterManager(ClusterProvider clusterProvider, ILogger logger = null)
+        public ClusterManager(ClusterProvider clusterProvider, ILogger logger = null)
         {
             this.clusterProvider = clusterProvider;
-            var opts = clusterProvider.serverOptions;
+            this.serverOptions = clusterProvider.serverOptions;
             var clusterFolder = "/cluster";
-            var clusterDataPath = opts.CheckpointDir + clusterFolder;
-            var deviceFactory = opts.GetInitializedDeviceFactory(clusterDataPath);
+            var clusterDataPath = serverOptions.CheckpointDir + clusterFolder;
+            var deviceFactory = serverOptions.GetInitializedDeviceFactory(clusterDataPath);
 
             clusterConfigDevice = deviceFactory.Get(new FileDescriptor(directoryName: "", fileName: "nodes.conf"));
             pool = new(1, (int)clusterConfigDevice.SectorSize);
 
             var clusterEndpoint = clusterProvider.storeWrapper.GetClusterEndpoint();
 
             this.logger = logger;
-            var recoverConfig = clusterConfigDevice.GetFileSize(0) > 0 && !opts.CleanClusterConfig;
+            var recoverConfig = clusterConfigDevice.GetFileSize(0) > 0 && !serverOptions.CleanClusterConfig;
 
-            tlsOptions = opts.TlsOptions;
-            if (!opts.CleanClusterConfig)
+            tlsOptions = serverOptions.TlsOptions;
+            if (!serverOptions.CleanClusterConfig)
                 logger?.LogInformation("Attempt to recover cluster config from: {configFilename}", clusterConfigDevice.FileName);
             else
                 logger?.LogInformation("Skipping recovery of local config due to CleanClusterConfig flag set");
@@ -84,10 +96,37 @@ public unsafe ClusterManager(ClusterProvider clusterProvider, ILogger logger = n
             clusterConnectionStore = new GarnetClusterConnectionStore(logger: logger);
             InitLocal(clusterEndpoint.Address.ToString(), clusterEndpoint.Port, recoverConfig);
             logger?.LogInformation("{NodeInfoStartup}", CurrentConfig.GetClusterInfo(clusterProvider).TrimEnd('\n'));
-            gossipDelay = TimeSpan.FromSeconds(opts.GossipDelay);
-            clusterTimeout = opts.ClusterTimeout <= 0 ? Timeout.InfiniteTimeSpan : TimeSpan.FromSeconds(opts.ClusterTimeout);
+            gossipDelay = TimeSpan.FromSeconds(serverOptions.GossipDelay);
+            clusterTimeout = serverOptions.ClusterTimeout <= 0 ? Timeout.InfiniteTimeSpan : TimeSpan.FromSeconds(serverOptions.ClusterTimeout);
             numActiveTasks = 0;
-            GossipSamplePercent = opts.GossipSamplePercent;
+            GossipSamplePercent = serverOptions.GossipSamplePercent;
+
+            // Run Background task
+            if (serverOptions.ClusterConfigFlushFrequencyMs > 0)
+                Task.Run(() => FlushTask());
+
+            async Task FlushTask()
+            {
+                var flushConfigFrequency = TimeSpan.FromMilliseconds(serverOptions.ClusterConfigFlushFrequencyMs);
+                try
+                {
+                    Interlocked.Increment(ref numActiveTasks);
+                    while (true)
+                    {
+                        if (ctsGossip.IsCancellationRequested)
+                            return;
+
+                        if (Interlocked.CompareExchange(ref flushCount, 0, 0) > 0)
+                            ClusterUtils.WriteInto(clusterConfigDevice, pool, 0, currentConfig.ToByteArray(), logger: logger);
+
+                        await Task.Delay(flushConfigFrequency, ctsGossip.Token).ConfigureAwait(false);
+                    }
+                }
+                finally
+                {
+                    Interlocked.Decrement(ref numActiveTasks);
+                }
+            }
         }
 
         /// <summary>
@@ -118,10 +157,23 @@ public void DisposeBackgroundTasks()
         public void Start()
             => TryStartGossipTasks();
 
+        /// <summary>
+        /// Flush current config to disk
+        /// </summary>
         public void FlushConfig()
         {
-            lock (this)
-                ClusterUtils.WriteInto(clusterConfigDevice, pool, 0, currentConfig.ToByteArray(), logger: logger);
+            if (serverOptions.ClusterConfigFlushFrequencyMs == -1)
+                return;
+
+            if (serverOptions.ClusterConfigFlushFrequencyMs > 0)
+                Interlocked.Increment(ref flushCount);
+            else
+            {
+                lock (this)
+                {
+                    ClusterUtils.WriteInto(clusterConfigDevice, pool, 0, currentConfig.ToByteArray(), logger: logger);
+                }
+            }
         }
 
         /// <summary>
@@ -157,6 +209,10 @@ private void InitLocal(string address, int port, bool recoverConfig)
             }
         }
 
+        /// <summary>
+        /// Implements CLUSTER INFO command
+        /// </summary>
+        /// <returns></returns>
         public string GetInfo()
         {
             var current = CurrentConfig;
@@ -175,6 +231,11 @@ public string GetInfo()
             return ClusterInfo;
         }
 
+        /// <summary>
+        /// Return range of slots from provided array of slots
+        /// </summary>
+        /// <param name="slots"></param>
+        /// <returns></returns>
         public static string GetRange(int[] slots)
         {
             var range = "> ";

libs/cluster/Server/Failover/ReplicaFailoverSession.cs

@@ -18,43 +18,11 @@ internal sealed partial class FailoverSession : IDisposable
         /// </summary>
         GarnetClient primaryClient = null;
 
-        /// <summary>
-        /// Set to true to re-use established gossip connections for failover.
-        /// Note connection might abruptly close due to timeout.
-        /// Increase gossip-delay to avoid shutting down connections prematurely during a failover.
-        /// </summary>
-        bool useGossipConnections = false;
-
         /// <summary>
         /// Send page size for GarnetClient
         /// </summary>
         const int sendPageSize = 1 << 17;
 
-        /// <summary>
-        /// Helper method to re-use gossip connection to perform the failover
-        /// </summary>
-        /// <param name="nodeId">Node-id to use for search the connection array</param>
-        /// <returns></returns>
-        /// <exception cref="GarnetException"></exception>
-        private async Task<GarnetClient> GetOrAddConnectionAsync(string nodeId)
-        {
-            _ = clusterProvider.clusterManager.clusterConnectionStore.GetConnection(nodeId, out var gsn);
-
-            var endpoint = oldConfig.GetEndpointFromNodeId(nodeId);
-            while (!clusterProvider.clusterManager.clusterConnectionStore.GetOrAdd(clusterProvider, endpoint, clusterProvider.serverOptions.TlsOptions, nodeId, out gsn, logger: logger))
-                _ = System.Threading.Thread.Yield();
-
-            if (gsn == null)
-            {
-                logger?.LogWarning("TryMeet: Could not establish connection to remote node [{nodeId} {endpoint}] failed", nodeId, endpoint);
-                return null;
-            }
-
-            await gsn.InitializeAsync();
-
-            return gsn.Client;
-        }
-
         /// <summary>
         /// Helper method to establish connection towards remote node
         /// </summary>
@@ -80,8 +48,7 @@ private async Task<GarnetClient> CreateConnectionAsync(string nodeId)
             }
             catch (Exception ex)
             {
-                if (!useGossipConnections)
-                    client?.Dispose();
+                client?.Dispose();
                 logger?.LogError(ex, "ReplicaFailoverSession.CreateConnection");
                 return null;
             }
@@ -93,9 +60,7 @@ private async Task<GarnetClient> CreateConnectionAsync(string nodeId)
         /// <param name="nodeId"></param>
         /// <returns></returns>
         private Task<GarnetClient> GetConnectionAsync(string nodeId)
-        {
-            return useGossipConnections ? GetOrAddConnectionAsync(nodeId) : CreateConnectionAsync(nodeId);
-        }
+            => CreateConnectionAsync(nodeId);
 
         /// <summary>
         /// Send stop writes message to PRIMARY
@@ -250,8 +215,7 @@ await client.Gossip(configByteArray).ContinueWith(t =>
             }
             finally
             {
-                if (!useGossipConnections)
-                    client?.Dispose();
+                client?.Dispose();
             }
         }