feat: add concurrent worker model option (#265)

Author: supercaracal

Gemfile

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

Rakefile

@@ -37,6 +37,14 @@ Rake::TestTask.new(:bench) do |t|
   t.test_files = ARGV.size > 1 ? ARGV[1..] : Dir['test/**/bench_*.rb']
 end
 
+Rake::TestTask.new(:ips) do |t|
+  t.libs << :lib
+  t.libs << :test
+  t.options = '-v'
+  t.warning = false
+  t.test_files = ARGV.size > 1 ? ARGV[1..] : Dir['test/**/ips_*.rb']
+end
+
 Rake::TestTask.new(:prof) do |t|
   t.libs << :lib
   t.libs << :test

lib/redis_client/cluster.rb

@@ -11,9 +11,9 @@ class Cluster
 
     attr_reader :config
 
-    def initialize(config, pool: nil, **kwargs)
+    def initialize(config, pool: nil, concurrent_worker_model: nil, **kwargs)
       @config = config
-      @concurrent_worker = ::RedisClient::Cluster::ConcurrentWorker.create
+      @concurrent_worker = ::RedisClient::Cluster::ConcurrentWorker.create(model: concurrent_worker_model)
       @router = ::RedisClient::Cluster::Router.new(config, @concurrent_worker, pool: pool, **kwargs)
       @command_builder = config.command_builder
     end

lib/redis_client/cluster/concurrent_worker.rb

@@ -64,7 +64,7 @@ def close
 
       def create(model: :on_demand)
         case model
-        when :on_demand then ::RedisClient::Cluster::ConcurrentWorker::OnDemand.new
+        when :on_demand, nil then ::RedisClient::Cluster::ConcurrentWorker::OnDemand.new
         when :pooled then ::RedisClient::Cluster::ConcurrentWorker::Pooled.new
         else raise ArgumentError, "Unknown model: #{model}"
         end

lib/redis_client/cluster_config.rb

@@ -27,7 +27,8 @@ def initialize(
       replica: false,
       replica_affinity: :random,
       fixed_hostname: '',
-      client_implementation: Cluster,
+      concurrent_worker_model: nil,
+      client_implementation: ::RedisClient::Cluster, # for redis gem
       **client_config
     )
 
@@ -38,6 +39,7 @@ def initialize(
       client_config = client_config.reject { |k, _| IGNORE_GENERIC_CONFIG_KEYS.include?(k) }
       @command_builder = client_config.fetch(:command_builder, ::RedisClient::CommandBuilder)
       @client_config = merge_generic_config(client_config, @node_configs)
+      @concurrent_worker_model = concurrent_worker_model
       @client_implementation = client_implementation
       @mutex = Mutex.new
     end
@@ -48,6 +50,7 @@ def dup
         replica: @replica,
         replica_affinity: @replica_affinity,
         fixed_hostname: @fixed_hostname,
+        concurrent_worker_model: @concurrent_worker_model,
         client_implementation: @client_implementation,
         **@client_config
       )
@@ -62,11 +65,16 @@ def read_timeout
     end
 
     def new_pool(size: 5, timeout: 5, **kwargs)
-      @client_implementation.new(self, pool: { size: size, timeout: timeout }, **kwargs)
+      @client_implementation.new(
+        self,
+        pool: { size: size, timeout: timeout },
+        concurrent_worker_model: @concurrent_worker_model,
+        **kwargs
+      )
     end
 
     def new_client(**kwargs)
-      @client_implementation.new(self, **kwargs)
+      @client_implementation.new(self, concurrent_worker_model: @concurrent_worker_model, **kwargs)
     end
 
     def per_node_key

test/ips_pipeline.rb

@@ -0,0 +1,56 @@
+# frozen_string_literal: true
+
+require 'benchmark/ips'
+require 'redis_cluster_client'
+require 'testing_constants'
+
+module BenchPipeline
+  module_function
+
+  ATTEMPTS = 100
+
+  def run
+    on_demand = make_client(:on_demand)
+    pooled = make_client(:pooled)
+    prepare(on_demand, pooled)
+    bench(on_demand, pooled)
+  end
+
+  def make_client(model)
+    ::RedisClient.cluster(
+      nodes: TEST_NODE_URIS,
+      fixed_hostname: TEST_FIXED_HOSTNAME,
+      concurrent_worker_model: model,
+      **TEST_GENERIC_OPTIONS
+    ).new_client
+  end
+
+  def prepare(on_demand, pooled)
+    on_demand.pipelined do |pi|
+      ATTEMPTS.times { |i| pi.call('SET', "key#{i}", "val#{i}") }
+    end
+
+    pooled.pipelined do |pi|
+      ATTEMPTS.times { |i| pi.call('SET', "key#{i}", "val#{i}") }
+    end
+  end
+
+  def bench(on_demand, pooled)
+    Benchmark.ips do |x|
+      x.time = 5
+      x.warmup = 1
+
+      x.report('on_demand') do
+        on_demand.pipelined { |pi| ATTEMPTS.times { |i| pi.call('GET', "key#{i}") } }
+      end
+
+      x.report('pooled') do
+        pooled.pipelined { |pi| ATTEMPTS.times { |i| pi.call('GET', "key#{i}") } }
+      end
+
+      x.compare!
+    end
+  end
+end
+
+BenchPipeline.run