Enhance performance to use thread and ensure thread safety (#28)

Author: supercaracal

README.md

@@ -20,6 +20,7 @@ gem 'redis-cluster-client'
 | `:fixed_hostname` | String | `nil` | required if client should connect to single endpoint with SSL |
 
 Also, [the other generic options](https://github.com/redis-rb/redis-client#configuration) can be passed.
+But `:url`, `:host`, `:port` and `:path` are ignored because they conflict with the `:nodes` option.
 
 ```ruby
 # The following examples are Docker containers on localhost.
@@ -88,6 +89,22 @@ cli.call('CLUSTER', 'KEYSLOT', 'key3')
 #=> 935
 ```
 
+Also, you can use the hash tag to bias keys to the same slot.
+
+```ruby
+cli.call('CLUSTER', 'KEYSLOT', '{key}1')
+#=> 12539
+
+cli.call('CLUSTER', 'KEYSLOT', '{key}2')
+#=> 12539
+
+cli.call('CLUSTER', 'KEYSLOT', '{key}3')
+#=> 12539
+
+cli.call('MGET', '{key}1', '{key}2', '{key}3')
+#=> [nil, nil, nil]
+```
+
 ## ACL
 The cluster client internally calls `COMMAND` and `CLUSTER NODES` commands to operate correctly.
 So please permit it like the followings.
@@ -97,6 +114,7 @@ So please permit it like the followings.
 cli1 = RedisClient.cluster.new_client
 
 # To create a user with permissions
+# Typically, user settings are configured in the config file for the server beforehand.
 cli1.call('ACL', 'SETUSER', 'foo', 'ON', '+COMMAND', '+CLUSTER|NODES', '+PING', '>mysecret')
 
 # To initialize client with the user

lib/redis_client/cluster.rb

@@ -15,7 +15,6 @@ class Pipeline
       def initialize(client)
         @client = client
         @grouped = Hash.new([].freeze)
-        @replies = []
         @size = 0
       end
 
@@ -41,27 +40,32 @@ def empty?
         @size.zero?
       end
 
-      # TODO: use concurrency
-      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength
-        @grouped.each do |node_key, rows|
-          node_key = node_key.nil? ? @client.instance_variable_get(:@node).primary_node_keys.sample : node_key
-          replies = @client.send(:find_node, node_key).pipelined do |pipeline|
-            rows.each do |row|
-              case row[1]
-              when :call then pipeline.call(*row[2], **row[3])
-              when :call_once then pipeline.call_once(*row[2], **row[3])
-              when :blocking_call then pipeline.blocking_call(row[2], *row[3], **row[4])
-              else raise NotImplementedError, row[1]
+      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity
+        all_replies = []
+        threads = @grouped.map do |k, v|
+          Thread.new(@client, k, v) do |client, node_key, rows|
+            Thread.pass
+
+            node_key = node_key.nil? ? client.instance_variable_get(:@node).primary_node_keys.sample : node_key
+            replies = client.send(:find_node, node_key).pipelined do |pipeline|
+              rows.each do |row|
+                case row[1]
+                when :call then pipeline.call(*row[2], **row[3])
+                when :call_once then pipeline.call_once(*row[2], **row[3])
+                when :blocking_call then pipeline.blocking_call(row[2], *row[3], **row[4])
+                else raise NotImplementedError, row[1]
+                end
               end
             end
-          end
 
-          raise ReplySizeError, "commands: #{rows.size}, replies: #{replies.size}" if rows.size != replies.size
+            raise ReplySizeError, "commands: #{rows.size}, replies: #{replies.size}" if rows.size != replies.size
 
-          rows.each_with_index { |row, idx| @replies[row.first] = replies[idx] }
+            rows.each_with_index { |row, idx| all_replies[row.first] = replies[idx] }
+          end
         end
 
-        @replies
+        threads.each(&:join)
+        all_replies
       end
     end
 
@@ -103,6 +107,7 @@ def initialize(config, pool: nil, **kwargs)
       @client_kwargs = kwargs
       @node = fetch_cluster_info!(@config, pool: @pool, **@client_kwargs)
       @command = ::RedisClient::Cluster::Command.load(@node)
+      @mutex = Mutex.new
     end
 
     def inspect
@@ -326,9 +331,13 @@ def assign_node(*command)
       find_node(node_key)
     end
 
-    def find_node_key(*command, primary_only: false)
+    def find_node_key(*command, primary_only: false) # rubocop:disable Metrics/MethodLength
       key = @command.extract_first_key(command)
-      return if key.empty?
+      if key.empty?
+        return @node.primary_node_keys.sample if @command.should_send_to_primary?(command) || primary_only
+
+        return @node.replica_node_keys.sample
+      end
 
       slot = ::RedisClient::Cluster::KeySlotConverter.convert(key)
       return unless @node.slot_exists?(slot)
@@ -350,13 +359,15 @@ def find_node(node_key)
     end
 
     def update_cluster_info!(node_key = nil)
-      unless node_key.nil?
-        host, port = ::RedisClient::Cluster::NodeKey.split(node_key)
-        @config.add_node(host, port)
-      end
+      @mutex.synchronize do
+        unless node_key.nil?
+          host, port = ::RedisClient::Cluster::NodeKey.split(node_key)
+          @config.add_node(host, port)
+        end
 
-      @node.each(&:close)
-      @node = fetch_cluster_info!(@config, pool: @pool, **@client_kwargs)
+        @node.each(&:close)
+        @node = fetch_cluster_info!(@config, pool: @pool, **@client_kwargs)
+      end
     end
   end
 end

lib/redis_client/cluster/node.rb

@@ -79,6 +79,7 @@ def initialize(options, node_info: [], pool: nil, with_replica: false, **kwargs)
         @slots = build_slot_node_mappings(node_info)
         @replications = build_replication_mappings(node_info)
         @clients = build_clients(options, pool: pool, **kwargs)
+        @mutex = Mutex.new
       end
 
       def inspect
@@ -101,6 +102,12 @@ def primary_node_keys
         @clients.filter_map { |k, _| primary?(k) ? k : nil }.sort
       end
 
+      def replica_node_keys
+        return primary_node_keys if replica_disabled?
+
+        @clients.filter_map { |k, _| replica?(k) ? k : nil }.sort
+      end
+
       def find_by(node_key)
         @clients.fetch(node_key)
       rescue KeyError
@@ -163,7 +170,7 @@ def find_node_key_of_replica(slot)
       end
 
       def update_slot(slot, node_key)
-        @slots[slot] = node_key
+        @mutex.synchronize { @slots[slot] = node_key }
       end
 
       private
@@ -217,15 +224,17 @@ def build_clients(options, pool: nil, **kwargs)
       def try_map # rubocop:disable Metrics/MethodLength
         errors = {}
         results = {}
-
-        @clients.each do |node_key, client|
-          reply = yield(node_key, client)
-          results[node_key] = reply unless reply.nil?
-        rescue ::RedisClient::CommandError => e
-          errors[node_key] = e
-          next
+        threads = @clients.map do |k, v|
+          Thread.new(k, v) do |node_key, client|
+            Thread.pass
+            reply = yield(node_key, client)
+            results[node_key] = reply unless reply.nil?
+          rescue ::RedisClient::CommandError => e
+            errors[node_key] = e
+          end
         end
 
+        threads.each(&:join)
         return results if errors.empty?
 
         raise ::RedisClient::Cluster::CommandErrorCollection, errors

lib/redis_client/cluster_config.rb

@@ -15,14 +15,17 @@ class ClusterConfig
     VALID_SCHEMES = [DEFAULT_SCHEME, SECURE_SCHEME].freeze
     VALID_NODES_KEYS = %i[ssl username password host port db].freeze
     MERGE_CONFIG_KEYS = %i[ssl username password].freeze
+    IGNORE_GENERIC_CONFIG_KEYS = %i[url host port path].freeze
 
     InvalidClientConfigError = Class.new(::RedisClient::Error)
 
     def initialize(nodes: DEFAULT_NODES, replica: false, fixed_hostname: '', **client_config)
       @replica = true & replica
       @fixed_hostname = fixed_hostname.to_s
       @node_configs = build_node_configs(nodes.dup)
+      client_config = client_config.reject { |k, _| IGNORE_GENERIC_CONFIG_KEYS.include?(k) }
       @client_config = merge_generic_config(client_config, @node_configs)
+      @mutex = Mutex.new
     end
 
     def inspect
@@ -51,11 +54,11 @@ def use_replica?
     end
 
     def update_node(addrs)
-      @node_configs = build_node_configs(addrs)
+      @mutex.synchronize { @node_configs = build_node_configs(addrs) }
     end
 
     def add_node(host, port)
-      @node_configs << { host: host, port: port }
+      @mutex.synchronize { @node_configs << { host: host, port: port } }
     end
 
     def dup

test/redis_client/cluster/test_node.rb

@@ -175,6 +175,16 @@ def test_primary_node_keys
         assert_equal(want, got)
       end
 
+      def test_replica_node_keys
+        want = @test_node_info.filter_map { |info| info[:role] == 'master' ? info[:node_key] : nil }.sort
+        got = @test_node.replica_node_keys
+        assert_equal(want, got)
+
+        want = @test_node_info.filter_map { |info| info[:role] == 'slave' ? info[:node_key] : nil }.sort
+        got = @test_node_with_scale_read.replica_node_keys
+        assert_equal(want, got)
+      end
+
       def test_find_by
         @test_node_info.each do |info|
           msg = "Case: primary only: #{info[:node_key]}"