test-bot: speed up parallel dependent setup

Author: GunniBusch

Library/Homebrew/test_bot/formulae_dependents.rb

@@ -1,9 +1,22 @@
 # typed: strict
 # frozen_string_literal: true
 
+require "json"
+
 module Homebrew
   module TestBot
     class FormulaeDependents < TestFormulae
+      DependentFootprint = T.type_alias { T::Set[String] }
+      class ParallelDependentTest < T::Struct
+        const :pid, Integer
+        const :dependent_name, String
+        const :lock_set, DependentFootprint
+        const :result_reader, IO
+        const :ignore_failures, T::Boolean
+      end
+
+      MAX_PARALLEL_DEPENDENT_TESTS = 2
+
       sig { params(testing_formulae: T::Array[String]).returns(T::Array[String]) }
       attr_writer :testing_formulae
 
@@ -24,6 +37,7 @@ def initialize(tap:, git:, dry_run:, fail_fast:, verbose:)
         @testing_formulae_with_tested_dependents = T.let([], T::Array[String])
         @tested_dependents_list = T.let(nil, T.nilable(Pathname))
         @dependent_testing_formulae = T.let([], T::Array[String])
+        @prepared_dependency_formulae = T.let(Set.new, T::Set[String])
       end
 
       sig { params(args: Homebrew::Cmd::TestBotCmd::Args).void }
@@ -142,8 +156,12 @@ def dependent_formulae!(formula_name, args:)
           install_dependent(dependent, testable_dependents, args:) if bottled?(dependent)
         end
 
-        bottled_dependents.each do |dependent|
-          install_dependent(dependent, testable_dependents, args:)
+        if bottled_dependents.one?
+          bottled_dependents.each do |dependent|
+            install_dependent(dependent, testable_dependents, args:)
+          end
+        else
+          pipeline_bottled_dependents!(bottled_dependents, testable_dependents, args:)
         end
       end
 
@@ -256,15 +274,185 @@ def dependents_for_formula(formula, formula_name, args:)
         [source_dependents, bottled_dependents, testable_dependents]
       end
 
