Class: Homebrew::DownloadQueue Private

Inherits:
Object
  • Object
show all
Includes:
Utils::Output::Mixin
Defined in:
download_queue.rb

Overview

This class is part of a private API. This class may only be used in the Homebrew/brew repository. Third parties should avoid using this class if possible, as it may be removed or changed without warning.

Manages a queue of concurrent downloads with cooperative cancellation support.

Defined Under Namespace

Classes: Spinner

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Utils::Output::Mixin

#issue_reporting_message, #odebug, #odeprecated, #odie, #odisabled, #ofail, #oh1, #oh1_title, #ohai, #ohai_title, #onoe, #opoo, #opoo_outside_github_actions, #opoo_without_github_actions_annotation, #pretty_deprecated, #pretty_disabled, #pretty_duration, #pretty_install_status, #pretty_installed, #pretty_uninstalled, #pretty_unmarked, #pretty_upgradable, #pretty_warning

Constructor Details

#initialize(retries: 1, force: false, pour: false) ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Parameters:

  • retries (Integer) (defaults to: 1)
  • force (Boolean) (defaults to: false)
  • pour (Boolean) (defaults to: false) 


23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'download_queue.rb', line 23

def initialize(retries: 1, force: false, pour: false)
  @concurrency = T.let(EnvConfig.download_concurrency, Integer)
  @quiet = T.let(@concurrency > 1, T::Boolean)
  @tries = T.let(retries + 1, Integer)
  @force = force
  @pour = pour
  @pool = T.let(Concurrent::FixedThreadPool.new(concurrency), Concurrent::FixedThreadPool)
  @tty = T.let($stdout.tty?, T::Boolean)
  @dumb_tty = T.let(ENV["TERM"] == "dumb", T::Boolean)
  @spinner = T.let(nil, T.nilable(Spinner))
  @symlink_targets = T.let({}, T::Hash[Pathname, T::Set[Downloadable]])
  @downloads_by_location = T.let({}, T::Hash[Pathname, Concurrent::Promises::Future])
  @staged_downloads_by_location = T.let({}, T::Hash[Pathname, Concurrent::Promises::Future])
  @cancelled = T.let(Concurrent::AtomicBoolean.new(false), Concurrent::AtomicBoolean)
  @active_threads = T.let(Concurrent::Set.new, Concurrent::Set)
  @failed_downloads = T.let([], T::Array[Downloadable])
  @deferred_failure_messages = T.let([], T::Array[T.proc.void])
end

Instance Attribute Details

#failed_downloads ⇒ Array<Downloadable> (readonly)

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

The downloadables that failed in the last fetch, so callers can retry or skip only the packages that were actually affected.

Returns:



326
327
328
 
# File 'download_queue.rb', line 326

def failed_downloads
  @failed_downloads
end

Instance Method Details

#downloads ⇒ Hash{Downloadable => Concurrent::Promises::Future}

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

Returns:



346
347
348
 
# File 'download_queue.rb', line 346

def downloads
  @downloads ||= T.let({}, T.nilable(T::Hash[Downloadable, Concurrent::Promises::Future]))
end

#enqueue(downloadable, check_attestation: false, stage: pour) ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

This method returns an undefined value.

Parameters:

  • downloadable (Downloadable)
  • check_attestation (Boolean) (defaults to: false)
  • stage (Boolean) (defaults to: pour) 


49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'download_queue.rb', line 49

def enqueue(downloadable, check_attestation: false, stage: pour)
  @cancelled.make_false
  cached_location = downloadable.cached_download

  @symlink_targets[cached_location] ||= Set.new
  targets = @symlink_targets.fetch(cached_location)
  targets << downloadable

  download = @downloads_by_location[cached_location] ||= Concurrent::Promises.future_on(
    pool, RetryableDownload.new(downloadable, tries:),
    @cancelled, force, quiet, check_attestation
  ) do |download, cancelled, force, quiet, check_attestation|
    with_active_thread do
      raise CancelledDownloadError if cancelled.true?

      download.clear_cache if force
      if !force && downloadable.downloaded_and_valid?
        check_bottle_attestation(downloadable, check_attestation:)
        create_symlinks_for_shared_download(cached_location)
        next cached_location
      end

      downloaded_path = download.fetch(quiet:)
      raise CancelledDownloadError if cancelled.true?

      check_bottle_attestation(downloadable, check_attestation:)
      if downloaded_path != cached_location
        @symlink_targets[downloaded_path] ||= Set.new
        @symlink_targets.fetch(downloaded_path).merge(@symlink_targets.fetch(cached_location, Set.new))
      end
      create_symlinks_for_shared_download(downloaded_path)
      downloaded_path
    end
  end

  downloads[downloadable] = if stage
    stage_download = lambda do
      download.then_on(
        pool, downloadable, pour, @cancelled
      ) do |downloaded_path, queued_downloadable, queue_pour, cancelled|
        with_active_thread do
          raise CancelledDownloadError if cancelled.true?

          if queued_downloadable.stage_from_download_queue?(downloaded_path, pour: queue_pour)
            queued_downloadable.extracting!
            queued_downloadable.stage_from_download_queue(downloaded_path, pour: queue_pour)
            queued_downloadable.downloaded!
          end
          downloaded_path
        end
      end
    end

    if (staged_location = downloadable.staged_path_from_download_queue)
      @staged_downloads_by_location[staged_location] ||= stage_download.call
    else
      stage_download.call
    end
  else
    download
  end
