Separate Bundler download, installation, and metadata concurrency

lib/bundler/fetcher/compact_index.rb

@@ -113,7 +113,7 @@ def in_parallel(inputs, &blk)
       def bundle_worker(func = nil)
         @bundle_worker ||= begin
           worker_name = "Compact Index (#{display_uri.host})"
-          Bundler::Worker.new(Bundler.settings.processor_count, worker_name, func)
+          Bundler::Worker.new(Bundler.settings.metadata_parallelization, worker_name, func)
         end
         @bundle_worker.tap do |worker|
           worker.instance_variable_set(:@func, func) if func

lib/bundler/fetcher/gem_remote_fetcher.rb

@@ -8,7 +8,7 @@ class GemRemoteFetcher < Gem::RemoteFetcher
       def initialize(*)
         super
 
-        @pool_size = Bundler.settings.installation_parallelization
+        @pool_size = Bundler.settings.download_parallelization
         ssl_ca_cert = Bundler.settings[:ssl_ca_cert]
         @cert_files << ssl_ca_cert if ssl_ca_cert
       end

lib/bundler/installer/parallel_installer.rb

@@ -64,9 +64,10 @@ def self.call(*args, **kwargs)
 
     attr_reader :size
 
-    def initialize(installer, all_specs, size, standalone, force, local: false, skip: nil)
+    def initialize(installer, all_specs, size, standalone, force, local: false, skip: nil, download_size: Bundler.settings.download_parallelization)
       @installer = installer
       @size = size
+      @download_size = download_size
       @standalone = standalone
       @force = force
       @local = local
@@ -91,7 +92,7 @@ def call
         Gem::Specification.reset
       end
 
-      if @size > 1
+      if @size > 1 || @download_size > 1
         install_with_worker
       else
         install_serially
@@ -100,6 +101,7 @@ def call
       handle_error if failed_specs.any?
       @specs
     ensure
+      download_worker_pool&.stop
       worker_pool&.stop
     end
 
@@ -170,17 +172,18 @@ def install_serially
       end
     end
 
+    def download_worker_pool
+      @download_worker_pool ||= Bundler::Worker.new(@download_size, "Gem Downloader",
+        ->(spec_install, worker_num) { do_download(spec_install, worker_num) }, response_queue: response_queue)
+    end
+
     def worker_pool
-      @worker_pool ||= Bundler::Worker.new @size, "Parallel Installer", lambda {|spec_install, worker_num|
-        case spec_install.state
-        when :enqueued
-          do_download(spec_install, worker_num)
-        when :installable
-          do_install(spec_install, worker_num)
-        else
-          spec_install
-        end
-      }
+      @worker_pool ||= Bundler::Worker.new(@size, "Parallel Installer",
+        ->(spec_install, worker_num) { do_install(spec_install, worker_num) }, response_queue: response_queue)
+    end
+
+    def response_queue
+      @response_queue ||= Thread::Queue.new
     end
 
     def do_download(spec_install, worker_num)
@@ -218,24 +221,24 @@ def do_install(spec_install, worker_num)
       spec_install
     end
 
-    # Dequeue a spec and save its post-install message and then enqueue the
-    # remaining specs.
-    # Some specs might've had to wait til this spec was installed to be
-    # processed so the call to `enqueue_specs` is important after every
-    # dequeue.
+    # Process one completed download or installation. Downloads can finish
+    # before their dependencies are installed, so check all downloaded specs
+    # after each completion and enqueue any that are now installable.
     def process_specs(installed_specs)
       spec = worker_pool.deq
 
       if spec.installed?
         installed_specs[spec.name] = true
-        return
       elsif spec.failed?
         return
-      elsif spec.ready_to_install?(installed_specs)
-        spec.state = :installable
       end
 
