From 36d082b1e20220fdebd680b350f2fff649c577ce Mon Sep 17 00:00:00 2001 From: Joshua Young Date: Mon, 17 Aug 2026 09:20:44 +1000 Subject: [PATCH] Separate Bundler I/O concurrency --- lib/bundler/fetcher/compact_index.rb | 2 +- lib/bundler/fetcher/gem_remote_fetcher.rb | 2 +- lib/bundler/installer/parallel_installer.rb | 54 +++++++++---------- lib/bundler/man/bundle-config.1 | 6 ++- lib/bundler/man/bundle-config.1.ronn | 8 ++- lib/bundler/man/bundle-install.1 | 2 +- lib/bundler/man/bundle-install.1.ronn | 4 +- lib/bundler/man/bundle-update.1 | 2 +- lib/bundler/man/bundle-update.1.ronn | 4 +- lib/bundler/settings.rb | 10 ++++ lib/bundler/source/rubygems.rb | 11 ++-- lib/bundler/worker.rb | 5 +- spec/bundler/fetcher/compact_index_spec.rb | 10 ++++ .../fetcher/gem_remote_fetcher_spec.rb | 6 +++ .../installer/parallel_installer_spec.rb | 39 +++++++++----- spec/bundler/settings_spec.rb | 34 ++++++++++++ spec/bundler/worker_spec.rb | 18 +++++++ 17 files changed, 161 insertions(+), 56 deletions(-) diff --git a/lib/bundler/fetcher/compact_index.rb b/lib/bundler/fetcher/compact_index.rb index 5fa6b96c20b3..0fcb9e32320b 100644 --- a/lib/bundler/fetcher/compact_index.rb +++ b/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 diff --git a/lib/bundler/fetcher/gem_remote_fetcher.rb b/lib/bundler/fetcher/gem_remote_fetcher.rb index d53a7ea52a8a..0781632957db 100644 --- a/lib/bundler/fetcher/gem_remote_fetcher.rb +++ b/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 diff --git a/lib/bundler/installer/parallel_installer.rb b/lib/bundler/installer/parallel_installer.rb index 9a46f408af6c..b7cc47d12024 100644 --- a/lib/bundler/installer/parallel_installer.rb +++ b/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 diff --git a/lib/bundler/man/bundle-config.1 b/lib/bundler/man/bundle-config.1 index 32291e048c1f..b59e13886902 100644 --- a/lib/bundler/man/bundle-config.1 +++ b/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 \fR\. .IP "\(bu" 4 \fBno_install\fR (\fBBUNDLE_NO_INSTALL\fR): Whether \fBbundle package\fR should skip installing gems\. diff --git a/lib/bundler/man/bundle-config.1.ronn b/lib/bundler/man/bundle-config.1.ronn index 6b52288783b5..f0e0204a788e 100644 --- a/lib/bundler/man/bundle-config.1.ronn +++ b/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. diff --git a/lib/bundler/man/bundle-install.1 b/lib/bundler/man/bundle-install.1 index 801768c7ecef..faf13fe29ae3 100644 --- a/lib/bundler/man/bundle-install.1 +++ b/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=\fR, \fB\-j=\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\. diff --git a/lib/bundler/man/bundle-install.1.ronn b/lib/bundler/man/bundle-install.1.ronn index 56fd8bdf42a1..e02995b076df 100644 --- a/lib/bundler/man/bundle-install.1.ronn +++ b/lib/bundler/man/bundle-install.1.ronn @@ -61,8 +61,8 @@ update process below under [CONSERVATIVE UPDATING][]. to this location. * `--jobs=`, `-j=`: - 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 diff --git a/lib/bundler/man/bundle-update.1 b/lib/bundler/man/bundle-update.1 index 94161083fc45..6a1345bdd267 100644 --- a/lib/bundler/man/bundle-update.1 +++ b/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=\fR, \fB\-j=\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=[]\fR Retry failed network or git requests for \fInumber\fR times\. diff --git a/lib/bundler/man/bundle-update.1.ronn b/lib/bundler/man/bundle-update.1.ronn index 72fbf054d157..381070bb2a6e 100644 --- a/lib/bundler/man/bundle-update.1.ronn +++ b/lib/bundler/man/bundle-update.1.ronn @@ -65,8 +65,8 @@ gem. Use the specified gemfile instead of [`Gemfile(5)`][Gemfile(5)]. * `--jobs=`, `-j=`: - 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=[]`: Retry failed network or git requests for times. diff --git a/lib/bundler/settings.rb b/lib/bundler/settings.rb index c1f8ecf824e1..44f1322ef383 100644 --- a/lib/bundler/settings.rb +++ b/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| diff --git a/lib/bundler/source/rubygems.rb b/lib/bundler/source/rubygems.rb index 8c21864b0828..63132e42842a 100644 --- a/lib/bundler/source/rubygems.rb +++ b/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) diff --git a/lib/bundler/worker.rb b/lib/bundler/worker.rb index 77f4f004aa69..f4e99de305d8 100644 --- a/lib/bundler/worker.rb +++ b/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 diff --git a/spec/bundler/fetcher/compact_index_spec.rb b/spec/bundler/fetcher/compact_index_spec.rb index 67582b73d6d7..1cbb2e17de98 100644 --- a/spec/bundler/fetcher/compact_index_spec.rb +++ b/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") } diff --git a/spec/bundler/fetcher/gem_remote_fetcher_spec.rb b/spec/bundler/fetcher/gem_remote_fetcher_spec.rb index a06cf0396c66..fa42e536982e 100644 --- a/spec/bundler/fetcher/gem_remote_fetcher_spec.rb +++ b/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 diff --git a/spec/bundler/installer/parallel_installer_spec.rb b/spec/bundler/installer/parallel_installer_spec.rb index bdfbbe6db34d..6bfa266ab115 100644 --- a/spec/bundler/installer/parallel_installer_spec.rb +++ b/spec/bundler/installer/parallel_installer_spec.rb @@ -61,21 +61,19 @@ 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 @@ -83,6 +81,21 @@ 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) diff --git a/spec/bundler/settings_spec.rb b/spec/bundler/settings_spec.rb index 5e1aaaa55511..b79a0a52acf8 100644 --- a/spec/bundler/settings_spec.rb +++ b/spec/bundler/settings_spec.rb @@ -169,6 +169,40 @@ end end + describe "#download_parallelization" do + it "defaults to three times installation parallelization" do + allow(settings).to receive(:installation_parallelization).and_return(2) + + expect(settings.download_parallelization).to eq(6) + end + + it "caps the default at eight" do + allow(settings).to receive(:installation_parallelization).and_return(4) + + expect(settings.download_parallelization).to eq(8) + end + + it "uses configured download jobs" do + settings.temporary(download_jobs: 7) do + expect(settings.download_parallelization).to eq(7) + end + end + end + + describe "#metadata_parallelization" do + it "defaults to download parallelization" do + allow(settings).to receive(:download_parallelization).and_return(7) + + expect(settings.metadata_parallelization).to eq(7) + end + + it "uses configured metadata jobs" do + settings.temporary(metadata_jobs: 9) do + expect(settings.metadata_parallelization).to eq(9) + end + end + end + describe "#set_global" do context "when it's not possible to write to create the settings directory" do it "raises an PermissionError with explanation" do diff --git a/spec/bundler/worker_spec.rb b/spec/bundler/worker_spec.rb index 2ad2845e378c..a13e66abc113 100644 --- a/spec/bundler/worker_spec.rb +++ b/spec/bundler/worker_spec.rb @@ -18,6 +18,24 @@ expect { subject.enq "a" }.to raise_error(Bundler::ThreadCreationError, "Failed to create threads for the Spec Worker worker: error creating thread") end end + + context "with a shared response queue" do + it "allows workers to publish to the same queue" do + response_queue = Thread::Queue.new + workers = [ + described_class.new(1, "First", function, response_queue: response_queue), + described_class.new(1, "Second", function, response_queue: response_queue), + ] + + workers.first.enq("first") + workers.last.enq("second") + + responses = Array.new(2) { workers.first.deq.first } + expect(responses).to contain_exactly("first", "second") + ensure + workers&.each(&:stop) + end + end end describe "priority queue" do