end

#fetch(only: nil, heading: nil, allow_failures: false) ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

This method returns an undefined value.

Waits for and reports queued downloads. With only:, limits that to downloadables of the given class, leaving the rest enqueued and unreported for a later fetch, e.g. so dependency resolution can wait on bottle manifests without reporting in-flight bottles before their downloads heading has been printed. A heading: is printed only when there is something to report, so every report gets a heading and empty fetches stay silent. With allow_failures:, failures are still reported with a ✘ line but neither raise nor mark the fetch or run as failed, for metadata prefetches such as the bottle manifest of a version whose bottle has not been published yet, where dependency resolution just falls back to a full install; known-bad cached files from checksum mismatches are still removed.

Parameters:

  • only (T::Class[Downloadable], nil) (defaults to: nil)
  • heading (String, nil) (defaults to: nil)
  • allow_failures (Boolean) (defaults to: false) 


128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
 
# File 'download_queue.rb', line 128

def fetch(only: nil, heading: nil, allow_failures: false)
  @failed_downloads = []
  @deferred_failure_messages = []
  context_before_fetch = Context.current
  fetchable_downloads = if only
    downloads.select { |downloadable, _| downloadable.is_a?(only) }
  else
    downloads
  end
  return if fetchable_downloads.empty?

  if heading
    if tty
      oh1 heading, truncate: false
      $stdout.flush
    else
      # Keep the heading off parsed stdout (e.g. `brew info --json | jq`)
      # and on the same stream as the non-TTY report lines below.
      $stderr.puts oh1_title(heading, truncate: false)
    end
  end

  if concurrency == 1
    fetchable_downloads.each do |downloadable, promise|
      promise.wait!
    rescue CancelledDownloadError
      next
    rescue ChecksumMismatchError => e
      if allow_failures
        report_tolerated_failure(downloadable)
        # Remove the known-bad download so it cannot be reused.
        unlink_cached_download(downloadable)
        next
      end

      @failed_downloads << downloadable
      ofail "#{downloadable.download_queue_type} reports different checksum: #{e.expected}"
    rescue
      raise unless allow_failures

      report_tolerated_failure(downloadable)
    end
  else
    message_length_max = fetchable_downloads.keys.map do |download|
      download.download_queue_message.length
    end.max || 0
    remaining_downloads = fetchable_downloads.dup.to_a
    previous_pending_line_count = 0
    max_lines = [concurrency, Tty.height].min

    resolution = Concurrent::Event.new
    fetchable_downloads.each_value { |future| future.on_resolution! { resolution.set } }

    begin
      stdout_print_and_flush_if_tty Tty.hide_cursor

      output_message = lambda do |downloadable, future, last|
        status = status_from_future(future)
        exception = future.reason if future.rejected?
        next 1 if exception.is_a?(CancelledDownloadError)

        message = downloadable.download_queue_message
        if tty_with_cursor_move_support?
          message = message_with_progress(downloadable, future, message, message_length_max)
          stdout_print_and_flush "#{status} #{message}#{"\n" unless last}"
        elsif status
          $stderr.puts "#{status} #{message}"
        end

        if future.rejected? && allow_failures
          # Remove known-bad downloads so they cannot be reused, while
          # staying non-fatal for tolerated metadata prefetches.
          unlink_cached_download(downloadable) if exception.is_a?(ChecksumMismatchError)
        elsif future.rejected?
          if exception.is_a?(ChecksumMismatchError)
            @failed_downloads << downloadable
            actual = Digest::SHA256.file(downloadable.cached_download).hexdigest
            actual_message, expected_message = align_checksum_mismatch_message(downloadable.download_queue_type)

            report_or_defer_failure do
              ofail "#{actual_message} #{exception.expected}"
              puts "#{expected_message} #{actual}"
            end
          elsif exception.is_a?(CannotInstallFormulaError)
            unlink_cached_download(downloadable)
            raise exception
          elsif bottle_manifest_error?(downloadable, exception)
            # Fatal: unlike a missing blob (which then fails to stage), a
            # stale blob would still pour without the manifest tab that
            # drives relocation, so abort rather than stage a broken keg.
            raise exception
          else
            failure_message = if exception.is_a?(DownloadError) && exception.cause.is_a?(ErrorDuringExecution)
              cause = T.cast(exception.cause, ErrorDuringExecution)
              if (stderr_output = cause.stderr.presence)
                "#{stderr_output}#{cause.message}"
              else
                cause.message
              end
            else
              future.reason.to_s
            end
            @failed_downloads << downloadable
            report_or_defer_failure { ofail failure_message }
          end
        end

        1
      end

      until remaining_downloads.empty?
        begin
          stdout_print_and_flush_if_tty Tty.begin_synchronized_update

          finished_states = [:fulfilled, :rejected]

          finished_downloads, remaining_downloads = remaining_downloads.partition do |_, future|
            finished_states.include?(future.state)
          end

          finished_downloads.each do |downloadable, future|
            previous_pending_line_count -= 1
            output_message.call(downloadable, future, false)
            stdout_print_and_flush_if_tty Tty.clear_to_end
          end

          previous_pending_line_count = 0
          remaining_downloads.each_with_index do |(downloadable, future), i|
            break if previous_pending_line_count >= max_lines

            last = i == max_lines - 1 || i == remaining_downloads.count - 1
            previous_pending_line_count += output_message.call(downloadable, future, last)
            stdout_print_and_flush_if_tty Tty.clear_to_end
          end

          if previous_pending_line_count.positive?
            if (previous_pending_line_count - 1).zero?
              stdout_print_and_flush_if_tty Tty.move_cursor_beginning
            else
              stdout_print_and_flush_if_tty Tty.move_cursor_up_beginning(previous_pending_line_count - 1)
            end
          end

          stdout_print_and_flush_if_tty Tty.end_synchronized_update

          next if remaining_downloads.empty?

          resolution.reset
          # A download may resolve between the partition above and this
          # reset: re-check before waiting to avoid a lost wakeup.
          next if remaining_downloads.any? { |_, future| finished_states.include?(future.state) }

          # Wake as soon as any download resolves; the timeout only sets
          # the redraw cadence for spinner and progress bars on TTYs.
          resolution.wait(tty_with_cursor_move_support? ? 0.05 : 1)
        # `Interrupt` inherits from `Exception`, so rescue it to restore the TTY.
        rescue Exception # rubocop:disable Lint/RescueException
          if previous_pending_line_count.positive?
            stdout_print_and_flush_if_tty Tty.move_cursor_down(previous_pending_line_count - 1)
          end

          raise
        end
      end
    ensure
      stdout_print_and_flush_if_tty Tty.end_synchronized_update
      stdout_print_and_flush_if_tty Tty.show_cursor
      @deferred_failure_messages.each(&:call)
    end
  end
