Merge pull request rails#48988 from jdelStrother/redis-distr-pipeline

Add Redis::Distributed support for pipelined write_multi

activesupport/lib/active_support/cache/redis_cache_store.rb

@@ -153,7 +153,6 @@ def initialize(error_handler: DEFAULT_ERROR_HANDLER, **redis_options)
 
         @max_key_bytesize = MAX_KEY_BYTESIZE
         @error_handler = error_handler
-        @supports_pipelining = true
 
         super(universal_options)
       end
@@ -294,15 +293,16 @@ def stats
       end
 
       private
-        def pipelined(&block)
-          if @supports_pipelining
-            redis.then { |c| c.pipelined(&block) }
-          else
-            redis.then(&block)
-          end
-        rescue Redis::Distributed::CannotDistribute
-          @supports_pipelining = false
-          retry
+        def pipeline_entries(entries, &block)
+          redis.then { |c|
+            if c.is_a?(Redis::Distributed)
+              entries.group_by { |k, _v| c.node_for(k) }.each do |node, sub_entries|
+                node.pipelined { |pipe| yield(pipe, sub_entries) }
+              end
+            else
+              c.pipelined { |pipe| yield(pipe, entries) }
+            end
+          }
         end
 
         # Store provider interface:
@@ -387,10 +387,10 @@ def write_multi_entries(entries, expires_in: nil, race_condition_ttl: nil, **opt
           return if entries.empty?
 
           failsafe :write_multi_entries do
-            pipelined do |pipeline|
+            pipeline_entries(entries) do |pipeline, sharded_entries|
               options = options.dup
               options[:pipeline] = pipeline
-              entries.each do |key, entry|
+              sharded_entries.each do |key, entry|
                 write_entry key, entry, **options
               end
             end