-      worker_pool.enq(spec, priority: spec.enqueue_with_priority?)
+      @specs.each do |candidate|
+        next unless candidate.ready_to_install?(installed_specs)
+
+        candidate.state = :installable
+        worker_pool.enq(candidate, priority: candidate.enqueue_with_priority?)
+      end
     end
 
     def finished_installing?
@@ -274,11 +277,8 @@ def require_tree_for_spec(spec)
       t
     end
 
-    # Keys in the remains hash represent uninstalled gems specs.
-    # We enqueue all gem specs that do not have any dependencies.
-    # Later we call this lambda again to install specs that depended on
-    # previously installed specifications. We continue until all specs
-    # are installed.
+    # Queue every missing spec for download. `process_specs` schedules each
+    # downloaded spec for installation once its dependencies are installed.
     def enqueue_specs(installed_specs)
       @specs.each do |spec|
         if spec.installed?
@@ -287,7 +287,7 @@ def enqueue_specs(installed_specs)
         end
 
         spec.state = :enqueued
-        worker_pool.enq spec
+        download_worker_pool.enq spec
       end
     end
   end

lib/bundler/man/bundle-config.1

@@ -117,6 +117,8 @@ Cooldown filtering depends on the gem server providing a per\-version \fBcreated
 .IP "\(bu" 4
 \fBdisable_version_check\fR (\fBBUNDLE_DISABLE_VERSION_CHECK\fR): Stop Bundler from checking if a newer Bundler version is available on rubygems\.org\.
 .IP "\(bu" 4
+\fBdownload_jobs\fR (\fBBUNDLE_DOWNLOAD_JOBS\fR): The number of gems Bundler can download in parallel\. Defaults to three times the number of installation jobs, capped at eight\.
+.IP "\(bu" 4
 \fBforce_ruby_platform\fR (\fBBUNDLE_FORCE_RUBY_PLATFORM\fR): Ignore the current machine's platform and install only \fBruby\fR platform gems\. As a result, gems with native extensions will be compiled from source\.
 .IP "\(bu" 4
 \fBfrozen\fR (\fBBUNDLE_FROZEN\fR): Disallow any automatic changes to \fBGemfile\.lock\fR\. Bundler commands will be blocked unless the lockfile can be installed exactly as written\. Usually this will happen when changing the \fBGemfile\fR manually and forgetting to update the lockfile through \fBbundle lock\fR or \fBbundle install\fR\.
@@ -135,12 +137,14 @@ Cooldown filtering depends on the gem server providing a per\-version \fBcreated
 .IP "\(bu" 4
 \fBinit_gems_rb\fR (\fBBUNDLE_INIT_GEMS_RB\fR): Generate a \fBgems\.rb\fR instead of a \fBGemfile\fR when running \fBbundle init\fR\.
 .IP "\(bu" 4
-\fBjobs\fR (\fBBUNDLE_JOBS\fR): The number of gems Bundler can download and install in parallel\. Defaults to the number of available processors\.
+\fBjobs\fR (\fBBUNDLE_JOBS\fR): The number of parallel installation jobs\. Defaults to the number of available processors\.
 .IP "\(bu" 4
 \fBlockfile\fR (\fBBUNDLE_LOCKFILE\fR): The path to the lockfile that bundler should use\. By default, Bundler adds \fB\.lock\fR to the end of the \fBgemfile\fR entry\. Can be set to \fBfalse\fR in the Gemfile to disable lockfile creation entirely (see gemfile(5))\.
 .IP "\(bu" 4
 \fBlockfile_checksums\fR (\fBBUNDLE_LOCKFILE_CHECKSUMS\fR): Whether Bundler should include a checksums section in new lockfiles, to protect from compromised gem sources\. Defaults to true\. Bundler's own checksum is only included when its \fB\.gem\fR file is cached, which may not be the case when Bundler is installed as a default gem\.
 .IP "\(bu" 4
