Merge pull request #97 from supercaracal/reduce-consumption-of-threads

Reduce consumption of threads

Author: supercaracal

lib/redis_client/cluster/node.rb

@@ -13,6 +13,7 @@ class Node
       MIN_SLOT = 0
       MAX_SLOT = SLOT_SIZE - 1
       MAX_STARTUP_SAMPLE = 37
+      MAX_THREADS = Integer(ENV.fetch('MAX_THREADS', 5))
       IGNORE_GENERIC_CONFIG_KEYS = %i[url host port path].freeze
 
       ReloadNeeded = Class.new(::RedisClient::Error)
@@ -39,18 +40,21 @@ def load_info(options, **kwargs) # rubocop:disable Metrics/AbcSize, Metrics/Cycl
           errors = Array.new(startup_size)
           startup_options = options.to_a.sample(MAX_STARTUP_SAMPLE).to_h
           startup_nodes = ::RedisClient::Cluster::Node.new(startup_options, **kwargs)
-          threads = startup_nodes.each_with_index.map do |raw_client, idx|
-            Thread.new(raw_client, idx) do |cli, i|
-              Thread.pass
-              reply = cli.call('CLUSTER', 'NODES')
-              node_info_list[i] = parse_node_info(reply)
-            rescue StandardError => e
-              errors[i] = e
-            ensure
-              cli&.close
+          startup_nodes.each_slice(MAX_THREADS * 2).with_index do |chuncked_startup_nodes, chuncked_idx|
+            threads = chuncked_startup_nodes.each_with_index.map do |raw_client, idx|
+              Thread.new(raw_client, (MAX_THREADS * 2 * chuncked_idx) + idx) do |cli, i|
+                Thread.pass
+                reply = cli.call('CLUSTER', 'NODES')
+                node_info_list[i] = parse_node_info(reply)
+              rescue StandardError => e
+                errors[i] = e
+              ensure
+                cli&.close
+              end
             end
+            threads.each(&:join)
           end
-          threads.each(&:join)
+
           raise ::RedisClient::Cluster::InitialSetupError, errors if node_info_list.all?(&:nil?)
 
           grouped = node_info_list.compact.group_by do |rows|
@@ -267,17 +271,20 @@ def build_clients(options, pool: nil, **kwargs)
       def try_map # rubocop:disable Metrics/MethodLength
         results = {}
         errors = {}
-        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 StandardError => e
-            errors[node_key] = e
+        @clients.each_slice(MAX_THREADS * 2) do |chuncked_clients|
+          threads = chuncked_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 StandardError => e
+              errors[node_key] = e
+            end
           end
+
+          threads.each(&:join)
         end
 
-        threads.each(&:join)
         [results, errors]
       end
     end

lib/redis_client/cluster/pipeline.rb

@@ -7,6 +7,7 @@ class RedisClient
   class Cluster
     class Pipeline
       ReplySizeError = Class.new(::RedisClient::Error)
+      MAX_THREADS = Integer(ENV.fetch('MAX_THREADS', 5))
 
       def initialize(router, command_builder)
         @router = router
@@ -62,27 +63,30 @@ def empty?
       end
 
       # TODO: https://github.com/redis-rb/redis-cluster-client/issues/37 handle redirections
-      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength
+      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity
         all_replies = Array.new(@size)
         errors = {}
-        threads = @grouped.map do |k, v|
-          Thread.new(@router, k, v) do |router, node_key, rows|
-            Thread.pass
-            replies = router.find_node(node_key).pipelined do |pipeline|
-              rows.each do |(_size, *row, block)|
-                pipeline.send(*row, &block)
+        @grouped.each_slice(MAX_THREADS * 2) do |chuncked_grouped|
+          threads = chuncked_grouped.map do |k, v|
+            Thread.new(@router, k, v) do |router, node_key, rows|
+              Thread.pass
+              replies = router.find_node(node_key).pipelined do |pipeline|
+                rows.each do |(_size, *row, block)|
+                  pipeline.send(*row, &block)
+                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| all_replies[row.first] = replies[idx] }
-          rescue StandardError => e
-            errors[node_key] = e
+              rows.each_with_index { |row, idx| all_replies[row.first] = replies[idx] }
+            rescue StandardError => e
+              errors[node_key] = e
+            end
           end
