perf: reduce memory allocations for pipelining (#123)

Author: supercaracal

.github/workflows/test.yaml

@@ -338,3 +338,32 @@ jobs:
           done
       - name: Stop containers
         run: docker compose -f $DOCKER_COMPOSE_FILE down
+  memory-profile:
+    name: Memory Profile
+    timeout-minutes: 15
+    strategy:
+      fail-fast: false
+    runs-on: ubuntu-latest
+    env:
+      REDIS_VERSION: '7.0.1'
+      DOCKER_COMPOSE_FILE: 'compose.yaml'
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v3
+      - name: Set up Ruby
+        uses: ruby/setup-ruby@v1
+        with:
+          ruby-version: '3.1'
+          bundler-cache: true
+      - name: Pull Docker images
+        run: docker pull redis:$REDIS_VERSION
+      - name: Run containers
+        run: docker compose -f $DOCKER_COMPOSE_FILE up -d
+      - name: Wait for Redis cluster to be ready
+        run: bundle exec rake wait
+      - name: Print containers
+        run: docker compose -f $DOCKER_COMPOSE_FILE ps
+      - name: Run memory profiler
+        run: bundle exec rake prof
+      - name: Stop containers
+        run: docker compose -f $DOCKER_COMPOSE_FILE down

Gemfile

@@ -4,6 +4,7 @@ source 'https://rubygems.org'
 gemspec name: 'redis-cluster-client'
 
 gem 'hiredis-client', '~> 0.6'
+gem 'memory_profiler'
 gem 'minitest'
 gem 'rake'
 gem 'rubocop'

Rakefile

@@ -34,6 +34,14 @@ Rake::TestTask.new(:bench) do |t|
   t.test_files = ARGV.size > 1 ? ARGV[1..] : Dir['test/**/bench_*.rb']
 end
 
+Rake::TestTask.new(:prof) do |t|
+  t.libs << :lib
+  t.libs << :test
+  t.options = '-v'
+  t.warning = false
+  t.test_files = ARGV.size > 1 ? ARGV[1..] : Dir['test/**/prof_*.rb']
+end
+
 desc 'Wait for cluster to be ready'
 task :wait do
   $LOAD_PATH.unshift(File.expand_path('test', __dir__))

lib/redis_client/cluster.rb

@@ -78,7 +78,8 @@ def zscan(key, *args, **kwargs, &block)
     end
 
     def pipelined
-      pipeline = ::RedisClient::Cluster::Pipeline.new(@router, @command_builder)
+      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
 

lib/redis_client/cluster/command.rb

@@ -72,14 +72,14 @@ def pick_details(details)
       end
 
       def dig_details(command, key)
-        name = command&.flatten&.first.to_s.downcase
+        name = command&.flatten&.first.to_s.downcase # OPTIMIZE: prevent allocation for string
         return if name.empty? || !@details.key?(name)
 
         @details.fetch(name).fetch(key)
       end
 
       def determine_first_key_position(command) # rubocop:disable Metrics/CyclomaticComplexity, Metrics/MethodLength
-        case command&.flatten&.first.to_s.downcase
+        case command&.flatten&.first.to_s.downcase # OPTIMIZE: prevent allocation for string
         when 'eval', 'evalsha', 'zinterstore', 'zunionstore' then 3
         when 'object' then 2
         when 'memory'

lib/redis_client/cluster/pipeline.rb

@@ -9,54 +9,48 @@ class Pipeline
       ReplySizeError = Class.new(::RedisClient::Error)
       MAX_THREADS = Integer(ENV.fetch('REDIS_CLIENT_MAX_THREADS', 5))
 
-      def initialize(router, command_builder)
+      def initialize(router, command_builder, seed: Random.new_seed)
         @router = router
         @command_builder = command_builder
-        @grouped = Hash.new([].freeze)
+        @grouped = {}
         @size = 0
-        @seed = Random.new_seed
+        @seed = seed
       end
 
       def call(*args, **kwargs, &block)
         command = @command_builder.generate(args, kwargs)
         node_key = @router.find_node_key(command, seed: @seed)
-        @grouped[node_key] += [[@size, :call_v, command, block]]
-        @size += 1
+        add_line(node_key, [@size, :call_v, command, block])
       end
 
       def call_v(args, &block)
         command = @command_builder.generate(args)
         node_key = @router.find_node_key(command, seed: @seed)
-        @grouped[node_key] += [[@size, :call_v, command, block]]
-        @size += 1
+        add_line(node_key, [@size, :call_v, command, block])
       end
 
       def call_once(*args, **kwargs, &block)
         command = @command_builder.generate(args, kwargs)
         node_key = @router.find_node_key(command, seed: @seed)
-        @grouped[node_key] += [[@size, :call_once_v, command, block]]
-        @size += 1
+        add_line(node_key, [@size, :call_once_v, command, block])
       end
 
       def call_once_v(args, &block)
         command = @command_builder.generate(args)
         node_key = @router.find_node_key(command, seed: @seed)
-        @grouped[node_key] += [[@size, :call_once_v, command, block]]
-        @size += 1
+        add_line(node_key, [@size, :call_once_v, command, block])
       end
 
       def blocking_call(timeout, *args, **kwargs, &block)
         command = @command_builder.generate(args, kwargs)
         node_key = @router.find_node_key(command, seed: @seed)
-        @grouped[node_key] += [[@size, :blocking_call_v, timeout, command, block]]
-        @size += 1
+        add_line(node_key, [@size, :blocking_call_v, timeout, command, block])
       end
 
       def blocking_call_v(timeout, args, &block)
         command = @command_builder.generate(args)
         node_key = @router.find_node_key(command, seed: @seed)
-        @grouped[node_key] += [[@size, :blocking_call_v, timeout, command, block]]
-        @size += 1
+        add_line(node_key, [@size, :blocking_call_v, timeout, command, block])
       end
 
       def empty?
@@ -92,6 +86,14 @@ def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Met
 
         raise ::RedisClient::Cluster::ErrorCollection, errors
       end
+
+      private
+
+      def add_line(node_key, line)
+        @grouped[node_key] = [] unless @grouped.key?(node_key)
+        @grouped[node_key] << line
+        @size += 1
+      end
     end
   end
 end

test/prof_mem.rb

@@ -0,0 +1,74 @@
+# frozen_string_literal: true
+
+require 'memory_profiler'
+require 'redis_cluster_client'
+require 'testing_constants'
+
+module ProfMem
+  module_function
+
+  ATTEMPT_COUNT = 1000
+
+  def run
+    %w[primary_only scale_read_random scale_read_latency pooled].each do |cli_type|
+      print "################################################################################\n"
+      print "# #{cli_type}\n"
+      print "################################################################################\n"
+      print "\n"
+
+      profile do
+        send("new_#{cli_type}_client".to_sym).pipelined do |pi|
+          ATTEMPT_COUNT.times { |i| pi.call('SET', "key#{i}", i) }
+          ATTEMPT_COUNT.times { |i| pi.call('GET', "key#{i}") }
+        end
+      end
+    end
+  end
+
+  def profile(&block)
+    # https://github.com/SamSaffron/memory_profiler
+    report = ::MemoryProfiler.report(top: 10, &block)
+    report.pretty_print(color_output: true, normalize_paths: true)
+  end
+
+  def new_primary_only_client
+    ::RedisClient.cluster(
+      nodes: TEST_NODE_URIS,
+      fixed_hostname: TEST_FIXED_HOSTNAME,
+      **TEST_GENERIC_OPTIONS
+    ).new_client
+  end
+
+  def new_scale_read_random_client
+    ::RedisClient.cluster(
+      nodes: TEST_NODE_URIS,
+      replica: true,
+      replica_affinity: :random,
+      fixed_hostname: TEST_FIXED_HOSTNAME,
+      **TEST_GENERIC_OPTIONS
+    ).new_client
+  end
+
+  def new_scale_read_latency_client
+    ::RedisClient.cluster(
+      nodes: TEST_NODE_URIS,
+      replica: true,
+      replica_affinity: :latency,
+      fixed_hostname: TEST_FIXED_HOSTNAME,
+      **TEST_GENERIC_OPTIONS
+    ).new_client
+  end
+
+  def new_pooled_client
+    ::RedisClient.cluster(
+      nodes: TEST_NODE_URIS,
+      fixed_hostname: TEST_FIXED_HOSTNAME,
+      **TEST_GENERIC_OPTIONS
+    ).new_pool(
+      timeout: TEST_TIMEOUT_SEC,
+      size: 2
+    )
+  end
+end
+
+ProfMem.run