fix: rename option for concurrency (#270)

Author: supercaracal

README.md

@@ -26,6 +26,7 @@ gem 'redis-cluster-client'
 | `:replica` | Boolean | `false` | `true` if client should use scale read feature |
 | `:replica_affinity` | Symbol or String | `:random` | scale reading strategy, `:random`, `random_with_primary` or `:latency` are valid |
 | `:fixed_hostname` | String | `nil` | required if client should connect to single endpoint with SSL |
+| `:concurrency` | Hash | `{ model: :on_demand, size: 5}` | concurrency settings, `:on_demand`, `:pooled` and `:none` are valid models, size is a max number of workers, `:none` model is no concurrency, Please choose the one suited your environment if needed. |
 
 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.

lib/redis_client/cluster.rb

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

lib/redis_client/cluster/concurrent_worker.rb

@@ -2,11 +2,11 @@
 
 require 'redis_client/cluster/concurrent_worker/on_demand'
 require 'redis_client/cluster/concurrent_worker/pooled'
+require 'redis_client/cluster/concurrent_worker/none'
 
 class RedisClient
   class Cluster
     module ConcurrentWorker
-      MAX_WORKERS = Integer(ENV.fetch('REDIS_CLIENT_MAX_THREADS', 5))
       InvalidNumberOfTasks = Class.new(::RedisClient::Error)
 
       class Group
@@ -30,55 +30,53 @@ def done
           end
         end
 
-        def initialize(worker:, size:)
-          raise ArgumentError, "the size must be positive: #{size} given" unless size.positive?
-
+        def initialize(worker:, queue:, size:)
           @worker = worker
-          @result_queue = SizedQueue.new(size)
+          @queue = queue
+          @size = size
           @count = 0
         end
 
         def push(id, *args, **kwargs, &block)
-          raise InvalidNumberOfTasks, "max size reached: #{@count}" if @count == @result_queue.max
+          raise InvalidNumberOfTasks, "max size reached: #{@count}" if @count == @size
 
-          task = Task.new(id: id, queue: @result_queue, args: args, kwargs: kwargs, block: block)
+          task = Task.new(id: id, queue: @queue, args: args, kwargs: kwargs, block: block)
           @worker.push(task)
           @count += 1
           nil
         end
 
         def each
-          raise InvalidNumberOfTasks, "expected: #{@result_queue.max}, actual: #{@count}" if @count != @result_queue.max
+          raise InvalidNumberOfTasks, "expected: #{@size}, actual: #{@count}" if @count != @size
 
-          @result_queue.max.times do
-            task = @result_queue.pop
+          @size.times do
+            task = @queue.pop
             yield(task.id, task.result)
           end
 
           nil
         end
 
         def close
-          @result_queue.clear
-          @result_queue.close
+          @queue.clear
+          @queue.close if @queue.respond_to?(:close)
           @count = 0
           nil
         end
       end
 
       module_function
 
-      def create(model: :on_demand)
+      def create(model: :on_demand, size: 5)
+        size = size.positive? ? size : 5
+
         case model
-        when :on_demand, nil then ::RedisClient::Cluster::ConcurrentWorker::OnDemand.new
-        when :pooled then ::RedisClient::Cluster::ConcurrentWorker::Pooled.new
+        when :on_demand, nil then ::RedisClient::Cluster::ConcurrentWorker::OnDemand.new(size: size)
+        when :pooled then ::RedisClient::Cluster::ConcurrentWorker::Pooled.new(size: size)
+        when :none then ::RedisClient::Cluster::ConcurrentWorker::None.new
         else raise ArgumentError, "Unknown model: #{model}"
         end
       end
-
-      def size
-        MAX_WORKERS.positive? ? MAX_WORKERS : 5
-      end
     end
   end
 end

lib/redis_client/cluster/concurrent_worker/none.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+class RedisClient
+  class Cluster
+    module ConcurrentWorker
+      class None
+        def new_group(size:)
+          ::RedisClient::Cluster::ConcurrentWorker::Group.new(
+            worker: self,
+            queue: Array.new(size),
+            size: size
+          )
+        end
+
+        def push(task)
+          task.exec
+        end
+
+        def close; end
+      end
+    end
+  end
+end

lib/redis_client/cluster/concurrent_worker/on_demand.rb

@@ -4,12 +4,16 @@ class RedisClient
   class Cluster
     module ConcurrentWorker
       class OnDemand
-        def initialize
-          @q = SizedQueue.new(::RedisClient::Cluster::ConcurrentWorker.size)
+        def initialize(size:)
+          @q = SizedQueue.new(size)
         end
 
         def new_group(size:)
-          ::RedisClient::Cluster::ConcurrentWorker::Group.new(worker: self, size: size)
+          ::RedisClient::Cluster::ConcurrentWorker::Group.new(
+            worker: self,
+            queue: SizedQueue.new(size),
+            size: size
+          )
         end
 
         def push(task)

lib/redis_client/cluster/concurrent_worker/pooled.rb

@@ -10,14 +10,19 @@ module ConcurrentWorker
       # It is a fixed size but we can modify the size with some environment variables.
       # So it consumes memory 1 MB multiplied a number of workers.
       class Pooled
-        def initialize
+        def initialize(size:)
+          @size = size
           setup
         end
 
         def new_group(size:)
           reset if @pid != ::RedisClient::PIDCache.pid
           ensure_workers if @workers.first.nil?
-          ::RedisClient::Cluster::ConcurrentWorker::Group.new(worker: self, size: size)
+          ::RedisClient::Cluster::ConcurrentWorker::Group.new(
+            worker: self,
+            queue: SizedQueue.new(size),
+            size: size
+          )
         end
 
         def push(task)
@@ -37,7 +42,7 @@ def close
 
         def setup
           @q = Queue.new
-          @workers = Array.new(::RedisClient::Cluster::ConcurrentWorker.size)
+          @workers = Array.new(@size)
           @pid = ::RedisClient::PIDCache.pid
         end
 

lib/redis_client/cluster_config.rb

@@ -17,6 +17,7 @@ class ClusterConfig
     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
+    MAX_WORKERS = Integer(ENV.fetch('REDIS_CLIENT_MAX_THREADS', 5))
 
     InvalidClientConfigError = Class.new(::RedisClient::Error)
 
@@ -27,7 +28,7 @@ def initialize(
       replica: false,
       replica_affinity: :random,
       fixed_hostname: '',
-      concurrent_worker_model: nil,
+      concurrency: nil,
       client_implementation: ::RedisClient::Cluster, # for redis gem
       **client_config
     )
@@ -39,7 +40,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
+      @concurrency = merge_concurrency_option(concurrency)
       @client_implementation = client_implementation
       @mutex = Mutex.new
     end
@@ -50,7 +51,7 @@ def dup
         replica: @replica,
         replica_affinity: @replica_affinity,
         fixed_hostname: @fixed_hostname,
-        concurrent_worker_model: @concurrent_worker_model,
+        concurrency: @concurrency,
         client_implementation: @client_implementation,
         **@client_config
       )
