From 3869101b5aa9245a01c869467304f177f55a7501 Mon Sep 17 00:00:00 2001 From: lullu57 Date: Wed, 4 Mar 2026 15:31:17 -0500 Subject: [PATCH 1/5] Add configurable Slurm submission workflow with storage and email alerts --- README.md | 48 +++++ .../slurm_experiment_list.example.txt | 4 + scripts/slurm/run_kerneltuner_array.sbatch | 188 ++++++++++++++++++ scripts/slurm/submit_kerneltuner.sh | 170 ++++++++++++++++ 4 files changed, 410 insertions(+) create mode 100644 configs/experiments/slurm_experiment_list.example.txt create mode 100755 scripts/slurm/run_kerneltuner_array.sbatch create mode 100755 scripts/slurm/submit_kerneltuner.sh diff --git a/README.md b/README.md index 55ef940..3b6af62 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,54 @@ The initial workflow is document-driven: 3. Implement modules against the specs in `docs/specs/`. 4. Keep deviations explicit through ADRs rather than ad hoc code changes. +## Slurm Submission Helpers + +For cluster execution, the repo includes reusable Slurm scripts: + +- `scripts/slurm/run_kerneltuner_array.sbatch`: array worker that maps one array task to one experiment YAML. +- `scripts/slurm/submit_kerneltuner.sh`: submit wrapper that computes array size from a list file. + +Example list file: + +- `configs/experiments/slurm_experiment_list.example.txt` + +Basic usage: + +```bash +chmod +x scripts/slurm/submit_kerneltuner.sh +cp configs/experiments/slurm_experiment_list.example.txt configs/experiments/slurm_experiment_list.txt +scripts/slurm/submit_kerneltuner.sh \ + --list configs/experiments/slurm_experiment_list.txt \ + --partition gpunodes \ + --gpu-type rtx_a2000 \ + --gpus 1 \ + --time 0-04:00 \ + --cpus 4 \ + --mem 24GB \ + --log-dir /scratch/scratch-space/expires-xxxx/$USER/kerneltuner_logs \ + --scratch-root /scratch/scratch-space/expires-xxxx/$USER/kerneltuner \ + --artifact-root /scratch/scratch-space/expires-xxxx/$USER/kerneltuner_artifacts \ + --mail-user you@example.com \ + --mail-type BEGIN,END,FAIL \ + --alert-email you@example.com \ + --alert-on-start +``` + +Useful environment overrides: + +- `RUN_COMMAND_TEMPLATE` (default: `ktune run-experiment --experiment "{experiment}"`) +- `INSTALL_PACKAGES` (`0` to skip pip install in jobs) +- `EXTRA_PIP_PACKAGES` (space-separated extra pip packages) +- `SKIP_IF_ARTIFACTS_EXIST` (`0` to force reruns) +- `DRY_RUN=1` (worker-level dry run) +- `ALERT_EMAIL`, `ALERT_ON_START`, `ALERT_ON_END`, `ALERT_ON_FAIL` for worker-level alerts + +Notes: + +- `--artifact-root` overrides `artifact_root` in experiment YAML at submission time. +- If no scratch path exists on the node, jobs fall back to `/.scratch/$USER`. +- Slurm native email (`--mail-user/--mail-type`) and worker-level alerts can be used together. + ## What This Repo Is Not - It is not a Triton compiler redesign effort. diff --git a/configs/experiments/slurm_experiment_list.example.txt b/configs/experiments/slurm_experiment_list.example.txt new file mode 100644 index 0000000..d777083 --- /dev/null +++ b/configs/experiments/slurm_experiment_list.example.txt @@ -0,0 +1,4 @@ +# One experiment YAML path per line. +# Blank lines and lines starting with '#' are ignored. + +configs/experiments/gemm_smoke.example.yaml diff --git a/scripts/slurm/run_kerneltuner_array.sbatch b/scripts/slurm/run_kerneltuner_array.sbatch new file mode 100755 index 0000000..05fc86b --- /dev/null +++ b/scripts/slurm/run_kerneltuner_array.sbatch @@ -0,0 +1,188 @@ +#!/bin/bash --login +#SBATCH --job-name=kerneltuner +#SBATCH --partition=gpunodes +#SBATCH --time=0-02:00 +#SBATCH -c 4 +#SBATCH --mem=24GB +#SBATCH --gres=gpu:1 +#SBATCH --array=0-0 +#SBATCH --output=slurm_jobs/kerneltuner_%A_%a.out +#SBATCH --error=slurm_jobs/kerneltuner_%A_%a.err + +set -euo pipefail + +send_alert() { + local phase="$1" + local details="$2" + local email="${ALERT_EMAIL:-}" + if [[ -z "$email" ]]; then + return 0 + fi + local subject_prefix="${ALERT_SUBJECT_PREFIX:-[KernelTuner]}" + local subject="${subject_prefix} ${phase} job=${SLURM_JOB_ID:-local} task=${SLURM_ARRAY_TASK_ID:-0}" + local body + body=$( + cat </dev/null 2>&1; then + printf '%s\n' "$body" | mail -s "$subject" "$email" || true + elif command -v sendmail >/dev/null 2>&1; then + { + printf 'To: %s\n' "$email" + printf 'Subject: %s\n' "$subject" + printf '\n%s\n' "$body" + } | sendmail -t || true + fi +} + +on_exit() { + local exit_code=$? + if [[ $exit_code -ne 0 && "${ALERT_ON_FAIL:-1}" == "1" ]]; then + send_alert "FAIL" "exit_code=${exit_code}" + fi +} +trap on_exit EXIT + +if [[ $# -lt 1 ]]; then + echo "Usage: $0 [workspace_root] [venv_name]" >&2 + exit 2 +fi + +EXPERIMENT_LIST_FILE="$1" +WORKSPACE_ROOT="${2:-$SLURM_SUBMIT_DIR}" +VENV_NAME="${3:-kerneltuner_env}" + +if [[ ! -f "$EXPERIMENT_LIST_FILE" ]]; then + echo "ERROR: Experiment list file not found: $EXPERIMENT_LIST_FILE" >&2 + exit 1 +fi + +mkdir -p "$WORKSPACE_ROOT/slurm_jobs" + +mapfile -t EXPERIMENTS < <(awk 'NF && $1 !~ /^#/' "$EXPERIMENT_LIST_FILE") + +if [[ "${#EXPERIMENTS[@]}" -eq 0 ]]; then + echo "ERROR: No experiments found in $EXPERIMENT_LIST_FILE" >&2 + exit 1 +fi + +TASK_ID="${SLURM_ARRAY_TASK_ID:-0}" +if (( TASK_ID < 0 || TASK_ID >= ${#EXPERIMENTS[@]} )); then + echo "ERROR: SLURM_ARRAY_TASK_ID=$TASK_ID out of range (0..$((${#EXPERIMENTS[@]} - 1)))" >&2 + exit 1 +fi + +EXPERIMENT_CFG="${EXPERIMENTS[$TASK_ID]}" +if [[ ! -f "$EXPERIMENT_CFG" ]]; then + echo "ERROR: Experiment config not found: $EXPERIMENT_CFG" >&2 + exit 1 +fi + +resolve_artifact_root() { + local cfg_path="$1" + local raw_root + raw_root="$(awk -F': *' '$1=="artifact_root"{print $2; exit}' "$cfg_path" | tr -d "\"'[:space:]")" + if [[ -z "$raw_root" ]]; then + raw_root="artifacts" + fi + if [[ "$raw_root" = /* ]]; then + printf '%s\n' "$raw_root" + else + printf '%s/%s\n' "$WORKSPACE_ROOT" "$raw_root" + fi +} + +echo ">>> KernelTuner Slurm Task <<<" +echo "Working directory: $WORKSPACE_ROOT" +echo "Experiment list: $EXPERIMENT_LIST_FILE" +echo "Array task id: $TASK_ID" +echo "Experiment config: $EXPERIMENT_CFG" +if [[ "${ALERT_ON_START:-0}" == "1" ]]; then + send_alert "START" "experiment_cfg=${EXPERIMENT_CFG}" +fi + +EXPERIMENT_ID="$(awk -F': *' '$1=="experiment_id"{print $2; exit}' "$EXPERIMENT_CFG" | tr -d "\"'[:space:]")" +if [[ -z "${EXPERIMENT_ID}" ]]; then + EXPERIMENT_ID="$(basename "$EXPERIMENT_CFG" .yaml)" +fi + +EXPERIMENT_TO_RUN="$EXPERIMENT_CFG" + +SCRATCH_ROOT="${SCRATCH_ROOT:-}" +if [[ -z "$SCRATCH_ROOT" ]]; then + SCRATCH_ROOT="$(ls -d /scratch/scratch-space/expires-* 2>/dev/null | head -n 1 || true)" +fi +if [[ -z "$SCRATCH_ROOT" ]]; then + SCRATCH_ROOT="$WORKSPACE_ROOT/.scratch/$USER" +fi +mkdir -p "$SCRATCH_ROOT" +echo "Scratch root: $SCRATCH_ROOT" + +if [[ -n "${ARTIFACT_ROOT_OVERRIDE:-}" ]]; then + mkdir -p "$SCRATCH_ROOT/kerneltuner_tmp" + OVERRIDE_CFG="$SCRATCH_ROOT/kerneltuner_tmp/${EXPERIMENT_ID}_${SLURM_JOB_ID:-local}_${TASK_ID}.yaml" + awk -v override="$ARTIFACT_ROOT_OVERRIDE" ' + BEGIN { seen = 0 } + /^artifact_root:/ { print "artifact_root: " override; seen = 1; next } + { print } + END { if (seen == 0) print "artifact_root: " override } + ' "$EXPERIMENT_CFG" > "$OVERRIDE_CFG" + EXPERIMENT_TO_RUN="$OVERRIDE_CFG" + echo "Using artifact_root override: $ARTIFACT_ROOT_OVERRIDE" +fi + +if [[ "${SKIP_IF_ARTIFACTS_EXIST:-1}" == "1" ]]; then + ARTIFACT_ROOT_RESOLVED="$(resolve_artifact_root "$EXPERIMENT_TO_RUN")" + if compgen -G "$ARTIFACT_ROOT_RESOLVED/$EXPERIMENT_ID/*/summary.json" > /dev/null; then + echo ">>> Found existing summary for experiment_id=$EXPERIMENT_ID; skipping." + if [[ "${ALERT_ON_END:-0}" == "1" ]]; then + send_alert "SKIP" "existing_summary=1 experiment_id=${EXPERIMENT_ID}" + fi + exit 0 + fi +fi + +VENV_PATH="$SCRATCH_ROOT/$VENV_NAME" +if [[ ! -d "$VENV_PATH" ]]; then + echo "Creating virtual environment at $VENV_PATH" + python3 -m venv "$VENV_PATH" +fi + +source "$VENV_PATH/bin/activate" + +if [[ "${INSTALL_PACKAGES:-1}" == "1" ]]; then + echo "Installing dependencies into $VENV_PATH" + pip install --upgrade pip + pip install --no-cache-dir -e "$WORKSPACE_ROOT" pyyaml pandas pyarrow + if [[ -n "${EXTRA_PIP_PACKAGES:-}" ]]; then + pip install --no-cache-dir ${EXTRA_PIP_PACKAGES} + fi +fi + +export PYTHONPATH="$WORKSPACE_ROOT/src:${PYTHONPATH:-}" + +RUN_COMMAND_TEMPLATE="${RUN_COMMAND_TEMPLATE:-ktune run-experiment --experiment \"{experiment}\"}" +EXPERIMENT_TOKEN='{experiment}' +WORKSPACE_TOKEN='{workspace}' +RUN_COMMAND="${RUN_COMMAND_TEMPLATE//$EXPERIMENT_TOKEN/$EXPERIMENT_TO_RUN}" +RUN_COMMAND="${RUN_COMMAND//$WORKSPACE_TOKEN/$WORKSPACE_ROOT}" + +echo "Run command: $RUN_COMMAND" +if [[ "${DRY_RUN:-0}" == "1" ]]; then + echo "DRY_RUN=1 set; not executing." + exit 0 +fi + +bash -lc "$RUN_COMMAND" + +echo ">>> Task complete" +if [[ "${ALERT_ON_END:-1}" == "1" ]]; then + send_alert "SUCCESS" "experiment_cfg=${EXPERIMENT_TO_RUN}" +fi diff --git a/scripts/slurm/submit_kerneltuner.sh b/scripts/slurm/submit_kerneltuner.sh new file mode 100755 index 0000000..c68c69f --- /dev/null +++ b/scripts/slurm/submit_kerneltuner.sh @@ -0,0 +1,170 @@ +#!/usr/bin/env bash +set -euo pipefail + +usage() { + cat <<'EOF' +Usage: + scripts/slurm/submit_kerneltuner.sh --list [options] + +Options: + --list Required. Text file with one experiment YAML path per line. + --job-name Slurm job name. Default: kerneltuner + --partition Slurm partition. Default: gpunodes + --time Slurm time limit. Default: 0-02:00 + --cpus CPUs per task. Default: 4 + --mem Memory per task. Default: 24GB + --gpu-type GPU type for --gres (e.g. rtx_a2000) + --gpus Number of GPUs. Default: 1 + --mail-user Email for Slurm notifications (optional) + --mail-type Slurm mail type string (optional) + --venv-name Virtualenv directory name. Default: kerneltuner_env + --workspace Workspace root. Default: current directory + --log-dir Directory for Slurm .out/.err files. Default: /slurm_jobs + --scratch-root Override SCRATCH_ROOT used by worker script + --artifact-root Override experiment artifact_root for this submission + --alert-email Send worker alerts to this email (mail/sendmail on node) + --alert-on-start Enable start alert from worker + --alert-on-end Enable success/skip alert from worker + --alert-on-fail Enable failure alert from worker (default on) + --dry-run Print sbatch command without submitting + +Environment vars passed through to worker script: + RUN_COMMAND_TEMPLATE e.g. 'ktune run-experiment --experiment "{experiment}"' + DRY_RUN Set to 1 for worker-level dry run + INSTALL_PACKAGES Set to 0 to skip pip install + EXTRA_PIP_PACKAGES Extra pip packages to install in job env + SKIP_IF_ARTIFACTS_EXIST Set to 0 to disable skip-on-existing-summary behavior +EOF +} + +LIST_FILE="" +JOB_NAME="kerneltuner" +PARTITION="gpunodes" +TIME_LIMIT="0-02:00" +CPUS="4" +MEMORY="24GB" +GPU_TYPE="" +GPU_COUNT="1" +MAIL_USER="" +MAIL_TYPE="" +VENV_NAME="kerneltuner_env" +WORKSPACE_ROOT="$(pwd)" +LOG_DIR="" +SCRATCH_ROOT_OVERRIDE="" +ARTIFACT_ROOT_OVERRIDE="" +ALERT_EMAIL="" +ALERT_ON_START="0" +ALERT_ON_END="1" +ALERT_ON_FAIL="1" +DRY_RUN_SUBMIT="0" + +while [[ $# -gt 0 ]]; do + case "$1" in + --list) LIST_FILE="${2:-}"; shift 2 ;; + --job-name) JOB_NAME="${2:-}"; shift 2 ;; + --partition) PARTITION="${2:-}"; shift 2 ;; + --time) TIME_LIMIT="${2:-}"; shift 2 ;; + --cpus) CPUS="${2:-}"; shift 2 ;; + --mem) MEMORY="${2:-}"; shift 2 ;; + --gpu-type) GPU_TYPE="${2:-}"; shift 2 ;; + --gpus) GPU_COUNT="${2:-}"; shift 2 ;; + --mail-user) MAIL_USER="${2:-}"; shift 2 ;; + --mail-type) MAIL_TYPE="${2:-}"; shift 2 ;; + --venv-name) VENV_NAME="${2:-}"; shift 2 ;; + --workspace) WORKSPACE_ROOT="${2:-}"; shift 2 ;; + --log-dir) LOG_DIR="${2:-}"; shift 2 ;; + --scratch-root) SCRATCH_ROOT_OVERRIDE="${2:-}"; shift 2 ;; + --artifact-root) ARTIFACT_ROOT_OVERRIDE="${2:-}"; shift 2 ;; + --alert-email) ALERT_EMAIL="${2:-}"; shift 2 ;; + --alert-on-start) ALERT_ON_START="1"; shift ;; + --alert-on-end) ALERT_ON_END="1"; shift ;; + --alert-on-fail) ALERT_ON_FAIL="1"; shift ;; + --dry-run) DRY_RUN_SUBMIT="1"; shift ;; + -h|--help) usage; exit 0 ;; + *) echo "Unknown option: $1" >&2; usage; exit 2 ;; + esac +done + +if [[ -z "$LIST_FILE" ]]; then + echo "ERROR: --list is required" >&2 + usage + exit 2 +fi + +if [[ ! -f "$LIST_FILE" ]]; then + echo "ERROR: list file not found: $LIST_FILE" >&2 + exit 1 +fi + +readarray -t EXP_LINES < <(awk 'NF && $1 !~ /^#/' "$LIST_FILE") +NUM_EXPERIMENTS="${#EXP_LINES[@]}" +if [[ "$NUM_EXPERIMENTS" -eq 0 ]]; then + echo "ERROR: no experiment entries found in $LIST_FILE" >&2 + exit 1 +fi + +ARRAY_SPEC="0-$((NUM_EXPERIMENTS - 1))" +if [[ -z "$LOG_DIR" ]]; then + LOG_DIR="$WORKSPACE_ROOT/slurm_jobs" +fi +mkdir -p "$LOG_DIR" + +if [[ -n "$GPU_TYPE" ]]; then + GRES="gpu:${GPU_TYPE}:${GPU_COUNT}" +else + GRES="gpu:${GPU_COUNT}" +fi + +SBATCH_CMD=( + sbatch + "--job-name=$JOB_NAME" + "--partition=$PARTITION" + "--time=$TIME_LIMIT" + "--cpus-per-task=$CPUS" + "--mem=$MEMORY" + "--gres=$GRES" + "--array=$ARRAY_SPEC" + "--output=$LOG_DIR/${JOB_NAME}_%A_%a.out" + "--error=$LOG_DIR/${JOB_NAME}_%A_%a.err" +) + +if [[ -n "$MAIL_USER" ]]; then + SBATCH_CMD+=("--mail-user=$MAIL_USER") +fi +if [[ -n "$MAIL_TYPE" ]]; then + SBATCH_CMD+=("--mail-type=$MAIL_TYPE") +fi + +EXPORT_VARS=("ALL") +if [[ -n "$SCRATCH_ROOT_OVERRIDE" ]]; then + EXPORT_VARS+=("SCRATCH_ROOT=$SCRATCH_ROOT_OVERRIDE") +fi +if [[ -n "$ARTIFACT_ROOT_OVERRIDE" ]]; then + EXPORT_VARS+=("ARTIFACT_ROOT_OVERRIDE=$ARTIFACT_ROOT_OVERRIDE") +fi +if [[ -n "$ALERT_EMAIL" ]]; then + EXPORT_VARS+=("ALERT_EMAIL=$ALERT_EMAIL") +fi +EXPORT_VARS+=("ALERT_ON_START=$ALERT_ON_START") +EXPORT_VARS+=("ALERT_ON_END=$ALERT_ON_END") +EXPORT_VARS+=("ALERT_ON_FAIL=$ALERT_ON_FAIL") +SBATCH_CMD+=("--export=$(IFS=,; echo "${EXPORT_VARS[*]}")") + +SBATCH_CMD+=( + "$WORKSPACE_ROOT/scripts/slurm/run_kerneltuner_array.sbatch" + "$LIST_FILE" + "$WORKSPACE_ROOT" + "$VENV_NAME" +) + +echo "Submitting $NUM_EXPERIMENTS experiments as array: $ARRAY_SPEC" +echo "sbatch command:" +printf ' %q' "${SBATCH_CMD[@]}" +echo + +if [[ "$DRY_RUN_SUBMIT" == "1" ]]; then + echo "Dry run enabled; not submitting." + exit 0 +fi + +"${SBATCH_CMD[@]}" From 8280b12e933f55f3530d96d900171f6974b33b49 Mon Sep 17 00:00:00 2001 From: lullu57 Date: Mon, 23 Mar 2026 13:41:17 -0400 Subject: [PATCH 2/5] Fix benchmark fairness and reportability semantics --- configs/counters/compute_lite.yaml | 1 + configs/counters/memory_lite.yaml | 1 + configs/studies/validation_phase.yaml | 2 +- src/kernel_tuner/analysis/comparison.py | 39 ++++- src/kernel_tuner/analysis/opportunities.py | 14 +- src/kernel_tuner/analysis/reporting.py | 41 ++++- src/kernel_tuner/baselines/strategies.py | 16 +- src/kernel_tuner/benchmark/harness.py | 64 +++++--- src/kernel_tuner/common/schema.py | 3 + src/kernel_tuner/experiments/orchestrator.py | 111 ++++++++++---- src/kernel_tuner/profiling/adapter.py | 151 +++++++++++++++---- src/kernel_tuner/profiling/compatibility.py | 66 ++++++-- src/kernel_tuner/selector/engine.py | 24 ++- tests/unit/test_phase2_analysis.py | 77 +++++++++- tests/unit/test_profiling.py | 94 ++++++++++++ tests/unit/test_reporting.py | 111 +++++++++++++- 16 files changed, 695 insertions(+), 120 deletions(-) create mode 100644 tests/unit/test_profiling.py diff --git a/configs/counters/compute_lite.yaml b/configs/counters/compute_lite.yaml index 0f6bd7c..4d79137 100644 --- a/configs/counters/compute_lite.yaml +++ b/configs/counters/compute_lite.yaml @@ -3,6 +3,7 @@ description: Compute-oriented Nsight Compute counter set for reportable GEMM stu tool: ncu replay_mode: kernel target_processes: all +kernel_name_regex: matmul_kernel kernel_family_filters: - gemm counters: diff --git a/configs/counters/memory_lite.yaml b/configs/counters/memory_lite.yaml index 8f2ef0a..07cf04e 100644 --- a/configs/counters/memory_lite.yaml +++ b/configs/counters/memory_lite.yaml @@ -3,6 +3,7 @@ description: Memory-oriented Nsight Compute counter set for reportable LayerNorm tool: ncu replay_mode: kernel target_processes: all +kernel_name_regex: layer_norm_kernel kernel_family_filters: - layernorm counters: diff --git a/configs/studies/validation_phase.yaml b/configs/studies/validation_phase.yaml index 77e15a8..529add9 100644 --- a/configs/studies/validation_phase.yaml +++ b/configs/studies/validation_phase.yaml @@ -1,7 +1,7 @@ study_id: validation_phase hypotheses: - hypothesis_id: H1 - description: Cheap compile signals prune obvious losers, but do not reliably rank near-tied GEMM configs. + description: Cheap compile signals prune obvious losers, but produce noticeably unstable ranking quality on representative GEMM workloads. clauses: - left: source: strategy_rows diff --git a/src/kernel_tuner/analysis/comparison.py b/src/kernel_tuner/analysis/comparison.py index 0eaf4e6..6754faf 100644 --- a/src/kernel_tuner/analysis/comparison.py +++ b/src/kernel_tuner/analysis/comparison.py @@ -73,7 +73,11 @@ def compare_runs( store.initialize_manifest(manifest) try: run_payloads = _resolve_run_payloads(study_spec, study_path) + if not run_payloads: + raise ValueError("no study runs matched the configured groups and filters") strategy_rows = _build_strategy_rows(run_payloads) + if strategy_rows.empty: + raise ValueError("no held-out strategy rows were available after loading matched study runs") stability_report = _build_stability_report(strategy_rows) hypothesis_results = _evaluate_hypotheses(study_spec, strategy_rows, stability_report) opportunity_catalog = _aggregate_opportunities(run_payloads) @@ -288,7 +292,10 @@ def _build_strategy_rows(run_payloads: list[dict[str, Any]]) -> pd.DataFrame: if not counter_availability.empty else {} ) - for workload_class in sorted(held_out["workload_class"].dropna().unique()): + workload_classes = sorted(held_out["workload_class"].dropna().unique()) + if not workload_classes: + continue + for workload_class in workload_classes: subset = held_out[held_out["workload_class"] == workload_class].copy() pivot = subset.pivot(index="shape_id", columns="strategy_id", values="latency_median_us") for strategy_id in pivot.columns: @@ -486,10 +493,12 @@ def _evaluate_clause( right_value = clause.right_constant supported = _compare_hypothesis_values(left_value, right_value, clause.comparator, clause.minimum_delta) comparator_name = clause.comparator.value if hasattr(clause.comparator, "value") else str(clause.comparator) + delta = left_value - right_value evidence = ( f"{clause.left.metric}={left_value:.4f} " f"{comparator_name} " - f"{right_value:.4f} with minimum_delta={clause.minimum_delta:.4f}" + f"{right_value:.4f} with minimum_delta={clause.minimum_delta:.4f} " + f"(observed_delta={delta:.4f})" ) return {"status": "ok", "supported": supported, "evidence": evidence} @@ -543,9 +552,9 @@ def _compare_hypothesis_values( if name == HypothesisComparator.GREATER_THAN_OR_EQUAL.value: return left >= right + minimum_delta if name == HypothesisComparator.LESS_THAN.value: - return left < right - minimum_delta + return left < right + minimum_delta if name == HypothesisComparator.LESS_THAN_OR_EQUAL.value: - return left <= right - minimum_delta + return left <= right + minimum_delta raise ValueError(f"unsupported hypothesis comparator '{comparator}'") @@ -563,17 +572,37 @@ def _aggregate_opportunities(run_payloads: list[dict[str, Any]]) -> pd.DataFrame if not rows: return pd.DataFrame() combined = pd.concat(rows, ignore_index=True) + combined["weighted_regret_sum"] = ( + pd.to_numeric(combined.get("avg_regret_to_best_measured"), errors="coerce").fillna(0.0) + * pd.to_numeric(combined.get("regret_weight"), errors="coerce").fillna(0.0) + ) aggregated = ( combined.groupby("opportunity_tag", dropna=False) .agg( occurrences=("occurrences", "sum"), selected_regret_count=("selected_regret_count", "sum"), + regret_weight=("regret_weight", "sum"), + weighted_regret_sum=("weighted_regret_sum", "sum"), avg_regret_to_best_measured=("avg_regret_to_best_measured", "mean"), + kernel_ids=("kernel_ids", lambda values: ",".join(sorted({item for value in values for item in str(value).split(",") if item}))), + workload_classes=("workload_classes", lambda values: ",".join(sorted({item for value in values for item in str(value).split(",") if item}))), + strategy_ids=("strategy_ids", lambda values: ",".join(sorted({item for value in values for item in str(value).split(",") if item}))), + run_ids=("run_ids", lambda values: ",".join(sorted({item for value in values for item in str(value).split(",") if item}))), + config_ids=("config_ids", lambda values: ",".join(sorted({item for value in values for item in str(value).split(",") if item}))), recommended_actions=("recommended_actions", lambda values: "; ".join(sorted(set(values)))), ) .reset_index() .sort_values(by=["selected_regret_count", "occurrences", "opportunity_tag"], ascending=[False, False, True]) ) + aggregated["avg_regret_to_best_measured"] = aggregated.apply( + lambda row: ( + row["weighted_regret_sum"] / row["regret_weight"] + if row["regret_weight"] and not pd.isna(row["regret_weight"]) + else row["avg_regret_to_best_measured"] + ), + axis=1, + ) + aggregated = aggregated.drop(columns=["weighted_regret_sum"]) return aggregated @@ -660,6 +689,7 @@ def _build_evidence_bundle( hypothesis_results: pd.DataFrame, run_payloads: list[dict[str, Any]], ) -> dict[str, Any]: + opportunity_catalog = _aggregate_opportunities(run_payloads) return { "run_count": len(run_payloads), "runs": [ @@ -675,6 +705,7 @@ def _build_evidence_bundle( "strategy_summary": _strategy_summary(strategy_rows), "stability_report": stability_report.to_dict(orient="records"), "hypothesis_results": hypothesis_results.to_dict(orient="records"), + "opportunity_catalog": opportunity_catalog.to_dict(orient="records"), } diff --git a/src/kernel_tuner/analysis/opportunities.py b/src/kernel_tuner/analysis/opportunities.py index f8c862a..730525f 100644 --- a/src/kernel_tuner/analysis/opportunities.py +++ b/src/kernel_tuner/analysis/opportunities.py @@ -7,11 +7,7 @@ import pandas as pd -from kernel_tuner.common.schema import ( - BottleneckSignatureRecord, - CounterAvailabilityRecord, - ExperimentSpec, -) +from kernel_tuner.common.schema import BottleneckSignatureRecord, CounterAvailabilityRecord, ExperimentSpec, ProfileStatus def _quantile_bucket(series: pd.Series, value: float | None) -> str: @@ -40,7 +36,9 @@ def build_counter_availability_records( return [] usable = profile_measurements[ - profile_measurements["profile_status"].isin(["success", "unsupported_counter"]) + profile_measurements["profile_status"].isin( + [ProfileStatus.SUCCESS, ProfileStatus.UNSUPPORTED_COUNTER] + ) ].copy() if usable.empty: return [] @@ -265,9 +263,13 @@ def build_opportunity_catalog(signatures: pd.DataFrame) -> pd.DataFrame: "opportunity_tag": opportunity_tag, "occurrences": len(subset), "selected_regret_count": int((subset["held_out_outcome"] == "selected_regret").sum()), + "regret_weight": int(subset["regret_to_best_measured"].dropna().shape[0]), "avg_regret_to_best_measured": subset["regret_to_best_measured"].dropna().mean(), "kernel_ids": ",".join(sorted(subset["kernel_id"].dropna().unique())), "workload_classes": ",".join(sorted(value for value in subset["workload_class"].dropna().unique())), + "strategy_ids": ",".join(sorted(value for value in subset["strategy_id"].dropna().astype(str).unique())), + "run_ids": ",".join(sorted(value for value in subset["run_id"].dropna().astype(str).unique())), + "config_ids": ",".join(sorted(value for value in subset["config_id"].dropna().astype(str).unique())), "recommended_actions": "; ".join(action), } ) diff --git a/src/kernel_tuner/analysis/reporting.py b/src/kernel_tuner/analysis/reporting.py index a5b867e..c6cfb81 100644 --- a/src/kernel_tuner/analysis/reporting.py +++ b/src/kernel_tuner/analysis/reporting.py @@ -38,9 +38,11 @@ def _load_required_manifest(run_dir: Path): return RunStore.from_run_dir(run_dir).load_manifest() -def _load_table(store: RunStore, logical_name: str) -> pd.DataFrame: +def _load_table(store: RunStore, logical_name: str, *, required: bool = False) -> pd.DataFrame: path = store.run_dir / f"{logical_name}.parquet" if not path.exists(): + if required: + raise FileNotFoundError(f"run directory '{store.run_dir}' is missing required artifact '{logical_name}.parquet'") return pd.DataFrame() return store.load_table(logical_name) @@ -247,10 +249,14 @@ def summarize_run(run_dir: str | Path) -> dict[str, object]: experiment_spec = load_experiment_spec(source_experiment_path) kernel_spec = load_kernel_spec(kernel_config_path(experiment_spec.kernels[0], source_experiment_path)) - compile_signals = _load_table(store, "compile_signals") - runtime_measurements = _load_table(store, "runtime_measurements") - profile_measurements = _load_table(store, "profile_measurements") - selection_decisions = _load_table(store, "selection_decisions") + compile_signals = _load_table(store, "compile_signals", required=True) + runtime_measurements = _load_table(store, "runtime_measurements", required=True) + profile_measurements = _load_table( + store, + "profile_measurements", + required=bool(experiment_spec.counter_set_id), + ) + selection_decisions = _load_table(store, "selection_decisions", required=True) selection_decisions = _decode_jsonish_columns(selection_decisions, ["score_map", "calibration_metadata"]) profile_measurements = _decode_jsonish_columns(profile_measurements, ["counter_map", "profiler_metadata"]) @@ -400,12 +406,30 @@ def summarize_run(run_dir: str | Path) -> dict[str, object]: and "shape_id" in runtime_measurements.columns else 0 ) + policy = experiment_spec.reportability_policy + held_out_per_class = ( + held_out_per_shape.groupby("workload_class", dropna=False)["shape_id"].nunique().to_dict() + if not held_out_per_shape.empty and "workload_class" in held_out_per_shape.columns + else {} + ) + comparison_classes = {value for value in comparison_class_by_strategy.values() if value is not None} + matched_budget_only = bool(comparison_classes) and comparison_classes == {"matched_budget"} + decision_statuses = set(selection_decisions["decision_status"].dropna().astype(str)) if "decision_status" in selection_decisions.columns else set() + has_budget_limited_decision = any(status.endswith("budget_limited") for status in decision_statuses) is_reportable = bool( experiment_spec.study_kind == "reportable" - and held_out_shape_count > 0 - and all(value == "matched_budget" for value in comparison_class_by_strategy.values()) + and held_out_shape_count >= policy.minimum_held_out_shapes + and matched_budget_only + and not has_budget_limited_decision and (not counter_compatibility or counter_compatibility.get("acceptable", True)) and (not counter_availability_ok or all(counter_availability_ok.values())) + and ( + policy.minimum_held_out_per_workload_class == 0 + or ( + held_out_per_class + and all(count >= policy.minimum_held_out_per_workload_class for count in held_out_per_class.values()) + ) + ) ) winner_rates = ( @@ -454,6 +478,7 @@ def summarize_run(run_dir: str | Path) -> dict[str, object]: "profile_failures": profile_failures, "comparison_class_by_strategy": comparison_class_by_strategy, "held_out_shape_count": held_out_shape_count, + "held_out_shape_count_by_workload_class": held_out_per_class, "counter_availability": counter_availability_ok, "winner_rates": winner_rates, "opportunity_counts": summarize_opportunity_counts(signatures_frame), @@ -473,6 +498,8 @@ def summarize_run(run_dir: str | Path) -> dict[str, object]: "target": experiment_spec.analysis_settings.reportability_target, "is_reportable": is_reportable, "comparison_class": "matched_budget" if is_reportable else "non_comparable", + "matched_budget_only": matched_budget_only, + "budget_limited_decision_present": has_budget_limited_decision, "counter_set_accepted": all(counter_availability_ok.values()) if counter_availability_ok else True, "counter_compatibility": counter_compatibility, }, diff --git a/src/kernel_tuner/baselines/strategies.py b/src/kernel_tuner/baselines/strategies.py index 99cd3b2..c10b51f 100644 --- a/src/kernel_tuner/baselines/strategies.py +++ b/src/kernel_tuner/baselines/strategies.py @@ -11,7 +11,9 @@ BaselineMode, CandidateConfig, ComparisonClass, + MeasurementPhase, RuntimeMeasurement, + RuntimeStatus, SelectionDecision, ) from kernel_tuner.config_space.generator import config_dict_from_record @@ -145,7 +147,17 @@ def run_baseline_mode( runtime_records.extend(request_benchmark(config_id)) runtime_scores = aggregate_runtime_scores(runtime_records) selected = min(runtime_scores, key=lambda config_id: (runtime_scores[config_id], config_id)) if runtime_scores else None - decision_status = "selected" if selected is not None else "failed_no_successful_measurements" + consumed_benchmarks = sum( + 1 + for record in runtime_records + if record.measurement_phase == MeasurementPhase.CALIBRATION + and record.status != RuntimeStatus.SKIPPED_BUDGET + ) + budget_limited = any(record.status == RuntimeStatus.SKIPPED_BUDGET for record in runtime_records) + if selected is None: + decision_status = "failed_no_successful_measurements" + else: + decision_status = "selected_budget_limited" if budget_limited else "selected" return SelectionDecision( run_id=run_id, strategy_id=strategy_id, @@ -159,7 +171,7 @@ def run_baseline_mode( ranked_config_ids=benchmark_ids, pruned_config_ids=[], candidates_considered=len(ordered_ids), - benchmarks_requested=len(benchmark_ids), + benchmarks_requested=consumed_benchmarks, profiles_requested=0, decision_wall_clock_s=time.perf_counter() - started, rationale_summary="; ".join(rationale), diff --git a/src/kernel_tuner/benchmark/harness.py b/src/kernel_tuner/benchmark/harness.py index 4390986..d086088 100644 --- a/src/kernel_tuner/benchmark/harness.py +++ b/src/kernel_tuner/benchmark/harness.py @@ -67,6 +67,14 @@ def _percentile(sorted_samples: list[float], percentile: float) -> float: return sorted_samples[index] +def _classify_launch_failure(exc: Exception) -> RuntimeStatus: + text = str(exc).lower() + compile_markers = ["compile", "ptx", "ptxas", "nvrtc", "llvm", "codegen"] + if any(marker in text for marker in compile_markers): + return RuntimeStatus.COMPILE_FAILED + return RuntimeStatus.RUNTIME_FAILED + + def benchmark_candidate( *, run_id: str, @@ -113,14 +121,46 @@ def benchmark_candidate( shape_id=shape.shape_id, config_id=candidate.config_id, settings=settings, - status=RuntimeStatus.COMPILE_FAILED, + status=_classify_launch_failure(exc), measurement_order_index=measurement_order_index, error_message=str(exc), ) ) - reference = kernel.reference_impl(inputs, kernel.spec.correctness_policy) - if not kernel.validate_output(output, reference, kernel.spec.correctness_policy): + try: + reference = kernel.reference_impl(inputs, kernel.spec.correctness_policy) + if not kernel.validate_output(output, reference, kernel.spec.correctness_policy): + return BenchmarkOutcome( + measurement=_status_measurement( + run_id=run_id, + strategy_id=strategy_id, + measurement_phase=measurement_phase, + kernel_id=kernel.spec.kernel_id, + shape_id=shape.shape_id, + config_id=candidate.config_id, + settings=settings, + status=RuntimeStatus.RUNTIME_FAILED, + measurement_order_index=measurement_order_index, + error_message="correctness validation failed", + ) + ) + + stream = torch.cuda.Stream() + with torch.cuda.stream(stream): + for _ in range(settings.warmup_iterations): + kernel.run_kernel(inputs, shape, config) + stream.synchronize() + + samples: list[float] = [] + for _ in range(settings.timed_iterations): + start = torch.cuda.Event(enable_timing=True) + end = torch.cuda.Event(enable_timing=True) + start.record(stream) + kernel.run_kernel(inputs, shape, config) + end.record(stream) + end.synchronize() + samples.append(start.elapsed_time(end) * 1000.0) + except Exception as exc: return BenchmarkOutcome( measurement=_status_measurement( run_id=run_id, @@ -132,26 +172,10 @@ def benchmark_candidate( settings=settings, status=RuntimeStatus.RUNTIME_FAILED, measurement_order_index=measurement_order_index, - error_message="correctness validation failed", + error_message=str(exc), ) ) - stream = torch.cuda.Stream() - with torch.cuda.stream(stream): - for _ in range(settings.warmup_iterations): - kernel.run_kernel(inputs, shape, config) - stream.synchronize() - - samples: list[float] = [] - for _ in range(settings.timed_iterations): - start = torch.cuda.Event(enable_timing=True) - end = torch.cuda.Event(enable_timing=True) - start.record(stream) - kernel.run_kernel(inputs, shape, config) - end.record(stream) - end.synchronize() - samples.append(start.elapsed_time(end) * 1000.0) - sorted_samples = sorted(samples) median_us = statistics.median(sorted_samples) mean_us = statistics.fmean(sorted_samples) diff --git a/src/kernel_tuner/common/schema.py b/src/kernel_tuner/common/schema.py index eb41ac0..4141bef 100644 --- a/src/kernel_tuner/common/schema.py +++ b/src/kernel_tuner/common/schema.py @@ -61,6 +61,9 @@ class ProfileStatus(StrEnum): SUCCESS = "success" UNSUPPORTED_COUNTER = "unsupported_counter" TOOL_UNAVAILABLE = "tool_unavailable" + TIMEOUT = "timeout" + INVOCATION_FAILED = "invocation_failed" + NO_PROFILE_DATA = "no_profile_data" PROFILE_FAILED = "profile_failed" SKIPPED_BUDGET = "skipped_budget" diff --git a/src/kernel_tuner/experiments/orchestrator.py b/src/kernel_tuner/experiments/orchestrator.py index 08f7671..2d7f087 100644 --- a/src/kernel_tuner/experiments/orchestrator.py +++ b/src/kernel_tuner/experiments/orchestrator.py @@ -114,9 +114,20 @@ def _validate_environment(spec: ExperimentSpec, environment) -> None: def _profile_shapes(spec: ExperimentSpec, calibration_shapes) -> list: if not calibration_shapes: return [] + + def _shape_work_score(shape) -> tuple[int, str]: + score = 1 + for value in shape.dimensions.values(): + score *= value + return score, shape.shape_id + + def _representative_shape(shapes): + ordered = sorted(shapes, key=_shape_work_score) + return ordered[len(ordered) // 2] + policy = spec.profile_policy if policy is None: - return [calibration_shapes[0]] + return [_representative_shape(calibration_shapes)] mode = policy.shape_sampling_mode if mode == "all_calibration": selected = list(calibration_shapes) @@ -128,12 +139,12 @@ def _profile_shapes(spec: ExperimentSpec, calibration_shapes) -> list: for shape in calibration_shapes: by_class.setdefault(shape.workload_class or "__unlabeled__", []).append(shape) selected = [ - sorted(shapes, key=lambda shape: shape.shape_id)[0] + _representative_shape(shapes) for _, shapes in sorted(by_class.items()) if shapes ] else: - selected = [sorted(calibration_shapes, key=lambda shape: shape.shape_id)[0]] + selected = [_representative_shape(calibration_shapes)] max_shapes = policy.max_shapes_per_config if policy else None if max_shapes is not None: selected = selected[:max_shapes] @@ -217,6 +228,12 @@ def _validate_reportability_contract( ) +def _strategy_deadline(limit_s: float | None) -> float | None: + if limit_s is None: + return None + return time.perf_counter() + limit_s + + @contextmanager def _isolated_caches(spec: ExperimentSpec, store: RunStore): previous = os.environ.get("TRITON_CACHE_DIR") @@ -293,6 +310,7 @@ def __init__( profile_shapes, selector_revision: SelectorRevisionSpec | None, deadline_s: float | None, + measurement_order_counter: list[int], ) -> None: self.store = store self.run_id = run_id @@ -306,6 +324,9 @@ def __init__( self.counter_set = counter_set self.selector_revision = selector_revision self.deadline_s = deadline_s + self.measurement_order_counter = measurement_order_counter + self.remaining_benchmarks = experiment_spec.budgets.max_benchmarks + self.remaining_profiles = experiment_spec.budgets.max_profiles self._candidate_lookup = { (candidate.shape_id, candidate.config_id): candidate for candidate in candidate_records } @@ -313,11 +334,21 @@ def __init__( self._profile_cache: dict[str, list[ProfileMeasurement]] = {} self.runtime_records: list[RuntimeMeasurement] = [] self.profile_records: list[ProfileMeasurement] = [] - self._measurement_order_index = 0 + + def _next_measurement_order_index(self) -> int: + index = self.measurement_order_counter[0] + self.measurement_order_counter[0] += 1 + return index def _budget_exhausted(self) -> bool: return self.deadline_s is not None and time.perf_counter() >= self.deadline_s + def _benchmark_budget_exhausted(self) -> bool: + return self.remaining_benchmarks <= 0 + + def _profile_budget_exhausted(self) -> bool: + return self.remaining_profiles <= 0 + def benchmark_calibration(self, config_id: str) -> list[RuntimeMeasurement]: return self._benchmark_shapes(config_id, self.calibration_shapes, MeasurementPhase.CALIBRATION) @@ -325,13 +356,14 @@ def benchmark_held_out(self, config_id: str) -> list[RuntimeMeasurement]: return self._benchmark_shapes(config_id, self.held_out_shapes, MeasurementPhase.HELD_OUT) def _benchmark_shapes(self, config_id: str, shapes, phase: MeasurementPhase) -> list[RuntimeMeasurement]: - cache_key = (config_id, phase) + cache_key = (config_id, phase, tuple(shape.shape_id for shape in shapes)) if cache_key in self._runtime_cache: return self._runtime_cache[cache_key] records: list[RuntimeMeasurement] = [] for index, shape in enumerate(shapes): - if self._budget_exhausted(): + enforce_budget = phase == MeasurementPhase.CALIBRATION + if enforce_budget and (self._budget_exhausted() or self._benchmark_budget_exhausted()): skipped = RuntimeMeasurement( run_id=self.run_id, strategy_id=self.strategy_id, @@ -349,14 +381,20 @@ def _benchmark_shapes(self, config_id: str, shapes, phase: MeasurementPhase) -> throughput_unit=None, status=RuntimeStatus.SKIPPED_BUDGET, timing_backend=self.experiment_spec.benchmark_settings.timing_backend, - measurement_order_index=self._measurement_order_index, - error_message="wall_clock_limit_s exhausted before benchmark", + measurement_order_index=self._next_measurement_order_index(), + error_message=( + "wall_clock_limit_s exhausted before benchmark" + if self._budget_exhausted() + else "benchmark budget exhausted before measurement" + ), ) records.append(skipped) self.runtime_records.append(skipped) - self._measurement_order_index += 1 continue candidate = self._candidate_lookup[(shape.shape_id, config_id)] + if enforce_budget: + self.remaining_benchmarks -= 1 + measurement_order_index = self._next_measurement_order_index() outcome = benchmark_candidate( run_id=self.run_id, strategy_id=self.strategy_id, @@ -366,7 +404,7 @@ def _benchmark_shapes(self, config_id: str, shapes, phase: MeasurementPhase) -> settings=self.experiment_spec.benchmark_settings, seed=self.experiment_spec.seed + index, measurement_phase=phase, - measurement_order_index=self._measurement_order_index, + measurement_order_index=measurement_order_index, ) measurement = _register_raw_samples( self.store, @@ -377,7 +415,6 @@ def _benchmark_shapes(self, config_id: str, shapes, phase: MeasurementPhase) -> ) records.append(measurement) self.runtime_records.append(measurement) - self._measurement_order_index += 1 self._runtime_cache[cache_key] = records return records @@ -389,7 +426,7 @@ def profile_calibration(self, config_id: str) -> list[ProfileMeasurement]: measurements: list[ProfileMeasurement] = [] for shape in self.profile_shapes: - if self._budget_exhausted(): + if self._budget_exhausted() or self._profile_budget_exhausted(): measurement = ProfileMeasurement( run_id=self.run_id, strategy_id=self.strategy_id, @@ -398,12 +435,17 @@ def profile_calibration(self, config_id: str) -> list[ProfileMeasurement]: config_id=config_id, counter_set_id=self.counter_set.counter_set_id, profile_status=ProfileStatus.SKIPPED_BUDGET, - notes="wall_clock_limit_s exhausted before profiling", + notes=( + "wall_clock_limit_s exhausted before profiling" + if self._budget_exhausted() + else "profile budget exhausted before measurement" + ), ) self.profile_records.append(measurement) measurements.append(measurement) continue candidate = self._candidate_lookup[(shape.shape_id, config_id)] + self.remaining_profiles -= 1 outcome = profile_candidate( run_id=self.run_id, strategy_id=self.strategy_id, @@ -519,11 +561,6 @@ def run_experiment( kernel = resolve_kernel(kernel_spec) profile_shapes = _profile_shapes(experiment_spec, calibration_shapes) - deadline_s = ( - time.perf_counter() + experiment_spec.budgets.wall_clock_limit_s - if experiment_spec.budgets.wall_clock_limit_s is not None - else None - ) candidate_records = generate_candidate_records(experiment_spec, experiment_path=experiment_path) store.write_table("candidates", candidate_records) @@ -551,10 +588,10 @@ def run_experiment( record for record in compile_signal_records if record.shape_id in {shape.shape_id for shape in calibration_shapes} ] - runtime_measurements: list[RuntimeMeasurement] = [] - profile_measurements: list[ProfileMeasurement] = [] selection_decisions = [] + strategy_brokers: dict[str, StrategyBroker] = {} warnings: list[str] = [] + measurement_order_counter = [0] for selector_mode in experiment_spec.selector_modes: strategy_id = _mode_name(selector_mode) @@ -571,7 +608,8 @@ def run_experiment( counter_set=counter_set, profile_shapes=profile_shapes, selector_revision=selector_revision, - deadline_s=deadline_s, + deadline_s=_strategy_deadline(experiment_spec.budgets.wall_clock_limit_s), + measurement_order_counter=measurement_order_counter, ) with _isolated_caches(experiment_spec, store): decision = run_selector_mode( @@ -586,15 +624,12 @@ def run_experiment( request_profile=broker.profile_calibration if counter_set is not None else None, selector_revision=selector_revision, ) - if decision.selected_config_id and held_out_shapes: - broker.benchmark_held_out(decision.selected_config_id) decision.comparison_class = _comparison_class_for_run( experiment_spec, counter_compatibility, ) + strategy_brokers[strategy_id] = broker selection_decisions.append(decision) - runtime_measurements.extend(broker.runtime_records) - profile_measurements.extend(broker.profile_records) if decision.decision_status.startswith("failed"): warnings.append(f"{strategy_id}: {decision.decision_status}") @@ -613,7 +648,8 @@ def run_experiment( counter_set=None, profile_shapes=[], selector_revision=None, - deadline_s=deadline_s, + deadline_s=_strategy_deadline(experiment_spec.budgets.wall_clock_limit_s), + measurement_order_counter=measurement_order_counter, ) with _isolated_caches(experiment_spec, store): decision = run_baseline_mode( @@ -627,18 +663,33 @@ def run_experiment( default_config=kernel_spec.default_config, request_benchmark=broker.benchmark_calibration, ) - if decision.selected_config_id and held_out_shapes: - broker.benchmark_held_out(decision.selected_config_id) decision.comparison_class = _comparison_class_for_run( experiment_spec, counter_compatibility, ) + strategy_brokers[strategy_id] = broker selection_decisions.append(decision) - runtime_measurements.extend(broker.runtime_records) - profile_measurements.extend(broker.profile_records) if decision.decision_status.startswith("failed"): warnings.append(f"{strategy_id}: {decision.decision_status}") + if held_out_shapes: + held_out_plan = [ + (decision.strategy_id, strategy_brokers[decision.strategy_id], decision.selected_config_id) + for decision in selection_decisions + if decision.selected_config_id and decision.strategy_id in strategy_brokers + ] + for index, shape in enumerate(held_out_shapes): + ordered_plan = held_out_plan if index % 2 == 0 else list(reversed(held_out_plan)) + for _, broker, config_id in ordered_plan: + broker._benchmark_shapes(config_id, [shape], MeasurementPhase.HELD_OUT) + + runtime_measurements: list[RuntimeMeasurement] = [] + profile_measurements: list[ProfileMeasurement] = [] + for strategy_id in [decision.strategy_id for decision in selection_decisions]: + broker = strategy_brokers[strategy_id] + runtime_measurements.extend(broker.runtime_records) + profile_measurements.extend(broker.profile_records) + store.write_table("runtime_measurements", runtime_measurements) store.write_table("profile_measurements", profile_measurements) store.write_table("selection_decisions", selection_decisions) diff --git a/src/kernel_tuner/profiling/adapter.py b/src/kernel_tuner/profiling/adapter.py index 4b75574..827ddfb 100644 --- a/src/kernel_tuner/profiling/adapter.py +++ b/src/kernel_tuner/profiling/adapter.py @@ -8,9 +8,10 @@ import re import subprocess import time -from dataclasses import dataclass +from functools import lru_cache from pathlib import Path from typing import Any +from dataclasses import dataclass from kernel_tuner.common.config import ( counter_set_path, @@ -55,7 +56,7 @@ def _maybe_float(value: str | None) -> float | None: def _extract_csv_rows(stdout: str) -> list[dict[str, str]]: lines = stdout.splitlines() - header_index = next((index for index, line in enumerate(lines) if line.startswith('"ID","Process ID"')), None) + header_index = next((index for index, line in enumerate(lines) if '"Kernel Name"' in line), None) if header_index is None: return [] reader = csv.DictReader(io.StringIO("\n".join(lines[header_index:]))) @@ -66,16 +67,35 @@ def _choose_kernel_row( rows: list[dict[str, str]], *, kernel_name_regex: str | None, -) -> dict[str, str] | None: +) -> tuple[dict[str, str] | None, dict[str, Any]]: if not rows: - return None + return None, {"kernel_attribution_status": "no_rows", "kernel_name_candidates": [], "matched_row_count": 0} - filtered = rows + kernel_names = sorted({row.get("Kernel Name", "") for row in rows if row.get("Kernel Name")}) + filtered: list[dict[str, str]] = rows if kernel_name_regex: pattern = re.compile(kernel_name_regex) matched = [row for row in rows if pattern.search(row.get("Kernel Name", ""))] - if matched: - filtered = matched + if not matched: + return None, { + "kernel_attribution_status": "regex_no_match", + "kernel_name_candidates": kernel_names, + "matched_row_count": 0, + } + matched_names = sorted({row.get("Kernel Name", "") for row in matched if row.get("Kernel Name")}) + if len(matched_names) > 1: + return None, { + "kernel_attribution_status": "regex_ambiguous", + "kernel_name_candidates": matched_names, + "matched_row_count": len(matched), + } + filtered = matched + elif len(kernel_names) > 1: + return None, { + "kernel_attribution_status": "ambiguous_without_regex", + "kernel_name_candidates": kernel_names, + "matched_row_count": len(rows), + } duration_key = "gpu__time_duration.sum" filtered = sorted( @@ -85,7 +105,14 @@ def _choose_kernel_row( row.get("Kernel Name", ""), ), ) - return filtered[-1] if filtered else None + return ( + filtered[-1] if filtered else None, + { + "kernel_attribution_status": "matched", + "kernel_name_candidates": kernel_names, + "matched_row_count": len(filtered), + }, + ) def _parse_counter_map( @@ -95,12 +122,67 @@ def _parse_counter_map( kernel_name_regex: str | None, ) -> tuple[dict[str, float | None], list[str], str | None]: rows = _extract_csv_rows(stdout) - row = _choose_kernel_row(rows, kernel_name_regex=kernel_name_regex) + row, diagnostics = _choose_kernel_row(rows, kernel_name_regex=kernel_name_regex) if row is None: - return ({counter: None for counter in counters}, counters, None) + return ( + {counter: None for counter in counters}, + counters, + None, + { + "row_count": len(rows), + **diagnostics, + }, + ) counter_map = {counter: _maybe_float(row.get(counter)) for counter in counters} missing = [counter for counter, value in counter_map.items() if value is None] - return counter_map, missing, row.get("Kernel Name") + return ( + counter_map, + missing, + row.get("Kernel Name"), + { + "row_count": len(rows), + **diagnostics, + }, + ) + + +@lru_cache(maxsize=1) +def _ncu_version() -> str | None: + try: + completed = subprocess.run( + ["ncu", "--version"], + check=False, + capture_output=True, + text=True, + ) + except FileNotFoundError: + return None + if completed.returncode != 0: + return None + lines = [line.strip() for line in completed.stdout.splitlines() if line.strip()] + return lines[-1] if lines else None + + +def _base_profiler_metadata( + *, + command: list[str], + counter_set: CounterSetSpec, + kernel_name_regex: str | None, + experiment_spec: ExperimentSpec, +) -> dict[str, Any]: + return { + "command": command, + "requested_counters": list(counter_set.counters), + "target_processes": counter_set.target_processes or "all", + "replay_mode": counter_set.replay_mode or experiment_spec.profiling_settings.replay_mode, + "kernel_name_regex": kernel_name_regex, + "ncu_args": list(counter_set.ncu_args), + "minimum_availability": counter_set.minimum_availability, + "diagnostic_only": counter_set.diagnostic_only, + "timeout_s": experiment_spec.profiling_settings.timeout_s, + "cooldown_s": experiment_spec.profiling_settings.cooldown_s, + "ncu_version": _ncu_version(), + } def _build_profile_command( @@ -152,6 +234,13 @@ def profile_candidate( ) command = _build_profile_command(counter_set, experiment_spec.profiling_settings, payload) command_started = time.perf_counter() + kernel_name_regex = counter_set.kernel_name_regex or experiment_spec.profiling_settings.kernel_name_regex + base_metadata = _base_profiler_metadata( + command=command, + counter_set=counter_set, + kernel_name_regex=kernel_name_regex, + experiment_spec=experiment_spec, + ) try: completed = subprocess.run( command, @@ -169,7 +258,11 @@ def profile_candidate( config_id=candidate.config_id, counter_set_id=counter_set.counter_set_id, profile_status=ProfileStatus.TOOL_UNAVAILABLE, - profiler_metadata={"command": command}, + profiler_metadata={ + **base_metadata, + "returncode": None, + "duration_s": time.perf_counter() - command_started, + }, notes="ncu executable not found", ) if experiment_spec.profiling_settings.cooldown_s > 0.0: @@ -183,11 +276,11 @@ def profile_candidate( shape_id=shape.shape_id, config_id=candidate.config_id, counter_set_id=counter_set.counter_set_id, - profile_status=ProfileStatus.PROFILE_FAILED, + profile_status=ProfileStatus.TIMEOUT, profiler_metadata={ - "command": command, + **base_metadata, + "returncode": None, "duration_s": time.perf_counter() - command_started, - "timeout_s": experiment_spec.profiling_settings.timeout_s, }, notes=f"ncu timed out: {exc}", ) @@ -195,21 +288,26 @@ def profile_candidate( time.sleep(experiment_spec.profiling_settings.cooldown_s) return ProfileOutcome(measurement=measurement, stdout=exc.stdout or "", stderr=exc.stderr or "") - kernel_name_regex = counter_set.kernel_name_regex or experiment_spec.profiling_settings.kernel_name_regex - counter_map, missing_counters, matched_kernel_name = _parse_counter_map( + counter_map, missing_counters, matched_kernel_name, diagnostics = _parse_counter_map( completed.stdout, counter_set.counters, kernel_name_regex=kernel_name_regex, ) - unsupported = bool(missing_counters) or bool( + unsupported_metric = bool( re.search(r"(unknown metric|unsupported metric|not supported)", completed.stderr, re.IGNORECASE) ) status = ProfileStatus.SUCCESS notes = None - if completed.returncode != 0: - status = ProfileStatus.PROFILE_FAILED + if completed.returncode != 0 and unsupported_metric: + status = ProfileStatus.UNSUPPORTED_COUNTER + notes = completed.stderr.strip() or completed.stdout.strip() or "unsupported counter" + elif completed.returncode != 0: + status = ProfileStatus.INVOCATION_FAILED notes = completed.stderr.strip() or completed.stdout.strip() or "ncu invocation failed" - elif unsupported: + elif diagnostics.get("matched_row_count", 0) == 0: + status = ProfileStatus.NO_PROFILE_DATA + notes = f"no attributable profiler row: {diagnostics.get('kernel_attribution_status')}" + elif missing_counters: status = ProfileStatus.UNSUPPORTED_COUNTER notes = "missing or unsupported counters: " + ", ".join(sorted(missing_counters)) @@ -223,13 +321,12 @@ def profile_candidate( profile_status=status, counter_map=counter_map, profiler_metadata={ - "command": command, + **base_metadata, "returncode": completed.returncode, "duration_s": time.perf_counter() - command_started, "matched_kernel_name": matched_kernel_name, - "kernel_name_regex": kernel_name_regex, - "replay_mode": counter_set.replay_mode or experiment_spec.profiling_settings.replay_mode, - "cooldown_s": experiment_spec.profiling_settings.cooldown_s, + "missing_counters": missing_counters, + **diagnostics, }, notes=notes, ) @@ -254,7 +351,9 @@ def profile_experiment( measurements = [] for index, shape in enumerate(selected_shapes): shape_candidates = [candidate for candidate in candidates if candidate.shape_id == shape.shape_id] - candidate = shape_candidates[0] + candidate = next((item for item in shape_candidates if item.is_valid), None) + if candidate is None: + candidate = shape_candidates[0] outcome = profile_candidate( run_id="standalone", strategy_id="profile_cli", diff --git a/src/kernel_tuner/profiling/compatibility.py b/src/kernel_tuner/profiling/compatibility.py index a80d633..b854d2b 100644 --- a/src/kernel_tuner/profiling/compatibility.py +++ b/src/kernel_tuner/profiling/compatibility.py @@ -28,7 +28,12 @@ def validate_counter_set_for_experiment( return None counter_set = load_counter_set(counter_set_path(counter_set_id, experiment_path)) kernel_spec = load_kernel_spec(kernel_config_path(experiment_spec.kernels[0], experiment_path)) - return validate_counter_set(counter_set, kernel_family=kernel_spec.family) + return validate_counter_set( + counter_set, + kernel_family=kernel_spec.family, + supports_profiling=kernel_spec.supports_profiling, + require_reportable_constraints=experiment_spec.study_kind == "reportable", + ) def validate_counter_set_from_path(experiment_path: str | Path) -> dict[str, object]: @@ -37,24 +42,47 @@ def validate_counter_set_from_path(experiment_path: str | Path) -> dict[str, obj return record.model_dump(mode="json") if record is not None else {} -def validate_counter_set(counter_set, *, kernel_family: str) -> CounterCompatibilityRecord: +def validate_counter_set( + counter_set, + *, + kernel_family: str, + supports_profiling: bool = True, + require_reportable_constraints: bool = False, +) -> CounterCompatibilityRecord: family_allowed = not counter_set.kernel_family_filters or kernel_family in counter_set.kernel_family_filters requested = list(counter_set.counters) available: list[str] = [] missing = list(requested) - notes = None + notes_parts: list[str] = [] backend = "ncu_query_metrics" + + if counter_set.tool != "ncu": + backend = "unsupported_tool" + notes_parts.append(f"unsupported profiling tool '{counter_set.tool}'") + if not supports_profiling: + notes_parts.append("kernel does not advertise profiling support") + if counter_set.diagnostic_only: + notes_parts.append("counter set is diagnostic-only") + if require_reportable_constraints and not counter_set.kernel_name_regex: + notes_parts.append("reportable counter sets require kernel_name_regex for defensible attribution") + if counter_set.kernel_name_regex: + try: + re.compile(counter_set.kernel_name_regex) + except re.error as exc: + notes_parts.append(f"invalid kernel_name_regex: {exc}") + try: - completed = subprocess.run( - ["ncu", "--query-metrics"], - check=False, - capture_output=True, - text=True, - ) + completed = None + if backend == "ncu_query_metrics": + completed = subprocess.run( + ["ncu", "--query-metrics"], + check=False, + capture_output=True, + text=True, + ) except FileNotFoundError: backend = "tool_unavailable" - notes = "ncu executable not found" - completed = None + notes_parts.append("ncu executable not found") if completed is not None and completed.returncode == 0: metrics_blob = f"{completed.stdout}\n{completed.stderr}" @@ -63,13 +91,19 @@ def validate_counter_set(counter_set, *, kernel_family: str) -> CounterCompatibi missing = [counter for counter in requested if counter not in discovered] elif completed is not None: backend = "ncu_query_failed" - notes = completed.stderr.strip() or completed.stdout.strip() or "ncu --query-metrics failed" + notes_parts.append(completed.stderr.strip() or completed.stdout.strip() or "ncu --query-metrics failed") availability = (len(available) / len(requested)) if requested else 1.0 - acceptable = family_allowed and availability >= counter_set.minimum_availability - if counter_set.diagnostic_only: - acceptable = False - notes = (notes + "; " if notes else "") + "counter set is diagnostic-only" + acceptable = ( + family_allowed + and supports_profiling + and counter_set.tool == "ncu" + and availability >= counter_set.minimum_availability + and (not counter_set.diagnostic_only) + and (not require_reportable_constraints or bool(counter_set.kernel_name_regex)) + and not any(part.startswith("invalid kernel_name_regex") for part in notes_parts) + ) + notes = "; ".join(notes_parts) or None return CounterCompatibilityRecord( counter_set_id=counter_set.counter_set_id, diff --git a/src/kernel_tuner/selector/engine.py b/src/kernel_tuner/selector/engine.py index 8ccfaa6..b3617e3 100644 --- a/src/kernel_tuner/selector/engine.py +++ b/src/kernel_tuner/selector/engine.py @@ -564,13 +564,27 @@ def run_selector_mode( runtime_scores = aggregate_runtime_scores(runtime_records) selected = _select_with_tolerance(runtime_scores, benchmark_order) + consumed_benchmarks = sum( + 1 + for record in runtime_records + if record.measurement_phase == MeasurementPhase.CALIBRATION + and record.status != RuntimeStatus.SKIPPED_BUDGET + ) + consumed_profiles = sum( + 1 + for record in profiled_records + if record.profile_status != ProfileStatus.SKIPPED_BUDGET + ) + budget_limited = any( + record.status == RuntimeStatus.SKIPPED_BUDGET for record in runtime_records + ) or any(record.profile_status == ProfileStatus.SKIPPED_BUDGET for record in profiled_records) if selected is None: decision_status = "failed_no_successful_measurements" rationale.append("no successful calibration measurements were available") else: - decision_status = "selected" + decision_status = "selected_budget_limited" if budget_limited else "selected" rationale.append( - f"benchmarked {len(benchmark_ids)} configs and selected the best score within a 2% tie band" + f"consumed {consumed_benchmarks} calibration benchmark rows and selected the best score within a 2% tie band" ) return SelectionDecision( @@ -585,10 +599,8 @@ def run_selector_mode( ranked_config_ids=benchmark_order, pruned_config_ids=pruned_ids, candidates_considered=len(candidate_groups), - benchmarks_requested=len(benchmark_ids), - profiles_requested=len( - {record.config_id for record in profiled_records if record.profile_status != ProfileStatus.SKIPPED_BUDGET} - ), + benchmarks_requested=consumed_benchmarks, + profiles_requested=consumed_profiles, decision_wall_clock_s=time.perf_counter() - started, rationale_summary="; ".join(rationale), decision_status=decision_status, diff --git a/tests/unit/test_phase2_analysis.py b/tests/unit/test_phase2_analysis.py index 847a54c..f188313 100644 --- a/tests/unit/test_phase2_analysis.py +++ b/tests/unit/test_phase2_analysis.py @@ -4,11 +4,19 @@ import pandas as pd from kernel_tuner.analysis.comparison import ( + _aggregate_opportunities, _build_stability_report, _build_strategy_rows, + _compare_hypothesis_values, _evaluate_hypotheses, ) -from kernel_tuner.common.schema import HypothesisClause, HypothesisMetricRef, HypothesisSpec, StudySpec +from kernel_tuner.common.schema import ( + HypothesisClause, + HypothesisComparator, + HypothesisMetricRef, + HypothesisSpec, + StudySpec, +) from kernel_tuner.analysis.opportunities import ( build_bottleneck_signatures, build_counter_availability_records, @@ -256,6 +264,73 @@ def test_evaluate_hypotheses_uses_clause_based_metrics(): assert results.iloc[0]["status"] == "supported" +def test_less_than_or_equal_comparator_allows_small_regression_within_delta(): + supported = _compare_hypothesis_values( + 1.01, + 1.00, + HypothesisComparator.LESS_THAN_OR_EQUAL, + 0.02, + ) + + assert supported is True + + +def test_aggregate_opportunities_uses_weighted_regret_and_preserves_provenance(): + aggregated = _aggregate_opportunities( + [ + { + "group_id": "gemm_representative", + "summary": {"run_id": "run_a"}, + "experiment_spec": type("Spec", (), {"experiment_id": "gemm_reportable"})(), + "opportunity_catalog": pd.DataFrame( + [ + { + "opportunity_tag": "selector_revision_candidate", + "occurrences": 2, + "selected_regret_count": 1, + "regret_weight": 2, + "avg_regret_to_best_measured": 0.20, + "kernel_ids": "gemm", + "workload_classes": "square_compute", + "strategy_ids": "prune_rank", + "run_ids": "run_a", + "config_ids": "cfg_a", + "recommended_actions": "investigate", + } + ] + ), + }, + { + "group_id": "gemm_representative", + "summary": {"run_id": "run_b"}, + "experiment_spec": type("Spec", (), {"experiment_id": "gemm_reportable"})(), + "opportunity_catalog": pd.DataFrame( + [ + { + "opportunity_tag": "selector_revision_candidate", + "occurrences": 1, + "selected_regret_count": 1, + "regret_weight": 1, + "avg_regret_to_best_measured": 0.05, + "kernel_ids": "gemm", + "workload_classes": "m_dominant", + "strategy_ids": "prune_rank_revised", + "run_ids": "run_b", + "config_ids": "cfg_b", + "recommended_actions": "investigate", + } + ] + ), + }, + ] + ) + + row = aggregated.iloc[0] + assert row["avg_regret_to_best_measured"] == (0.20 * 2 + 0.05 * 1) / 3 + assert row["run_ids"] == "run_a,run_b" + assert row["strategy_ids"] == "prune_rank,prune_rank_revised" + + def test_opportunity_catalog_contains_expected_template(): experiment_spec = load_experiment_spec(Path("configs/experiments/gemm_development.yaml")) compile_signals = pd.DataFrame( diff --git a/tests/unit/test_profiling.py b/tests/unit/test_profiling.py new file mode 100644 index 0000000..32a6d77 --- /dev/null +++ b/tests/unit/test_profiling.py @@ -0,0 +1,94 @@ +from datetime import datetime, timezone +from pathlib import Path +import subprocess + +import pandas as pd + +from kernel_tuner.common.config import load_counter_set, load_experiment_spec +from kernel_tuner.common.provenance import capture_environment_metadata, capture_invocation_metadata +from kernel_tuner.common.schema import CandidateConfig, Manifest, ProfileMeasurement, ProfileStatus +from kernel_tuner.profiling.adapter import ProfileOutcome, profile_experiment +from kernel_tuner.profiling.compatibility import validate_counter_set + + +def test_validate_counter_set_requires_kernel_regex_for_reportable(monkeypatch): + counter_set = load_counter_set(Path("configs/counters/compute_lite.yaml")) + counter_set.kernel_name_regex = None + + monkeypatch.setattr( + subprocess, + "run", + lambda *args, **kwargs: subprocess.CompletedProcess( + args=args[0], + returncode=0, + stdout="\n".join(counter_set.counters), + stderr="", + ), + ) + + record = validate_counter_set( + counter_set, + kernel_family="gemm", + supports_profiling=True, + require_reportable_constraints=True, + ) + + assert record.acceptable is False + assert "kernel_name_regex" in (record.notes or "") + + +def test_profile_experiment_prefers_valid_candidate(monkeypatch): + experiment_spec = load_experiment_spec(Path("configs/experiments/gemm_reportable.yaml")) + shape = experiment_spec.shapes[0] + picked: list[str] = [] + + invalid = CandidateConfig( + experiment_id=experiment_spec.experiment_id, + kernel_id=experiment_spec.kernels[0], + shape_id=shape.shape_id, + config_id="cfg_invalid", + config={"block_m": 64}, + is_valid=False, + ) + valid = CandidateConfig( + experiment_id=experiment_spec.experiment_id, + kernel_id=experiment_spec.kernels[0], + shape_id=shape.shape_id, + config_id="cfg_valid", + config={"block_m": 128}, + is_valid=True, + ) + + monkeypatch.setattr( + "kernel_tuner.profiling.adapter.generate_candidate_records", + lambda *args, **kwargs: [invalid, valid], + ) + monkeypatch.setattr( + "kernel_tuner.experiments.orchestrator._shape_split", + lambda spec: ([shape], []), + ) + monkeypatch.setattr( + "kernel_tuner.experiments.orchestrator._profile_shapes", + lambda spec, calibration_shapes: calibration_shapes, + ) + + def fake_profile_candidate(**kwargs): + picked.append(kwargs["candidate"].config_id) + measurement = ProfileMeasurement( + run_id="standalone", + strategy_id="profile_cli", + kernel_id=kwargs["kernel_id"], + shape_id=kwargs["shape"].shape_id, + config_id=kwargs["candidate"].config_id, + counter_set_id=kwargs["counter_set"].counter_set_id, + profile_status=ProfileStatus.SUCCESS, + counter_map={}, + ) + return ProfileOutcome(measurement=measurement) + + monkeypatch.setattr("kernel_tuner.profiling.adapter.profile_candidate", fake_profile_candidate) + + result = profile_experiment(experiment_spec) + + assert picked == ["cfg_valid"] + assert result["measurements"][0]["config_id"] == "cfg_valid" diff --git a/tests/unit/test_reporting.py b/tests/unit/test_reporting.py index b229f4e..42b58b8 100644 --- a/tests/unit/test_reporting.py +++ b/tests/unit/test_reporting.py @@ -1,8 +1,15 @@ import math +from datetime import datetime, timezone +from pathlib import Path import pandas as pd +import pytest -from kernel_tuner.analysis.reporting import _pairwise_speedups +from kernel_tuner.analysis.reporting import _load_table, _pairwise_speedups, summarize_run +from kernel_tuner.common.config import load_experiment_spec +from kernel_tuner.common.provenance import capture_environment_metadata, capture_invocation_metadata +from kernel_tuner.common.schema import Manifest +from kernel_tuner.storage import RunStore def test_pairwise_speedups_uses_geometric_mean_of_per_shape_ratios(): @@ -26,3 +33,105 @@ def test_pairwise_speedups_uses_geometric_mean_of_per_shape_ratios(): expected = math.sqrt((10.0 / 5.0) * (20.0 / 30.0)) assert speedup == expected + + +def test_load_table_raises_for_required_missing_artifact(tmp_path): + store = RunStore(tmp_path, "exp", "run") + + with pytest.raises(FileNotFoundError): + _load_table(store, "runtime_measurements", required=True) + + +def test_summarize_run_marks_budget_limited_runs_non_reportable(tmp_path): + experiment_spec = load_experiment_spec(Path("configs/experiments/gemm_reportable.yaml")) + store = RunStore(tmp_path, experiment_spec.experiment_id, "run_001") + manifest = Manifest( + experiment_id=experiment_spec.experiment_id, + run_id="run_001", + created_at_utc=datetime.now(timezone.utc), + environment=capture_environment_metadata("."), + invocation=capture_invocation_metadata( + "pytest", + experiment_config_path=str(Path("configs/experiments/gemm_reportable.yaml").resolve()), + ), + artifact_files=[], + ) + store.initialize_manifest(manifest) + store.write_experiment_spec(experiment_spec) + + pd.DataFrame( + [ + { + "kernel_id": experiment_spec.kernels[0], + "shape_id": experiment_spec.shapes[0].shape_id, + "config_id": "cfg_a", + "compile_status": "success", + "compile_success": True, + "occupancy_estimate": 0.5, + "register_count": 64, + "shared_memory_bytes": 1024, + } + ] + ).to_parquet(store.run_dir / "compile_signals.parquet", index=False) + pd.DataFrame( + [ + { + "measurement_phase": "held_out", + "status": "success", + "shape_id": experiment_spec.shapes[0].shape_id, + "strategy_id": "default_config", + "latency_median_us": 10.0, + "throughput_value": 1.0, + "config_id": "cfg_default", + }, + { + "measurement_phase": "held_out", + "status": "success", + "shape_id": experiment_spec.shapes[0].shape_id, + "strategy_id": "prune_rank", + "latency_median_us": 9.0, + "throughput_value": 1.0, + "config_id": "cfg_a", + }, + ] + ).to_parquet(store.run_dir / "runtime_measurements.parquet", index=False) + pd.DataFrame( + [ + { + "strategy_id": "prune_rank", + "counter_set_id": experiment_spec.counter_set_id, + "profile_status": "success", + "counter_map": '{"sm__warps_active.avg.pct_of_peak_sustained_active": 1.0}', + "config_id": "cfg_a", + } + ] + ).to_parquet(store.run_dir / "profile_measurements.parquet", index=False) + pd.DataFrame( + [ + { + "strategy_id": "default_config", + "selected_config_id": "cfg_default", + "benchmarks_requested": 1, + "profiles_requested": 0, + "decision_status": "selected", + "comparison_class": "matched_budget", + }, + { + "strategy_id": "prune_rank", + "selected_config_id": "cfg_a", + "benchmarks_requested": 1, + "profiles_requested": 1, + "decision_status": "selected_budget_limited", + "comparison_class": "matched_budget", + }, + ] + ).to_parquet(store.run_dir / "selection_decisions.parquet", index=False) + (store.run_dir / "counter_compatibility.json").write_text( + '{"acceptable": true, "counter_set_id": "compute_lite"}', + encoding="utf-8", + ) + + summary = summarize_run(store.run_dir) + + assert summary["reportability"]["is_reportable"] is False + assert summary["reportability"]["budget_limited_decision_present"] is True From 4e918716ad282000b505a6c13a4e190e12f8622d Mon Sep 17 00:00:00 2001 From: lullu57 Date: Sat, 28 Mar 2026 14:57:41 -0400 Subject: [PATCH 3/5] Fix profiling tool resolution and smoke counters --- README.md | 3 +- configs/counters/default_calibration.yaml | 2 +- docs/gpu_job_guide.md | 6 +- ...-03-28_profiling_reliability_validation.md | 90 +++++++++++++++++++ scripts/slurm/run_kerneltuner_array.sbatch | 2 +- src/kernel_tuner/common/provenance.py | 16 +++- src/kernel_tuner/profiling/adapter.py | 17 +++- src/kernel_tuner/profiling/compatibility.py | 8 +- tests/integration/test_slurm_scripts.py | 5 ++ tests/unit/test_profiling.py | 46 ++++++++++ tests/unit/test_profiling_compatibility.py | 49 ++++++++++ 11 files changed, 229 insertions(+), 15 deletions(-) create mode 100644 docs/research/logs/2026-03-28_profiling_reliability_validation.md diff --git a/README.md b/README.md index 43fa736..5f72234 100644 --- a/README.md +++ b/README.md @@ -84,8 +84,7 @@ On the pinned GPU environment: ```bash export KTUNE_SCRATCH=/scratch/scratch-space/expires-xxxx/$USER/kerneltuner -scripts/bootstrap_env.sh "$KTUNE_SCRATCH/venv-py312" -source "$KTUNE_SCRATCH/venv-py312/bin/activate" +source scripts/bootstrap_env.sh "$KTUNE_SCRATCH/venv-py312" ktune validate-kernel --kernel configs/kernels/gemm.yaml ktune run-experiment --experiment configs/experiments/gemm_smoke.yaml diff --git a/configs/counters/default_calibration.yaml b/configs/counters/default_calibration.yaml index 9e26099..5b8a661 100644 --- a/configs/counters/default_calibration.yaml +++ b/configs/counters/default_calibration.yaml @@ -4,7 +4,7 @@ tool: ncu replay_mode: kernel target_processes: all counters: - - smsp__average_warps_issue_stalled_long_scoreboard_per_issue_active.avg + - smsp__warp_issue_stalled_long_scoreboard_per_warp_active.pct - sm__warps_active.avg.pct_of_peak_sustained_active - smsp__inst_executed.sum minimum_availability: 0.9 diff --git a/docs/gpu_job_guide.md b/docs/gpu_job_guide.md index 9688cfb..5d4b891 100644 --- a/docs/gpu_job_guide.md +++ b/docs/gpu_job_guide.md @@ -144,15 +144,13 @@ else export KTUNE_SCRATCH="/tmp/$USER/kerneltuner" fi -scripts/bootstrap_env.sh "$KTUNE_SCRATCH/venv-py312" -source "$KTUNE_SCRATCH/venv-py312/bin/activate" +source scripts/bootstrap_env.sh "$KTUNE_SCRATCH/venv-py312" ``` If you want to choose the virtualenv path explicitly: ```bash -scripts/bootstrap_env.sh /scratch/scratch-space/expires-xxxx/$USER/kerneltuner/venv-py312 -source /scratch/scratch-space/expires-xxxx/$USER/kerneltuner/venv-py312/bin/activate +source scripts/bootstrap_env.sh /scratch/scratch-space/expires-xxxx/$USER/kerneltuner/venv-py312 ``` The bootstrap script: diff --git a/docs/research/logs/2026-03-28_profiling_reliability_validation.md b/docs/research/logs/2026-03-28_profiling_reliability_validation.md new file mode 100644 index 0000000..7093459 --- /dev/null +++ b/docs/research/logs/2026-03-28_profiling_reliability_validation.md @@ -0,0 +1,90 @@ +# Profiling Reliability Validation On gpunode2 + +Purpose: record the profiling reliability fix, the counter-set audit, and the first reportable reruns on the pinned validation host. +Status: Log +Update Rule: append-only; do not rewrite past observations except to fix factual errors. +Feeds Paper Sections: operational methodology, profiling validity, limitations. +Depends On: [../04_signal_and_profiling_plan.md](../04_signal_and_profiling_plan.md), [../06_hypotheses_and_ablation_plan.md](../06_hypotheses_and_ablation_plan.md), [../08_evidence_registry.md](../08_evidence_registry.md), [../../04_experiment_protocol.md](../../04_experiment_protocol.md) + +## What Changed + +- preserved the CUDA toolchain exports by making `source scripts/bootstrap_env.sh ` the canonical launch contract for both manual GPU-shell runs and the Slurm worker path +- updated profiler invocation and provenance capture to resolve tool paths through `CUDA_HOME/bin` before falling back to `PATH` +- tightened profiling diagnostics so artifacts distinguish missing profiler binaries, missing queried metrics, invocation failures, kernel attribution failures, and runs that return null counter values +- replaced the fragile `default_calibration` long-scoreboard counter with a queryable and populated variant on `gpunode2` + +## Original Failure + +- baseline smoke run: `artifacts/gemm_smoke/run_20260328T153717Z_fb053320` +- observed failure: `counter_compatibility.notes` reported `ncu executable not found` +- root cause: `bootstrap_env.sh` was executed in a subprocess, so the exported CUDA tool paths were lost before `ktune` launched profiling + +## Intermediate Validation + +- post-bootstrap-fix smoke run: `artifacts/gemm_smoke/run_20260328T155619Z_7fd01b07` +- result: profiler path was fixed and `counter_compatibility.validation_backend` became `ncu_query_metrics` +- remaining issue: `smsp__average_warps_issue_stalled_long_scoreboard_per_issue_active.avg` populated `0/4` rows and caused `unsupported_counter` failures in the calibration profile path +- interpretation: the environment bug was resolved, but the default smoke counter set still contained a metric that was operationally fragile on the pinned host + +## Counter Audit Outcome + +### `default_calibration` + +- replaced `smsp__average_warps_issue_stalled_long_scoreboard_per_issue_active.avg` +- with `smsp__warp_issue_stalled_long_scoreboard_per_warp_active.pct` +- reason: the original metric queried successfully but returned null values in live smoke profiling on `gpunode2` + +### `compute_lite` + +- validation command reported `6/6` queried metrics available +- reportable GEMM run populated every configured counter for profiled strategies +- no counter replacement was required + +### `memory_lite` + +- validation command reported `6/6` queried metrics available +- reportable LayerNorm run populated every configured counter for profiled strategies +- no counter replacement was required + +## Rerun Results + +### Smoke + +- rerun: `artifacts/gemm_smoke/run_20260328T182208Z_78c5f7dc` +- `profile_failures` reduced to `{"success": 4}` +- `counter_set_accepted` was `true` +- all calibration counters populated `4/4` rows in `counter_availability_report.csv` + +### GEMM reportable + +- rerun: `artifacts/gemm_reportable/run_20260328T182338Z_6e92e07c` +- `terminal_status`: `success` +- `profile_failures`: `{"skipped_budget": 24, "success": 8}` +- `counter_set_accepted`: `true` +- all `compute_lite` counters populated `4/4` rows for `prune_rank_profiled` and `prune_rank_revised` +- `reportability.is_reportable`: `false` +- limiting reason: `budget_limited_decision_present` + +### LayerNorm reportable + +- rerun: `artifacts/layernorm_reportable/run_20260328T182601Z_5351e70a` +- `terminal_status`: `success` +- `profile_failures`: `{"success": 8, "skipped_budget": 8}` +- `counter_set_accepted`: `true` +- all `memory_lite` counters populated `4/4` rows for `prune_rank_profiled` and `prune_rank_revised` +- `reportability.is_reportable`: `false` +- limiting reason: `budget_limited_decision_present` + +## Interpretation + +- profiling is now operationally trustworthy on `gpunode2` for the audited smoke and reportable paths +- the resolved problem was environment propagation, not a fundamental incompatibility between Nsight Compute and the pinned host +- the remaining obstacle to study-level hypothesis testing is reportability semantics under matched-budget runs, not counter availability or profiler invocation +- this means the profiling hypothesis can now be tested operationally, but the current `validation_phase` study spec still does not admit these first reruns as study evidence because it requires reportable runs and broader repeat/seed coverage + +## Immediate Follow-up + +- treat the profiling tool-path issue as closed for PR `#11` +- keep `default_calibration` as a smoke-only compatibility set with intentionally robust counters +- address `budget_limited_decision_present` semantics before expecting `compare-runs --spec configs/studies/validation_phase.yaml` to ingest these reruns +- expand reportable validation coverage to satisfy the study spec's repeat and seed requirements before drawing hypothesis-level conclusions diff --git a/scripts/slurm/run_kerneltuner_array.sbatch b/scripts/slurm/run_kerneltuner_array.sbatch index d778ec2..5ce46a1 100755 --- a/scripts/slurm/run_kerneltuner_array.sbatch +++ b/scripts/slurm/run_kerneltuner_array.sbatch @@ -172,7 +172,7 @@ fi VENV_PATH="$SCRATCH_ROOT/$VENV_NAME" if [[ "${INSTALL_PACKAGES:-1}" == "1" ]]; then echo "Bootstrapping environment into $VENV_PATH" - "$WORKSPACE_ROOT/scripts/bootstrap_env.sh" "$VENV_PATH" + source "$WORKSPACE_ROOT/scripts/bootstrap_env.sh" "$VENV_PATH" else if [[ ! -d "$VENV_PATH" ]]; then echo "ERROR: virtual environment missing at $VENV_PATH and INSTALL_PACKAGES=0" >&2 diff --git a/src/kernel_tuner/common/provenance.py b/src/kernel_tuner/common/provenance.py index 3633a6d..4b87c42 100644 --- a/src/kernel_tuner/common/provenance.py +++ b/src/kernel_tuner/common/provenance.py @@ -27,6 +27,16 @@ def _run_command(args: list[str], cwd: str | Path | None = None) -> str | None: return completed.stdout.strip() +def resolve_tool_path(name: str) -> str: + cuda_home = os.environ.get("CUDA_HOME") + if cuda_home: + candidate = Path(cuda_home) / "bin" / name + if candidate.exists(): + return str(candidate) + resolved = which(name) + return resolved or name + + def _import_version(module_name: str) -> str | None: try: module = __import__(module_name) @@ -85,7 +95,7 @@ def _capture_git(repo_root: str | Path) -> dict[str, str | bool | None]: def _capture_ncu_version() -> str | None: - output = _run_command(["ncu", "--version"]) + output = _run_command([resolve_tool_path("ncu"), "--version"]) if output: line = output.splitlines()[-1].strip() return line @@ -104,8 +114,8 @@ def _cache_roots() -> dict[str, str]: def _tool_paths() -> dict[str, str]: tools = {} for name in ["python3", "ncu", "nsys", "nvcc"]: - path = which(name) - if path: + path = resolve_tool_path(name) + if path != name: tools[name] = path return tools diff --git a/src/kernel_tuner/profiling/adapter.py b/src/kernel_tuner/profiling/adapter.py index 5053a2a..2b59831 100644 --- a/src/kernel_tuner/profiling/adapter.py +++ b/src/kernel_tuner/profiling/adapter.py @@ -20,7 +20,7 @@ load_experiment_spec, load_kernel_spec, ) -from kernel_tuner.common.provenance import python_command +from kernel_tuner.common.provenance import python_command, resolve_tool_path from kernel_tuner.common.schema import ( CandidateConfig, CounterSetSpec, @@ -149,8 +149,9 @@ def _parse_counter_map( @lru_cache(maxsize=1) def _ncu_version() -> str | None: try: + ncu = resolve_tool_path("ncu") completed = subprocess.run( - ["ncu", "--version"], + [ncu, "--version"], check=False, capture_output=True, text=True, @@ -172,6 +173,7 @@ def _base_profiler_metadata( ) -> dict[str, Any]: return { "command": command, + "resolved_profiler_path": command[0], "requested_counters": list(counter_set.counters), "target_processes": counter_set.target_processes or "all", "replay_mode": counter_set.replay_mode or experiment_spec.profiling_settings.replay_mode, @@ -190,8 +192,9 @@ def _build_profile_command( profiling_settings, payload: str, ) -> list[str]: + ncu = resolve_tool_path("ncu") command = [ - "ncu", + ncu, "--csv", "--page", "raw", @@ -262,6 +265,7 @@ def profile_candidate( **base_metadata, "returncode": None, "duration_s": time.perf_counter() - command_started, + "failure_reason": "tool_unavailable", }, notes="ncu executable not found", ) @@ -281,6 +285,7 @@ def profile_candidate( **base_metadata, "returncode": None, "duration_s": time.perf_counter() - command_started, + "failure_reason": "timeout", }, notes=f"ncu timed out: {exc}", ) @@ -298,18 +303,23 @@ def profile_candidate( ) status = ProfileStatus.SUCCESS notes = None + failure_reason = None attribution_failed = matched_kernel_name is None or diagnostics.get("kernel_attribution_status") != "matched" if completed.returncode != 0 and unsupported_metric: status = ProfileStatus.UNSUPPORTED_COUNTER + failure_reason = "unsupported_metric" notes = completed.stderr.strip() or completed.stdout.strip() or "unsupported counter" elif completed.returncode != 0: status = ProfileStatus.INVOCATION_FAILED + failure_reason = "invocation_failed" notes = completed.stderr.strip() or completed.stdout.strip() or "ncu invocation failed" elif attribution_failed: status = ProfileStatus.NO_PROFILE_DATA + failure_reason = f"kernel_attribution_{diagnostics.get('kernel_attribution_status')}" notes = f"no attributable profiler row: {diagnostics.get('kernel_attribution_status')}" elif missing_counters: status = ProfileStatus.UNSUPPORTED_COUNTER + failure_reason = "missing_counter_values" notes = "missing or unsupported counters: " + ", ".join(sorted(missing_counters)) measurement = ProfileMeasurement( @@ -327,6 +337,7 @@ def profile_candidate( "duration_s": time.perf_counter() - command_started, "matched_kernel_name": matched_kernel_name, "missing_counters": missing_counters, + "failure_reason": failure_reason, **diagnostics, }, notes=notes, diff --git a/src/kernel_tuner/profiling/compatibility.py b/src/kernel_tuner/profiling/compatibility.py index c650860..9333d62 100644 --- a/src/kernel_tuner/profiling/compatibility.py +++ b/src/kernel_tuner/profiling/compatibility.py @@ -13,6 +13,7 @@ load_experiment_spec, load_kernel_spec, ) +from kernel_tuner.common.provenance import resolve_tool_path from kernel_tuner.common.schema import CounterCompatibilityRecord, ExperimentSpec @@ -85,8 +86,9 @@ def validate_counter_set( try: completed = None if backend == "ncu_query_metrics": + ncu = resolve_tool_path("ncu") completed = subprocess.run( - ["ncu", "--query-metrics"], + [ncu, "--query-metrics"], check=False, capture_output=True, text=True, @@ -100,6 +102,10 @@ def validate_counter_set( discovered = set(re.findall(r"([A-Za-z0-9_.]+)", metrics_blob)) available = [counter for counter in requested if _metric_match(counter, discovered)] missing = [counter for counter in requested if counter not in available] + if missing: + notes_parts.append( + "missing or unsupported queried metrics: " + ", ".join(sorted(missing)) + ) elif completed is not None: backend = "ncu_query_failed" notes_parts.append(completed.stderr.strip() or completed.stdout.strip() or "ncu --query-metrics failed") diff --git a/tests/integration/test_slurm_scripts.py b/tests/integration/test_slurm_scripts.py index a42ee63..1947951 100644 --- a/tests/integration/test_slurm_scripts.py +++ b/tests/integration/test_slurm_scripts.py @@ -20,3 +20,8 @@ def test_submit_script_dry_run_supports_nodelist(): text=True, ) assert "--nodelist=gpunode2" in result.stdout + + +def test_worker_script_sources_bootstrap_env_so_cuda_paths_persist(): + script = Path("scripts/slurm/run_kerneltuner_array.sbatch").read_text() + assert 'source "$WORKSPACE_ROOT/scripts/bootstrap_env.sh" "$VENV_PATH"' in script diff --git a/tests/unit/test_profiling.py b/tests/unit/test_profiling.py index 3f3e046..6f35b7c 100644 --- a/tests/unit/test_profiling.py +++ b/tests/unit/test_profiling.py @@ -171,3 +171,49 @@ def test_profile_candidate_marks_ambiguous_kernel_attribution_as_no_profile_data assert outcome.measurement.profile_status == ProfileStatus.NO_PROFILE_DATA assert outcome.measurement.profiler_metadata["kernel_attribution_status"] == "regex_ambiguous" + + +def test_profile_candidate_records_missing_counter_failure_reason(monkeypatch): + experiment_spec = load_experiment_spec(Path("configs/experiments/gemm_reportable.yaml")) + counter_set = load_counter_set(Path("configs/counters/compute_lite.yaml")) + shape = experiment_spec.shapes[0] + candidate = CandidateConfig( + experiment_id=experiment_spec.experiment_id, + kernel_id=experiment_spec.kernels[0], + shape_id=shape.shape_id, + config_id="cfg_valid", + config={"block_m": 128}, + is_valid=True, + ) + + stdout = "\n".join( + [ + '"ID","Process ID","Process Name","Host Name","Kernel Name","gpu__time_duration.sum","sm__warps_active.avg.pct_of_peak_sustained_active"', + '1,10,"python","host","matmul_kernel",5.0,1.0', + ] + ) + + monkeypatch.setattr( + subprocess, + "run", + lambda *args, **kwargs: subprocess.CompletedProcess( + args=args[0], + returncode=0, + stdout=stdout, + stderr="", + ), + ) + + outcome = profile_candidate( + run_id="run_001", + strategy_id="prune_rank_profiled", + kernel_id=experiment_spec.kernels[0], + shape=shape, + candidate=candidate, + counter_set=counter_set, + experiment_spec=experiment_spec, + ) + + assert outcome.measurement.profile_status == ProfileStatus.UNSUPPORTED_COUNTER + assert outcome.measurement.profiler_metadata["failure_reason"] == "missing_counter_values" + assert "missing or unsupported counters" in (outcome.measurement.notes or "") diff --git a/tests/unit/test_profiling_compatibility.py b/tests/unit/test_profiling_compatibility.py index 442920d..672c100 100644 --- a/tests/unit/test_profiling_compatibility.py +++ b/tests/unit/test_profiling_compatibility.py @@ -1,7 +1,9 @@ +import os from pathlib import Path from subprocess import CompletedProcess from kernel_tuner.common.config import load_counter_set +from kernel_tuner.common.provenance import resolve_tool_path from kernel_tuner.profiling.compatibility import validate_counter_set @@ -44,3 +46,50 @@ def test_shared_diag_uses_queryable_suffix_metrics(): "l1tex__data_pipe_lsu_wavefronts_mem_shared.avg", "smsp__warp_issue_stalled_short_scoreboard_per_warp_active.pct", ] + + +def test_default_calibration_uses_queryable_suffix_metrics(): + counter_set = load_counter_set(Path("configs/counters/default_calibration.yaml")) + + assert counter_set.counters == [ + "smsp__warp_issue_stalled_long_scoreboard_per_warp_active.pct", + "sm__warps_active.avg.pct_of_peak_sustained_active", + "smsp__inst_executed.sum", + ] + + +def test_resolve_tool_path_prefers_cuda_home_bin(tmp_path, monkeypatch): + cuda_bin = tmp_path / "cuda" / "bin" + cuda_bin.mkdir(parents=True) + ncu = cuda_bin / "ncu" + ncu.write_text("") + monkeypatch.setenv("CUDA_HOME", str(tmp_path / "cuda")) + monkeypatch.setattr("kernel_tuner.common.provenance.which", lambda name: None) + + assert resolve_tool_path("ncu") == str(ncu) + + +def test_validate_counter_set_records_missing_metrics_in_notes(monkeypatch): + counter_set = load_counter_set(Path("configs/counters/compute_lite.yaml")) + + monkeypatch.setattr( + "kernel_tuner.profiling.compatibility.subprocess.run", + lambda *args, **kwargs: CompletedProcess( + args=args[0], + returncode=0, + stdout="\n".join( + [ + "sm__warps_active", + "smsp__inst_executed", + ] + ), + stderr="", + ), + ) + + record = validate_counter_set(counter_set, kernel_family="gemm") + + assert record.acceptable is False + assert record.validation_backend == "ncu_query_metrics" + assert "missing or unsupported queried metrics" in (record.notes or "") + assert "smsp__inst_executed_pipe_tensor_op_hmma.avg" in (record.notes or "") From 3fb1b6f747ed921243c4550aa2375d50ae965ce8 Mon Sep 17 00:00:00 2001 From: lullu57 Date: Sat, 28 Mar 2026 14:57:49 -0400 Subject: [PATCH 4/5] Treat matched-budget runs as reportable evidence --- src/kernel_tuner/analysis/comparison.py | 91 ++++++++++++++---- src/kernel_tuner/analysis/reporting.py | 7 +- tests/unit/test_phase2_analysis.py | 60 ++++++++++++ tests/unit/test_reporting.py | 118 +++++++++++++++++++++--- 4 files changed, 241 insertions(+), 35 deletions(-) diff --git a/src/kernel_tuner/analysis/comparison.py b/src/kernel_tuner/analysis/comparison.py index 215dd56..91649ad 100644 --- a/src/kernel_tuner/analysis/comparison.py +++ b/src/kernel_tuner/analysis/comparison.py @@ -72,9 +72,12 @@ def compare_runs( ) store.initialize_manifest(manifest) try: - run_payloads = _resolve_run_payloads(study_spec, study_path) + run_payloads, filter_diagnostics = _resolve_run_payloads(study_spec, study_path) if not run_payloads: - raise ValueError("no study runs matched the configured groups and filters") + raise ValueError( + "no study runs matched the configured groups and filters" + + _format_filter_diagnostics(filter_diagnostics) + ) strategy_rows = _build_strategy_rows(run_payloads) if strategy_rows.empty: raise ValueError("no held-out strategy rows were available after loading matched study runs") @@ -156,18 +159,43 @@ def validate_study_from_path(study_path: str | Path) -> dict[str, object]: } -def _resolve_run_payloads(study_spec: StudySpec, study_path: str | Path | None) -> list[dict[str, Any]]: +def _resolve_run_payloads( + study_spec: StudySpec, + study_path: str | Path | None, +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: payloads: list[dict[str, Any]] = [] + diagnostics: list[dict[str, Any]] = [] seen_run_dirs: set[Path] = set() for group in study_spec.run_groups: + group_stats = { + "group_id": group.group_id, + "candidate_run_count": 0, + "matched_run_count": 0, + "excluded_reportability": 0, + "excluded_environment": 0, + "excluded_kernel_family": 0, + "excluded_selector_version": 0, + "excluded_selector_revision_id": 0, + "excluded_counter_set_id": 0, + "excluded_budget_id": 0, + "excluded_execution_mode": 0, + "excluded_seed": 0, + "excluded_repeat_index": 0, + } for run_dir in _resolve_group_run_dirs(group, study_spec, study_path): if run_dir in seen_run_dirs: continue seen_run_dirs.add(run_dir) + group_stats["candidate_run_count"] += 1 payload = _load_run_payload(run_dir, group.group_id) - if _passes_filters(payload, study_spec): + accepted, reason = _passes_filters(payload, study_spec) + if accepted: payloads.append(payload) - return payloads + group_stats["matched_run_count"] += 1 + elif reason is not None: + group_stats[reason] += 1 + diagnostics.append(group_stats) + return payloads, diagnostics def _resolve_group_run_dirs(group, study_spec: StudySpec, study_path: str | Path | None) -> list[Path]: @@ -233,37 +261,64 @@ def _load_run_payload(run_dir: Path, group_id: str) -> dict[str, Any]: } -def _passes_filters(payload: dict[str, Any], study_spec: StudySpec) -> bool: +def _passes_filters(payload: dict[str, Any], study_spec: StudySpec) -> tuple[bool, str | None]: summary = payload["summary"] manifest = payload["manifest"] labels = payload.get("run_labels") or {} group_id = payload["group_id"] group = next((item for item in study_spec.run_groups if item.group_id == group_id), None) if study_spec.reportability_filter and not summary.get("reportability", {}).get("is_reportable", False): - return False + return False, "excluded_reportability" for field, expected in study_spec.environment_filter.items(): actual = getattr(manifest.environment, field, None) if actual != expected: - return False + return False, "excluded_environment" if group is None: - return True + return True, None if group.kernel_family and labels.get("kernel_family") != group.kernel_family: - return False + return False, "excluded_kernel_family" if group.selector_version and labels.get("selector_version") != group.selector_version: - return False + return False, "excluded_selector_version" if group.selector_revision_id and labels.get("selector_revision_id") != group.selector_revision_id: - return False + return False, "excluded_selector_revision_id" if group.counter_set_id and labels.get("counter_set_id") != group.counter_set_id: - return False + return False, "excluded_counter_set_id" if group.budget_id and labels.get("budget_id") != group.budget_id: - return False + return False, "excluded_budget_id" if group.execution_mode and labels.get("execution_mode") != group.execution_mode: - return False + return False, "excluded_execution_mode" if group.seeds and labels.get("seed") not in group.seeds: - return False + return False, "excluded_seed" if group.repeat_indices and labels.get("repeat_index") not in group.repeat_indices: - return False - return True + return False, "excluded_repeat_index" + return True, None + + +def _format_filter_diagnostics(filter_diagnostics: list[dict[str, Any]]) -> str: + if not filter_diagnostics: + return "" + parts: list[str] = [] + for row in filter_diagnostics: + details = [ + f"candidates={row['candidate_run_count']}", + f"matched={row['matched_run_count']}", + ] + for key, label in [ + ("excluded_reportability", "reportability"), + ("excluded_environment", "environment"), + ("excluded_kernel_family", "kernel_family"), + ("excluded_selector_version", "selector_version"), + ("excluded_selector_revision_id", "selector_revision_id"), + ("excluded_counter_set_id", "counter_set_id"), + ("excluded_budget_id", "budget_id"), + ("excluded_execution_mode", "execution_mode"), + ("excluded_seed", "seed"), + ("excluded_repeat_index", "repeat_index"), + ]: + if row[key]: + details.append(f"{label}={row[key]}") + parts.append(f"{row['group_id']}(" + ", ".join(details) + ")") + return ": " + "; ".join(parts) def _build_strategy_rows(run_payloads: list[dict[str, Any]]) -> pd.DataFrame: diff --git a/src/kernel_tuner/analysis/reporting.py b/src/kernel_tuner/analysis/reporting.py index 4402d24..baeba03 100644 --- a/src/kernel_tuner/analysis/reporting.py +++ b/src/kernel_tuner/analysis/reporting.py @@ -424,7 +424,6 @@ def summarize_run(run_dir: str | Path) -> dict[str, object]: experiment_spec.study_kind == "reportable" and held_out_shape_count >= policy.minimum_held_out_shapes and matched_budget_only - and not has_budget_limited_decision and (not counter_compatibility or counter_compatibility.get("acceptable", True)) and (not counter_availability_ok or all(counter_availability_ok.values())) and ( @@ -467,6 +466,10 @@ def summarize_run(run_dir: str | Path) -> dict[str, object]: interpretation_notes.append("No usable profiler counter availability rows were recorded.") elif not availability_frame.empty and not availability_frame["acceptable"].all(): interpretation_notes.append("At least one requested profiler counter failed the availability threshold.") + if has_budget_limited_decision: + interpretation_notes.append( + "At least one strategy exhausted its matched budget before evaluating every requested profile or benchmark." + ) result = ExperimentResult( schema_version=SCHEMA_VERSION, @@ -501,7 +504,7 @@ def summarize_run(run_dir: str | Path) -> dict[str, object]: reportability={ "target": experiment_spec.analysis_settings.reportability_target, "is_reportable": is_reportable, - "comparison_class": "matched_budget" if is_reportable else "non_comparable", + "comparison_class": "matched_budget" if matched_budget_only else "non_comparable", "matched_budget_only": matched_budget_only, "budget_limited_decision_present": has_budget_limited_decision, "counter_set_accepted": bool(counter_compatibility.get("acceptable", True)) diff --git a/tests/unit/test_phase2_analysis.py b/tests/unit/test_phase2_analysis.py index d50fd45..433f015 100644 --- a/tests/unit/test_phase2_analysis.py +++ b/tests/unit/test_phase2_analysis.py @@ -9,6 +9,8 @@ _build_strategy_rows, _compare_hypothesis_values, _evaluate_hypotheses, + _format_filter_diagnostics, + _passes_filters, ) from kernel_tuner.common.schema import ( HypothesisClause, @@ -179,6 +181,64 @@ def test_build_strategy_rows_preserves_kernel_family_and_workload_class(): assert set(rows["workload_class"]) == {"small_batch"} +def test_passes_filters_accepts_reportable_budget_limited_runs(): + study = StudySpec( + study_id="study", + hypotheses=[], + run_groups=[], + reportability_filter=True, + ) + payload = { + "group_id": "group", + "summary": { + "reportability": { + "is_reportable": True, + "budget_limited_decision_present": True, + } + }, + "manifest": Manifest( + experiment_id="exp", + run_id="run_001", + created_at_utc=datetime.now(timezone.utc), + environment=capture_environment_metadata("."), + invocation=capture_invocation_metadata("pytest"), + artifact_files=[], + ), + "run_labels": {}, + } + + accepted, reason = _passes_filters(payload, study) + + assert accepted is True + assert reason is None + + +def test_format_filter_diagnostics_mentions_seed_and_repeat_exclusions(): + message = _format_filter_diagnostics( + [ + { + "group_id": "gemm_representative", + "candidate_run_count": 3, + "matched_run_count": 0, + "excluded_reportability": 0, + "excluded_environment": 0, + "excluded_kernel_family": 0, + "excluded_selector_version": 0, + "excluded_selector_revision_id": 0, + "excluded_counter_set_id": 0, + "excluded_budget_id": 0, + "excluded_execution_mode": 0, + "excluded_seed": 2, + "excluded_repeat_index": 1, + } + ] + ) + + assert "gemm_representative" in message + assert "seed=2" in message + assert "repeat_index=1" in message + + def test_evaluate_hypotheses_uses_clause_based_metrics(): strategy_rows = pd.DataFrame( [ diff --git a/tests/unit/test_reporting.py b/tests/unit/test_reporting.py index da83288..9421264 100644 --- a/tests/unit/test_reporting.py +++ b/tests/unit/test_reporting.py @@ -139,7 +139,7 @@ def test_summarize_run_falls_back_to_manifest_experiment_config(tmp_path): assert summary["run_id"] == "run_001" -def test_summarize_run_marks_budget_limited_runs_non_reportable(tmp_path): +def test_summarize_run_allows_budget_limited_runs_when_other_reportability_requirements_pass(tmp_path): experiment_spec = load_experiment_spec(Path("configs/experiments/gemm_reportable.yaml")) store = RunStore(tmp_path, experiment_spec.experiment_id, "run_001") manifest = Manifest( @@ -170,6 +170,105 @@ def test_summarize_run_marks_budget_limited_runs_non_reportable(tmp_path): } ] ).to_parquet(store.run_dir / "compile_signals.parquet", index=False) + held_out_shape_ids = [ + experiment_spec.shapes[0].shape_id, + experiment_spec.shapes[3].shape_id, + experiment_spec.shapes[6].shape_id, + experiment_spec.shapes[9].shape_id, + ] + runtime_rows = [] + for idx, shape_id in enumerate(held_out_shape_ids): + runtime_rows.extend( + [ + { + "measurement_phase": "held_out", + "status": "success", + "shape_id": shape_id, + "strategy_id": "default_config", + "latency_median_us": 10.0 + idx, + "throughput_value": 1.0, + "config_id": "cfg_default", + }, + { + "measurement_phase": "held_out", + "status": "success", + "shape_id": shape_id, + "strategy_id": "prune_rank", + "latency_median_us": 9.0 + idx, + "throughput_value": 1.0, + "config_id": "cfg_a", + }, + ] + ) + pd.DataFrame(runtime_rows).to_parquet(store.run_dir / "runtime_measurements.parquet", index=False) + pd.DataFrame( + [ + { + "strategy_id": "prune_rank", + "counter_set_id": experiment_spec.counter_set_id, + "profile_status": "success", + "counter_map": ( + '{"sm__warps_active.avg.pct_of_peak_sustained_active": 1.0, ' + '"smsp__inst_executed.sum": 2.0, ' + '"smsp__inst_executed_pipe_tensor_op_hmma.avg": 3.0, ' + '"smsp__pipe_tensor_op_hmma_cycles_active.avg": 4.0, ' + '"smsp__warp_issue_stalled_math_pipe_throttle_per_warp_active.pct": 5.0, ' + '"smsp__warp_issue_stalled_long_scoreboard_per_warp_active.pct": 6.0}' + ), + "config_id": "cfg_a", + } + ] + ).to_parquet(store.run_dir / "profile_measurements.parquet", index=False) + pd.DataFrame( + [ + { + "strategy_id": "default_config", + "selected_config_id": "cfg_default", + "benchmarks_requested": 1, + "profiles_requested": 0, + "decision_status": "selected", + "comparison_class": "matched_budget", + }, + { + "strategy_id": "prune_rank", + "selected_config_id": "cfg_a", + "benchmarks_requested": 1, + "profiles_requested": 1, + "decision_status": "selected_budget_limited", + "comparison_class": "matched_budget", + }, + ] + ).to_parquet(store.run_dir / "selection_decisions.parquet", index=False) + (store.run_dir / "counter_compatibility.json").write_text( + '{"acceptable": true, "counter_set_id": "compute_lite"}', + encoding="utf-8", + ) + + summary = summarize_run(store.run_dir) + + assert summary["reportability"]["is_reportable"] is True + assert summary["reportability"]["comparison_class"] == "matched_budget" + assert summary["reportability"]["budget_limited_decision_present"] is True + + +def test_summarize_run_marks_mixed_comparison_class_runs_non_reportable(tmp_path): + experiment_spec = load_experiment_spec(Path("configs/experiments/gemm_reportable.yaml")) + store = RunStore(tmp_path, experiment_spec.experiment_id, "run_mixed") + manifest = Manifest( + experiment_id=experiment_spec.experiment_id, + run_id="run_mixed", + created_at_utc=datetime.now(timezone.utc), + environment=capture_environment_metadata("."), + invocation=capture_invocation_metadata( + "pytest", + experiment_config_path=str(Path("configs/experiments/gemm_reportable.yaml").resolve()), + ), + artifact_files=[], + ) + store.initialize_manifest(manifest) + store.write_experiment_spec(experiment_spec) + + _write_minimal_required_tables(store, include_profile=True) pd.DataFrame( [ { @@ -192,17 +291,6 @@ def test_summarize_run_marks_budget_limited_runs_non_reportable(tmp_path): }, ] ).to_parquet(store.run_dir / "runtime_measurements.parquet", index=False) - pd.DataFrame( - [ - { - "strategy_id": "prune_rank", - "counter_set_id": experiment_spec.counter_set_id, - "profile_status": "success", - "counter_map": '{"sm__warps_active.avg.pct_of_peak_sustained_active": 1.0}', - "config_id": "cfg_a", - } - ] - ).to_parquet(store.run_dir / "profile_measurements.parquet", index=False) pd.DataFrame( [ { @@ -218,8 +306,8 @@ def test_summarize_run_marks_budget_limited_runs_non_reportable(tmp_path): "selected_config_id": "cfg_a", "benchmarks_requested": 1, "profiles_requested": 1, - "decision_status": "selected_budget_limited", - "comparison_class": "matched_budget", + "decision_status": "selected", + "comparison_class": "non_comparable", }, ] ).to_parquet(store.run_dir / "selection_decisions.parquet", index=False) @@ -231,7 +319,7 @@ def test_summarize_run_marks_budget_limited_runs_non_reportable(tmp_path): summary = summarize_run(store.run_dir) assert summary["reportability"]["is_reportable"] is False - assert summary["reportability"]["budget_limited_decision_present"] is True + assert summary["reportability"]["comparison_class"] == "non_comparable" def test_summarize_run_marks_counter_set_unaccepted_when_compatibility_fails(tmp_path): From aeef9ce9cd27dfc3464594619b4d7d0164848e7b Mon Sep 17 00:00:00 2001 From: lullu57 Date: Sat, 28 Mar 2026 15:37:08 -0400 Subject: [PATCH 5/5] Prefer run-local specs when summarizing campaign runs --- src/kernel_tuner/analysis/comparison.py | 2 +- src/kernel_tuner/analysis/reporting.py | 2 +- tests/unit/test_phase2_analysis.py | 56 +++++++++++++++++++++++++ tests/unit/test_reporting.py | 25 +++++++++++ 4 files changed, 83 insertions(+), 2 deletions(-) diff --git a/src/kernel_tuner/analysis/comparison.py b/src/kernel_tuner/analysis/comparison.py index 91649ad..392dc7a 100644 --- a/src/kernel_tuner/analysis/comparison.py +++ b/src/kernel_tuner/analysis/comparison.py @@ -234,7 +234,7 @@ def _load_run_payload(run_dir: Path, group_id: str) -> dict[str, Any]: summarize_run(run_dir) summary = json.loads(summary_path.read_text(encoding="utf-8")) source_experiment_path = run_dir / "experiment_spec.yaml" - if manifest.invocation.experiment_config_path: + if not source_experiment_path.exists() and manifest.invocation.experiment_config_path: candidate = Path(manifest.invocation.experiment_config_path) if candidate.exists(): source_experiment_path = candidate diff --git a/src/kernel_tuner/analysis/reporting.py b/src/kernel_tuner/analysis/reporting.py index baeba03..ab6e42e 100644 --- a/src/kernel_tuner/analysis/reporting.py +++ b/src/kernel_tuner/analysis/reporting.py @@ -242,7 +242,7 @@ def summarize_run(run_dir: str | Path) -> dict[str, object]: source_experiment_path = experiment_spec_path if not experiment_spec_path.exists(): source_experiment_path = None - if manifest.invocation.experiment_config_path: + if source_experiment_path is None and manifest.invocation.experiment_config_path: candidate = Path(manifest.invocation.experiment_config_path) if candidate.exists(): source_experiment_path = candidate diff --git a/tests/unit/test_phase2_analysis.py b/tests/unit/test_phase2_analysis.py index 433f015..5943321 100644 --- a/tests/unit/test_phase2_analysis.py +++ b/tests/unit/test_phase2_analysis.py @@ -10,6 +10,7 @@ _compare_hypothesis_values, _evaluate_hypotheses, _format_filter_diagnostics, + _load_run_payload, _passes_filters, ) from kernel_tuner.common.schema import ( @@ -27,6 +28,7 @@ from kernel_tuner.common.config import load_experiment_spec, load_kernel_spec from kernel_tuner.common.provenance import capture_environment_metadata, capture_invocation_metadata from kernel_tuner.common.schema import Manifest +from kernel_tuner.storage import RunStore def test_counter_availability_records_apply_threshold(): @@ -239,6 +241,60 @@ def test_format_filter_diagnostics_mentions_seed_and_repeat_exclusions(): assert "repeat_index=1" in message +def test_load_run_payload_prefers_run_local_experiment_spec_over_manifest_path(tmp_path): + original_spec = load_experiment_spec(Path("configs/experiments/gemm_reportable.yaml")) + mutated_spec = original_spec.model_copy(deep=True) + mutated_spec.seed = 43 + store = RunStore(tmp_path / "artifacts", mutated_spec.experiment_id, "run_payload") + manifest = Manifest( + experiment_id=mutated_spec.experiment_id, + run_id="run_payload", + created_at_utc=datetime.now(timezone.utc), + environment=capture_environment_metadata("."), + invocation=capture_invocation_metadata( + "pytest", + experiment_config_path=str(Path("configs/experiments/gemm_reportable.yaml").resolve()), + ), + artifact_files=[], + ) + store.initialize_manifest(manifest) + store.write_experiment_spec(mutated_spec) + (store.run_dir / "summary.json").write_text( + '{"run_id":"run_payload","reportability":{"is_reportable":true}}', + encoding="utf-8", + ) + pd.DataFrame( + [{"strategy_id": "prune_rank", "selected_config_id": "cfg_a"}] + ).to_parquet(store.run_dir / "selection_decisions.parquet", index=False) + pd.DataFrame( + [ + { + "measurement_phase": "held_out", + "status": "success", + "shape_id": mutated_spec.shapes[0].shape_id, + "strategy_id": "prune_rank", + "latency_median_us": 9.0, + "config_id": "cfg_a", + } + ] + ).to_parquet(store.run_dir / "runtime_measurements.parquet", index=False) + pd.DataFrame( + [ + { + "shape_id": mutated_spec.shapes[0].shape_id, + "workload_class": mutated_spec.shapes[0].workload_class, + "strategy_id": "prune_rank", + "latency_median_us": 9.0, + "winner_on_shape": True, + } + ] + ).to_csv(store.run_dir / "held_out_per_shape.csv", index=False) + + payload = _load_run_payload(store.run_dir, "group") + + assert payload["experiment_spec"].seed == 43 + + def test_evaluate_hypotheses_uses_clause_based_metrics(): strategy_rows = pd.DataFrame( [ diff --git a/tests/unit/test_reporting.py b/tests/unit/test_reporting.py index 9421264..fdc9128 100644 --- a/tests/unit/test_reporting.py +++ b/tests/unit/test_reporting.py @@ -139,6 +139,31 @@ def test_summarize_run_falls_back_to_manifest_experiment_config(tmp_path): assert summary["run_id"] == "run_001" +def test_summarize_run_prefers_run_local_experiment_spec_over_manifest_path(tmp_path): + original_spec = load_experiment_spec(Path("configs/experiments/gemm_reportable.yaml")) + mutated_spec = original_spec.model_copy(deep=True) + mutated_spec.seed = 19 + store = RunStore(tmp_path / "artifacts", "test_experiment", "run_local_spec") + manifest = Manifest( + experiment_id="test_experiment", + run_id="run_local_spec", + created_at_utc=datetime.now(timezone.utc), + environment=capture_environment_metadata("."), + invocation=capture_invocation_metadata( + "pytest", + experiment_config_path=str(Path("configs/experiments/gemm_reportable.yaml").resolve()), + ), + artifact_files=[], + ) + store.initialize_manifest(manifest) + store.write_experiment_spec(mutated_spec) + _write_minimal_required_tables(store, include_profile=True) + + summary = summarize_run(store.run_dir) + + assert summary["aggregate_metrics"]["seed"] == 19 + + def test_summarize_run_allows_budget_limited_runs_when_other_reportability_requirements_pass(tmp_path): experiment_spec = load_experiment_spec(Path("configs/experiments/gemm_reportable.yaml")) store = RunStore(tmp_path, experiment_spec.experiment_id, "run_001")