Concurrent uploading in mirror services

Author: SGospodinov

Gemfile

@@ -5,9 +5,9 @@ gemspec
 gem "rake"
 gem "byebug"
 
-gem 'sqlite3'
-gem 'httparty'
-gem 'concurrent-ruby'
+gem "sqlite3"
+gem "httparty"
+gem "concurrent-ruby"
 
 gem "aws-sdk", "~> 2", require: false
 gem "google-cloud-storage", "~> 1.3", require: false

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,7 +1,9 @@
 require "active_support/core_ext/module/delegation"
+require "active_storage/async_uploader"
 require "concurrent/promise"
 
 class ActiveStorage::Service::MirrorService < ActiveStorage::Service
+  CHUNK_SIZE = 1024
   attr_reader :primary, :mirrors
 
   delegate :download, :exist?, :url, to: :primary
@@ -18,24 +20,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(CHUNK_SIZE)
+      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