perf: reduce load of servers in initialization as possible (#287)

Author: supercaracal

lib/redis_client/cluster/command.rb

@@ -7,6 +7,8 @@
 class RedisClient
   class Cluster
     class Command
+      SLOW_COMMAND_TIMEOUT = Float(ENV.fetch('REDIS_CLIENT_SLOW_COMMAND_TIMEOUT', -1))
+
       EMPTY_STRING = ''
       LEFT_BRACKET = '{'
       RIGHT_BRACKET = '}'
@@ -25,7 +27,10 @@ def load(nodes)
           cmd = errors = nil
 
           nodes&.each do |node|
+            regular_timeout = node.read_timeout
+            node.read_timeout = SLOW_COMMAND_TIMEOUT > 0.0 ? SLOW_COMMAND_TIMEOUT : regular_timeout
             reply = node.call('COMMAND')
+            node.read_timeout = regular_timeout
             commands = parse_command_reply(reply)
             cmd = ::RedisClient::Cluster::Command.new(commands)
             break

lib/redis_client/cluster/node.rb

@@ -13,11 +13,18 @@ class Cluster
     class Node
       include Enumerable
 
+      # It affects to strike a balance between load and stability in initialization or changed states.
+      MAX_STARTUP_SAMPLE = Integer(ENV.fetch('REDIS_CLIENT_MAX_STARTUP_SAMPLE', 3))
+
+      # It's used with slow queries of fetching meta data like CLUSTER NODES, COMMAND and so on.
+      SLOW_COMMAND_TIMEOUT = Float(ENV.fetch('REDIS_CLIENT_SLOW_COMMAND_TIMEOUT', -1))
+
+      # less memory consumption, but slow
+      USE_CHAR_ARRAY_SLOT = Integer(ENV.fetch('REDIS_CLIENT_USE_CHAR_ARRAY_SLOT', 1)) == 1
+
       SLOT_SIZE = 16_384
       MIN_SLOT = 0
       MAX_SLOT = SLOT_SIZE - 1
-      MAX_STARTUP_SAMPLE = Integer(ENV.fetch('REDIS_CLIENT_MAX_STARTUP_SAMPLE', 3))
-      USE_CHAR_ARRAY_SLOT = Integer(ENV.fetch('REDIS_CLIENT_USE_CHAR_ARRAY_SLOT', 1)) == 1 # less memory consumption, but slow
       IGNORE_GENERIC_CONFIG_KEYS = %i[url host port path].freeze
       DEAD_FLAGS = %w[fail? fail handshake noaddr noflags].freeze
       ROLE_FLAGS = %w[master slave].freeze
@@ -99,7 +106,10 @@ def load_info(options, concurrent_worker, **kwargs) # rubocop:disable Metrics/Ab
 
           startup_nodes.each_with_index do |raw_client, i|
             work_group.push(i, raw_client) do |client|
+              regular_timeout = client.read_timeout
+              client.read_timeout = SLOW_COMMAND_TIMEOUT > 0.0 ? SLOW_COMMAND_TIMEOUT : regular_timeout
               reply = client.call('CLUSTER', 'NODES')
+              client.read_timeout = regular_timeout
               parse_cluster_node_reply(reply)
             rescue StandardError => e
               e
@@ -209,10 +219,6 @@ def each(&block)
         @topology.clients.each_value(&block)
       end
 
-      def shuffled_nodes
-        @topology.clients.values.shuffle
-      end
-
       def sample
         @topology.clients.values.sample
       end
@@ -252,6 +258,10 @@ def clients_for_scanning(seed: nil)
         @topology.clients_for_scanning(seed: seed).values.sort_by { |c| "#{c.config.host}-#{c.config.port}" }
       end
 
+      def replica_clients
+        @topology.replica_clients.values
+      end
+
       def find_node_key_of_primary(slot)
         return if slot.nil?
 

lib/redis_client/cluster/pub_sub.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'redis_client'
+require 'redis_client/cluster/normalized_cmd_name'
 
 class RedisClient
   class Cluster

lib/redis_client/cluster/router.rb

@@ -22,7 +22,7 @@ def initialize(config, concurrent_worker, pool: nil, **kwargs)
         @pool = pool
         @client_kwargs = kwargs
         @node = fetch_cluster_info(@config, @concurrent_worker, pool: @pool, **@client_kwargs)
-        @command = ::RedisClient::Cluster::Command.load(@node.shuffled_nodes)
+        @command = ::RedisClient::Cluster::Command.load(@node.replica_clients.shuffle)
         @mutex = Mutex.new
         @command_builder = @config.command_builder
       end