From dadcdbd35185258074bbea1b8056d107e85d567e Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Wed, 15 Jul 2026 22:32:40 +0800 Subject: [PATCH] Warm community rank runs before measurement --- .../best_effort/fixed_cohort_rank.py | 97 ++++++++++++++++--- .../best_effort/test_fixed_cohort_rank.py | 28 ++++++ 2 files changed, 114 insertions(+), 11 deletions(-) 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 4206272..95ab0e2 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 @@ -56,6 +56,7 @@ COHORT_SIZE = 64 COHORT_SEED = 2026071501 CONFIG_ORDER_SEED = 2026071502 RATE_ORDER_SEED = 2026071503 +WARMUP_SEED = 2026071504 INPUT_BINS = (0, 1024, 2048, 4096, 8192, 16384, 32769) OFFERED_RATES = (0.15, 0.25, 0.35, 0.50, 0.75, 1.00, 1.50) TARGET_PASS_RATE = 0.95 @@ -854,6 +855,12 @@ def prepare_community(args: argparse.Namespace) -> Path: "client_max_concurrency": community_grid.CLIENT_MAX_CONCURRENCY, "slo_early_stop": False, "one_server_launch_per_config": True, + "warmup": { + "method": "one_disjoint_request_per_input_length_bin", + "request_count": len(INPUT_BINS) - 1, + "seed": WARMUP_SEED, + "latencies_discarded": True, + }, "rate_order_randomized_within_config": True, "config_order_seed": args.config_order_seed, "rate_order_seed": args.rate_order_seed, @@ -891,6 +898,51 @@ def _materialize_real_requests( return requests +def _select_warmup_requests( + all_requests: list[TraceRequest], *, cohort_ids: set[str] +) -> list[TraceRequest]: + by_bin = [[] for _ in range(len(INPUT_BINS) - 1)] + for request in all_requests: + if request.row_id in cohort_ids or request.prompt_tokens_hint is None: + continue + input_tokens = int(request.prompt_tokens_hint) + if not 0 <= input_tokens <= 32768: + continue + by_bin[_input_bin(input_tokens)].append(request) + selected = [] + for bin_index, candidates in enumerate(by_bin): + if not candidates: + raise ValueError(f"no disjoint warmup request in input bin {bin_index}") + selected.append( + min( + candidates, + key=lambda request: sha256_text( + f"{WARMUP_SEED}|{bin_index}|{request.row_id}" + ), + ) + ) + return [replace(request, arrival_s=0.0) for request in selected] + + +def _run_real_replay( + requests: list[TraceRequest], *, recipe: Any, study: Any, max_elapsed_s: float +) -> tuple[list[Any], bool, str]: + return _replay_requests( + requests, + base_url=recipe.base_url, + timeout_s=recipe.request_timeout_s, + max_concurrency=study.trace.max_concurrency, + # target=0 makes the mathematical SLO early-stop bound unreachable + # while retaining the existing replay engine. + target_pass_rate=0.0, + max_lag_s=None, + max_elapsed_s=max_elapsed_s, + evaluate_outcome=lambda outcome: type( + "NoEarlyStopEvaluation", (), {"passed": bool(outcome.success)} + )(), + ) + + def _real_request_records( requests: list[TraceRequest], outcomes: list[Any] ) -> list[dict[str, Any]]: @@ -971,6 +1023,37 @@ def run_community_config( healthcheck_path=recipe.healthcheck_path, ready_timeout_s=recipe.ready_timeout_s, ) + cohort_ids = { + str(row["source_row_index"]) for row in protocol["cohort"] + } + warmup_requests = _select_warmup_requests( + all_requests, cohort_ids=cohort_ids + ) + warmup_started = time.time() + warmup_outcomes, warmup_early_stopped, warmup_stop_reason = _run_real_replay( + warmup_requests, + recipe=recipe, + study=study, + max_elapsed_s=1200.0, + ) + warmup_records = _real_request_records(warmup_requests, warmup_outcomes) + warmup_result = { + "status": "completed" + if not warmup_early_stopped + and all(item["success"] for item in warmup_records) + else "failed", + "elapsed_seconds": time.time() - warmup_started, + "early_stopped": warmup_early_stopped, + "early_stop_reason": warmup_stop_reason, + "cohort_disjoint": not bool( + cohort_ids & {item["request_id"] for item in warmup_records} + ), + "requests": warmup_records, + } + write_json(run_dir / "warmup.json", warmup_result) + if warmup_result["status"] != "completed": + raise RuntimeError(f"server warmup failed: {warmup_result}") + time.sleep(2.0) for rate in record["rate_order"]: rate = float(rate) load_dir = run_dir / rate_key(rate) @@ -988,19 +1071,11 @@ def run_community_config( cohort=protocol["cohort"], ) load_started = time.time() - outcomes, early_stopped, early_stop_reason = _replay_requests( + outcomes, early_stopped, early_stop_reason = _run_real_replay( requests, - base_url=recipe.base_url, - timeout_s=recipe.request_timeout_s, - max_concurrency=study.trace.max_concurrency, - # target=0 makes the mathematical SLO early-stop bound - # unreachable while retaining the existing replay engine. - target_pass_rate=0.0, - max_lag_s=None, + recipe=recipe, + study=study, max_elapsed_s=max(float(rate_record["duration_s"]) + 900.0, 1200.0), - evaluate_outcome=lambda outcome: type( - "NoEarlyStopEvaluation", (), {"passed": bool(outcome.success)} - )(), ) request_records = _real_request_records(requests, outcomes) score = score_requests(request_records) diff --git a/runs/frontier-multicase-sufficiency-v0/best_effort/test_fixed_cohort_rank.py b/runs/frontier-multicase-sufficiency-v0/best_effort/test_fixed_cohort_rank.py index 966a912..0dbbd11 100644 --- a/runs/frontier-multicase-sufficiency-v0/best_effort/test_fixed_cohort_rank.py +++ b/runs/frontier-multicase-sufficiency-v0/best_effort/test_fixed_cohort_rank.py @@ -141,3 +141,31 @@ def test_community_study_uses_affine_primary_slo(tmp_path: Path) -> None: } assert payload["trace"]["request_mode"] == "raw_completion" assert payload["trace"]["completion_tokens_override"] == 1 + + +def test_warmup_is_disjoint_and_covers_each_input_bin() -> None: + lengths = [100, 1500, 3000, 6000, 12000, 24000] + requests = [] + cohort_ids = set() + for bin_index, length in enumerate(lengths): + for suffix in ("cohort", "warmup"): + row_id = f"{bin_index}-{suffix}" + if suffix == "cohort": + cohort_ids.add(row_id) + requests.append( + MODULE.TraceRequest( + row_id=row_id, + arrival_s=10.0, + sampling_u=0.5, + body={"prompt": row_id}, + prompt_tokens_hint=length, + completion_tokens_hint=1, + ) + ) + selected = MODULE._select_warmup_requests(requests, cohort_ids=cohort_ids) + assert len(selected) == len(MODULE.INPUT_BINS) - 1 + assert not cohort_ids & {request.row_id for request in selected} + assert [MODULE._input_bin(int(request.prompt_tokens_hint)) for request in selected] == list( + range(len(MODULE.INPUT_BINS) - 1) + ) + assert {request.arrival_s for request in selected} == {0.0}