+\fBmetadata_jobs\fR (\fBBUNDLE_METADATA_JOBS\fR): The number of compact index metadata requests Bundler can make in parallel\. Defaults to the number of download jobs\.
+.IP "\(bu" 4
 \fBno_build_extension\fR (\fBBUNDLE_NO_BUILD_EXTENSION\fR): Whether Bundler should skip building native extensions during installation\. When set, gems are installed without compiling their C extensions\. To build extensions later, unset this setting and run \fBbundle pristine <gem>\fR\.
 .IP "\(bu" 4
 \fBno_install\fR (\fBBUNDLE_NO_INSTALL\fR): Whether \fBbundle package\fR should skip installing gems\.

lib/bundler/man/bundle-config.1.ronn

@@ -190,6 +190,9 @@ learn more about their operation in [bundle install(1)](bundle-install.1.html).
 * `disable_version_check` (`BUNDLE_DISABLE_VERSION_CHECK`):
    Stop Bundler from checking if a newer Bundler version is available on
    rubygems.org.
+* `download_jobs` (`BUNDLE_DOWNLOAD_JOBS`):
+   The number of gems Bundler can download in parallel. Defaults to three times
+   the number of installation jobs, capped at eight.
 * `force_ruby_platform` (`BUNDLE_FORCE_RUBY_PLATFORM`):
    Ignore the current machine's platform and install only `ruby` platform gems.
    As a result, gems with native extensions will be compiled from source.
@@ -222,7 +225,7 @@ learn more about their operation in [bundle install(1)](bundle-install.1.html).
 * `init_gems_rb` (`BUNDLE_INIT_GEMS_RB`):
    Generate a `gems.rb` instead of a `Gemfile` when running `bundle init`.
 * `jobs` (`BUNDLE_JOBS`):
-   The number of gems Bundler can download and install in parallel.
+   The number of parallel installation jobs.
    Defaults to the number of available processors.
 * `lockfile` (`BUNDLE_LOCKFILE`):
    The path to the lockfile that bundler should use. By default, Bundler adds
@@ -232,6 +235,9 @@ learn more about their operation in [bundle install(1)](bundle-install.1.html).
    Whether Bundler should include a checksums section in new lockfiles, to protect from compromised gem sources. Defaults to true.
    Bundler's own checksum is only included when its `.gem` file is cached, which
    may not be the case when Bundler is installed as a default gem.
+* `metadata_jobs` (`BUNDLE_METADATA_JOBS`):
+   The number of compact index metadata requests Bundler can make in parallel.
+   Defaults to the number of download jobs.
 * `no_build_extension` (`BUNDLE_NO_BUILD_EXTENSION`):
    Whether Bundler should skip building native extensions during installation.
    When set, gems are installed without compiling their C extensions.

lib/bundler/man/bundle-install.1

@@ -26,7 +26,7 @@ Bundler will not call Rubygems' API endpoint (default) but download and cache a
 The location of the Gemfile(5) which Bundler should use\. This defaults to a Gemfile(5) in the current working directory\. In general, Bundler will assume that the location of the Gemfile(5) is also the project's root and will try to find \fBGemfile\.lock\fR and \fBvendor/cache\fR relative to this location\.
 .TP
 \fB\-\-jobs=<number>\fR, \fB\-j=<number>\fR
-The maximum number of parallel download and install jobs\. The default is the number of available processors\.
+The maximum number of parallel installation jobs\. The default is the number of available processors\.
 .TP
 \fB\-\-local\fR
 Do not attempt to connect to \fBrubygems\.org\fR\. Instead, Bundler will use the gems already present in Rubygems' cache or in \fBvendor/cache\fR\. Note that if an appropriate platform\-specific gem exists on \fBrubygems\.org\fR it will not be found\.

lib/bundler/man/bundle-install.1.ronn

@@ -61,8 +61,8 @@ update process below under [CONSERVATIVE UPDATING][].
   to this location.
 
 * `--jobs=<number>`, `-j=<number>`:
