experiment: add per-gpu workload control

2026-07-20 18:48:22 +08:00
parent a033a72195
commit 809ad9ffef
8 changed files with 330 additions and 7 deletions
--- a/runs/frontier-workload-regime-taxonomy-v0/jobs_simulator_per_gpu_v1.toml
+++ b/runs/frontier-workload-regime-taxonomy-v0/jobs_simulator_per_gpu_v1.toml
@@ -0,0 +1,33 @@
+version = 1
+
+[[jobs]]
+name = "workload-regime-per-gpu-sim-tp1-v1-20260720"
+gpus = 1
+gpu_model = "H20"
+hosts = ["dash1"]
+command = "TRACES_ROOT=/home/admin/cpfs/wjh/aituner/workload-regime-20260720/traces-per-gpu-low SIM_GENERATION=per-gpu-v1 bash runs/frontier-workload-regime-taxonomy-v0/run_simulator_group.sh 1 both"
+artifacts = []
+
+[[jobs]]
+name = "workload-regime-per-gpu-sim-tp2-v1-20260720"
+gpus = 1
+gpu_model = "H20"
+hosts = ["dash2"]
+command = "TRACES_ROOT=/home/admin/cpfs/wjh/aituner/workload-regime-20260720/traces-per-gpu-low SIM_GENERATION=per-gpu-v1 bash runs/frontier-workload-regime-taxonomy-v0/run_simulator_group.sh 2 both"
+artifacts = []
+
+[[jobs]]
+name = "workload-regime-per-gpu-sim-tp4-noprefix-v1-20260720"
+gpus = 1
+gpu_model = "H20"
+hosts = ["dash3"]
+command = "TRACES_ROOT=/home/admin/cpfs/wjh/aituner/workload-regime-20260720/traces-per-gpu-low SIM_GENERATION=per-gpu-v1 bash runs/frontier-workload-regime-taxonomy-v0/run_simulator_group.sh 4 false"
+artifacts = []
+
+[[jobs]]
+name = "workload-regime-per-gpu-sim-tp4-prefix-v1-20260720"
+gpus = 1
+gpu_model = "H20"
+hosts = ["dash4"]
+command = "TRACES_ROOT=/home/admin/cpfs/wjh/aituner/workload-regime-20260720/traces-per-gpu-low SIM_GENERATION=per-gpu-v1 bash runs/frontier-workload-regime-taxonomy-v0/run_simulator_group.sh 4 true"
+artifacts = []
--- a/runs/frontier-workload-regime-taxonomy-v0/materialize_workload_regimes.py
+++ b/runs/frontier-workload-regime-taxonomy-v0/materialize_workload_regimes.py
@@ -91,6 +91,10 @@ def vector_sha256(rows: Iterable[dict[str, Any]]) -> str:
    return digest.hexdigest()


+def rho_label(rho: float) -> str:
+    return f"rho{format(rho, '.12g').replace('.', 'p')}"
+
+
 def load_source(
    public_path: Path | None, private_path: Path
 ) -> list[dict[str, Any]]:
@@ -377,8 +381,7 @@ def main() -> None:
                target_rate=target_rate,
                fixed_token_ids=fixed_token_ids,
            )
