Implement parallel downloads.

Library/Homebrew/Gemfile

@@ -81,4 +81,4 @@ gem "plist"
 gem "ruby-macho"
 gem "sorbet-runtime"
 gem "warning"
-gem 'whirly'
+gem "whirly"

Library/Homebrew/cask/download.rb

@@ -8,7 +8,7 @@
 
 module Cask
   # A download corresponding to a {Cask}.
-  class Download < ::Downloadable
+  class Download < Downloadable
     include Context
 
     attr_reader :cask
@@ -20,6 +20,11 @@ def initialize(cask, quarantine: nil)
       @quarantine = quarantine
     end
 
+    sig { override.returns(String) }
+    def name
+      cask.token
+    end
+
     sig { override.returns(T.nilable(::URL)) }
     def url
       return if cask.url.nil?
@@ -88,6 +93,11 @@ def download_name
       cask.token
     end
 
+    sig { override.returns(String) }
+    def download_type
+      "cask"
+    end
+
     private
 
     def quarantine(path)

Library/Homebrew/cmd/fetch.rb

@@ -5,6 +5,8 @@
 require "formula"
 require "fetch"
 require "cask/download"
+require "retryable_download"
+require "whirly"
 
 module Homebrew
   module Cmd
@@ -25,6 +27,7 @@ class FetchCmd < AbstractCommand
                             "(Pass `all` to download for all architectures.)"
         flag   "--bottle-tag=",
                description: "Download a bottle for given tag."
+        flag "--concurrency=", description: "Number of concurrent downloads.", hidden: true
         switch "--HEAD",
                description: "Fetch HEAD version instead of stable version."
         switch "-f", "--force",
@@ -67,10 +70,23 @@ class FetchCmd < AbstractCommand
         named_args [:formula, :cask], min: 1
       end
 
+      def concurrency
+        @concurrency ||= args.concurrency&.to_i || 1
+      end
+
+      def download_queue
+        @download_queue ||= begin
+          require "download_queue"
+          DownloadQueue.new(concurrency)
+        end
+      end
+
       sig { override.void }
       def run
         Formulary.enable_factory_cache!
 
+        Homebrew.install_gem! "concurrent-ruby"
+
         bucket = if args.deps?
           args.named.to_formulae_and_casks.flat_map do |formula_or_cask|
             case formula_or_cask
@@ -125,13 +141,10 @@ def run
                       next
                     end
 
-                    begin
-                      bottle.fetch_tab
-                    rescue DownloadError
-                      retry if retry_fetch?(bottle)
-                      raise
+                    if (manifest_resource = bottle.github_packages_manifest_resource)
+                      fetch_downloadable(manifest_resource)
                     end
-                    fetch_formula(bottle)
+                    fetch_downloadable(bottle)
                   rescue Interrupt
                     raise
                   rescue => e
@@ -147,14 +160,14 @@ def run
 
                 next if fetched_bottle
 
-                fetch_formula(formula)
+                fetch_downloadable(formula.resource)
 
                 formula.resources.each do |r|
-                  fetch_resource(r)
-                  r.patches.each { |p| fetch_patch(p) if p.external? }
+                  fetch_downloadable(r)
+                  r.patches.each { |patch| fetch_downloadable(patch.resource) if patch.external? }
                 end
 
-                formula.patchlist.each { |p| fetch_patch(p) if p.external? }
+                formula.patchlist.each { |patch| fetch_downloadable(patch.resource) if patch.external? }
               end
             end
           else
@@ -176,81 +189,43 @@ def run
                 quarantine = true if quarantine.nil?
 
                 download = Cask::Download.new(cask, quarantine:)
-                fetch_cask(download)
+                fetch_downloadable(download)
               end
             end
           end
         end
-      end
 
-      private
+        downloads.each do |downloadable, promise|
+          message = "#{downloadable.download_type.capitalize} #{downloadable.name}"
+          if concurrency > 1
+            Whirly.start spinner: "arc", status: message
+          else
+            puts message
+          end
 
-      def fetch_resource(resource)
-        puts "Resource: #{resource.name}"
-        fetch_fetchable resource
-      rescue ChecksumMismatchError => e
-        retry if retry_fetch?(resource)
-        opoo "Resource #{resource.name} reports different sha256: #{e.expected}"
-      end
+          promise.wait!
 
-      def fetch_formula(formula)
-        fetch_fetchable(formula)
-      rescue ChecksumMismatchError => e
-        retry if retry_fetch?(formula)
-        opoo "Formula reports different sha256: #{e.expected}"
-      end
+          Whirly.configure stop: "✔︎"
+          Whirly.stop if args.concurrency
+        rescue ChecksumMismatchError => e
+          Whirly.configure stop: "✘"
+          Whirly.stop if args.concurrency
 
-      def fetch_cask(cask_download)
-        fetch_fetchable(cask_download)
-      rescue ChecksumMismatchError => e
-        retry if retry_fetch?(cask_download)
-        opoo "Cask reports different sha256: #{e.expected}"
-      end
+          opoo "#{downloadable.download_type.capitalize} reports different checksum: #{e.expected}"
+          Homebrew.failed = true if downloadable.is_a?(Resource::Patch)
+        end
 
-      def fetch_patch(patch)
-        fetch_fetchable(patch)
-      rescue ChecksumMismatchError => e
-        opoo "Patch reports different sha256: #{e.expected}"
-        Homebrew.failed = true
+        download_queue.shutdown
       end
 
-      def retry_fetch?(formula)
-        @fetch_tries ||= Hash.new { |h, k| h[k] = 1 }
-        if args.retry? && (@fetch_tries[formula] < FETCH_MAX_TRIES)
-          wait = 2 ** @fetch_tries[formula]
-          remaining = FETCH_MAX_TRIES - @fetch_tries[formula]
-          what = Utils.pluralize("tr", remaining, plural: "ies", singular: "y")
-
-          ohai "Retrying download in #{wait}s... (#{remaining} #{what} left)"
-          sleep wait
+      private
 