+
+          threads.each(&:join)
         end
 
-        threads.each(&:join)
         return all_replies if errors.empty?
 
         raise ::RedisClient::Cluster::ErrorCollection, errors

test/redis_client/test_cluster.rb

@@ -29,7 +29,7 @@ def test_inspect
       def test_call
         assert_raises(ArgumentError) { @client.call }
 
-        (0..9).each do |i|
+        10.times do |i|
           assert_equal('OK', @client.call('SET', "key#{i}", i), "Case: SET: key#{i}")
           wait_for_replication
           assert_equal(i.to_s, @client.call('GET', "key#{i}"), "Case: GET: key#{i}")
@@ -44,7 +44,7 @@ def test_call
       def test_call_once
         assert_raises(ArgumentError) { @client.call_once }
 
-        (0..9).each do |i|
+        10.times do |i|
           assert_equal('OK', @client.call_once('SET', "key#{i}", i), "Case: SET: key#{i}")
           wait_for_replication
           assert_equal(i.to_s, @client.call_once('GET', "key#{i}"), "Case: GET: key#{i}")
@@ -72,7 +72,7 @@ def test_blocking_call
       def test_scan
         assert_raises(ArgumentError) { @client.scan }
 
-        (0..9).each { |i| @client.call('SET', "key#{i}", i) }
+        10.times { |i| @client.call('SET', "key#{i}", i) }
         wait_for_replication
         want = (0..9).map { |i| "key#{i}" }
         got = []
@@ -81,8 +81,8 @@ def test_scan
       end
 
       def test_sscan
-        (0..9).each do |i|
-          (0..9).each { |j| @client.call('SADD', "key#{i}", "member#{j}") }
+        10.times do |i|
+          10.times { |j| @client.call('SADD', "key#{i}", "member#{j}") }
           wait_for_replication
           want = (0..9).map { |j| "member#{j}" }
           got = []
@@ -92,8 +92,8 @@ def test_sscan
       end
 
       def test_hscan
-        (0..9).each do |i|
-          (0..9).each { |j| @client.call('HSET', "key#{i}", "field#{j}", j) }
+        10.times do |i|
+          10.times { |j| @client.call('HSET', "key#{i}", "field#{j}", j) }
           wait_for_replication
           want = (0..9).map { |j| "field#{j}" }
           got = []
@@ -103,8 +103,8 @@ def test_hscan
       end
 
       def test_zscan
-        (0..9).each do |i|
-          (0..9).each { |j| @client.call('ZADD', "key#{i}", j, "member#{j}") }
+        10.times do |i|
+          10.times { |j| @client.call('ZADD', "key#{i}", j, "member#{j}") }
           wait_for_replication
           want = (0..9).map { |j| "member#{j}" }
           got = []
@@ -118,18 +118,18 @@ def test_pipelined
 
         want = (0..9).map { 'OK' } + (1..3).to_a + %w[PONG] + (0..9).map(&:to_s) + [%w[list 2]]
         got = @client.pipelined do |pipeline|
-          (0..9).each { |i| pipeline.call('SET', "string#{i}", i) }
-          (0..2).each { |i| pipeline.call('RPUSH', 'list', i) }
+          10.times { |i| pipeline.call('SET', "string#{i}", i) }
+          3.times { |i| pipeline.call('RPUSH', 'list', i) }
           pipeline.call_once('PING')
-          (0..9).each { |i| pipeline.call('GET', "string#{i}") }
+          10.times { |i| pipeline.call('GET', "string#{i}") }
           pipeline.blocking_call(0.2, 'BRPOP', 'list', '0.1')
         end
 
         assert_equal(want, got)
       end
 
       def test_pubsub
-        (0..9).each do |i|
+        10.times do |i|
           pubsub = @client.pubsub
           pubsub.call('SUBSCRIBE', "channel#{i}")
           assert_equal(['subscribe', "channel#{i}", 1], pubsub.next_event(0.1))
@@ -154,7 +154,7 @@ def test_close
       end
 
       def test_dedicated_commands
-        (0..9).each { |i| @client.call('SET', "key#{i}", i) }
+        10.times { |i| @client.call('SET', "key#{i}", i) }
         wait_for_replication
         [
           { command: %w[ACL HELP], is_a: Array },