-  The maximum number of parallel download and install jobs. The default is the
-  number of available processors.
+  The maximum number of parallel installation jobs. The default is the number
+  of available processors.
 
 * `--local`:
   Do not attempt to connect to `rubygems.org`. Instead, Bundler will use the

lib/bundler/man/bundle-update.1

@@ -39,7 +39,7 @@ Fall back to using the single\-file index of all gems\.
 Use the specified gemfile instead of [\fBGemfile(5)\fR][Gemfile(5)]\.
 .TP
 \fB\-\-jobs=<number>\fR, \fB\-j=<number>\fR
-Specify the number of jobs to run in parallel\. The default is the number of available processors\.
+Specify the number of installation jobs to run in parallel\. The default is the number of available processors\.
 .TP
 \fB\-\-retry=[<number>]\fR
 Retry failed network or git requests for \fInumber\fR times\.

lib/bundler/man/bundle-update.1.ronn

@@ -65,8 +65,8 @@ gem.
   Use the specified gemfile instead of [`Gemfile(5)`][Gemfile(5)].
 
 * `--jobs=<number>`, `-j=<number>`:
-  Specify the number of jobs to run in parallel. The default is the number of
-  available processors.
+  Specify the number of installation jobs to run in parallel. The default is
+  the number of available processors.
 
 * `--retry=[<number>]`:
   Retry failed network or git requests for <number> times.

lib/bundler/settings.rb