@@ -61,20 +62,20 @@ def inspect
     end
 
     def read_timeout
-      @client_config[:read_timeout] || @client_config[:timeout] || RedisClient::Config::DEFAULT_TIMEOUT
+      @client_config[:read_timeout] || @client_config[:timeout] || ::RedisClient::Config::DEFAULT_TIMEOUT
     end
 
     def new_pool(size: 5, timeout: 5, **kwargs)
       @client_implementation.new(
         self,
         pool: { size: size, timeout: timeout },
-        concurrent_worker_model: @concurrent_worker_model,
+        concurrency: @concurrency,
         **kwargs
       )
     end
 
     def new_client(**kwargs)
-      @client_implementation.new(self, concurrent_worker_model: @concurrent_worker_model, **kwargs)
+      @client_implementation.new(self, concurrency: @concurrency, **kwargs)
     end
 
     def per_node_key
@@ -104,6 +105,15 @@ def add_node(host, port)
 
     private
 
+    def merge_concurrency_option(option)
+      case option
+      when Hash
+        option = option.transform_keys(&:to_sym)
+        { size: MAX_WORKERS }.merge(option)
+      else { size: MAX_WORKERS }
+      end
+    end
+
     def build_node_configs(addrs)
       configs = Array[addrs].flatten.filter_map { |addr| parse_node_addr(addr) }
       raise InvalidClientConfigError, '`nodes` option is empty' if configs.empty?

test/ips_pipeline.rb

@@ -12,13 +12,15 @@ module IpsPipeline
   def run
     on_demand = make_client(:on_demand)
     pooled = make_client(:pooled)
+    none = make_client(:none)
     envoy = make_client_for_envoy
     cluster_proxy = make_client_for_cluster_proxy
-    prepare(on_demand, pooled, envoy, cluster_proxy)
+    prepare(on_demand, pooled, none, envoy, cluster_proxy)
     print_letter('pipelined')
     bench(
       ondemand: on_demand,
       pooled: pooled,
+      none: none,
       envoy: envoy,
       cproxy: cluster_proxy
     )
@@ -30,7 +32,7 @@ def make_client(model)
       replica: true,
       replica_affinity: :random,
       fixed_hostname: TEST_FIXED_HOSTNAME,
-      concurrent_worker_model: model,
+      concurrency: { model: model },
       **TEST_GENERIC_OPTIONS
     ).new_client
   end

test/redis_client/cluster/concurrent_worker/test_none.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+require 'testing_helper'
+require 'redis_client/cluster/concurrent_worker/mixin'
+
+class RedisClient
+  class Cluster
+    module ConcurrentWorker
+      class TestNone < TestingWrapper
+        include Mixin
+
+        private
+
+        def model
+          :none
+        end
+      end
+    end
+  end
+end