Implement concurrent downloads in brew fetch.

Library/Homebrew/Gemfile

@@ -78,6 +78,7 @@ end
 
 # vendored gems (no group)
 gem "addressable"
+gem "concurrent-ruby"
 gem "patchelf"
 gem "plist"
 gem "ruby-macho"

Library/Homebrew/Gemfile.lock

@@ -9,6 +9,7 @@ GEM
     coderay (1.1.3)
     commander (5.0.0)
       highline (~> 3.0.0)
+    concurrent-ruby (1.3.4)
     diff-lcs (1.5.1)
     docile (1.4.1)
     elftools (1.3.1)
@@ -161,6 +162,7 @@ PLATFORMS
 
 DEPENDENCIES
   addressable
+  concurrent-ruby
   json_schemer
   kramdown
   method_source

Library/Homebrew/api/download.rb

@@ -12,7 +12,9 @@ def symlink_location
       end
     end
 
-    class Download < Downloadable
+    class Download
+      include Downloadable
+
       sig {
         params(
           url:      String,
@@ -29,14 +31,29 @@ def initialize(url, checksum, mirrors: [], cache: nil)
         @cache = cache
       end
 
+      sig { override.returns(API::DownloadStrategy) }
+      def downloader
+        T.cast(super, API::DownloadStrategy)
+      end
+
+      sig { override.returns(String) }
+      def name
+        download_name
+      end
+
+      sig { override.returns(String) }
+      def download_type
+        "API source"
+      end
+
       sig { override.returns(Pathname) }
       def cache
         @cache || super
       end
 
       sig { returns(Pathname) }
       def symlink_location
-        T.cast(downloader, API::DownloadStrategy).symlink_location
+        downloader.symlink_location
       end
     end
   end

Library/Homebrew/cask/download.rb

@@ -8,7 +8,9 @@
 
 module Cask
   # A download corresponding to a {Cask}.
-  class Download < ::Downloadable
+  class Download
+    include Downloadable
+
     include Context
 
     attr_reader :cask
@@ -20,6 +22,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 +95,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,7 @@
 require "formula"
 require "fetch"
 require "cask/download"
+require "retryable_download"
 
 module Homebrew
   module Cmd
@@ -25,6 +26,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,6 +69,49 @@ 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
+
+      class Spinner
+        FRAMES = [
+          "⠋",
+          "⠙",
+          "⠚",
+          "⠞",
+          "⠖",
+          "⠦",
+          "⠴",
+          "⠲",
+          "⠳",
+          "⠓",
+        ].freeze
+
+        sig { void }
+        def initialize
+          @start = Time.now
+          @i = 0
+        end
+
+        sig { returns(String) }
+        def to_s
+          now = Time.now
+          if @start + 0.1 < now
+            @start = now
+            @i = (@i + 1) % FRAMES.count
+          end
+
+          FRAMES.fetch(@i)
+        end
+      end
+
       sig { override.void }
       def run
         Formulary.enable_factory_cache!
@@ -125,13 +170,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 +189,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 +218,115 @@ 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
+        if concurrency == 1
+          downloads.each do |downloadable, promise|
+            promise.wait!
+          rescue ChecksumMismatchError => e
+            opoo "#{downloadable.download_type.capitalize} reports different checksum: #{e.expected}"
+            Homebrew.failed = true if downloadable.is_a?(Resource::Patch)
+          end
+        else
+          spinner = Spinner.new
+          remaining_downloads = downloads.dup
+          previous_pending_line_count = 0
+
+          begin
+            $stdout.print Tty.hide_cursor
+            $stdout.flush
+
+            output_message = lambda do |downloadable, future|
+              status = case future.state
+              when :fulfilled
+                "#{Tty.green}✔︎#{Tty.reset}"
+              when :rejected
+                "#{Tty.red}✘#{Tty.reset}"
+              when :pending, :processing
+                "#{Tty.blue}#{spinner}#{Tty.reset}"
+              else
+                raise future.state.to_s
+              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
+              message = "#{downloadable.download_type.capitalize} #{downloadable.name}"
+              $stdout.puts "#{status} #{message}"
+              $stdout.flush
 
-      def fetch_formula(formula)
-        fetch_fetchable(formula)
-      rescue ChecksumMismatchError => e
-        retry if retry_fetch?(formula)
-        opoo "Formula reports different sha256: #{e.expected}"
-      end
+              if future.rejected? && (e = future.reason).is_a?(ChecksumMismatchError)
+                opoo "#{downloadable.download_type.capitalize} reports different checksum: #{e.expected}"
+                Homebrew.failed = true if downloadable.is_a?(Resource::Patch)
+                next 2
+              end
 
-      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
+              1
+            end
 
-      def fetch_patch(patch)
-        fetch_fetchable(patch)
-      rescue ChecksumMismatchError => e
-        opoo "Patch reports different sha256: #{e.expected}"
-        Homebrew.failed = true
-      end
+            until remaining_downloads.empty?
+              begin
+                finished_states = [:fulfilled, :rejected]
 
-      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")
+                finished_downloads, remaining_downloads = remaining_downloads.partition do |_, future|
+                  finished_states.include?(future.state)
+                end
 
-          ohai "Retrying download in #{wait}s... (#{remaining} #{what} left)"
-          sleep wait
+                finished_downloads.each do |downloadable, future|
+                  previous_pending_line_count -= 1
+                  $stdout.print Tty.clear_to_end
+                  $stdout.flush
+                  output_message.call(downloadable, future)
+                end
 
-          formula.clear_cache
-          @fetch_tries[formula] += 1
-          true
-        else
-          Homebrew.failed = true
-          false
-        end
-      end
+                previous_pending_line_count = 0
+                remaining_downloads.each do |downloadable, future|
+                  # FIXME: Allow printing full terminal height.
+                  break if previous_pending_line_count >= [concurrency, (Tty.height - 1)].min
+
+                  $stdout.print Tty.clear_to_end
+                  $stdout.flush
+                  previous_pending_line_count += output_message.call(downloadable, future)
+                end
+
+                if previous_pending_line_count.positive?
+                  $stdout.print Tty.move_cursor_up_beginning(previous_pending_line_count)
+                  $stdout.flush
+                end
 
-      def fetch_fetchable(formula)
-        formula.clear_cache if args.force?
+                sleep 0.05
+              rescue Interrupt
+                remaining_downloads.each do |_, future|
+                  # FIXME: Implement cancellation of running downloads.
+                end
 
-        already_fetched = formula.cached_download.exist?
+                if previous_pending_line_count.positive?
+                  $stdout.print Tty.move_cursor_down(previous_pending_line_count - 1)
+                  $stdout.flush
+                end
 
-        begin
-          download = formula.fetch(verify_download_integrity: false)
-        rescue DownloadError
-          retry if retry_fetch?(formula)
-          raise
+                raise
+              end
+            end
+          ensure
+            $stdout.print Tty.show_cursor
+            $stdout.flush
+          end
         end
 
-        return unless download.file?
+        download_queue.shutdown
+      end
 
-        puts "Downloaded to: #{download}" unless already_fetched
-        puts "SHA256: #{download.sha256}"
+      private
+
+      def downloads
+        @downloads ||= {}
+      end
 
-        formula.verify_download_integrity(download)
+      def fetch_downloadable(downloadable)
+        tries = args.retry? ? {} : { tries: 1 }
+        downloads[downloadable] ||= download_queue.enqueue(RetryableDownload.new(downloadable, **tries), force: args.force?)
       end
     end
   end

Library/Homebrew/cmd/install.rb

@@ -207,7 +207,6 @@ def run
         end
 
         if casks.any?
-
           if args.dry_run?
             if (casks_to_install = casks.reject(&:installed?).presence)
               ohai "Would install #{::Utils.pluralize("cask", casks_to_install.count, include_count: true)}:"