Concurrent uploading in mirror services

Author: SGospodinov

lib/active_storage/async_uploader.rb

@@ -0,0 +1,34 @@
+require "concurrent/promise"
+
+class ActiveStorage::AsyncUploader
+  class << self
+    def result uploaders
+      promises = uploaders.map(&:promise)
+      Concurrent::Promise.zip(*promises).value!
+    end
+  end
+
+  attr_reader :promise
+
+  def initialize service, key, checksum: nil
+    @data = ""
+    @eof = false
+    @promise = Concurrent::Promise.execute do
+      until eof? do; end
+      service.upload key, StringIO.new(@data), checksum: checksum
+    end
+  end
+
+  def eof?
+    @eof
+  end
+
+  def write chunk
+    @data << chunk
+    @eof = false
+  end
+
+  def close
+    @eof = true
+  end
+end

lib/active_storage/service/mirror_service.rb

@@ -1,4 +1,5 @@
 require "active_support/core_ext/module/delegation"
+require "active_storage/async_uploader"
 require "concurrent/promise"
 
 class ActiveStorage::Service::MirrorService < ActiveStorage::Service
@@ -18,24 +19,29 @@ def initialize(primary:, mirrors:)
   end
 
   def upload(key, io, checksum: nil)
-    each_service.collect do |service|
-      service.upload key, io.tap(&:rewind), checksum: checksum
+    uploaders = each_service.collect do |service|
+      ActiveStorage::AsyncUploader.new(service, key, checksum: checksum)
     end
+    io.rewind
+    while chunk = io.read(1024)
+      uploaders.each { |uploader| uploader.write(chunk) }
+    end
+    ActiveStorage::AsyncUploader.result(uploaders.each(&:close))
   end
 
   def delete(key)
-    perform_across_services :delete, key
+    perform_async_across_services :delete, key
   end
 
   private
     def each_service(&block)
       [ primary, *mirrors ].each(&block)
     end
 
-    def perform_across_services(method, *args)
-      promises = services.collect do |service|
+    def perform_async_across_services(method, *args)
+      promises = each_service.collect do |service|
         Concurrent::Promise.execute { service.public_send method, *args }
       end
-      Concurrent::Promise.zip(*promises).value
+      Concurrent::Promise.zip(*promises).value!
     end
 end