[Tuning] Share mp_tuner typed statuses and add a central tuning policy

.github/scripts/split_tests.sh

@@ -56,7 +56,9 @@ if [[ "$TEST_TYPE" == "aiter" ]]; then
             printf '%s\n' \
                 "$TEST_DIR/tuning_tests/test_csv_validation.py" \
                 "$TEST_DIR/tuning_tests/test_config_shape_collision.py" \
-                "$TEST_DIR/tuning_tests/test_mixed_mxfp_tuning.py"
+                "$TEST_DIR/tuning_tests/test_mixed_mxfp_tuning.py" \
+                "$TEST_DIR/tuning_tests/test_mp_tuner_fault.py" \
+                "$TEST_DIR/tuning_tests/test_tuning_policy.py"
         } | LC_ALL=C sort -u
     )
 elif [[ "$TEST_TYPE" == "triton" ]]; then
@@ -183,6 +185,8 @@ if [[ "$TEST_TYPE" == "aiter" ]]; then
     FILE_TIMES[op_tests/tuning_tests/test_config_shape_collision.py]=4
     FILE_TIMES[op_tests/tuning_tests/test_csv_validation.py]=4
     FILE_TIMES[op_tests/tuning_tests/test_mixed_mxfp_tuning.py]=4
+    FILE_TIMES[op_tests/tuning_tests/test_mp_tuner_fault.py]=84
+    FILE_TIMES[op_tests/tuning_tests/test_tuning_policy.py]=4
     FILE_TIMES[op_tests/test_fused_qk_rmsnorm_per_token_quant.py]=5
     FILE_TIMES[op_tests/test_groupnorm.py]=5
     FILE_TIMES[op_tests/test_indexer_k_quant_and_cache.py]=5

.github/workflows/tuning-tests.yaml

@@ -93,7 +93,7 @@ jobs:
             -e PYTHONUNBUFFERED=1 \
             -w /workspace \
             tuning_tests_level01 \
-            bash -lc "set -euo pipefail; python3 -m unittest op_tests.tuning_tests.test_csv_validation op_tests.tuning_tests.test_tuner_infra op_tests.tuning_tests.test_mp_tuner_logic op_tests.tuning_tests.test_config_shape_collision op_tests.tuning_tests.test_a6w6_tuning op_tests.tuning_tests.test_mixed_mxfp_tuning -v 2>&1 | tee /workspace/${LOG_DIR}/level01.log"
+            bash -lc "set -euo pipefail; python3 -m unittest op_tests.tuning_tests.test_csv_validation op_tests.tuning_tests.test_tuner_infra op_tests.tuning_tests.test_mp_tuner_logic op_tests.tuning_tests.test_tuning_policy op_tests.tuning_tests.test_config_shape_collision op_tests.tuning_tests.test_a6w6_tuning op_tests.tuning_tests.test_mixed_mxfp_tuning -v 2>&1 | tee /workspace/${LOG_DIR}/level01.log"
 
       - name: Upload level 0+1 artifacts
         if: always()
@@ -176,7 +176,7 @@ jobs:
             -e PYTHONUNBUFFERED=1 \
             -w /workspace \
             tuning_tests_pipeline \
-            bash -lc "set -euo pipefail; python3 -m unittest op_tests.tuning_tests.test_tune_pipeline -v 2>&1 | tee /workspace/${LOG_DIR}/tune-pipeline.log"
+            bash -lc "set -euo pipefail; python3 -m unittest op_tests.tuning_tests.test_tune_pipeline op_tests.tuning_tests.test_mp_tuner_fault -v 2>&1 | tee /workspace/${LOG_DIR}/tune-pipeline.log"
 
       - name: Upload pipeline artifacts
         if: always()

aiter/jit/utils/file_baton.py

@@ -114,7 +114,14 @@ def _pid_alive(pid):
             return False
         except PermissionError:
             return True  # exists but owned by another user
-        return True
+        # A zombie answers signal 0 until its parent reaps it, but has already
+        # exited and will never release(), wedging the build for everyone after.
+        try:
+            with open(f"/proc/{pid}/stat", "rb") as stat_file:
+                # comm may contain spaces and parens; state follows the last ')'
+                return stat_file.read().rpartition(b")")[2].split()[0] != b"Z"
+        except (OSError, IndexError):
+            return True  # no procfs: trust the signal
 
     def _is_stale(self):
         """A lock is stale if its recorded holder is dead, or if it carries no

aiter/utility/base_tuner.py

@@ -16,6 +16,11 @@
 
 from aiter import dtypes, logger
 from aiter.jit.utils.chip_info import get_gfx_runtime as _chip_get_gfx
+from aiter.utility.tuning_policy import (
+    DEFAULT_MEASUREMENT,
+    DEFAULT_PROMOTION,
+    DEFAULT_RUN,
+)
 
 INVALID_TIME = -1
 
@@ -26,10 +31,17 @@ def _read_csv(filepath, **kwargs):
     """
     df = pd.read_csv(filepath, **kwargs)
     df.columns = df.columns.str.strip()
