fix: use fiber local storage instead of thread local storage (#185)

Author: supercaracal

lib/redis_client/cluster/node.rb

@@ -85,24 +85,24 @@ def load_info(options, **kwargs) # rubocop:disable Metrics/AbcSize, Metrics/Cycl
           startup_nodes.each_slice(MAX_THREADS).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 * chuncked_idx) + idx) do |cli, i|
-                Thread.current.thread_variable_set(:index, i)
+                Thread.current[:index] = i
                 reply = cli.call('CLUSTER', 'NODES')
-                Thread.current.thread_variable_set(:info, parse_cluster_node_reply(reply))
+                Thread.current[:info] = parse_cluster_node_reply(reply)
               rescue StandardError => e
-                Thread.current.thread_variable_set(:error, e)
+                Thread.current[:error] = e
               ensure
                 cli&.close
               end
             end
 
             threads.each do |t|
               t.join
-              if t.thread_variable?(:info)
+              if t.key?(:info)
                 node_info_list ||= Array.new(startup_size)
-                node_info_list[t.thread_variable_get(:index)] = t.thread_variable_get(:info)
-              elsif t.thread_variable?(:error)
+                node_info_list[t[:index]] = t[:info]
+              elsif t.key?(:error)
                 errors ||= Array.new(startup_size)
-                errors[t.thread_variable_get(:index)] = t.thread_variable_get(:error)
+                errors[t[:index]] = t[:error]
               end
             end
           end
@@ -303,22 +303,22 @@ def try_map(clients) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComple
         clients.each_slice(MAX_THREADS) do |chuncked_clients|
           threads = chuncked_clients.map do |k, v|
             Thread.new(k, v) do |node_key, client|
-              Thread.current.thread_variable_set(:node_key, node_key)
+              Thread.current[:node_key] = node_key
               reply = yield(node_key, client)
-              Thread.current.thread_variable_set(:result, reply)
+              Thread.current[:result] = reply
             rescue StandardError => e
-              Thread.current.thread_variable_set(:error, e)
+              Thread.current[:error] = e
             end
           end
 
           threads.each do |t|
             t.join
-            if t.thread_variable?(:result)
+            if t.key?(:result)
               results ||= {}
-              results[t.thread_variable_get(:node_key)] = t.thread_variable_get(:result)
-            elsif t.thread_variable?(:error)
+              results[t[:node_key]] = t[:result]
+            elsif t.key?(:error)
               errors ||= {}
-              errors[t.thread_variable_get(:node_key)] = t.thread_variable_get(:error)
+              errors[t[:node_key]] = t[:error]
             end
           end
         end

lib/redis_client/cluster/node/latency_replica.rb

@@ -43,7 +43,7 @@ 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.current.thread_variable_set(:node_key, node_key)
+                Thread.current[:node_key] = node_key
 
                 min = DUMMY_LATENCY_NSEC
                 MEASURE_ATTEMPT_COUNT.times do
@@ -53,15 +53,15 @@ def measure_latencies(clients) # rubocop:disable Metrics/AbcSize
                   min = duration if duration < min
                 end
 
-                Thread.current.thread_variable_set(:latency, min)
+                Thread.current[:latency] = min
               rescue StandardError
-                Thread.current.thread_variable_set(:latency, DUMMY_LATENCY_NSEC)
+                Thread.current[:latency] = DUMMY_LATENCY_NSEC
               end
             end
 
             threads.each do |t|
               t.join
-              acc[t.thread_variable_get(:node_key)] = t.thread_variable_get(:latency)
+              acc[t[:node_key]] = t[:latency]
             end
           end
         end

lib/redis_client/cluster/pipeline.rb

@@ -150,34 +150,34 @@ def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Met
         @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.current.thread_variable_set(:node_key, nk)