@@ -45,7 +45,9 @@ class Settings
 
     NUMBER_KEYS = %w[
       cooldown
+      download_jobs
       jobs
+      metadata_jobs
       redirect
       retry
       ssl_verify_mode
@@ -307,10 +309,18 @@ def app_cache_path
       @app_cache_path ||= self[:cache_path] || "vendor/cache"
     end
 
+    def download_parallelization
+      self[:download_jobs] || [installation_parallelization * 3, 8].min
+    end
+
     def installation_parallelization
       self[:jobs] || processor_count
     end
 
+    def metadata_parallelization
+      self[:metadata_jobs] || download_parallelization
+    end
+
     def validate!
       all.each do |raw_key|
         [@local_config, @env_config, @global_config].each do |settings|

lib/bundler/source/rubygems.rb

@@ -25,6 +25,7 @@ def initialize(options = {})
         @checksum_store = Checksum::Store.new
         @gem_installers = {}
         @gem_installers_mutex = Mutex.new
+        @remote_spec_for_mutex = Mutex.new
         @remote_specs_mutex = Mutex.new
 
         cooldown = options["cooldown"]
@@ -437,12 +438,14 @@ def remote_specs
       # Looks up a single spec in the remote sources, fetching only its own
       # name when the full remote index is not already materialized.
       def remote_spec_for(spec)
-        return remote_specs.search(spec).first if @remote_specs || api_fetchers.empty?
+        @remote_spec_for_mutex.synchronize do
+          return remote_specs.search(spec).first if @remote_specs || api_fetchers.empty?
 
-        index = Index.build do |idx|
-          fetch_names(api_fetchers, [spec.name], idx)
+          index = Index.build do |idx|
+            fetch_names(api_fetchers, [spec.name], idx)
+          end
+          index.search(spec).first
         end
-        index.search(spec).first
       end
 
       def fetch_names(fetchers, dependency_names, index)

lib/bundler/worker.rb

@@ -19,11 +19,12 @@ def initialize(exn)
     # @param size [Integer] Size of pool
     # @param name [String] name the name of the worker
     # @param func [Proc] job to run in inside the worker pool
-    def initialize(size, name, func)
+    # @param response_queue [Thread::Queue] queue that receives completed jobs
+    def initialize(size, name, func, response_queue: Thread::Queue.new)
       @name = name
       @request_queue = Thread::Queue.new
       @request_queue_with_priority = Thread::Queue.new
-      @response_queue = Thread::Queue.new
+      @response_queue = response_queue
       @func = func
       @size = size
       @threads = nil

spec/bundler/fetcher/compact_index_spec.rb

@@ -18,6 +18,16 @@
     allow(compact_index).to receive(:compact_index_client).and_return(compact_index_client)
   end
 
+  describe "#bundle_worker" do
+    it "uses metadata jobs for the worker pool size" do
+      Bundler.settings.temporary(metadata_jobs: 9) do
+        worker = compact_index.send(:bundle_worker)
+
+        expect(worker.instance_variable_get(:@size)).to eq(9)
+      end
+    end
+  end
+
   describe "#specs_for_names" do
     let(:thread_list) { Thread.list.select {|thread| thread.status == "run" } }
     let(:thread_inspection) { thread_list.map {|th| "  * #{th}:\n    #{th.backtrace_locations.join("\n    ")}" }.join("\n") }

spec/bundler/fetcher/gem_remote_fetcher_spec.rb

@@ -7,6 +7,12 @@
 
 RSpec.describe Bundler::Fetcher::GemRemoteFetcher do
   describe "#initialize" do
+    it "uses download jobs for the connection pool size" do
+      Bundler.settings.temporary(download_jobs: 7) do
+        expect(subject.instance_variable_get(:@pool_size)).to eq(7)
+      end
+    end
+
     context "when ssl_ca_cert setting is not set" do
       before do
         allow(Bundler.settings).to receive(:[]).and_call_original

spec/bundler/installer/parallel_installer_spec.rb

@@ -61,28 +61,41 @@
     end
     let(:installer) { Bundler::Installer.new(bundled_app, definition) }
 
-    it "queues native extensions in priority" do
-      parallel_installer = Bundler::ParallelInstaller.new(installer, definition.specs, 2, false, true)
-      worker_pool = parallel_installer.send(:worker_pool)
-      expected = 6 # Enqueue to download bundler and the 2 gems. Enqueue to install Bundler and the 2 gems.
-
-      expect(worker_pool).to receive(:enq).exactly(expected).times.and_wrap_original do |original_enq, spec, opts|
-        unless opts.nil? # Enqueued for download, no priority
-          if spec.name == "gem_with_extension"
-            expect(opts).to eq({ priority: true })
-          else
-            expect(opts).to eq({ priority: false })
-          end
+    it "prioritizes native extensions for installation" do
+      parallel_installer = Bundler::ParallelInstaller.new(installer, definition.specs, 2, false, true, download_size: 6)
+      download_worker_pool = parallel_installer.send(:download_worker_pool)
+      install_worker_pool = parallel_installer.send(:worker_pool)
+
+      expect(download_worker_pool).to receive(:enq).exactly(3).times.and_call_original
+      expect(install_worker_pool).to receive(:enq).exactly(3).times.and_wrap_original do |original_enq, spec, opts|
+        if spec.name == "gem_with_extension"
+          expect(opts).to eq({ priority: true })
+        else
+          expect(opts).to eq({ priority: false })
         end
 
-        opts ||= {}
         original_enq.call(spec, **opts)
       end
 
       parallel_installer.call
     end
   end
 
+  describe "worker pools" do
+    it "uses separate sizes for download and installation workers" do
+      parallel_installer = described_class.new(nil, [], 2, false, false, download_size: 6)
+
+      download_worker_pool = parallel_installer.send(:download_worker_pool)
+      install_worker_pool = parallel_installer.send(:worker_pool)
+
+      expect(download_worker_pool.instance_variable_get(:@size)).to eq(6)
+      expect(install_worker_pool.instance_variable_get(:@size)).to eq(2)
+    ensure
+      download_worker_pool&.stop
+      install_worker_pool&.stop
+    end
+  end
+
   describe "connect to make jobserver" do
     before do
       unless Gem::Installer.private_method_defined?(:build_jobs)