# `Interrupt` inherits from `Exception`, so rescue it to cancel active workers
# even when it arrives before fetch setup completes.
rescue Exception # rubocop:disable Lint/RescueException
  cancel
  raise
ensure
  # Restore the pre-parallel fetch context to avoid quiet state bleeding out
  # from threads, and clear queue state even when a fatal download error
  # aborts the fetch above.
  Context.current = context_before_fetch if context_before_fetch

  if only
    # Keep unfetched downloads (and their location dedup entries) queued
    # for the next fetch.
    fetchable_downloads.each_key { |downloadable| downloads.delete(downloadable) }
    fetched_staging = fetchable_downloads.each_value.to_set
    @staged_downloads_by_location.delete_if { |_, future| fetched_staging.include?(future) }
  else
    downloads.clear
    @downloads_by_location.clear
    @staged_downloads_by_location.clear
    @symlink_targets.clear
  end
end

#shutdown ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

This method returns an undefined value.



340
341
342
343
 
# File 'download_queue.rb', line 340

def shutdown
  pool.shutdown
  pool.wait_for_termination
end

#stdout_print_and_flush(message) ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

This method returns an undefined value.

Parameters:



334
335
336
337
 
# File 'download_queue.rb', line 334

def stdout_print_and_flush(message)
  $stdout.print(message)
  $stdout.flush
end

#stdout_print_and_flush_if_tty(message) ⇒ void

This method is part of a private API. This method may only be used in the Homebrew/brew repository. Third parties should avoid using this method if possible, as it may be removed or changed without warning.

This method returns an undefined value.

Parameters:



329
330
331
 
# File 'download_queue.rb', line 329

def stdout_print_and_flush_if_tty(message)
  stdout_print_and_flush(message) if tty_with_cursor_move_support?
end