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 @@
  end
  end
 
- class Download < Downloadable
+ class Download
+ include Downloadable
+
  sig {
  params(
  url: String,
@@ -29,14 +31,29 @@
  @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 @@
  @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 @@
  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 @@
  "(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 @@
  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 @@
  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 @@
 
  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,117 @@
  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
 
- def fetch_fetchable(formula)
- formula.clear_cache if args.force?
+ $stdout.print Tty.clear_to_end
+ $stdout.flush
+ previous_pending_line_count += output_message.call(downloadable, future)
+ end
 
- already_fetched = formula.cached_download.exist?
+ if previous_pending_line_count.positive?
+ $stdout.print Tty.move_cursor_up_beginning(previous_pending_line_count)
+ $stdout.flush
+ end
+
+ sleep 0.05
+ rescue Interrupt
+ remaining_downloads.each do |_, future|
+ # FIXME: Implement cancellation of running downloads.
+ end
 
- begin
- download = formula.fetch(verify_download_integrity: false)
- rescue DownloadError
- retry if retry_fetch?(formula)
- raise
+ if previous_pending_line_count.positive?
+ $stdout.print Tty.move_cursor_down(previous_pending_line_count - 1)
+ $stdout.flush
+ end
+
+ 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
 
- formula.verify_download_integrity(download)
+ def downloads
+ @downloads ||= {}
+ end
+
+ def fetch_downloadable(downloadable)
+ downloads[downloadable] ||= begin
+ tries = args.retry? ? {} : { tries: 1 }
+ download_queue.enqueue(RetryableDownload.new(downloadable, **tries), force: args.force?)
+ end
  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)}:"