-          formula.clear_cache
-          @fetch_tries[formula] += 1
-          true
-        else
-          Homebrew.failed = true
-          false
-        end
+      def downloads
+        @downloads ||= {}
       end
 
-      def fetch_fetchable(formula)
-        formula.clear_cache if args.force?
-
-        already_fetched = formula.cached_download.exist?
-
-        begin
-          download = formula.fetch(verify_download_integrity: false)
-        rescue DownloadError
-          retry if retry_fetch?(formula)
-          raise
-        end
-
-        return unless download.file?
-
-        puts "Downloaded to: #{download}" unless already_fetched
-        puts "SHA256: #{download.sha256}"
-
-        formula.verify_download_integrity(download)
+      def fetch_downloadable(downloadable)
+        downloads[downloadable] ||= download_queue.enqueue(RetryableDownload.new(downloadable))
       end
     end
   end

Library/Homebrew/download_queue.rb

@@ -0,0 +1,28 @@
+# typed: true
+# frozen_string_literal: true
+
+require "downloadable"
+require "concurrent"
+
+module Homebrew
+  class DownloadQueue
+    private attr_reader :pool
+
+    sig { params(size: Integer).void }
+    def initialize(size = 1)
+      @pool = Concurrent::FixedThreadPool.new(size)
+    end
+
+    sig { params(downloadable: Downloadable).returns(Concurrent::Promise) }
+    def enqueue(downloadable)
+      Concurrent::Promise.execute(executor: pool) do
+        downloadable.fetch(quiet: pool.max_length > 1)
+      end
+    end
+
+    sig { void }
+    def shutdown
+      pool.shutdown
+    end
+  end
+end

Library/Homebrew/downloadable.rb

@@ -39,6 +39,16 @@ def freeze
     super
   end
 
+  sig { returns(String) }
+  def name
+    ""
+  end
+
+  sig { returns(String) }
+  def download_type
+    self.class.name.split("::").last.gsub(/([[:lower:]])([[:upper:]])/, '\1 \2').downcase
+  end
+
   sig { returns(T::Boolean) }
   def downloaded?
     cached_download.exist?
@@ -78,11 +88,15 @@ def downloader
     end
   end
 
-  sig { params(verify_download_integrity: T::Boolean, timeout: T.nilable(T.any(Integer, Float))).returns(Pathname) }
-  def fetch(verify_download_integrity: true, timeout: nil)
+  sig {
+    params(verify_download_integrity: T::Boolean, timeout: T.nilable(T.any(Integer, Float)),
+           quiet: T::Boolean).returns(Pathname)
+  }
+  def fetch(verify_download_integrity: true, timeout: nil, quiet: false)
     cache.mkpath
 
     begin
+      downloader.quiet! if quiet
       downloader.fetch(timeout:)
     rescue ErrorDuringExecution, CurlDownloadStrategyError => e
       raise DownloadError.new(self, e)

Library/Homebrew/formula.rb

@@ -564,7 +564,7 @@ def synced_with_other_formulae?
     params(name: String, klass: T.class_of(Resource), block: T.nilable(T.proc.bind(Resource).void))
       .returns(T.nilable(Resource))
   }
-  def resource(name, klass = Resource, &block) = active_spec.resource(name, klass, &block)
+  def resource(name = T.unsafe(nil), klass = T.unsafe(nil), &block) = active_spec.resource(*name, *klass, &block)
 
   # Old names for the formula.
   #
@@ -2750,8 +2750,12 @@ def on_system_blocks_exist?
     self.class.on_system_blocks_exist? || @on_system_blocks_exist
   end
 
-  def fetch(verify_download_integrity: true)
-    active_spec.fetch(verify_download_integrity:)
+  sig {
+    params(verify_download_integrity: T::Boolean, timeout: T.nilable(T.any(Integer, Float)),
+           quiet: T::Boolean).returns(Pathname)
+  }
+  def fetch(verify_download_integrity: true, timeout: nil, quiet: false)
+    active_spec.fetch(verify_download_integrity:, timeout:, quiet:)
   end
 
   def verify_download_integrity(filename)

Library/Homebrew/patch.rb

@@ -106,7 +106,7 @@ class ExternalPatch
 
   def initialize(strip, &block)
     @strip    = strip
-    @resource = Resource::PatchResource.new(&block)
+    @resource = Resource::Patch.new(&block)
   end
 
   sig { returns(T::Boolean) }

Library/Homebrew/resource.rb

@@ -140,7 +140,15 @@ def files(*files)
     Partial.new(self, files)
   end
 
-  def fetch(verify_download_integrity: true)
+  sig {
+    override
+      .params(
+        verify_download_integrity: T::Boolean,
+        timeout:                   T.nilable(T.any(Integer, Float)),
+        quiet:                     T.nilable(T::Boolean),
+      ).returns(Pathname)
+  }
+  def fetch(verify_download_integrity: true, timeout: nil, quiet: false)
     fetch_patches
 
     super
@@ -260,6 +268,13 @@ def determine_url_mirrors
     [*extra_urls, *super].uniq
   end
 
+  # A resource for a formula.
+  class Formula < Resource
+    def name
+      super || owner&.name
+    end
+  end
+
   # A resource containing a Go package.
   class Go < Resource
     def stage(target, &block)
@@ -320,7 +335,7 @@ def tab
   end
 
   # A resource containing a patch.
-  class PatchResource < Resource
+  class Patch < Resource
     attr_reader :patch_files
 
     def initialize(&block)