diff --git a/dsl/extdeps/rust/cargo_build.dag b/dsl/extdeps/rust/cargo_build.dag index d4ecdf37382..72e85a0d3b2 100644 --- a/dsl/extdeps/rust/cargo_build.dag +++ b/dsl/extdeps/rust/cargo_build.dag @@ -12,15 +12,23 @@ data extdeps_external_authority_anchor: ExternalAuthority = ExternalAuthority { data cargo_compile_env: Map = empty_map() +fn cargo_build_jobs_flag(count: Int) -> List { + ["--jobs", "{count}"] +} + +fn cargo_nextest_build_jobs_flag(count: Int) -> List { + ["--build-jobs", "{count}"] +} + service cargo.Build { operation Build { - input { extra_args: List = [], env: Map = cargo_compile_env } + input { extra_args: List = [], env: Map = cargo_compile_env, build_jobs_args: List = [] } output { success: Bool from "exit_success" stdout: String from "stdout" stderr: String from "stderr" } - transport shell { argv: ["cargo", "build", extra_args] } + transport shell { argv: ["cargo", "build", build_jobs_args, extra_args] } exit { 0 => Unit 101 => String "Compilation failed" @@ -43,13 +51,13 @@ service cargo.Build { } operation Nextest { - input { extra_args: List = [] } + input { extra_args: List = [], build_jobs_args: List = [] } output { success: Bool from "exit_success" stdout: String from "stdout" stderr: String from "stderr" } - transport shell { argv: ["cargo", "nextest", "run", extra_args] } + transport shell { argv: ["cargo", "nextest", "run", build_jobs_args, extra_args] } exit { 0 => Unit 100 => String "Tests failed" diff --git a/dsl/gunbc/ci_compile_jobs.dag b/dsl/gunbc/ci_compile_jobs.dag new file mode 100644 index 00000000000..49ebd2e78d0 --- /dev/null +++ b/dsl/gunbc/ci_compile_jobs.dag @@ -0,0 +1,63 @@ +module gunbc.ci_compile_jobs + +import std.types { Int, String } +import std.measure { ByteSize, HardwareThreadCount, hardware_thread_count_value } +import std.realization_width { process_memory_aware_spawn_width } +import gunbc.ci_compile_measurement { + CompileJobMeasurement, CompileJobMeasured, CompileJobUnmeasured, + gunbc_ci_compile_measurement, gunbc_ci_compile_conservative_fallback_jobs +} +import extdeps.cpu.ampere { altra_max_m12830_catalog } + +type DerivedCompileJobs + = CompileJobs { count: Int } + | CompileJobsRefuse { reason: String } + +fn gunbc_ci_compile_hardware_threads() -> HardwareThreadCount { + altra_max_m12830_catalog.threads +} + +fn ci_compile_jobs_for( + measurement: CompileJobMeasurement, + hardware_threads: HardwareThreadCount, + memory_budget: ByteSize, + force_serial: Bool +) -> DerivedCompileJobs { + if force_serial { + CompileJobs { count: 1 } + } else { + match measurement { + CompileJobUnmeasured => CompileJobsRefuse { + reason: "ci_compile_jobs: gunbc.ci_compile_measurement is Unmeasured — no committed per-rustc peak / procs-per-job / pids budget. Refusing to guess a job count (DESIGN §5); stamp a real fleet measurement first." + } + CompileJobMeasured { per_rustc_peak: peak, procs_per_job: ppj, pids_budget: pids } => CompileJobs { + count: hardware_thread_count_value( + t: process_memory_aware_spawn_width( + hardware_threads: hardware_threads, + memory_budget: memory_budget, + per_job_peak: peak, + pids_budget: pids, + procs_per_job: ppj, + conservative_fallback_width: gunbc_ci_compile_conservative_fallback_jobs + ) + ) + } + } + } +} + +fn ci_compile_jobs(memory_budget: ByteSize, force_serial: Bool) -> DerivedCompileJobs { + ci_compile_jobs_for( + measurement: gunbc_ci_compile_measurement, + hardware_threads: gunbc_ci_compile_hardware_threads(), + memory_budget: memory_budget, + force_serial: force_serial + ) +} + +fn compile_jobs_count_or_serial(jobs: DerivedCompileJobs) -> Int { + match jobs { + CompileJobs { count: c } => c + CompileJobsRefuse { reason: _ } => 1 + } +} diff --git a/dsl/gunbc/ci_compile_measurement.dag b/dsl/gunbc/ci_compile_measurement.dag new file mode 100644 index 00000000000..45ce8b99f5e --- /dev/null +++ b/dsl/gunbc/ci_compile_measurement.dag @@ -0,0 +1,12 @@ +module gunbc.ci_compile_measurement + +import std.measure { ByteSize } +import std.types { Int } + +type CompileJobMeasurement + = CompileJobMeasured { per_rustc_peak: ByteSize, procs_per_job: Int, pids_budget: Int } + | CompileJobUnmeasured + +data gunbc_ci_compile_measurement: CompileJobMeasurement = CompileJobUnmeasured + +data gunbc_ci_compile_conservative_fallback_jobs: Int = 1 diff --git a/dsl/std/realization.dag b/dsl/std/realization.dag index be493ab3637..5f7e125c5e9 100644 --- a/dsl/std/realization.dag +++ b/dsl/std/realization.dag @@ -51,6 +51,13 @@ import std.realization_width { witness_unreadable_budget_falls_back_conservative, witness_safety_fraction_is_load_bearing, witness_subunit_budget_admits_one, + process_bounded_job_count, + process_memory_aware_spawn_width, + process_memory_aware_width_value, + witness_pids_term_binds_below_cores_and_memory, + witness_memory_term_still_binds_under_three_term_fold, + witness_unreadable_pids_budget_falls_back_conservative, + witness_subunit_pids_budget_admits_one, } type Independence = Independent | Dependent | Unknown diff --git a/dsl/std/realization_width.dag b/dsl/std/realization_width.dag index 97b65ae8571..b83e02dd918 100644 --- a/dsl/std/realization_width.dag +++ b/dsl/std/realization_width.dag @@ -102,6 +102,39 @@ fn memory_aware_spawn_width( } } +fn process_bounded_job_count(pids_budget: Int, procs_per_job: Int) -> Int { + if procs_per_job <= 0 { + 1 + } else { + let fits = pids_budget / procs_per_job + if fits < 1 { 1 } else { fits } + } +} + +fn process_memory_aware_spawn_width( + hardware_threads: HardwareThreadCount, + memory_budget: ByteSize, + per_job_peak: ByteSize, + pids_budget: Int, + procs_per_job: Int, + conservative_fallback_width: Int +) -> HardwareThreadCount { + let mem_bytes = byte_size_count(b: memory_budget) + if mem_bytes == 0 { + hardware_thread_count(count: conservative_fallback_width) + } else if pids_budget == 0 { + hardware_thread_count(count: conservative_fallback_width) + } else { + let cores = hardware_thread_count_value(t: hardware_threads) + let mem_width = memory_bounded_shard_count( + memory_budget: memory_budget, + per_shard_peak: per_job_peak + ) + let pids_width = process_bounded_job_count(pids_budget: pids_budget, procs_per_job: procs_per_job) + hardware_thread_count(count: int_min(a: int_min(a: cores, b: mem_width), b: pids_width)) + } +} + fn width_fold_objective_goals() -> List { [LowerIsBetter, LowerIsBetter] } @@ -241,3 +274,67 @@ fn witness_subunit_budget_admits_one() -> Bool { conservative_fallback_width: 4 ) == 1 } + +fn process_memory_aware_width_value( + hardware_threads: HardwareThreadCount, + memory_budget: ByteSize, + per_job_peak: ByteSize, + pids_budget: Int, + procs_per_job: Int, + conservative_fallback_width: Int +) -> Int { + hardware_thread_count_value( + t: process_memory_aware_spawn_width( + hardware_threads: hardware_threads, + memory_budget: memory_budget, + per_job_peak: per_job_peak, + pids_budget: pids_budget, + procs_per_job: procs_per_job, + conservative_fallback_width: conservative_fallback_width + ) + ) +} + +fn witness_pids_term_binds_below_cores_and_memory() -> Bool { + process_memory_aware_width_value( + hardware_threads: hardware_thread_count(128), + memory_budget: byte_size(137438953472), + per_job_peak: byte_size(2100028416), + pids_budget: 24, + procs_per_job: 3, + conservative_fallback_width: 1 + ) == 8 +} + +fn witness_memory_term_still_binds_under_three_term_fold() -> Bool { + process_memory_aware_width_value( + hardware_threads: hardware_thread_count(128), + memory_budget: byte_size(6442450944), + per_job_peak: byte_size(2100028416), + pids_budget: 4096, + procs_per_job: 3, + conservative_fallback_width: 1 + ) == 2 +} + +fn witness_unreadable_pids_budget_falls_back_conservative() -> Bool { + process_memory_aware_width_value( + hardware_threads: hardware_thread_count(128), + memory_budget: byte_size(137438953472), + per_job_peak: byte_size(2100028416), + pids_budget: 0, + procs_per_job: 3, + conservative_fallback_width: 4 + ) == 4 +} + +fn witness_subunit_pids_budget_admits_one() -> Bool { + process_memory_aware_width_value( + hardware_threads: hardware_thread_count(128), + memory_budget: byte_size(137438953472), + per_job_peak: byte_size(2100028416), + pids_budget: 2, + procs_per_job: 3, + conservative_fallback_width: 4 + ) == 1 +} diff --git a/dsl/test/claim/ci_compile_jobs_witness_test.dag b/dsl/test/claim/ci_compile_jobs_witness_test.dag new file mode 100644 index 00000000000..e1936c2bd9e --- /dev/null +++ b/dsl/test/claim/ci_compile_jobs_witness_test.dag @@ -0,0 +1,131 @@ +module test.claim.ci_compile_jobs_witness + +import std.types { Int, Bool, String, List } +import std.measure { hardware_thread_count, ByteSize, byte_size } +import gunbc.ci_compile_measurement { CompileJobMeasurement, CompileJobMeasured } +import gunbc.ci_compile_jobs { + DerivedCompileJobs, CompileJobs, CompileJobsRefuse, + ci_compile_jobs, ci_compile_jobs_for, compile_jobs_count_or_serial +} +import extdeps.cargo_build { cargo_nextest_build_jobs_flag, cargo_build_jobs_flag } +import tools.rust_gates_ci { rust_gate_build_jobs_args } + +fn join_space(xs: List) -> String { + fold(xs, init: "", f: (acc, p) => if acc == "" { p } else { concat(concat(acc, " "), p) }) +} + +fn jobs_count_or(d: DerivedCompileJobs, default: Int) -> Int { + match d { + CompileJobs { count: c } => c + CompileJobsRefuse { reason: _ } => default + } +} + +fn jobs_is_refuse(d: DerivedCompileJobs) -> Bool { + match d { + CompileJobsRefuse { reason: _ } => true + CompileJobs { count: _ } => false + } +} + +fn sample_measured() -> CompileJobMeasurement { + measured_with_pids(p: 4096) +} + +fn measured_with_pids(p: Int) -> CompileJobMeasurement { + CompileJobMeasured { per_rustc_peak: hypothetical_peak(), procs_per_job: 3, pids_budget: p } +} + +fn hypothetical_peak() -> ByteSize { + byte_size(2100028416) +} + +fn witness_committed_unmeasured_refuses() -> Bool { + jobs_is_refuse(ci_compile_jobs(memory_budget: byte_size(137438953472), force_serial: false)) +} + +fn witness_pids_budget_binds_the_count() -> Bool { + jobs_count_or( + ci_compile_jobs_for( + measurement: measured_with_pids(p: 24), + hardware_threads: hardware_thread_count(128), + memory_budget: byte_size(137438953472), + force_serial: false + ), + 0 + ) == 8 +} + +fn witness_memory_budget_binds_the_count() -> Bool { + jobs_count_or( + ci_compile_jobs_for( + measurement: sample_measured(), + hardware_threads: hardware_thread_count(128), + memory_budget: byte_size(6442450944), + force_serial: false + ), + 0 + ) == 2 +} + +fn witness_force_serial_overrides_unmeasured() -> Bool { + jobs_count_or(ci_compile_jobs(memory_budget: byte_size(137438953472), force_serial: true), 0) == 1 +} + +fn witness_force_serial_beats_wide_budget() -> Bool { + jobs_count_or( + ci_compile_jobs_for( + measurement: sample_measured(), + hardware_threads: hardware_thread_count(128), + memory_budget: byte_size(1099511627776), + force_serial: true + ), + 0 + ) == 1 +} + +fn witness_unreadable_budget_falls_back_serial() -> Bool { + jobs_count_or( + ci_compile_jobs_for( + measurement: sample_measured(), + hardware_threads: hardware_thread_count(128), + memory_budget: byte_size(0), + force_serial: false + ), + 0 + ) == 1 +} + +fn witness_nextest_flag_spelling() -> Bool { + join_space(xs: cargo_nextest_build_jobs_flag(count: 8)) == "--build-jobs 8" +} + +fn witness_build_flag_spelling() -> Bool { + join_space(xs: cargo_build_jobs_flag(count: 8)) == "--jobs 8" +} + +fn witness_count_or_serial_passes_measured_count() -> Bool { + compile_jobs_count_or_serial(jobs: CompileJobs { count: 8 }) == 8 +} + +fn witness_count_or_serial_refuse_falls_to_serial() -> Bool { + compile_jobs_count_or_serial(jobs: CompileJobsRefuse { reason: "unmeasured" }) == 1 +} + +fn witness_live_rust_gate_is_failclosed_serial() -> Bool { + join_space(xs: rust_gate_build_jobs_args()) == "--build-jobs 1" +} + +test fn ci_compile_jobs_witnesses() -> Bool { + witness_committed_unmeasured_refuses() + && witness_pids_budget_binds_the_count() + && witness_memory_budget_binds_the_count() + && witness_force_serial_overrides_unmeasured() + && witness_force_serial_beats_wide_budget() + && witness_unreadable_budget_falls_back_serial() + && witness_nextest_flag_spelling() + && witness_build_flag_spelling() + && witness_count_or_serial_passes_measured_count() + && witness_count_or_serial_refuse_falls_to_serial() + && witness_live_rust_gate_is_failclosed_serial() +} diff --git a/dsl/test/claim/realization_width_witness_test.dag b/dsl/test/claim/realization_width_witness_test.dag index 94b767c9c7c..46d86a9bb01 100644 --- a/dsl/test/claim/realization_width_witness_test.dag +++ b/dsl/test/claim/realization_width_witness_test.dag @@ -21,6 +21,10 @@ import std.realization { witness_unreadable_budget_falls_back_conservative, witness_safety_fraction_is_load_bearing, witness_subunit_budget_admits_one, + witness_pids_term_binds_below_cores_and_memory, + witness_memory_term_still_binds_under_three_term_fold, + witness_unreadable_pids_budget_falls_back_conservative, + witness_subunit_pids_budget_admits_one, } fn synthetic_plan_total() -> CostAccount { @@ -124,4 +128,8 @@ test fn realization_width_witnesses() -> Bool { && witness_unreadable_budget_falls_back_conservative() && witness_safety_fraction_is_load_bearing() && witness_subunit_budget_admits_one() + && witness_pids_term_binds_below_cores_and_memory() + && witness_memory_term_still_binds_under_three_term_fold() + && witness_unreadable_pids_budget_falls_back_conservative() + && witness_subunit_pids_budget_admits_one() } diff --git a/dsl/tools/rust_gates_ci.dag b/dsl/tools/rust_gates_ci.dag index 001aafd67b9..3bf2da7caed 100644 --- a/dsl/tools/rust_gates_ci.dag +++ b/dsl/tools/rust_gates_ci.dag @@ -3,17 +3,29 @@ module tools.rust_gates_ci import std.process { ProcessExit, ExitSuccess, ExitFailure } import extdeps.git import extdeps.cargo_build +import extdeps.cargo_build { cargo_nextest_build_jobs_flag } import tools.rust_stage0_gates { should_run_gates } import gunbc.ci_spec { gunbc_ci_spec, DiffMergeBase, DiffTwoDot, } +import gunbc.ci_compile_jobs { ci_compile_jobs, compile_jobs_count_or_serial } +import gunbc.operator_fleet { srv1_host } +import product.compute_fabric { compute_host_ram_bytes_total } + +fn rust_gate_build_jobs_args() -> List { + let jobs = ci_compile_jobs( + memory_budget: compute_host_ram_bytes_total(host: srv1_host), + force_serial: false + ) + cargo_nextest_build_jobs_flag(count: compile_jobs_count_or_serial(jobs: jobs)) +} func run_gates() -> ProcessExit { fmt = cargo.Build.Fmt(extra_args: ["--all", "--check"]) clippy = cargo.Build.Clippy(extra_args: ["--all-targets"], lint_args: ["-D", "warnings"]) - test = cargo.Build.Nextest(extra_args: ["-p", "v1-compiler-tests"]) + test = cargo.Build.Nextest(extra_args: ["-p", "v1-compiler-tests"], build_jobs_args: rust_gate_build_jobs_args()) return if fmt.success && clippy.success && test.success { ExitSuccess } else {