-    df = df.loc[:, ~df.columns.str.startswith("Unnamed:")]
+    df = df.loc[:, ~df.columns.str.startswith("Unnamed:")].copy()
     str_cols = df.select_dtypes(include=["object"]).columns
-    for col in str_cols:
-        df[col] = df[col].apply(lambda v: v.strip() if isinstance(v, str) else v)
+    if len(str_cols):
+        df = df.assign(
+            **{
+                col: df[col].map(
+                    lambda value: value.strip() if isinstance(value, str) else value
+                )
+                for col in str_cols
+            }
+        )
     df.dropna(how="all", inplace=True)
     return df
 
@@ -39,19 +51,15 @@ class TunerCommon:
         "verbose": False,
         "tune_file": "",
         "untune_file": "",
-        "errRatio": 0.05,
-        "batch": 100,
+        "errRatio": DEFAULT_MEASUREMENT.err_ratio,
+        "batch": DEFAULT_RUN.batch,
         "profile_file": "",  # for all results
-        # Per-task watchdog (seconds). A worker killed by a GPU memory-access
-        # fault leaves its in-flight task unresolvable; with no timeout the whole
-        # run hangs. A generous default lets the existing reaping path drop the
-        # lost task and restart the pool, while staying comfortably above any
-        # legitimate shape-group tuning time so healthy tasks are never falsely
-        # reaped. Override with --timeout for tighter/looser bounds.
-        "timeout": 1800,
-        "warmup": 5,  # 5 warmup iters for profiling
-        "iters": 101,  # 101 run iters for profiling
-        "min_improvement_pct": 3.0,  # only write shapes improved by >= N%
+        # Override with --timeout for tighter/looser bounds.
+        "timeout": DEFAULT_MEASUREMENT.timeout,
+        "warmup": DEFAULT_MEASUREMENT.warmup,
+        "iters": DEFAULT_MEASUREMENT.iters,
+        # only write shapes improved by >= N%
+        "min_improvement_pct": DEFAULT_PROMOTION.min_improvement_pct,
     }
     dtype2bpe_dict: ClassVar[dict[str, Any]] = {
         dtypes.fp16: 2,
@@ -236,7 +244,9 @@ def _setup_common_arguments(self):
             "--min_improvement_pct",
             dest="min_improvement_pct",
             type=float,
-            default=defaults.get("min_improvement_pct", 3.0),
+            default=defaults.get(
+                "min_improvement_pct", DEFAULT_PROMOTION.min_improvement_pct
+            ),
             help="With --compare --update_improved, update tuned CSV only when a valid pre/post benchmark shows at least this percent improvement. Shapes with no valid pre-run baseline but passing post-run are still allowed to update.",
         )
 
@@ -246,6 +256,19 @@ def parse_args(self):
             self.parser.error("--update_improved requires --compare")
         return args
 
+    @staticmethod
+    def measurement_kwargs(args, use_cuda_event=False):
+        """Timing keyword arguments for the function an mp_tuner task measures.
+
+        --warmup and --iters only take effect if they reach the timed call, so
+        a task that passes {} silently measures with run_perftest's defaults.
+        """
+        return {
+            "num_warmup": args.warmup,
+            "num_iters": args.iters,
+            "use_cuda_event": use_cuda_event,
+        }
+
     @abstractmethod
     def _setup_specific_arguments(self):
         """set specific arguments"""
@@ -669,7 +692,11 @@ def _get_run_config_err_ratio_limit(self, row, args):
         not falsely flag kernels whose error fluctuates slightly across seeds.
         """
         default_limit = float(
-            getattr(args, "errRatio", self.ARG_DEFAULTS.get("errRatio", 0.05))
+            getattr(
+                args,
+                "errRatio",
+                self.ARG_DEFAULTS.get("errRatio", DEFAULT_MEASUREMENT.err_ratio),
+            )
         )
         default_desc = f"--errRatio={default_limit:.6g}"
         if row is None or not hasattr(row, "get"):