-            rho_label = f"rho{rho:.2f}".replace(".", "p")
-            root = args.output_root / family.name / rho_label
+            root = args.output_root / family.name / rho_label(rho)
            cases.append(
                write_case(
                    root,
--- a/runs/frontier-workload-regime-taxonomy-v0/project_per_gpu_load.py
+++ b/runs/frontier-workload-regime-taxonomy-v0/project_per_gpu_load.py
@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+"""Project base workload cases to a constant per-GPU request-rate contract."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--source-manifest", type=Path, required=True)
+    parser.add_argument("--output-root", type=Path, required=True)
+    parser.add_argument("--tp", type=int, action="append", default=None)
+    return parser.parse_args()
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def rho_label(rho: float) -> str:
+    return format(rho, ".12g").replace(".", "p")
+
+
+def project_case(
+    case: dict[str, Any], output_root: Path, tp: int
+) -> dict[str, Any]:
+    source_public = Path(case["public_csv"])
+    source_private = Path(case["private_jsonl"])
+    if sha256(source_public) != case["public_csv_sha256"]:
+        raise ValueError(f"source public digest mismatch: {source_public}")
+    if sha256(source_private) != case["private_jsonl_sha256"]:
+        raise ValueError(f"source private digest mismatch: {source_private}")
+
+    case_root = output_root / f"tp{tp}" / case["family"] / f"rho{rho_label(case['rho'])}"
+    public_root = case_root / "public"
+    private_root = case_root / "private"
+    public_root.mkdir(parents=True, exist_ok=True)
+    private_root.mkdir(parents=True, exist_ok=True)
+    public_path = public_root / "frontier.csv"
+    private_path = private_root / "real_requests.jsonl"
+
+    with source_public.open(newline="") as source:
+        reader = csv.DictReader(source)
+        rows = list(reader)
+        fieldnames = reader.fieldnames
+    if not rows or fieldnames is None:
+        raise ValueError(f"empty public source: {source_public}")
+    with public_path.open("w", newline="") as output:
+        writer = csv.DictWriter(output, fieldnames=fieldnames, lineterminator="\n")
+        writer.writeheader()
+        for row in rows:
+            row["arrived_at"] = f"{float(row['arrived_at']) / tp:.12f}"
+            writer.writerow(row)
+
+    private_rows = [
+        json.loads(line) for line in source_private.open() if line.strip()
+    ]
+    if len(private_rows) != len(rows):
+        raise ValueError(f"public/private row mismatch: {source_public}")
+    with private_path.open("w") as output:
+        for row in private_rows:
+            row["arrived_at"] = float(row["arrived_at"]) / tp
+            output.write(json.dumps(row, separators=(",", ":")) + "\n")
+
+    projected = dict(case)
+    projected.update(
+        {
+            "load_contract": "constant_per_gpu_request_rate",
+            "tp": tp,
+            "rho_per_gpu": case["rho"],
+            "per_gpu_offered_request_rate": case["global_offered_request_rate"],
+            "global_offered_request_rate": case["global_offered_request_rate"] * tp,
+            "empirical_interarrival_rate": case["empirical_interarrival_rate"] * tp,
+            "decode_offered_tokens_per_second_per_gpu": case[
+                "decode_offered_tokens_per_second"
+            ],
+            "decode_offered_tokens_per_second": case[
+                "decode_offered_tokens_per_second"
+            ]
+            * tp,
+            "last_arrival_s": case["last_arrival_s"] / tp,
+            "source_case_public_csv": str(source_public),
+            "source_case_public_csv_sha256": case["public_csv_sha256"],
+            "source_case_private_jsonl": str(source_private),
+            "source_case_private_jsonl_sha256": case["private_jsonl_sha256"],
+            "public_csv": str(public_path.resolve()),
+            "public_csv_sha256": sha256(public_path),
+            "private_jsonl": str(private_path.resolve()),
+            "private_jsonl_sha256": sha256(private_path),
+        }
+    )
+    (public_root / "manifest.json").write_text(
+        json.dumps(projected, indent=2, sort_keys=True) + "\n"
+    )
+    return projected
+
+
+def main() -> None:
+    args = parse_args()
+    tps = args.tp or [1, 2, 4]
+    if any(tp not in (1, 2, 4) for tp in tps):
+        raise ValueError("TP must be 1, 2, or 4")
+    suite = json.loads(args.source_manifest.read_text())
+    projected = [
+        project_case(case, args.output_root, tp)
+        for tp in tps
+        for case in suite["cases"]
+    ]
+    output = {
+        "schema": "frontier-workload-regime-per-gpu-v1",
+        "load_contract": "constant_per_gpu_request_rate",
+        "source_manifest": str(args.source_manifest.resolve()),
+        "source_manifest_sha256": sha256(args.source_manifest),
+        "tensor_parallel_sizes": tps,
+        "cases": projected,
+    }
+    args.output_root.mkdir(parents=True, exist_ok=True)
+    path = args.output_root / "manifest.json"
+    path.write_text(json.dumps(output, indent=2, sort_keys=True) + "\n")
+    print(path)
+
+
+if __name__ == "__main__":
+    main()
--- a/runs/frontier-workload-regime-taxonomy-v0/run_real_pilot.sh
+++ b/runs/frontier-workload-regime-taxonomy-v0/run_real_pilot.sh
@@ -0,0 +1,74 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+FAMILY="${1:?usage: run_real_pilot.sh FAMILY RHO TP MNS SERVER_PORT}"
+RHO="${2:?usage: run_real_pilot.sh FAMILY RHO TP MNS SERVER_PORT}"
+TP="${3:?usage: run_real_pilot.sh FAMILY RHO TP MNS SERVER_PORT}"
+MNS="${4:?usage: run_real_pilot.sh FAMILY RHO TP MNS SERVER_PORT}"
+SERVER_PORT="${5:?usage: run_real_pilot.sh FAMILY RHO TP MNS SERVER_PORT}"
+
+case "${TP}" in
+  1|2|4) ;;
+  *) echo "ERROR: TP must be 1, 2, or 4" >&2; exit 2 ;;
+esac
+case "${MNS}" in
+  8|16|32|64) ;;
+  *) echo "ERROR: MNS must be 8, 16, 32, or 64" >&2; exit 2 ;;
+esac
+
+EXPERIMENT_ROOT="${EXPERIMENT_ROOT:-/home/admin/cpfs/wjh/aituner/workload-regime-20260720}"
+TRACES_MANIFEST="${TRACES_MANIFEST:-${EXPERIMENT_ROOT}/traces/manifest.json}"
+VENV_ROOT="${VENV_ROOT:-/home/admin/cpfs/wjh/venvs/vllm-0.20.0-cu129-workload-regime}"
+RUNNER_ROOT="${RUNNER_ROOT:-/home/admin/cpfs/wjh/aituner/aituner-graph-piecewise-bdc357d/runs/frontier-fidelity-envelope-v1}"
+RUNNER="${RUNNER_ROOT}/run_qwen30_exact_trace_real_anchor.sh"
+MODEL_ROOT="${MODEL_ROOT:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B}"
+OUTPUT_GENERATION="${OUTPUT_GENERATION:-v1}"
+
+for path in "${TRACES_MANIFEST}" "${VENV_ROOT}/bin/vllm" "${RUNNER}" \
+  "${MODEL_ROOT}/config.json"; do
+  [[ -e "${path}" ]] || { echo "ERROR: missing ${path}" >&2; exit 1; }
+done
+
+mapfile -t CASE_FIELDS < <(
+  jq -r --arg family "${FAMILY}" --argjson rho "${RHO}" '
+    .cases
+    | map(select(.family == $family and ((.rho - $rho) | fabs) < 1e-12))
+    | .[]
+    | [.private_jsonl, (.prefix_caching | tostring), .private_jsonl_sha256]
+    | @tsv
+  ' "${TRACES_MANIFEST}"
+)
+[[ "${#CASE_FIELDS[@]}" -eq 1 ]] || {
+  echo "ERROR: expected one manifest case for family=${FAMILY} rho=${RHO}, got ${#CASE_FIELDS[@]}" >&2
+  exit 1
+}
+IFS=$'\t' read -r REQUESTS_FILE PREFIX_CACHING EXPECTED_REQUESTS_SHA \
+  <<< "${CASE_FIELDS[0]}"
+[[ -f "${REQUESTS_FILE}" ]] || { echo "ERROR: missing ${REQUESTS_FILE}" >&2; exit 1; }
+ACTUAL_REQUESTS_SHA="$(sha256sum "${REQUESTS_FILE}" | cut -d' ' -f1)"
+[[ "${ACTUAL_REQUESTS_SHA}" == "${EXPECTED_REQUESTS_SHA}" ]] || {
+  echo "ERROR: private trace digest mismatch" >&2
+  exit 1
+}
+
+RHO_LABEL="${RHO//./p}"
+TRACE_LABEL="${FAMILY}-rho${RHO_LABEL}-tp${TP}-mns${MNS}"
+OUTPUT_ROOT="${EXPERIMENT_ROOT}/real-pilot-${OUTPUT_GENERATION}/${TRACE_LABEL}"
+[[ ! -e "${OUTPUT_ROOT}/results/result.json" ]] || {
+  echo "ERROR: refusing to overwrite completed ${OUTPUT_ROOT}" >&2
+  exit 1
+}
+
+FLASHINFER_WORKSPACE_BASE="${FLASHINFER_WORKSPACE_BASE:-/tmp/wjh/flashinfer-workload-regime-vllm020-cu129}"
+mkdir -p "${FLASHINFER_WORKSPACE_BASE}"
+echo "WORKLOAD_REGIME_REAL_PILOT family=${FAMILY} rho=${RHO} tp=${TP} mns=${MNS} prefix=${PREFIX_CACHING} requests_sha256=${ACTUAL_REQUESTS_SHA} venv=${VENV_ROOT} output=${OUTPUT_ROOT}"
+
+cd "${RUNNER_ROOT}"
+OUTPUT_ROOT="${OUTPUT_ROOT}" REQUESTS_FILE="${REQUESTS_FILE}" \
+TP="${TP}" MNS="${MNS}" TRACE_LABEL="${TRACE_LABEL}" \
+SERVER_PORT="${SERVER_PORT}" PREFIX_CACHING="${PREFIX_CACHING}" \
+VENV_ROOT="${VENV_ROOT}" MODEL_ROOT="${MODEL_ROOT}" \
+FLASHINFER_WORKSPACE_BASE="${FLASHINFER_WORKSPACE_BASE}" \
+SERVER_READY_ATTEMPTS="${SERVER_READY_ATTEMPTS:-300}" \
+bash "${RUNNER}"
--- a/runs/frontier-workload-regime-taxonomy-v0/run_simulator_group.sh
+++ b/runs/frontier-workload-regime-taxonomy-v0/run_simulator_group.sh
@@ -47,9 +47,9 @@ run_mode() {
  fi

  mapfile -t families < <(
-    jq -r --argjson prefix "${prefix_mode}" '
+    jq -r --argjson prefix "${prefix_mode}" --argjson tp "${TP}" '
      .cases
-      | map(select(.prefix_caching == $prefix))
+      | map(select(.prefix_caching == $prefix and ((.tp // $tp) == $tp)))
      | map(.family)
      | unique
      | .[]
@@ -75,9 +75,9 @@ run_mode() {
    traces=()
    command=()
    mapfile -t traces < <(
-      jq -r --arg family "${family}" '
+      jq -r --arg family "${family}" --argjson tp "${TP}" '
        .cases
-        | map(select(.family == $family))
+        | map(select(.family == $family and ((.tp // $tp) == $tp)))
        | sort_by(.global_offered_request_rate)
        | .[]
        | "\(.family)-rho\(.rho | tostring | gsub("\\."; "p"))=\(.public_csv)"