+              Thread.current[: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(:replies, replies)
+              Thread.current[:replies] = replies
             rescue ::RedisClient::Cluster::Pipeline::RedirectionNeeded => e
-              Thread.current.thread_variable_set(:redirection_needed, e)
+              Thread.current[:redirection_needed] = e
             rescue StandardError => e
-              Thread.current.thread_variable_set(:error, e)
+              Thread.current[:error] = e
             end
           end
 
           threads.each(&:join)
           threads.each do |t|
-            if t.thread_variable?(:replies)
+            if t.key?(:replies)
               all_replies ||= Array.new(@size)
-              @pipelines[t.thread_variable_get(:node_key)]
+              @pipelines[t[:node_key]]
                 .outer_indices
-                .each_with_index { |outer, inner| all_replies[outer] = t.thread_variable_get(:replies)[inner] }
-            elsif t.thread_variable?(:redirection_needed)
+                .each_with_index { |outer, inner| all_replies[outer] = t[:replies][inner] }
+            elsif t.key?(:redirection_needed)
               all_replies ||= Array.new(@size)
-              pipeline = @pipelines[t.thread_variable_get(:node_key)]
-              err = t.thread_variable_get(:redirection_needed)
+              pipeline = @pipelines[t[:node_key]]
+              err = t[:redirection_needed]
               err.indices.each { |i| err.replies[i] = handle_redirection(err.replies[i], pipeline, i) }
               pipeline.outer_indices.each_with_index { |outer, inner| all_replies[outer] = err.replies[inner] }
-            elsif t.thread_variable?(:error)
+            elsif t.key?(:error)
               errors ||= {}
-              errors[t.thread_variable_get(:node_key)] = t.thread_variable_get(:error)
+              errors[t[:node_key]] = t[:error]
             end
           end
         end

test/redis_client/cluster/test_normalized_cmd_name.rb

@@ -67,21 +67,21 @@ def test_thread_safety
 
         threads = attempts.each_with_index.map do |_, i|
           Thread.new do
-            Thread.current.thread_variable_set(:index, i)
+            Thread.current[:index] = i
             got = if i.even?
                     @subject.get_by_command(%w[SET foo bar])
                   else
                     @subject.clear ? 'set' : 'clear failed'
                   end
-            Thread.current.thread_variable_set(:got, got)
+            Thread.current[:got] = got
           rescue StandardError => e
-            Thread.current.thread_variable_set(:got, "#{e.class.name}: #{e.message}")
+            Thread.current[:got] = "#{e.class.name}: #{e.message}"
           end
         end
 
         threads.each do |t|
           t.join
-          attempts[t.thread_variable_get(:index)] = t.thread_variable_get(:got)
+          attempts[t[:index]] = t[:got]
         end
 
         attempts.each { |got| assert_equal('set', got) }

test/test_concurrency.rb

@@ -56,12 +56,12 @@ def test_threading
         ATTEMPTS.times { MAX_THREADS.times { |i| @client.call('INCR', "key#{i}") } }
         ATTEMPTS.times { MAX_THREADS.times { |i| @client.call('DECR', "key#{i}") } }
       rescue StandardError => e
-        Thread.current.thread_variable_set(:error, e)
+        Thread.current[:error] = e
       end
     end
 
     threads.each(&:join)
-    threads.each { |t| assert_nil(t.thread_variable_get(:error)) }
+    threads.each { |t| assert_nil(t[:error]) }
     MAX_THREADS.times { |i| assert_equal(WANT, @client.call('GET', "key#{i}")) }
   end
 
@@ -71,12 +71,12 @@ def test_threading_with_pipelining
         @client.pipelined { |pi| ATTEMPTS.times { MAX_THREADS.times { |i| pi.call('INCR', "key#{i}") } } }
         @client.pipelined { |pi| ATTEMPTS.times { MAX_THREADS.times { |i| pi.call('DECR', "key#{i}") } } }
       rescue StandardError => e
-        Thread.current.thread_variable_set(:error, e)
+        Thread.current[:error] = e
       end
     end
 
     threads.each(&:join)
-    threads.each { |t| assert_nil(t.thread_variable_get(:error)) }
+    threads.each { |t| assert_nil(t[:error]) }
     MAX_THREADS.times { |i| assert_equal(WANT, @client.call('GET', "key#{i}")) }
   end