+      sig {
+        params(bottled_dependents: T::Array[Formula],
+               testable_dependents: T::Array[Formula], args: Homebrew::Cmd::TestBotCmd::Args).void
+      }
+      def pipeline_bottled_dependents!(bottled_dependents, testable_dependents, args:)
+        info_header "Running bottled dependents:"
+        puts bottled_dependents
+
+        dependent_dependency_sets = T.let(
+          bottled_dependents.to_h { |dependent| [dependent, dependent_dependencies(dependent, testable_dependents)] },
+          T::Hash[Formula, T::Array[Dependency]],
+        )
+        lock_sets = T.let(
+          bottled_dependents.to_h do |dependent|
+            [dependent, dependent_lock_set(dependent, dependent_dependency_sets.fetch(dependent))]
+          end,
+          T::Hash[Formula, DependentFootprint],
+        )
+        active_tests = T.let({}, T::Hash[Integer, ParallelDependentTest])
+        pending_dependents = T.let(
+          bottled_dependents.sort_by do |dependent|
+            [testable_dependents.include?(dependent) ? 0 : 1,
+             dependent_mutation_set(dependent, dependent_dependency_sets.fetch(dependent)).length]
+          end,
+          T::Array[Formula],
+        )
+
+        until pending_dependents.blank?
+          dependent = pending_dependents.find do |candidate|
+            mutation_set = dependent_mutation_set(candidate, dependent_dependency_sets.fetch(candidate))
+            active_tests.values.none? do |worker|
+              worker.lock_set.intersect?(mutation_set)
+            end
+          end
+
+          unless dependent
+            wait_for_parallel_dependent_test(active_tests)
+            next
+          end
+
+          pending_dependents.delete(dependent)
+
+          worker = install_dependent(
+            dependent,
+            testable_dependents,
+            dependency_closure: dependent_dependency_sets.fetch(dependent),
+            lock_set:           lock_sets.fetch(dependent),
+            parallel_test:      true,
+            args:,
+          )
+          next if worker.nil?
+
+          puts "Launching parallel test for #{dependent.full_name}"
+          active_tests[worker.pid] = worker
+          wait_for_parallel_dependent_test(active_tests) while active_tests.length >= MAX_PARALLEL_DEPENDENT_TESTS
+        end
+
+        wait_for_parallel_dependent_test(active_tests) until active_tests.blank?
+      end
+
+      sig { params(active_tests: T::Hash[Integer, ParallelDependentTest]).void }
+      def wait_for_parallel_dependent_test(active_tests)
+        pid, = Process.wait2
+        worker = active_tests.delete(pid)
+        return if worker.nil?
+
+        result_reader = worker.result_reader
+        result = T.let(JSON.parse(result_reader.read), T::Hash[String, T.untyped])
+        result_reader.close
+
+        status = T.cast(result.fetch("status"), String)
+        step = Step.new(
+          ["brew", "test", "--retry", "--verbose"],
+          named_args:      worker.dependent_name,
+          env:             {},
+          verbose:         false,
+          ignore_failures: worker.ignore_failures,
+          repository:,
+        )
+        step.instance_variable_set(:@status, status.to_sym)
+        step.instance_variable_set(:@start_time, Time.at(T.cast(result.fetch("start_time"), Float)))
+        step.instance_variable_set(:@end_time, Time.at(T.cast(result.fetch("end_time"), Float)))
+        steps << step
+
+        test "brew", "uninstall", "--force", "--ignore-dependencies", worker.dependent_name
+
+        if status == "passed"
+          @tested_dependents_list&.write(worker.dependent_name, mode: "a")
+          @tested_dependents_list&.write("\n", mode: "a")
+        end
+
+        exit 1 if @fail_fast && status == "failed"
+      end
+
+      sig {
+        params(
+          dependent:       Formula,
+          env:             T::Hash[String, String],
+          ignore_failures: T::Boolean,
+          lock_set:        DependentFootprint,
+        ).returns(ParallelDependentTest)
+      }
+      def launch_parallel_dependent_test(dependent, env:, ignore_failures:, lock_set:)
+        result_reader, result_writer = IO.pipe
+        pid = fork do
+          result_reader.close
+
+          # The parent is responsible for stopping the queue; the child must always report back.
+          original_fail_fast = @fail_fast
+          @fail_fast = false
+          test_header(:FormulaeDependents, method: "parallel_test(#{dependent.full_name})")
+          test "brew", "test", "--retry", "--verbose",
+               named_args:      dependent.full_name,
+               env:,
+               ignore_failures: ignore_failures
+          step = steps.fetch(-1)
+          result_writer.write(JSON.generate(
+                                status:     step.status.to_s,
+                                start_time: T.must(step.start_time).to_f,
+                                end_time:   T.must(step.end_time).to_f,
+                              ))
+        # The child must always report back to the parent, even on unexpected exceptions.
+        rescue Exception => e # rubocop:disable Lint/RescueException
+          $stderr.puts e.full_message(highlight: false, order: :top)
+          result_writer.write(JSON.generate(status: "failed", start_time: Time.now.to_f, end_time: Time.now.to_f))
+        ensure
+          @fail_fast = T.must(original_fail_fast)
+          result_writer.close unless result_writer.closed?
+          exit!
+        end
+
+        result_writer.close
+
+        ParallelDependentTest.new(
+          pid:,
+          dependent_name:  dependent.full_name,
+          lock_set:,
+          result_reader:,
+          ignore_failures:,
+        )
+      end
+
+      sig { params(formula: Formula, dependencies: T::Array[Dependency]).returns(DependentFootprint) }
+      def dependent_lock_set(formula, dependencies)
+        dependency_formulae = dependencies.map(&:to_formula)
+        recursive_formulae = [formula, *dependency_formulae].flat_map do |dep_formula|
+          [dep_formula, *recursive_dependency_formulae(dep_formula)]
+        end
+
+        Set.new(recursive_formulae.uniq(&:full_name).map(&:full_name))
+      end
+
+      sig { params(formula: Formula, dependencies: T::Array[Dependency]).returns(DependentFootprint) }
+      def dependent_mutation_set(formula, dependencies)
+        dependency_formulae = dependencies.filter_map do |dependency|
+          dependency_formula = dependency.to_formula
+          next if @prepared_dependency_formulae.include?(dependency_formula.name)
+          next if dependency.satisfied? && (dependency_formula.keg_only? || dependency_formula.linked?)
+
+          dependency_formula.full_name
+        end
+
+        Set.new([formula.full_name, *dependency_formulae])
+      end
+
       sig {
         params(
           dependent:           Formula,
           testable_dependents: T::Array[Formula],
           args:                Homebrew::Cmd::TestBotCmd::Args,
           build_from_source:   T::Boolean,
-        ).void
+          dependency_closure:  T.nilable(T::Array[Dependency]),
+          lock_set:            T.nilable(DependentFootprint),
+          parallel_test:       T::Boolean,
+        ).returns(T.nilable(ParallelDependentTest))
       }
-      def install_dependent(dependent, testable_dependents, args:, build_from_source: false)
+      def install_dependent(dependent, testable_dependents, args:, build_from_source: false, dependency_closure: nil,
+                            lock_set: nil,
+                            parallel_test: false)
         if @skip_candidates&.include?(dependent.full_name) &&
            artifact_cache_valid?(dependent, formulae_dependents: true)
           @tested_dependents_list&.write(dependent.full_name, mode: "a")
