perf: reduce memory allocations for pipelining (#130)

Author: supercaracal

.rubocop.yml

@@ -15,9 +15,18 @@ Metrics/AbcSize:
     - 'test/**/*'
 
 Metrics/ClassLength:
-  Max: 400
+  Max: 500
+
+Metrics/ModuleLength:
+  Max: 500
 
 Metrics/MethodLength:
+  Max: 50
+  Exclude:
+    - 'test/**/*'
+
+Metrics/BlockLength:
+  Max: 25
   Exclude:
     - 'test/**/*'
 

lib/redis_client/cluster.rb

@@ -81,7 +81,7 @@ def pipelined
       seed = @config.use_replica? && @config.replica_affinity == :random ? nil : Random.new_seed
       pipeline = ::RedisClient::Cluster::Pipeline.new(@router, @command_builder, seed: seed)
       yield pipeline
-      return [] if pipeline.empty? == 0
+      return [] if pipeline.empty?
 
       pipeline.execute
     end

lib/redis_client/cluster/command.rb

@@ -10,7 +10,7 @@ class Command
       EMPTY_STRING = ''
 
       class << self
-        def load(nodes) # rubocop:disable Metrics/MethodLength
+        def load(nodes)
           errors = []
           cmd = nil
           nodes&.each do |node|
@@ -32,10 +32,7 @@ def load(nodes) # rubocop:disable Metrics/MethodLength
 
         def parse_command_details(rows)
           rows&.reject { |row| row[0].nil? }.to_h do |row|
-            [
-              ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_name(row[0]),
-              { arity: row[1], flags: row[2], first: row[3], last: row[4], step: row[5] }
-            ]
+            [row[0].downcase, { arity: row[1], flags: row[2], first: row[3], last: row[4], step: row[5] }]
           end
         end
       end
@@ -85,7 +82,7 @@ def dig_details(command, key)
         @details.fetch(name).fetch(key)
       end
 
-      def determine_first_key_position(command) # rubocop:disable Metrics/CyclomaticComplexity, Metrics/MethodLength
+      def determine_first_key_position(command) # rubocop:disable Metrics/CyclomaticComplexity
         case ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_command(command)
         when 'eval', 'evalsha', 'zinterstore', 'zunionstore' then 3
         when 'object' then 2

lib/redis_client/cluster/node.rb

@@ -37,7 +37,7 @@ def build_connection_prelude
       end
 
       class << self
-        def load_info(options, **kwargs) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity
+        def load_info(options, **kwargs) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
           startup_size = options.size > MAX_STARTUP_SAMPLE ? MAX_STARTUP_SAMPLE : options.size
           node_info_list = errors = nil
           startup_options = options.to_a.sample(MAX_STARTUP_SAMPLE).to_h
@@ -83,7 +83,7 @@ def load_info(options, **kwargs) # rubocop:disable Metrics/AbcSize, Metrics/Cycl
 
         # @see https://redis.io/commands/cluster-nodes/
         # @see https://github.com/redis/redis/blob/78960ad57b8a5e6af743d789ed8fd767e37d42b8/src/cluster.c#L4660-L4683
-        def parse_node_info(info) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/MethodLength
+        def parse_node_info(info) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
           rows = info.split("\n").map(&:split)
           rows.each { |arr| arr[2] = arr[2].split(',') }
           rows.select! { |arr| arr[7] == 'connected' && (arr[2] & %w[fail? fail handshake noaddr noflags]).empty? }
@@ -242,7 +242,7 @@ def call_multiple_nodes!(clients, method, command, args, &block)
         raise ::RedisClient::Cluster::ErrorCollection, errors
       end
 
-      def try_map(clients) # rubocop:disable Metrics/MethodLength, Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
+      def try_map(clients) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
         results = errors = nil
         clients.each_slice(MAX_THREADS) do |chuncked_clients|
           threads = chuncked_clients.map do |k, v|

lib/redis_client/cluster/node/latency_replica.rb

@@ -39,9 +39,8 @@ def any_replica_node_key(seed: nil)
 
         private
 
-        def measure_latencies(clients) # rubocop:disable Metrics/MethodLength, Metrics/AbcSize
-          latencies = nil
-          clients.each_slice(::RedisClient::Cluster::Node::MAX_THREADS) do |chuncked_clients|
+        def measure_latencies(clients) # rubocop:disable Metrics/AbcSize
+          clients.each_slice(::RedisClient::Cluster::Node::MAX_THREADS).each_with_object({}) do |chuncked_clients, acc|
             threads = chuncked_clients.map do |k, v|
               Thread.new(k, v) do |node_key, client|
                 Thread.pass
@@ -63,12 +62,9 @@ def measure_latencies(clients) # rubocop:disable Metrics/MethodLength, Metrics/A
 
             threads.each do |t|
               t.join
-              latencies ||= {}
-              latencies[t.thread_variable_get(:node_key)] = t.thread_variable_get(:latency)
+              acc[t.thread_variable_get(:node_key)] = t.thread_variable_get(:latency)
             end
           end
-
-          latencies
         end
 
         def select_replica_clients(replications, clients)

lib/redis_client/cluster/normalized_cmd_name.rb

@@ -28,6 +28,7 @@ def get_by_name(name)
 
       def clear
         @mutex.synchronize { @cache.clear }
+        true
       end
 
       private

lib/redis_client/cluster/pipeline.rb

@@ -2,6 +2,8 @@
 
 require 'redis_client'
 require 'redis_client/cluster/errors'
+require 'redis_client/middlewares'
+require 'redis_client/pooled'
 
 class RedisClient
   class Cluster
@@ -12,65 +14,70 @@ class Pipeline
       def initialize(router, command_builder, seed: Random.new_seed)
         @router = router
         @command_builder = command_builder
-        @grouped = {}
-        @size = 0
         @seed = seed
+        @pipelines = @indices = nil
+        @size = 0
       end
 
       def call(*args, **kwargs, &block)
         command = @command_builder.generate(args, kwargs)
         node_key = @router.find_node_key(command, seed: @seed)
-        add_row(node_key, [@size, :call_v, command, block])
+        get_pipeline(node_key).call_v(command, &block)
+        index_pipeline(node_key)
       end
 
       def call_v(args, &block)
         command = @command_builder.generate(args)
         node_key = @router.find_node_key(command, seed: @seed)
-        add_row(node_key, [@size, :call_v, command, block])
+        get_pipeline(node_key).call_v(command, &block)
+        index_pipeline(node_key)
       end
 
       def call_once(*args, **kwargs, &block)
         command = @command_builder.generate(args, kwargs)
         node_key = @router.find_node_key(command, seed: @seed)
-        add_row(node_key, [@size, :call_once_v, command, block])
+        get_pipeline(node_key).call_once_v(command, &block)
+        index_pipeline(node_key)
       end
 
       def call_once_v(args, &block)
         command = @command_builder.generate(args)
         node_key = @router.find_node_key(command, seed: @seed)
-        add_row(node_key, [@size, :call_once_v, command, block])
+        get_pipeline(node_key).call_once_v(command, &block)
+        index_pipeline(node_key)
       end
 
       def blocking_call(timeout, *args, **kwargs, &block)
         command = @command_builder.generate(args, kwargs)
         node_key = @router.find_node_key(command, seed: @seed)
-        add_row(node_key, [@size, :blocking_call_v, timeout, command, block])
+        get_pipeline(node_key).blocking_call_v(timeout, command, &block)
+        index_pipeline(node_key)
       end
 
       def blocking_call_v(timeout, args, &block)
         command = @command_builder.generate(args)
         node_key = @router.find_node_key(command, seed: @seed)
-        add_row(node_key, [@size, :blocking_call_v, timeout, command, block])
+        get_pipeline(node_key).blocking_call_v(timeout, command, &block)
+        index_pipeline(node_key)
       end
 
       def empty?
         @size.zero?
       end
 
       # TODO: https://github.com/redis-rb/redis-cluster-client/issues/37 handle redirections
-      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity
+      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
         all_replies = errors = nil
-        @grouped.each_slice(MAX_THREADS) do |chuncked_grouped|
-          threads = chuncked_grouped.map do |k, v|
-            Thread.new(@router, k, v) do |router, node_key, rows|
+        @pipelines&.each_slice(MAX_THREADS) do |chuncked_pipelines|
+          threads = chuncked_pipelines.map do |node_key, pipeline|
+            Thread.new(node_key, pipeline) do |nk, pl|
               Thread.pass
-              replies = do_pipelining(router.find_node(node_key), rows)
-              raise ReplySizeError, "commands: #{rows.size}, replies: #{replies.size}" if rows.size != replies.size
+              Thread.current.thread_variable_set(:node_key, nk)
+              replies = do_pipelining(@router.find_node(nk), pl)
+              raise ReplySizeError, "commands: #{pl._size}, replies: #{replies.size}" if pl._size != replies.size
 
-              Thread.current.thread_variable_set(:rows, rows)
               Thread.current.thread_variable_set(:replies, replies)
             rescue StandardError => e
-              Thread.current.thread_variable_set(:node_key, node_key)
               Thread.current.thread_variable_set(:error, e)
             end
           end
@@ -79,7 +86,7 @@ def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Met
             t.join
             if t.thread_variable?(:replies)
               all_replies ||= Array.new(@size)
-              t.thread_variable_get(:rows).each_with_index { |r, i| all_replies[r.first] = t.thread_variable_get(:replies)[i] }
+              @indices[t.thread_variable_get(:node_key)].each_with_index { |gi, i| all_replies[gi] = t.thread_variable_get(:replies)[i] }
             elsif t.thread_variable?(:error)
               errors ||= {}
               errors[t.thread_variable_get(:node_key)] = t.thread_variable_get(:error)
@@ -94,21 +101,35 @@ def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Met
 
       private
 
-      def add_row(node_key, row)
-        @grouped[node_key] = [] unless @grouped.key?(node_key)
-        @grouped[node_key] << row
+      def get_pipeline(node_key)
+        @pipelines ||= {}
+        @pipelines[node_key] ||= ::RedisClient::Pipeline.new(@command_builder)
+      end
+
+      def index_pipeline(node_key)
+        @indices ||= {}
+        @indices[node_key] ||= []
+        @indices[node_key] << @size
         @size += 1
       end
 
-      def do_pipelining(node, rows)
-        node.pipelined do |pipeline|
-          rows.each do |row|
-            case row.size
-            when 4 then pipeline.send(row[1], row[2], &row[3])
-            when 5 then pipeline.send(row[1], row[2], row[3], &row[4])
-            end
+      def do_pipelining(client, pipeline)
+        case client
+        when ::RedisClient then send_pipeline(client, pipeline)
+        when ::RedisClient::Pooled then client.with { |cli| send_pipeline(cli, pipeline) }
+        else raise NotImplementedError, "#{client.class.name}#pipelined for cluster client"
+        end
+      end
+
+      def send_pipeline(client, pipeline)
+        results = client.send(:ensure_connected, retryable: pipeline._retryable?) do |connection|
+          commands = pipeline._commands
+          ::RedisClient::Middlewares.call_pipelined(commands, client.config) do
+            connection.call_pipelined(commands, pipeline._timeouts)
           end
         end
+
+        pipeline._coerce!(results)
       end
     end
   end

lib/redis_client/cluster/router.rb

@@ -25,7 +25,7 @@ def initialize(config, pool: nil, **kwargs)
         @command_builder = @config.command_builder
       end
 
-      def send_command(method, command, *args, &block) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity
+      def send_command(method, command, *args, &block) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
         cmd = ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_command(command)
         case cmd
         when 'acl', 'auth', 'bgrewriteaof', 'bgsave', 'quit', 'save'
@@ -65,7 +65,7 @@ def send_command(method, command, *args, &block) # rubocop:disable Metrics/AbcSi
 
       # @see https://redis.io/topics/cluster-spec#redirection-and-resharding
       #   Redirection and resharding
-      def try_send(node, method, command, args, retry_count: 3, &block) # rubocop:disable Metrics/MethodLength, Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
+      def try_send(node, method, command, args, retry_count: 3, &block) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
         if args.empty?
           # prevent memory allocation for variable-length args
           node.send(method, command, &block)
@@ -100,7 +100,7 @@ def try_send(node, method, command, args, retry_count: 3, &block) # rubocop:disa
         retry
       end
 
-      def try_delegate(node, method, *args, retry_count: 3, **kwargs, &block) # rubocop:disable Metrics/MethodLength, Metrics/AbcSize
+      def try_delegate(node, method, *args, retry_count: 3, **kwargs, &block) # rubocop:disable Metrics/AbcSize
         node.send(method, *args, **kwargs, &block)
       rescue ::RedisClient::CommandError => e
         raise if retry_count <= 0
@@ -129,7 +129,7 @@ def try_delegate(node, method, *args, retry_count: 3, **kwargs, &block) # ruboco
         retry
       end
 
-      def scan(*command, seed: nil, **kwargs) # rubocop:disable Metrics/MethodLength, Metrics/AbcSize
+      def scan(*command, seed: nil, **kwargs) # rubocop:disable Metrics/AbcSize
         command = @command_builder.generate(command, kwargs)
 
         command[1] = ZERO_CURSOR_FOR_SCAN if command.size == 1
@@ -270,7 +270,7 @@ def assign_asking_node(err_msg)
         find_node(node_key)
       end
 
-      def fetch_cluster_info(config, pool: nil, **kwargs) # rubocop:disable Metrics/MethodLength
+      def fetch_cluster_info(config, pool: nil, **kwargs)
         node_info = ::RedisClient::Cluster::Node.load_info(config.per_node_key, **kwargs)
         node_addrs = node_info.map { |info| ::RedisClient::Cluster::NodeKey.hashify(info[:node_key]) }
         config.update_node(node_addrs)

lib/redis_client/cluster_config.rb

@@ -114,7 +114,7 @@ def parse_node_addr(addr)
       end
     end
 
-    def parse_node_url(addr) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/MethodLength
+    def parse_node_url(addr) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
       return if addr.empty?
 
       uri = URI(addr)

test/redis_client/cluster/test_normalized_cmd_name.rb

@@ -63,27 +63,29 @@ def test_get_by_name
       end
 
       def test_thread_safety
-        results = Array.new(10, true)
-        threads = results.each_with_index.map do |_, i|
+        attempts = Array.new(100, true)
+        threads = attempts.each_with_index.map do |_, i|
           Thread.new do
             Thread.pass
             Thread.current.thread_variable_set(:index, i)
-            if i.even?
-              Thread.current.thread_variable_set(:result, @subject.get_by_command(%w[SET foo bar]) == 'set')
-            else
-              @subject.clear
-            end
+            got = if i.even?
+                    @subject.get_by_command(%w[SET foo bar]) == 'set'
+                  else
+                    @subject.clear
+                  end
+            Thread.current.thread_variable_set(:got, got)
           rescue StandardError
-            Thread.current.thread_variable_set(:result, false)
+            Thread.current.thread_variable_set(:got, false)
           end
         end
 
         threads.each do |t|
           t.join
-          results[t.thread_variable_get(:index)] = t.thread_variable_get(:result)
+          attempts[t.thread_variable_get(:index)] = t.thread_variable_get(:got)
         end
 
-        refute_includes(results, false)
+        refute_includes(attempts, false)
+        refute_includes(attempts, nil)
       end
     end
   end