diff --git a/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py b/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py index cb97cf0..4206272 100644 --- a/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py +++ b/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py @@ -21,6 +21,7 @@ import statistics import subprocess import sys import time +from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import asdict, replace from pathlib import Path from typing import Any, Iterable @@ -525,7 +526,7 @@ def run_frontier_load( config=config, probe_dir=load_dir, run_id=f"{config.name}_{rate_key(rate)}", - cache_root=args.output_root / "cache", + cache_root=args.output_root / "cache" / config.name, ) write_json(load_dir / "command.json", command) environment = os.environ.copy() @@ -652,15 +653,20 @@ def run_frontier(args: argparse.Namespace) -> Path: "end_to_end_action_specific_calibration": False, "real_serving_results_observed_before_freeze": False, }, + "orchestration": { + "max_parallel_configs": args.max_parallel_configs, + "cache_isolation": "one_cache_root_per_config", + "loads_within_config": "sequential", + }, "configs": [asdict(config) | {"name": config.name} for config in configs], } write_json(args.output_root / "run_manifest.json", run_manifest) - config_results = [] rate_records = sorted( protocol["rates"].values(), key=lambda item: item["offered_request_rate"] ) - for config in configs: + + def run_config(config: Any) -> dict[str, Any]: loads = [] for record in rate_records: result = run_frontier_load( @@ -686,7 +692,15 @@ def run_frontier(args: argparse.Namespace) -> Path: ) summary = {"config": asdict(config) | {"name": config.name}, "loads": loads} write_json(args.output_root / "results" / f"{config.name}.json", summary) - config_results.append(summary) + return summary + + results_by_name = {} + with ThreadPoolExecutor(max_workers=args.max_parallel_configs) as pool: + futures = {pool.submit(run_config, config): config for config in configs} + for future in as_completed(futures): + config = futures[future] + results_by_name[config.name] = future.result() + config_results = [results_by_name[config.name] for config in configs] freeze = { "schema": FRONTIER_SCHEMA, "status": "frozen_before_community_run", @@ -1370,6 +1384,7 @@ def parse_args() -> argparse.Namespace: frontier.add_argument("--tp4-capacity-artifact", type=Path, required=True) frontier.add_argument("--tp8-capacity-artifact", type=Path, required=True) frontier.add_argument("--cache-equivalence-artifact", type=Path, required=True) + frontier.add_argument("--max-parallel-configs", type=int, default=1) frontier.add_argument("--config", action="append") community = subparsers.add_parser("prepare-community")