@@ -290,23 +478,13 @@ def install_dependent(dependent, testable_dependents, args:, build_from_source:
         bottled_on_current_version = bottled?(dependent, no_older_versions: true)
         dependent_was_previously_installed = dependent.latest_version_installed?
 
-        dependent_dependencies = Dependency.expand(
-          dependent,
-          cache_key: "test-bot-dependent-dependencies-#{dependent.full_name}",
-        ) do |dep_dependent, dependency|
-          next if !dependency.build? && !dependency.test? && !dependency.optional?
-          next if dependency.test? &&
-                  dep_dependent == dependent &&
-                  !dependency.optional? &&
-                  testable_dependents.include?(dependent)
-
-          Dependency.prune
-        end
+        dependency_closure ||= dependent_dependencies(dependent, testable_dependents)
+        missing_dependencies = missing_dependency_names(dependency_closure)
 
         unless dependent_was_previously_installed
           build_args = []
 
-          fetch_formulae = dependent_dependencies.reject(&:satisfied?).map(&:name)
+          fetch_formulae = missing_dependencies.dup
 
           if build_from_source
             required_dependent_reqs = dependent.requirements.reject(&:optional?)
@@ -329,10 +507,11 @@ def install_dependent(dependent, testable_dependents, args:, build_from_source:
 
           unlink_conflicts dependent
 
-          test "brew", "install", *build_args, "--only-dependencies",
-               named_args:      dependent.full_name,
-               ignore_failures: !bottled_on_current_version,
-               env:             { "HOMEBREW_DEVELOPER" => nil }
+          install_missing_dependency_formulae(
+            missing_dependencies,
+            build_args:,
+            ignore_failures: !bottled_on_current_version,
+          )
 
           env = {}
           env["HOMEBREW_GIT_PATH"] = nil if build_from_source && required_dependent_deps.any? do |d|
@@ -352,7 +531,14 @@ def install_dependent(dependent, testable_dependents, args:, build_from_source:
           unlink_conflicts dependent
           test "brew", "link", dependent.full_name
         end
-        test "brew", "install", "--only-dependencies", dependent.full_name
+
+        # Install the exact missing closure once; for testable dependents this
+        # already includes direct test deps and their required recursive deps.
+        install_missing_dependency_formulae(
+          missing_dependency_names(dependency_closure),
+          ignore_failures: !bottled_on_current_version,
+        )
+
         test "brew", "linkage", "--test",
              named_args:      dependent.full_name,
              ignore_failures: !args.test_default_formula? && !bottled_on_current_version
@@ -367,9 +553,7 @@ def install_dependent(dependent, testable_dependents, args:, build_from_source:
         end
 
         if testable_dependents.include? dependent
-          test "brew", "install", "--only-dependencies", "--include-test", dependent.full_name
-
-          dependent_dependencies.each do |dependency|
+          dependency_closure.each do |dependency|
             dependency_f = dependency.to_formula
             next if dependency_f.keg_only?
             next if dependency_f.linked?
@@ -382,6 +566,15 @@ def install_dependent(dependent, testable_dependents, args:, build_from_source:
           env["HOMEBREW_GIT_PATH"] = nil if required_dependent_deps.any? do |d|
             d.name == "git" && (!d.build? || d.test?)
           end
+          if parallel_test
+            return launch_parallel_dependent_test(
+              dependent,
+              env:,
+              ignore_failures: !args.test_default_formula? && !bottled_on_current_version,
+              lock_set:        T.must(lock_set),
+            )
+          end
+
           test "brew", "test", "--retry", "--verbose",
                named_args:      dependent.full_name,
                env:,
@@ -426,6 +619,58 @@ def install_dependent(dependent, testable_dependents, args:, build_from_source:
             title: "#{dependent} should be bottled for #{os_string}!",
           )
         end
+
+        nil
+      end
+
+      sig { params(dependencies: T::Array[Dependency]).returns(T::Array[String]) }
+      def missing_dependency_names(dependencies)
+        dependencies.reject(&:satisfied?).map(&:name).reject do |formula_name|
+          @prepared_dependency_formulae.include?(formula_name)
+        end.uniq.sort
+      end
+
+      sig {
+        params(
+          formula_names:   T::Array[String],
+          ignore_failures: T::Boolean,
+          build_args:      T::Array[String],
+        ).void
+      }
+      def install_missing_dependency_formulae(formula_names, ignore_failures:, build_args: [])
+        return if formula_names.blank?
+
+        test "brew", "install", *build_args, "--formula", *formula_names,
+             ignore_failures:,
+             env:             { "HOMEBREW_DEVELOPER" => nil }
+        @prepared_dependency_formulae.merge(formula_names) if steps.fetch(-1).passed?
+      end
+
+      sig { params(dependent: Formula, testable_dependents: T::Array[Formula]).returns(T::Array[Dependency]) }
+      def dependent_dependencies(dependent, testable_dependents)
+        Dependency.expand(
+          dependent,
+          cache_key: "test-bot-dependent-dependencies-#{dependent.full_name}",
+        ) do |dep_dependent, dependency|
+          next if !dependency.build? && !dependency.test? && !dependency.optional?
+          next if dependency.test? &&
+                  dep_dependent == dependent &&
+                  !dependency.optional? &&
+                  testable_dependents.include?(dependent)
+
+          Dependency.prune
+        end
+      end
+
+      sig { params(formula: Formula).returns(T::Array[Formula]) }
+      def recursive_dependency_formulae(formula)
+        formula.recursive_dependencies.reject(&:optional?).map(&:to_formula)
+      rescue TapFormulaUnavailableError => e
+        raise if e.tap.installed?
+
+        e.tap.clear_cache
+        safe_system "brew", "tap", e.tap.name
+        retry
       end
 
       sig { params(formula: Formula).void }