Something went wrong. Try again.
My version of https://github.com/vercel-labs/fx
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179from __future__ import annotations
import dataclassesimport jsonimport mathimport osimport pathlibimport shleximport timeimport uuidfrom collections.abc import Callable, Mapping, Sequence
from scripts.pgso.model import PgsoError, sha256_filefrom scripts.pgso.pipeline import ( ArtifactSpec, PipelinePaths, apply_profile, build_instrumented, collect_instrumented_profile, emit_bitcode, link_candidate, merge_profile_batch, read_macos_minos,)from scripts.pgso.profile_supplement import ( MappedProfile, ProfileSupplement, create_mapped_benchmark_profile, create_profile_supplement, merge_profile_supplement, verify_supplement_functions,)from scripts.pgso.runner import emit_progress, hermetic_environment, run_checkedfrom scripts.pgso.toolchain import Toolchain
MINIMUM_SAMPLES = 50STARTUP_MINIMUM_SAMPLES = 1_000STARTUP_WARMUP_RUNS = 10STARTUP_MINIMUM_ROUNDS = 10STARTUP_MAX_RUNS_PER_ROUND = 10MAXIMUM_REGRESSION = 0.10REQUIRED_EVIDENCE = ( "identity", "runtime", "artifacts", "profile", "corpus", "startup", "heavy_workloads",)
STARTUP_COMMANDS = ( ("help", ("help",)), ("version", ("--version",)), ("status", ("status", "--json")), ("doctor", ("doctor", "--json")), ("sessions", ("sessions", "--json")),)
@dataclasses.dataclass(frozen=True)class Workload: name: str argv: tuple[str, ...]
@dataclasses.dataclass(frozen=True)class BenchmarkPlan: selector: str profile_module: str function_prefixes: tuple[str, ...] training_argvs: tuple[tuple[str, ...], ...] workloads: tuple[Workload, ...] microbenchmark: bool = False
@dataclasses.dataclass(frozen=True)class BenchmarkPair: selector: str control_binary: pathlib.Path candidate_binary: pathlib.Path bitcode_sha256: str merged_raw_profiles: int merged_profile: pathlib.Path | None = None profile_use_ir: pathlib.Path | None = None candidate_profile: pathlib.Path | None = None
@dataclasses.dataclass(frozen=True)class ProfileLinkedBenchmark: pair: BenchmarkPair supplement_path: pathlib.Path supplement: ProfileSupplement mapped_profile_text: pathlib.Path | None = None mapped_profile: MappedProfile | None = None
BENCHMARK_PLANS = ( BenchmarkPlan( selector="file_index", profile_module="file-index-bench", function_prefixes=("core.workspace.file_index.",), training_argvs=(("100000", "500"),), workloads=(Workload("file-index-100k", ("100000", "500")),), ), BenchmarkPlan( selector="ui_activity", profile_module="ui-activity-progress-bench", function_prefixes=( "core.output.activity_runtime.", "core.output.worker_status.", "ui.transcript.", "ui.render_engine.", ), training_argvs=((),), workloads=(Workload("ui-activity", ()),), microbenchmark=True, ), BenchmarkPlan( selector="approval_review", profile_module="approval-review-bench", function_prefixes=( "core.output.diff.", "core.permissions.approval_prompt.", "ui.approval_screen.", "ui.footer.approval_ui.", "ui.render_engine.transcript_blocks.", "ui.render_engine.transcript_measure.", "core.terminal.vt_emulator.", ), training_argvs=( ("transcript", "3", "1"), ("diff", "3", "1"), ("combined", "3", "1"), ("payload", "3", "1"), ), workloads=( Workload("approval-transcript", ("transcript", "1", "1")), Workload("approval-diff", ("diff", "1", "1")), Workload("approval-combined", ("combined", "1", "1")), Workload("approval-payload", ("payload", "1", "1")), ), ),)
def select_startup_commands( names: Sequence[str] | None,) -> tuple[tuple[str, tuple[str, ...]], ...]: if names is None: return STARTUP_COMMANDS if not names: raise PgsoError("startup assignment cannot be empty") if len(names) != len(set(names)): raise PgsoError("duplicate startup command") available = {name: argv for name, argv in STARTUP_COMMANDS} unknown = sorted(set(names) - set(available)) if unknown: raise PgsoError("unknown startup command: " + ", ".join(unknown)) requested = set(names) return tuple(item for item in STARTUP_COMMANDS if item[0] in requested)
def select_benchmark_plans( workload_names: Sequence[str] | None,) -> tuple[BenchmarkPlan, ...]: if workload_names is None: return BENCHMARK_PLANS if not workload_names: raise PgsoError("heavy workload assignment cannot be empty") if len(workload_names) != len(set(workload_names)): raise PgsoError("duplicate heavy workload") available = { workload.name for plan in BENCHMARK_PLANS for workload in plan.workloads } unknown = sorted(set(workload_names) - available) if unknown: raise PgsoError("unknown heavy workload: " + ", ".join(unknown)) requested = set(workload_names) return tuple( dataclasses.replace( plan, workloads=tuple( workload for workload in plan.workloads if workload.name in requested ), ) for plan in BENCHMARK_PLANS if any(workload.name in requested for workload in plan.workloads) )
@dataclasses.dataclass(frozen=True)class Comparison: control_samples: tuple[float, ...] candidate_samples: tuple[float, ...] control_p50: float control_p95: float candidate_p50: float candidate_p95: float p50_change: float p95_change: float passed: bool
@dataclasses.dataclass(frozen=True)class MeasurementResult: name: str argv: tuple[str, ...] requested_samples: int control_samples: tuple[float, ...] candidate_samples: tuple[float, ...] control_failures: int candidate_failures: int errors: tuple[str, ...] comparison: Comparison | None passed: bool
class EvidenceRecorder: def __init__( self, path: pathlib.Path, *, command: str, configuration: Mapping[str, object], ) -> None: self.path = path self.payload: dict[str, object] = { "schema_version": 1, "command": command, "configuration": dict(configuration), "status": "running", "stage": "initialize", "eligible": False, "error": None, "evidence": {}, "stages": [], } self._active_stage: tuple[str, float] | None = None self._write()
def _stage_elapsed(self, name: str) -> float: if self._active_stage is None or self._active_stage[0] != name: return 0.0 return time.monotonic() - self._active_stage[1]
def _write(self) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) temporary = self.path.with_name( f".{self.path.name}.{uuid.uuid4().hex}.tmp" ) temporary.write_text( json.dumps(self.payload, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) os.replace(temporary, self.path)
def stage( self, name: str, status: str, evidence: Mapping[str, object] | None = None, ) -> None: if status not in ("running", "passed"): raise PgsoError(f"invalid evidence stage status: {status}") if self.payload["status"] == "failed": raise PgsoError("cannot update a failed evidence manifest") self.payload["stage"] = name self.payload["status"] = "running" stages = self.payload["stages"] if not isinstance(stages, list): raise PgsoError("invalid in-memory evidence stage list") elapsed_seconds = self._stage_elapsed(name) if status == "passed" else None stage_entry: dict[str, object] = {"name": name, "status": status} if elapsed_seconds is not None: stage_entry["elapsed_seconds"] = elapsed_seconds stages.append(stage_entry) if evidence: recorded = self.payload["evidence"] if not isinstance(recorded, dict): raise PgsoError("invalid in-memory evidence mapping") recorded.update(evidence) self._write() if status == "running": self._active_stage = (name, time.monotonic()) emit_progress(f"stage started: {name}") else: self._active_stage = None if elapsed_seconds is None: raise PgsoError("passed stage is missing elapsed time") emit_progress(f"stage passed in {elapsed_seconds:.3f}s: {name}")
def fail(self, stage: str, error: BaseException) -> None: elapsed_seconds = self._stage_elapsed(stage) self.payload["stage"] = stage self.payload["status"] = "failed" self.payload["eligible"] = False self.payload["error"] = str(error) stages = self.payload["stages"] if isinstance(stages, list): stages.append( { "name": stage, "status": "failed", "elapsed_seconds": elapsed_seconds, } ) self._write() self._active_stage = None emit_progress( f"stage failed in {elapsed_seconds:.3f}s: {stage}: {error}" )
def complete(self) -> None: evidence = self.payload["evidence"] if not isinstance(evidence, dict): raise PgsoError("invalid in-memory evidence mapping") missing = tuple(key for key in REQUIRED_EVIDENCE if key not in evidence) if missing: raise PgsoError( "required evidence is incomplete: " + ", ".join(missing) ) self.payload["stage"] = "complete" self.payload["status"] = "passed" self.payload["eligible"] = True self.payload["error"] = None stages = self.payload["stages"] if isinstance(stages, list): stages.append({"name": "complete", "status": "passed"}) self._write()
def finish_partial(self, stage: str) -> None: if self.payload["status"] == "failed": raise PgsoError("cannot finish a failed evidence manifest") self.payload["stage"] = stage self.payload["status"] = "passed" self.payload["eligible"] = False self.payload["error"] = None stages = self.payload["stages"] if isinstance(stages, list): stages.append({"name": stage, "status": "passed"}) self._write()
def percentile(samples: Sequence[float], fraction: float) -> float: if not samples: raise PgsoError("percentile requires at least one sample") if not 0 < fraction <= 1: raise PgsoError("percentile fraction must be in (0, 1]") if not all(math.isfinite(sample) and sample >= 0 for sample in samples): raise PgsoError("samples must be finite nonnegative values") ordered = sorted(samples) rank = max(1, math.ceil(fraction * len(ordered))) return ordered[rank - 1]
def _change(control: float, candidate: float) -> float: if control <= 0: raise PgsoError("control percentile must be positive") return candidate / control - 1.0
def compare_samples( control_samples: Sequence[float], candidate_samples: Sequence[float], *, minimum_samples: int = MINIMUM_SAMPLES, maximum_regression: float = MAXIMUM_REGRESSION,) -> Comparison: if len(control_samples) < minimum_samples or len(candidate_samples) < minimum_samples: raise PgsoError( f"comparison requires at least {minimum_samples} samples per artifact" ) if maximum_regression < 0: raise PgsoError("maximum regression must be nonnegative") control = tuple(control_samples) candidate = tuple(candidate_samples) control_p50 = percentile(control, 0.50) control_p95 = percentile(control, 0.95) candidate_p50 = percentile(candidate, 0.50) candidate_p95 = percentile(candidate, 0.95) p50_change = _change(control_p50, candidate_p50) p95_change = _change(control_p95, candidate_p95) passed = ( candidate_p50 <= control_p50 * (1.0 + maximum_regression) and candidate_p95 <= control_p95 * (1.0 + maximum_regression) ) return Comparison( control_samples=control, candidate_samples=candidate, control_p50=control_p50, control_p95=control_p95, candidate_p50=candidate_p50, candidate_p95=candidate_p95, p50_change=p50_change, p95_change=p95_change, passed=passed, )
def _read_hyperfine_samples( path: pathlib.Path, *, expected_samples: int,) -> dict[str, tuple[float, ...]]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise PgsoError(f"could not read Hyperfine export: {path}: {error}") from error if not isinstance(payload, dict) or not isinstance(payload.get("results"), list): raise PgsoError("Hyperfine export is missing its results list")
parsed: dict[str, tuple[float, ...]] = {} for entry in payload["results"]: if not isinstance(entry, dict): raise PgsoError("Hyperfine result must be an object") label = entry.get("command") if label not in ("control", "candidate"): raise PgsoError(f"Hyperfine round contains unexpected label: {label!r}") if label in parsed: raise PgsoError(f"Hyperfine round contains duplicate label: {label}") raw_samples = entry.get("times") if not isinstance(raw_samples, list): raise PgsoError(f"Hyperfine {label} result is missing its samples") if len(raw_samples) != expected_samples: raise PgsoError( f"Hyperfine round returned {len(raw_samples)} {label} samples; " f"expected {expected_samples}" ) samples: list[float] = [] for index, sample in enumerate(raw_samples): if ( isinstance(sample, bool) or not isinstance(sample, (int, float)) or not math.isfinite(sample) or sample <= 0 ): raise PgsoError( f"Hyperfine {label} sample {index} must be finite and positive" ) samples.append(float(sample)) parsed[label] = tuple(samples)
missing = tuple(label for label in ("control", "candidate") if label not in parsed) if missing: raise PgsoError(f"Hyperfine round is missing label: {', '.join(missing)}") return parsed
def measure_alternating( *, name: str, control_binary: pathlib.Path, candidate_binary: pathlib.Path, argv: Sequence[str], samples: int, sample_runner: Callable[[str, pathlib.Path, tuple[str, ...], int], float],) -> MeasurementResult: if samples < MINIMUM_SAMPLES: raise PgsoError( f"qualification requires at least {MINIMUM_SAMPLES} samples" ) control_samples: list[float] = [] candidate_samples: list[float] = [] control_failures = 0 candidate_failures = 0 errors: list[str] = [] arguments = tuple(argv)
for sample_index in range(samples): order = ( (("control", control_binary), ("candidate", candidate_binary)) if sample_index % 2 == 0 else (("candidate", candidate_binary), ("control", control_binary)) ) for label, binary in order: try: elapsed = sample_runner(label, binary, arguments, sample_index) if not math.isfinite(elapsed) or elapsed <= 0: raise PgsoError("sample duration must be finite and positive") if label == "control": control_samples.append(elapsed) else: candidate_samples.append(elapsed) except Exception as error: errors.append(f"{label}[{sample_index}]: {error}") if label == "control": control_failures += 1 else: candidate_failures += 1
comparison: Comparison | None = None if ( control_failures == 0 and candidate_failures == 0 and len(control_samples) >= MINIMUM_SAMPLES and len(candidate_samples) >= MINIMUM_SAMPLES ): comparison = compare_samples(control_samples, candidate_samples) passed = comparison is not None and comparison.passed return MeasurementResult( name=name, argv=arguments, requested_samples=samples, control_samples=tuple(control_samples), candidate_samples=tuple(candidate_samples), control_failures=control_failures, candidate_failures=candidate_failures, errors=tuple(errors), comparison=comparison, passed=passed, )
def _external_control_link_argv( toolchain: Toolchain, object_path: pathlib.Path, binary_path: pathlib.Path,) -> tuple[str, ...]: return ( str(toolchain.zig), "cc", "-target", toolchain.target, "-O2", "-Wl,-dead_strip", "-s", str(object_path), "-o", str(binary_path), "-lc", )
def build_external_o2_control( toolchain: Toolchain, paths: PipelinePaths,) -> pathlib.Path: output = paths.root / "external-control" output.mkdir(parents=True, exist_ok=False) bitcode = output / paths.bitcode.name object_path = output / f"{paths.binary_name}.o" binary = output / paths.binary_name run_checked( ( str(toolchain.opt), "-passes=default<O2>", str(paths.bitcode), "-o", str(bitcode), ), cwd=paths.root, env=os.environ.copy(), timeout_s=900, log_path=paths.logs / "external-control-opt.json", require_empty_stderr=True, ) run_checked( ( str(toolchain.llc), "-filetype=obj", "-O=2", str(bitcode), "-o", str(object_path), ), cwd=paths.root, env=os.environ.copy(), timeout_s=900, log_path=paths.logs / "external-control-object.json", require_empty_stderr=True, ) run_checked( _external_control_link_argv(toolchain, object_path, binary), cwd=paths.root, env=os.environ.copy(), timeout_s=900, log_path=paths.logs / "external-control-link.json", require_empty_stderr=True, ) run_checked( (str(toolchain.strip), "-S", "-x", str(binary)), cwd=paths.root, env=os.environ.copy(), timeout_s=120, log_path=paths.logs / "external-control-strip.json", require_empty_stderr=True, ) if not binary.is_file() or binary.stat().st_size == 0: raise PgsoError(f"external O2 control is missing or empty: {binary}") return binary
def build_benchmark_pair( toolchain: Toolchain, repo_root: pathlib.Path, output_root: pathlib.Path, plan: BenchmarkPlan,) -> BenchmarkPair: paths = PipelinePaths.create(output_root, selector=plan.selector) spec = ArtifactSpec(repo_root=repo_root, selector=plan.selector) bitcode_sha256 = emit_bitcode(toolchain, spec, paths) external_control = build_external_o2_control(toolchain, paths) minimum_macos = read_macos_minos( toolchain, external_control, paths.logs / "control-load-commands.json", )
first_training = plan.training_argvs[0] raw_profiles = list( build_instrumented( toolchain, paths, minimum_macos, training_argv=first_training, training_name="training-1", ) ) for index, training_argv in enumerate(plan.training_argvs[1:], start=2): raw_profiles.extend( collect_instrumented_profile( toolchain, paths, training_argv=training_argv, training_name=f"training-{index}", timeout_s=300, ) ) merged_count = merge_profile_batch( toolchain, raw_profiles, paths.merged_profile, paths.logs / "merge-training.json", ) emit_bitcode( toolchain, spec, paths, expected_sha256=bitcode_sha256, ) apply_profile(toolchain, paths, bitcode_sha256) candidate = link_candidate( toolchain, paths, require_release_safe_evidence=False, ) return BenchmarkPair( selector=plan.selector, control_binary=external_control, candidate_binary=candidate, bitcode_sha256=bitcode_sha256, merged_raw_profiles=merged_count, merged_profile=paths.merged_profile, profile_use_ir=paths.profile_use_ir, )
def build_profile_linked_benchmarks( toolchain: Toolchain, repo_root: pathlib.Path, production_paths: PipelinePaths,) -> dict[str, ProfileLinkedBenchmark]: supplement_dir = production_paths.profiles / "supplements" supplement_dir.mkdir() linked: dict[str, ProfileLinkedBenchmark] = {} for plan in BENCHMARK_PLANS: pair = build_benchmark_pair( toolchain, repo_root, production_paths.root / "heavy" / plan.selector, plan, ) if pair.merged_profile is None or pair.profile_use_ir is None: raise PgsoError( f"benchmark profile evidence is incomplete: {plan.selector}" ) supplement_path = supplement_dir / f"{plan.selector}.proftext" supplement = create_profile_supplement( toolchain, production_profile=production_paths.merged_profile, benchmark_profile=pair.merged_profile, benchmark_ir=pair.profile_use_ir, output_text=supplement_path, source_module=plan.profile_module, destination_module="fx", allowed_prefixes=plan.function_prefixes, log_dir=( production_paths.logs / "supplements" / plan.selector ), ) linked[plan.selector] = ProfileLinkedBenchmark( pair=pair, supplement_path=supplement_path, supplement=supplement, ) for plan in BENCHMARK_PLANS: merge_profile_supplement( toolchain, production_profile=production_paths.merged_profile, supplement_text=linked[plan.selector].supplement_path, log_path=( production_paths.logs / "supplements" / plan.selector / "merge.json" ), ) return linked
def relink_profile_linked_benchmarks( toolchain: Toolchain, production_paths: PipelinePaths, linked_benchmarks: Mapping[str, ProfileLinkedBenchmark],) -> dict[str, ProfileLinkedBenchmark]: expected = tuple(plan.selector for plan in BENCHMARK_PLANS) if set(linked_benchmarks) != set(expected): raise PgsoError("profile-linked benchmark set is incomplete") relinked: dict[str, ProfileLinkedBenchmark] = {} for plan in BENCHMARK_PLANS: linked = linked_benchmarks[plan.selector] pair = linked.pair if pair.merged_profile is None: raise PgsoError( f"benchmark training profile is incomplete: {plan.selector}" ) pair_paths = PipelinePaths.open( production_paths.root / "heavy" / plan.selector, selector=plan.selector, ) mapped_text = pair_paths.profiles / "production.proftext" mapped_profile_path = pair_paths.profiles / "production.profdata" mapped = create_mapped_benchmark_profile( toolchain, production_profile=production_paths.merged_profile, benchmark_profile=pair.merged_profile, output_text=mapped_text, output_profile=mapped_profile_path, source_module="fx", destination_module=plan.profile_module, log_dir=pair_paths.logs / "production-profile-map", ) supplemented_functions = { name.removeprefix("fx;") for name in linked.supplement.function_names } if not supplemented_functions.issubset( set(mapped.compatible_function_names) ): raise PgsoError( f"mapped benchmark omitted supplemented functions: " f"{plan.selector}" ) apply_profile( toolchain, pair_paths, pair.bitcode_sha256, profile_path=mapped_profile_path, ) candidate = link_candidate( toolchain, pair_paths, require_release_safe_evidence=False, ) relinked[plan.selector] = dataclasses.replace( linked, pair=dataclasses.replace( pair, candidate_binary=candidate, profile_use_ir=pair_paths.profile_use_ir, candidate_profile=mapped_profile_path, ), mapped_profile_text=mapped_text, mapped_profile=mapped, ) return relinked
def profile_linked_benchmark_evidence( linked_benchmarks: Mapping[str, ProfileLinkedBenchmark], production_ir: str,) -> dict[str, object]: expected = tuple(plan.selector for plan in BENCHMARK_PLANS) if set(linked_benchmarks) != set(expected): raise PgsoError("profile-linked benchmark evidence is incomplete") evidence: dict[str, object] = {} for plan in BENCHMARK_PLANS: linked = linked_benchmarks[plan.selector] pair = linked.pair if ( pair.merged_profile is None or pair.profile_use_ir is None or pair.candidate_profile is None or linked.mapped_profile_text is None or linked.mapped_profile is None ): raise PgsoError( f"benchmark profile evidence is incomplete: {plan.selector}" ) function_modes = verify_supplement_functions( production_ir, linked.supplement.function_names, production_module="fx", ) benchmark_profile_names = tuple( f"{plan.profile_module};{name.removeprefix('fx;')}" for name in linked.supplement.function_names ) benchmark_modes = verify_supplement_functions( pair.profile_use_ir.read_text( encoding="utf-8", errors="replace", ), benchmark_profile_names, production_module=plan.profile_module, ) evidence[plan.selector] = { "profile_module": plan.profile_module, "bitcode_sha256": pair.bitcode_sha256, "merged_raw_profiles": pair.merged_raw_profiles, "benchmark_profile_sha256": sha256_file(pair.merged_profile), "candidate_profile_sha256": sha256_file( pair.candidate_profile ), "mapped_profile_sha256": sha256_file( linked.mapped_profile_text ), "mapped_compatible_functions": len( linked.mapped_profile.compatible_function_names ), "mapped_counter_total": linked.mapped_profile.total_counter_value, "control_sha256": sha256_file(pair.control_binary), "candidate_sha256": sha256_file(pair.candidate_binary), "supplement_sha256": sha256_file(linked.supplement_path), "normalized_counter_total": ( linked.supplement.total_counter_value ), "functions": function_modes, "benchmark_functions": benchmark_modes, } return evidence
def _measurement_environment(home: pathlib.Path) -> dict[str, str]: environment = hermetic_environment(home) environment.update( { "FX_AUTO_UPGRADE": "0", "FX_DISABLE_KEYCHAIN": "1", "FX_SKIP_ONBOARDING": "1", "FX_SOUND": "0", "HOME": str(home), "NO_COLOR": "1", } ) return environment
def measure_startup( *, repo_root: pathlib.Path, control_binary: pathlib.Path, candidate_binary: pathlib.Path, hyperfine_binary: pathlib.Path, output_dir: pathlib.Path, samples: int, timeout_s: float, command_names: Sequence[str] | None = None,) -> tuple[MeasurementResult, ...]: home = output_dir / "home" logs = output_dir / "logs" home.mkdir(parents=True, exist_ok=True) logs.mkdir(parents=True, exist_ok=True) results: list[MeasurementResult] = [] startup_samples = max(samples, STARTUP_MINIMUM_SAMPLES) startup_rounds = max( STARTUP_MINIMUM_ROUNDS, math.ceil(startup_samples / STARTUP_MAX_RUNS_PER_ROUND), ) samples_per_round, extra_samples = divmod(startup_samples, startup_rounds) round_samples = tuple( samples_per_round + (round_index < extra_samples) for round_index in range(startup_rounds) )
for command_name, command_argv in select_startup_commands(command_names): for label, binary in ( ("control", control_binary), ("candidate", candidate_binary), ): result = run_checked( (str(binary), *command_argv), cwd=repo_root, env=_measurement_environment(home), timeout_s=timeout_s, log_path=logs / f"{command_name}-preflight-{label}.json", require_empty_stderr=True, ) if not result.stdout.strip(): raise PgsoError(f"startup command produced empty stdout: {command_name}")
samples_by_label: dict[str, list[float]] = { "control": [], "candidate": [], } for round_index, count in enumerate(round_samples, start=1): order = ( ( ("control", control_binary), ("candidate", candidate_binary), ) if round_index % 2 == 1 else ( ("candidate", candidate_binary), ("control", control_binary), ) ) export_path = logs / f"{command_name}-round-{round_index}-samples.json" benchmark_argv: list[str] = [ str(hyperfine_binary), "--shell=none", "--style", "none", "--warmup", str(STARTUP_WARMUP_RUNS), "--runs", str(count), "--export-json", str(export_path), ] for label, binary in order: benchmark_argv.extend( ( "--command-name", label, shlex.join((str(binary), *command_argv)), ) ) run_checked( benchmark_argv, cwd=repo_root, env=_measurement_environment(home), timeout_s=timeout_s, log_path=logs / f"{command_name}-round-{round_index}.json", require_empty_stderr=False, ) round_results = _read_hyperfine_samples( export_path, expected_samples=count, ) for label, measured in round_results.items(): samples_by_label[label].extend(measured)
comparison = compare_samples( samples_by_label["control"], samples_by_label["candidate"], minimum_samples=startup_samples, ) results.append( MeasurementResult( name=f"startup-{command_name}", argv=command_argv, requested_samples=startup_samples, control_samples=tuple(samples_by_label["control"]), candidate_samples=tuple(samples_by_label["candidate"]), control_failures=0, candidate_failures=0, errors=(), comparison=comparison, passed=comparison.passed, ) ) return tuple(results)
def _measure_microbenchmark( name: str, control_binary: pathlib.Path, candidate_binary: pathlib.Path, hyperfine_binary: pathlib.Path, repo_root: pathlib.Path, output_dir: pathlib.Path, samples: int, timeout_s: float,) -> MeasurementResult: home = output_dir / "home" logs = output_dir / "logs" home.mkdir(parents=True, exist_ok=True) logs.mkdir(parents=True, exist_ok=True) for label, binary in (("control", control_binary), ("candidate", candidate_binary)): result = run_checked( (str(binary),), cwd=repo_root, env=_measurement_environment(home), timeout_s=timeout_s, log_path=logs / f"preflight-{label}.json", ) if not result.stdout.strip(): raise PgsoError(f"microbenchmark produced empty stdout: {name}")
control_samples: list[float] = [] candidate_samples: list[float] = [] for round_index in range(samples): export_path = logs / f"round-{round_index}-samples.json" benchmark_argv: list[str] = [ str(hyperfine_binary), "--shell=none", "--style", "none", "--warmup", "0", "--runs", "1", "--export-json", str(export_path), ] order = ( (("control", control_binary), ("candidate", candidate_binary)) if round_index % 2 == 0 else (("candidate", candidate_binary), ("control", control_binary)) ) for label, binary in order: benchmark_argv.extend( ( "--command-name", label, shlex.join((str(binary),)), ) ) run_checked( benchmark_argv, cwd=repo_root, env=_measurement_environment(home), timeout_s=timeout_s, log_path=logs / f"round-{round_index}.json", require_empty_stderr=False, ) round_results = _read_hyperfine_samples(export_path, expected_samples=1) control_samples.append(round_results["control"][0]) candidate_samples.append(round_results["candidate"][0])
comparison = compare_samples(tuple(control_samples), tuple(candidate_samples)) return MeasurementResult( name=name, argv=(), requested_samples=samples, control_samples=tuple(control_samples), candidate_samples=tuple(candidate_samples), control_failures=0, candidate_failures=0, errors=(), comparison=comparison, passed=comparison.passed, )
def measure_heavy_workloads( *, toolchain: Toolchain | None, repo_root: pathlib.Path, output_dir: pathlib.Path, samples: int, timeout_s: float, workload_names: Sequence[str] | None = None, prebuilt_pairs: Mapping[str, BenchmarkPair] | None = None, hyperfine_binary: pathlib.Path | None = None,) -> tuple[MeasurementResult, ...]: output_dir.mkdir(parents=True, exist_ok=False) home = output_dir / "home" logs = output_dir / "logs" home.mkdir() logs.mkdir() results: list[MeasurementResult] = []
for plan in select_benchmark_plans(workload_names): if prebuilt_pairs is None: if toolchain is None: raise PgsoError("heavy workload build requires an LLVM toolchain") pair = build_benchmark_pair( toolchain, repo_root, output_dir / plan.selector, plan, ) else: pair = prebuilt_pairs.get(plan.selector) if pair is None: raise PgsoError( f"missing prebuilt benchmark pair: {plan.selector}" ) if pair.selector != plan.selector: raise PgsoError( f"prebuilt benchmark selector mismatch: {plan.selector}" ) for label, binary in ( ("control", pair.control_binary), ("candidate", pair.candidate_binary), ): if not binary.is_file() or binary.stat().st_size == 0: raise PgsoError( f"prebuilt benchmark {label} is missing: {binary}" ) for workload in plan.workloads: workload_logs = logs / workload.name
def sample_runner( label: str, binary: pathlib.Path, argv: tuple[str, ...], sample_index: int, ) -> float: result = run_checked( (str(binary), *argv), cwd=repo_root, env=_measurement_environment(home), timeout_s=timeout_s, log_path=( workload_logs / f"{sample_index:03d}-{label}.json" ), require_empty_stderr=True, ) if not result.stdout.strip(): raise PgsoError( f"heavy workload produced empty stdout: {workload.name}" ) return result.elapsed_seconds
if plan.microbenchmark: if hyperfine_binary is None: raise PgsoError( f"microbenchmark measurement requires Hyperfine: {plan.selector}" ) results.append( _measure_microbenchmark( workload.name, pair.control_binary, pair.candidate_binary, hyperfine_binary, repo_root, workload_logs, samples, timeout_s, ) ) continue results.append( measure_alternating( name=workload.name, control_binary=pair.control_binary, candidate_binary=pair.candidate_binary, argv=workload.argv, samples=samples, sample_runner=sample_runner, ) ) return tuple(results)
def measurement_payload( results: Sequence[MeasurementResult],) -> list[dict[str, object]]: return [dataclasses.asdict(result) for result in results]
def require_measurements_passed( label: str, results: Sequence[MeasurementResult],) -> None: failed = tuple(result.name for result in results if not result.passed) if failed: raise PgsoError(f"{label} qualification failed: {', '.join(failed)}")