from __future__ import annotations
import dataclasses
import functools
import inspect
import json
import logging
import os
import time
from typing import TYPE_CHECKING
from typing import Callable
from typing import Literal
from typing import Protocol
from typing import Sequence
from typing import TypeVar
from typing import cast
from typing import get_args
import torch
from torch._environment import is_fbcode
from .. import exc
from .._compat import is_hip
from .._compat import supports_tf32_precision_on_amd
from .._compiler.backend_registry import get_backend
from .._compiler.backend_registry import list_backends
from ..autotuner.effort_profile import AutotuneEffort
from ..autotuner.effort_profile import InitialPopulation
from ..autotuner.effort_profile import get_effort_profile
from .cute.structural_policy import CUTE_STRUCTURAL_SETTINGS
from .cute.structural_policy import CuteStructuralOrigins
from .cute.structural_policy import CuteStructuralPolicy
from .cute.structural_policy import StructuralSettingOrigin
from .ref_mode import RefMode
if TYPE_CHECKING:
from ..autotuner.base_search import BaseAutotuner
from ..autotuner.base_search import BaseSearch
from ..autotuner.pattern_search import InitialPopulationStrategy
from .config import Config
from .cute.structural_config import CuteStructuralConfig
from .kernel import BoundKernel
_T = TypeVar("_T")
ConfigLike = Config | dict[str, object] | CuteStructuralConfig
class AutotunerFunction(Protocol):
def __call__(
self, bound_kernel: BoundKernel, args: Sequence[object], **kwargs: object
) -> BaseAutotuner: ...
DotPrecision = Literal["tf32", "tf32x3", "ieee", "default", "high", "highest"]
PrecompileMode = Literal["spawn", "fork"] | None
# How the autotuner compares fp8 outputs against the baseline: dequantized
# values within one fp8 ulp plus the tolerances, or exact bytes.
AutotuneFp8Accuracy = Literal["ulp", "bitwise"]
_TRUE_LITERALS = frozenset({"1", "true", "yes", "on"})
_FALSE_LITERALS = frozenset({"0", "false", "no", "off"})
def _resolve_warning_name(name: str) -> type[exc.BaseWarning]:
attr = name.strip()
if not attr:
raise ValueError("HELION_IGNORE_WARNINGS entries must be non-empty names")
try:
warning_cls = getattr(exc, attr)
except AttributeError as err:
raise ValueError(
f"HELION_IGNORE_WARNINGS entry {name!r} is not a warning defined in helion.exc"
) from err
if not isinstance(warning_cls, type) or not issubclass(
warning_cls, exc.BaseWarning
):
raise ValueError(
f"HELION_IGNORE_WARNINGS entry {name!r} does not refer to a helion.exc.BaseWarning subclass"
)
return warning_cls
def _get_ignore_warnings() -> list[type[exc.BaseWarning]]:
value = os.environ.get("HELION_IGNORE_WARNINGS")
if not value:
return []
result: list[type[exc.BaseWarning]] = []
for entry in value.split(","):
entry = entry.strip()
if not entry:
continue
result.append(_resolve_warning_name(entry))
return result
def _env_get_optional_int(var_name: str) -> int | None:
value = os.environ.get(var_name)
if value is None or (value := value.strip()) == "":
return None
try:
parsed = int(value)
except ValueError as err:
raise ValueError(f"{var_name} must be an integer, got {value!r}") from err
return parsed
def _env_get_int(var_name: str, default: int) -> int:
result = _env_get_optional_int(var_name)
return default if result is None else result
def _env_get_optional_float(var_name: str) -> float | None:
value = os.environ.get(var_name)
if value is None or (value := value.strip()) == "":
return None
try:
return float(value)
except ValueError as err:
raise ValueError(f"{var_name} must be a float, got {value!r}") from err
def _env_get_bool(var_name: str, default: bool) -> bool:
return _parse_env_bool(var_name, os.environ.get(var_name), default)
def _parse_env_bool(var_name: str, value: str | None, default: bool) -> bool:
if value is None or (value := value.strip()) == "":
return default
lowered = value.lower()
if lowered in _TRUE_LITERALS:
return True
if lowered in _FALSE_LITERALS:
return False
raise ValueError(
f"{var_name} must be one of {_TRUE_LITERALS | _FALSE_LITERALS}, got {value!r}"
)
def _env_get_literal(
var_name: str,
default: _T,
*,
mapping: dict[str, object],
) -> _T:
value = os.environ.get(var_name)
if value is None:
return default
value = value.strip()
if value in mapping:
return cast("_T", mapping[value])
if value == "":
return default
raise ValueError(
f"{var_name} must be one of {', '.join(sorted(mapping))}, got {value!r}"
)
def _env_get_str_list(var_name: str) -> list[str]:
value = os.environ.get(var_name)
if value is None or value == "":
return []
return [item.strip() for item in value.split(",")]
def _env_get_str(var_name: str, default: str) -> str:
value = os.environ.get(var_name)
if value is None or (value := value.strip()) == "":
return default
return value
def _env_get_str_or_none(var_name: str, default: str | None) -> str | None:
value = os.environ.get(var_name)
if value is None or (value := value.strip()) == "":
return default
return value
def _get_index_dtype() -> torch.dtype | None:
value = os.environ.get("HELION_INDEX_DTYPE")
if value is None or (token := value.strip()) == "":
return None
if token.lower() == "auto":
return None
try:
dtype = getattr(torch, token)
except AttributeError as err:
raise ValueError(
f"HELION_INDEX_DTYPE must map to a torch dtype attribute, got {value!r}"
) from err
if not isinstance(dtype, torch.dtype):
raise ValueError(f"HELION_INDEX_DTYPE {value!r} is not a torch.dtype")
return dtype
def _get_autotune_log_level() -> int:
value = os.environ.get("HELION_AUTOTUNE_LOG_LEVEL")
if value is None or value.strip() == "":
return logging.INFO
text = value.strip()
if text.lstrip("+-").isdigit():
return int(text)
upper = text.upper()
# pyrefly: ignore [deprecated]
level = logging.getLevelName(upper)
if isinstance(level, int):
return level
raise ValueError(
f"HELION_AUTOTUNE_LOG_LEVEL must be an integer or logging level name, got {value!r}"
)
def _get_autotune_log_path() -> str | None:
value = os.environ.get("HELION_AUTOTUNE_LOG")
if value is None or (value := value.strip()) == "":
return None
return value
def _get_autotune_config_overrides() -> dict[str, object]:
value = os.environ.get("HELION_AUTOTUNE_CONFIG_OVERRIDES")
if not value or (value := value.strip()) == "":
return {}
if not value.startswith("{") and os.path.exists(value):
value = open(value).read()
try:
parsed = json.loads(value)
except json.JSONDecodeError as err:
raise ValueError(
"HELION_AUTOTUNE_CONFIG_OVERRIDES must be valid JSON mapping of config keys to values"
) from err
if not isinstance(parsed, dict):
raise ValueError(
"HELION_AUTOTUNE_CONFIG_OVERRIDES must decode to a JSON dictionary"
)
return parsed
def _get_initial_population_strategy(
default: str,
setting_override: InitialPopulation | None = None,
) -> InitialPopulationStrategy:
"""
Get the initial population strategy, respecting setting and env var overrides.
Args:
default: The default strategy string from the effort profile ("from_random" or "from_best_available").
setting_override: Optional override from kernel decorator settings.
Returns:
The InitialPopulationStrategy enum value, considering overrides.
Raises:
ValueError: If the environment variable is set to an invalid value.
"""
from ..autotuner import initial_population_strategies
# Priority: setting_override > env var > effort profile default
if setting_override is not None:
strategy = initial_population_strategies.get(setting_override)
if strategy is None:
raise ValueError(
f"Invalid autotune_initial_population_strategy value: {setting_override!r}. "
f"Valid values are: {', '.join(initial_population_strategies.keys())}"
)
return strategy
env_value = os.environ.get("HELION_AUTOTUNER_INITIAL_POPULATION", "").lower()
if env_value == "":
# No override, use the default from effort profile
strategy = initial_population_strategies.get(default)
assert strategy is not None
return strategy
strategy = initial_population_strategies.get(env_value)
if strategy is None:
raise ValueError(
f"Invalid HELION_AUTOTUNER_INITIAL_POPULATION value: {env_value!r}. "
f"Valid values are: {', '.join(initial_population_strategies.keys())}"
)
return strategy
def default_autotuner_fn(
bound_kernel: BoundKernel, args: Sequence[object], **kwargs: object
) -> BaseAutotuner:
from ..autotuner import LFBOTreeSearch
from ..autotuner import cache_classes
from ..autotuner import search_algorithms
autotuner_name = _env_get_str("HELION_AUTOTUNER", "")
if not autotuner_name:
autotuner_cls = LFBOTreeSearch
else:
autotuner_cls = search_algorithms.get(autotuner_name)
if autotuner_cls is None:
raise ValueError(
f"Unknown HELION_AUTOTUNER value: {autotuner_name}, valid options are: "
f"{', '.join(search_algorithms.keys())}"
)
profile = get_effort_profile(bound_kernel.settings.autotune_effort)
parameters = inspect.signature(autotuner_cls.__init__).parameters
for k, v in autotuner_cls.get_kwargs_from_profile(
profile, bound_kernel.settings
).items():
if k not in kwargs and k in parameters:
kwargs[k] = v
def autotuner_factory() -> BaseSearch:
"""Build a fresh inner BaseSearch instance.
Used by the best-of-K cache layer to construct K independent
autotuner instances, one per trial.
"""
# pyrefly: ignore [bad-argument-type]
return autotuner_cls(bound_kernel, args, **kwargs)
cache_name = bound_kernel.settings.autotune_cache
cache_cls = cache_classes.get(cache_name)
if cache_cls is None:
raise ValueError(
f"Unknown HELION_AUTOTUNE_CACHE value: {cache_name}, valid options are: "
f"{', '.join(cache_classes.keys())}"
)
# The cache layer's best-of-K loop calls ``autotuner_factory`` once per
# trial. The first construction here is the autotuner the cache wraps
# for the K=1 path; the factory is also passed through so K>1 trials can
# build additional fresh autotuners.
return cache_cls(autotuner_factory(), autotuner_factory=autotuner_factory)
def _get_autotune_random_seed() -> int:
if (seed := _env_get_optional_int("HELION_AUTOTUNE_RANDOM_SEED")) is not None:
return seed
return int(time.time() * 1000) % 2**32
def _get_ref_mode() -> RefMode:
interpret = _env_get_bool("HELION_INTERPRET", False)
triton_interpret = os.environ.get("TRITON_INTERPRET") == "1"
if interpret and triton_interpret:
raise exc.IncompatibleInterpretModes
return RefMode.EAGER if interpret else RefMode.OFF
def _get_dot_precision() -> DotPrecision:
"""
Gets the dot precision setting from the TRITON_F32_DEFAULT environment variable.
Defaults to 'tf32', 'ieee' if rocm and not CDNA.
For Pallas, gets the precision setting from the JAX_DEFAULT_MATMUL_PRECISION environment variable, defaulting to 'default'.
"""
if _get_backend() == "pallas":
return _env_get_literal(
"JAX_DEFAULT_MATMUL_PRECISION",
cast("DotPrecision", "default"),
mapping={
"default": "default",
"high": "high",
"highest": "highest",
"bfloat16": "default",
"tensorfloat32": "default",
"float32": "default",
},
)
if is_hip():
default_precision = "tf32" if supports_tf32_precision_on_amd() else "ieee"
else:
default_precision = "tf32"
return _env_get_literal(
"TRITON_F32_DEFAULT",
cast("DotPrecision", default_precision),
mapping={k: k for k in get_args(DotPrecision)},
)
def _get_backend() -> str:
return _env_get_literal(
"HELION_BACKEND",
"triton",
mapping={name: name for name in list_backends()},
)
def is_pallas_interpret() -> bool:
"""Return True if pallas_interpret is enabled.
Checks the active CompileEnvironment first (if one exists),
then falls back to the HELION_PALLAS_INTERPRET env var.
"""
from helion._compiler.compile_environment import CompileEnvironment
if CompileEnvironment.has_current():
return CompileEnvironment.current().settings.pallas_interpret
return _env_get_bool("HELION_PALLAS_INTERPRET", False)
@dataclasses.dataclass
class _Settings:
# see __slots__ below for the doc strings that show up in help(Settings)
backend: str = dataclasses.field(default_factory=_get_backend)
ignore_warnings: list[type[exc.BaseWarning]] = dataclasses.field(
default_factory=_get_ignore_warnings
)
index_dtype: torch.dtype | None = dataclasses.field(
default_factory=_get_index_dtype
)
dot_precision: DotPrecision = dataclasses.field(default_factory=_get_dot_precision)
cute_rng_stream: Literal["auto", "word0", "philox4"] = "word0"
fast_math: bool = dataclasses.field(
default_factory=functools.partial(_env_get_bool, "HELION_FAST_MATH", False)
)
cute_flash_env_overrides: dict[str, str] = dataclasses.field(default_factory=dict)
static_shapes: bool = dataclasses.field(
default_factory=functools.partial(_env_get_bool, "HELION_STATIC_SHAPES", True)
)
persistent_reserved_sms: int = dataclasses.field(
default_factory=functools.partial(
_env_get_int,
"HELION_PERSISTENT_RESERVED_SMS",
0,
)
)
autotune_force_persistent: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool,
"HELION_AUTOTUNE_FORCE_PERSISTENT",
False,
)
)
distributed: bool = dataclasses.field(
default_factory=functools.partial(_env_get_bool, "HELION_DISTRIBUTED", False)
)
autotune_log_level: int = dataclasses.field(default_factory=_get_autotune_log_level)
autotune_log: str | None = dataclasses.field(default_factory=_get_autotune_log_path)
autotune_log_details: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_LOG_DETAILS", False
)
)
autotune_handoff: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_HANDOFF", False
)
)
autotune_handoff_budget_seconds: int = dataclasses.field(
default_factory=functools.partial(
_env_get_int, "HELION_AUTOTUNE_HANDOFF_BUDGET_SECONDS", 1500
)
)
autotune_compile_timeout: int = dataclasses.field(
default_factory=functools.partial(
_env_get_int, "HELION_AUTOTUNE_COMPILE_TIMEOUT", 60
)
)
autotune_benchmark_subprocess: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_BENCHMARK_SUBPROCESS", True
)
)
autotune_benchmark_timeout: int = dataclasses.field(
default_factory=functools.partial(
_env_get_int,
"HELION_AUTOTUNE_BENCHMARK_TIMEOUT",
# Higher default when the first benchmark call has to absorb a
# heavier subprocess cold-start (slower spawn/imports, CUPTI init).
90 if is_fbcode() else 30,
)
)
autotune_worker_start_timeout: int | None = dataclasses.field(
default_factory=functools.partial(
_env_get_optional_int, "HELION_AUTOTUNE_WORKER_START_TIMEOUT"
)
)
autotune_precompile: PrecompileMode = dataclasses.field(
default_factory=functools.partial(
_env_get_literal,
"HELION_AUTOTUNE_PRECOMPILE",
cast("PrecompileMode", "fork"),
mapping={
"spawn": "spawn",
"fork": "fork",
"": None,
"0": None,
},
)
)
autotune_precompile_jobs: int | None = dataclasses.field(
default_factory=functools.partial(
_env_get_optional_int,
"HELION_AUTOTUNE_PRECOMPILE_JOBS",
)
)
autotune_random_seed: int = dataclasses.field(
default_factory=_get_autotune_random_seed
)
autotune_best_of_k: int = dataclasses.field(
default_factory=functools.partial(_env_get_int, "HELION_AUTOTUNE_BEST_OF_K", 1)
)
autotune_accuracy_check: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_ACCURACY_CHECK", True
)
)
autotune_rebenchmark_threshold: float | None = dataclasses.field(
default_factory=functools.partial(
_env_get_optional_float,
"HELION_REBENCHMARK_THRESHOLD",
)
)
autotune_suspicious_rebenchmark_ratio: float | None = dataclasses.field(
default_factory=functools.partial(
_env_get_optional_float,
"HELION_AUTOTUNE_SUSPICIOUS_REBENCHMARK_RATIO",
)
)
autotune_search_acf: list[str] = dataclasses.field(
default_factory=functools.partial(
_env_get_str_list, "HELION_AUTOTUNE_SEARCH_ACF"
)
)
autotune_progress_bar: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_PROGRESS_BAR", True
)
)
autotune_max_generations: int | None = dataclasses.field(
default_factory=functools.partial(
_env_get_optional_int,
"HELION_AUTOTUNE_MAX_GENERATIONS",
)
)
autotune_budget_seconds: int | None = dataclasses.field(
default_factory=functools.partial(
_env_get_optional_int,
"HELION_AUTOTUNE_BUDGET_SECONDS",
)
)
autotune_ignore_errors: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_IGNORE_ERRORS", False
)
)
autotune_adaptive_timeout: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_ADAPTIVE_TIMEOUT", True
)
)
print_output_code: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_PRINT_OUTPUT_CODE", False
)
)
print_repro: bool = dataclasses.field(
default_factory=functools.partial(_env_get_bool, "HELION_PRINT_REPRO", False)
)
output_origin_lines: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_OUTPUT_ORIGIN_LINES", True
)
)
force_autotune: bool = dataclasses.field(
default_factory=functools.partial(_env_get_bool, "HELION_FORCE_AUTOTUNE", False)
)
autotune_config_overrides: dict[str, object] = dataclasses.field(
default_factory=_get_autotune_config_overrides
)
autotune_seed_configs: ConfigLike | Sequence[ConfigLike] | None = None
disable_autotuner_heuristics: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_DISABLE_AUTOTUNER_HEURISTICS", False
)
)
autotune_effort: AutotuneEffort = dataclasses.field(
default_factory=functools.partial(
_env_get_literal,
"HELION_AUTOTUNE_EFFORT",
cast("AutotuneEffort", "full"),
mapping={key: key for key in ("none", "quick", "full")},
)
)
allow_warp_specialize: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_ALLOW_WARP_SPECIALIZE", True
)
)
debug_dtype_asserts: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_DEBUG_DTYPE_ASSERTS", False
)
)
ref_mode: RefMode = dataclasses.field(default_factory=_get_ref_mode)
autotune_cache: str = dataclasses.field(
default_factory=functools.partial(
_env_get_str, "HELION_AUTOTUNE_CACHE", "LocalAutotuneCache"
)
)
autotuner_fn: AutotunerFunction = default_autotuner_fn
autotune_baseline_fn: Callable[..., object] | None = None
autotune_baseline_atol: float | None = None
autotune_baseline_rtol: float | None = None
autotune_fp8_accuracy: AutotuneFp8Accuracy = dataclasses.field(
default_factory=functools.partial(
_env_get_literal,
"HELION_AUTOTUNE_FP8_ACCURACY",
cast("AutotuneFp8Accuracy", "ulp"),
mapping={"ulp": "ulp", "bitwise": "bitwise"},
)
)
autotune_baseline_accuracy_check_fn: Callable[[object, object], None] | None = None
autotune_benchmark_fn: Callable[..., list[float]] | None = None
autotune_best_available_max_configs: int = dataclasses.field(
default_factory=functools.partial(
_env_get_int, "HELION_BEST_AVAILABLE_MAX_CONFIGS", 20
)
)
autotune_best_available_max_cache_scan: int = dataclasses.field(
default_factory=functools.partial(
_env_get_int, "HELION_BEST_AVAILABLE_MAX_CACHE_SCAN", 500
)
)
autotune_initial_population_strategy: InitialPopulation | None = None
torch_compile_fusion: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_TORCH_COMPILE_FUSION", False
)
)
autotune_with_torch_compile_fusion: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_WITH_TORCH_COMPILE_FUSION", False
)
)
autotune_config_filter: Callable[[Config], Config | None] | None = None
pallas_collective_id: int | None = None
pallas_interpret: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_PALLAS_INTERPRET", False
)
)
pallas_topk_recall_target: float = 0.99
cute_region_fission: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_REGION_FISSION", False
)
)
cute_materialize_transformed_operands: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_MATERIALIZE_TRANSFORMED_OPERANDS", False
)
)
cute_full_slice_matmul_tiling: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_FULL_SLICE_MATMUL_TILING", False
)
)
cute_segmented_matmul_tiling: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_SEGMENTED_MATMUL_TILING", False
)
)
cute_flatten_nested_reductions: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_FLATTEN_NESTED_REDUCTIONS", False
)
)
cute_graph_cse: bool = dataclasses.field(
default_factory=functools.partial(_env_get_bool, "HELION_CUTE_GRAPH_CSE", True)
)
cute_stage_unaligned_operands: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_STAGE_UNALIGNED_OPERANDS", True
)
)
cute_flat_edge_tma: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_FLAT_EDGE_TMA", True
)
)
cute_exact_division: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_EXACT_DIVISION", True
)
)
cute_promote_half_atomic_outputs: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_CUTE_PROMOTE_HALF_ATOMIC_OUTPUTS", False
)
)
triton_do_not_specialize: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_TRITON_DO_NOT_SPECIALIZE", False
)
)
autotune_log_search_space: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_LOG_SEARCH_SPACE", False
)
)
autotune_log_search_space_verbose: bool = dataclasses.field(
default_factory=functools.partial(
_env_get_bool, "HELION_AUTOTUNE_LOG_SEARCH_SPACE_VERBOSE", False
)
)
autotune_log_search_space_path: str | None = dataclasses.field(
default_factory=functools.partial(
_env_get_str_or_none, "HELION_AUTOTUNE_LOG_SEARCH_SPACE_PATH", None
)
)
[docs]
class Settings(_Settings):
"""
Settings can be passed to hl.kernel as kwargs and control the behavior of the
compilation process. Unlike a Config, settings are not auto-tuned and set by the user.
"""
# Not a dataclass field: public serialization, equality and existing cache
# fingerprints continue to contain only the effective setting values.
_cute_structural_origins: CuteStructuralOrigins
__slots__ = {
"backend": (
"Code generation backend. One of 'triton' (default), 'pallas' (JAX/Pallas), "
"'cute' (CUTLASS CuTe DSL), or 'metal' (Apple Metal MSL). "
"Set HELION_BACKEND=<backend> to override."
),
"ignore_warnings": (
"Subtypes of exc.BaseWarning to ignore when compiling. "
"Set HELION_IGNORE_WARNINGS=WarningA,WarningB (names from helion.exc) to configure via env."
),
"index_dtype": (
"The dtype to use for index variables. Default auto-selects torch.int32 or torch.int64 based on input sizes. "
"Override with HELION_INDEX_DTYPE=<dtype> (or set to 'auto')."
),
"dot_precision": "Precision for dot products. For Triton backend, see `triton.language.dot` (can be 'tf32', 'tf32x3', 'ieee'). For JAX/Pallas backend, accepted values emit Pallas default precision on TPU. Unified mappings exist so that any value can be used on any backend.",
"cute_rng_stream": (
"CuTe RNG policy: auto (CuTe default) uses philox4 for explicit hl.rand "
"and preserves other RNG operations. philox4 maps logical offset i "
"to Philox(seed, i // 4)[i % 4], independently of configuration. "
"Set word0 for the previous reproducible stream; other backends "
"default to word0. Explicit philox4 rejects unsupported RNG operations. "
"Override with HELION_CUTE_RNG_STREAM."
),
"fast_math": (
"If True, enable fast math approximations (Helion-level, including the CuTe "
"flash-attention polynomial exp2 packets, and Inductor-level). "
"May reduce numerical precision and change NaN/Inf behavior. "
"Set HELION_FAST_MATH=1 to enable."
),
"cute_flash_env_overrides": (
"CuTe flash-attention knob channel: HELION_CUTE_FLASH_* variables the "
"kernel resolves as if set in the environment. The channel is captured "
"once when the kernel binds (the process environment overlaid with this "
"mapping); later environment changes do not reach the bound kernel."
),
"static_shapes": (
"If True, use static shapes for all tensors. This is a performance optimization. "
"Set HELION_STATIC_SHAPES=0 to disable."
),
"persistent_reserved_sms": (
"Number of streaming multiprocessors to reserve when launching persistent kernels. "
"Set HELION_PERSISTENT_RESERVED_SMS=N (default 0) or pass persistent_reserved_sms=N to helion.kernel."
),
"autotune_force_persistent": (
"If True, restrict pid_type choices to persistent kernels only during config selection. "
"Set HELION_AUTOTUNE_FORCE_PERSISTENT=1 to force persistent kernel autotuning globally."
),
"distributed": (
"Force distributed compilation behavior (persistent-only PID types, signal-pad SM "
"limits, and process-group resolution). Normally auto-detected from symmetric-memory "
"tensor arguments; set this for distributed kernels whose symmetric memory is not passed "
"as a detectable tensor. Set HELION_DISTRIBUTED=1 to force globally."
),
"autotune_log_level": (
"Log level for autotuning using Python logging levels. Default is logging.INFO. "
"Use HELION_AUTOTUNE_LOG_LEVEL to override or set 0 to disable output."
),
"autotune_log": (
"Base filename for autotune logs. Set HELION_AUTOTUNE_LOG=/tmp/run to write "
"/tmp/run.csv and /tmp/run.log with per-config metrics and debug logs."
),
"autotune_log_details": (
"Opt-in (HELION_AUTOTUNE_LOG_DETAILS=1) to write /tmp/run.trace.jsonl "
"with config evaluations, rebenchmarks, selection, exact LLM prompts/"
"responses, and timestamps, plus the cost-model dataset sidecar "
"/tmp/run.meta.jsonl (kernel identity and configs keyed by config_id). "
"Off by default; needs autotune_log."
),
"autotune_handoff": (
"Hand configuration search to a native-source agent automatically. "
"Off by default; enable with HELION_AUTOTUNE_HANDOFF=1. "
"Uses the existing search budget or search completion, "
"then saves validated standalone source beside the autotune logs."
),
"autotune_handoff_budget_seconds": (
"Wall-clock budget for the native-source agent session, excluding "
"search and export. Default 1500 seconds; override with "
"HELION_AUTOTUNE_HANDOFF_BUDGET_SECONDS."
),
"autotune_compile_timeout": (
"Timeout in seconds for one autotune precompile subprocess "
"(autotune_precompile); a CuTe render worker past it is given up "
"and its config rendered in-process instead (never failed for the "
"timeout alone). Also the benchmark worker's deadline for a "
"candidate's compile and first call (the backend JIT; never below "
"autotune_benchmark_timeout, armed once the worker is ready so a "
"fresh worker's start is not the candidate's; lowered along with the "
"setting by autotune_adaptive_timeout): a candidate past it is killed and "
"recorded as failed to compile, and the search goes on. Default "
"is 60 seconds. In-process compiles cannot be interrupted by it; "
"the autotune budget bounds them instead."
),
"autotune_benchmark_subprocess": "Run the autotune benchmark phase in a long-lived spawn subprocess so a hung/slow kernel can be killed without losing autotune progress. Enabled by default. Set HELION_AUTOTUNE_BENCHMARK_SUBPROCESS=0 to disable.",
"autotune_benchmark_timeout": (
"Per-config wall-clock timeout in seconds for the subprocess "
"benchmark phase, counted once the candidate's compile and first "
"call are done (those have autotune_compile_timeout). Only applies "
"when autotune_benchmark_subprocess is enabled. Default 30 seconds, "
"raised automatically on environments with a heavier subprocess "
"cold-start."
),
"autotune_worker_start_timeout": (
"Seconds a fresh benchmark worker may take to start (the spawned "
"interpreter, its imports and the device initialization) before "
"it is killed and the candidate recorded as a worker failure. The "
"start is never charged to the candidate's compile deadline "
"(autotune_compile_timeout), which is armed once the worker "
"reports ready. Default None: twice the worker's compile "
"deadline, at least 120 seconds. Set "
"HELION_AUTOTUNE_WORKER_START_TIMEOUT to override."
),
"autotune_precompile": (
"Autotuner precompile mode: 'fork', 'spawn', or falsy/None to "
"disable. Defaults to 'fork' on non-Windows platforms. With a pool "
"a batch of configs is generated, precompiled in parallel, then "
"timed; without one (disabled, or a backend such as CuTe that has "
"no pool) each config is compiled and timed in turn, so results "
"arrive as the batch runs and the autotune budget can stop it "
"between configs. CuTe generates each candidate's code in forked "
"workers ahead of its turn in 'fork' mode (its only mode; 'spawn' "
"disables the workers), so the serial part per config is the "
"module load, the DSL JIT at first call and the timing."
),
"autotune_precompile_jobs": (
"Maximum concurrent Triton precompile processes, default to cpu "
"count. Also the number of CuTe render workers alive at once, "
"which default to the CPUs this process may run on, divided by "
"the autotuning processes of the host (LOCAL_WORLD_SIZE or the "
"process group size), capped at 32. Does not affect results or "
"the autotune cache key."
),
"autotune_random_seed": "Seed used for autotuner random number generation. Defaults to HELION_AUTOTUNE_RANDOM_SEED or a time-based seed.",
"autotune_best_of_k": (
"When > 1, run autotuning K times with different random seeds and "
"keep the config with the best benchmark performance. Each trial "
"uses ``seed = autotune_random_seed + i`` for i in range(K), so two "
"runs with the same base seed produce the same K configs. Cost is "
"K× the autotune wall time. Cache entries are keyed on K, so a "
"K≥2 result will not be overwritten by a later K=1 run and "
"vice versa. Default 1 (current single-trial behavior). Set "
"HELION_AUTOTUNE_BEST_OF_K=N to override."
),
"autotune_accuracy_check": (
"If True, validate candidate configs against the baseline kernel output "
"before accepting them during autotuning. Unless autotune_baseline_fn is "
"set, the baseline is the output of the config the kernel runs by default "
"(a compiler-promoted default when there is one, else the conservative "
"autotuning reference, which is also the fallback when the promoted "
"default fails to compile or run)."
),
"autotune_rebenchmark_threshold": "If a config is within threshold*best_perf, re-benchmark it to avoid outliers. Defaults to effort profile value. Set HELION_REBENCHMARK_THRESHOLD to override.",
"autotune_suspicious_rebenchmark_ratio": "When subprocess benchmarking is enabled, recheck rebenchmark timings below ratio*previous_timing before accepting them. Defaults to 0.9. Set HELION_AUTOTUNE_SUSPICIOUS_REBENCHMARK_RATIO=0 to disable.",
"autotune_search_acf": "List of PTXAS Advanced Controls Files (ACFs) to search during autotuning. ACFs are highly specialized configurations for specific hardware and use cases; when autotuning with ACFs, default -O3 is always considered. Empty list disables.",
"autotune_progress_bar": "If True, show progress bar during autotuning. Default is True. Set HELION_AUTOTUNE_PROGRESS_BAR=0 to disable.",
"autotune_max_generations": "Override the maximum number of generations for Pattern Search and Differential Evolution Search autotuning algorithms with HELION_AUTOTUNE_MAX_GENERATIONS=N or @helion.kernel(autotune_max_generations=N).",
"autotune_budget_seconds": (
"Wall-clock budget in seconds for the entire autotune. When the "
"budget is exceeded the search returns the best config found so "
"far. The budget is checked between generations and before each "
"config's compile (with a precompile pool also before each "
"config's timing; without one a config is timed right after its "
"compile). Once it is out nothing further is compiled or timed, "
"except that while no config has been measured successfully the "
"search still attempts up to 3 more candidates, so a budget that "
"runs out early returns a config when one of those works; the "
"overrun is bounded by those attempts (and, with a pool, by the "
"precompile wait of the batch in flight). Without a precompile "
"pool the initial population may take at most half of the search "
"phase (the budget minus the time kept for the final selection), "
"counted from the start of each of its batches: once a few "
"configs have been timed and one has succeeded, the measured cost "
"per config decides how many more are attempted and the rest are "
"skipped, never below a CuTe flash surface's structural coverage "
"rows, so slow-compiling kernels keep time for the search "
"generations. A population that fits its share is evaluated in "
"full. Set with HELION_AUTOTUNE_BUDGET_SECONDS=N or "
"@helion.kernel(autotune_budget_seconds=N). Default None "
"(no budget)."
),
"autotune_ignore_errors": (
"If True, skip logging and raising autotune errors. "
"Set HELION_AUTOTUNE_IGNORE_ERRORS=1 to enable globally. The "
"one-line summary of the initial population (configs measured, "
"failed, skipped for the budget, and the wall time per config) and "
"the end-of-run failure counts are logged either way."
),
"autotune_adaptive_timeout": (
"If True, set the compile timeout threshold to be smaller for Triton compilation,"
"based on a quantile of initial compile times (with a lower bound). Lower bound and quantile "
"are set by the effort profile. Set HELION_AUTOTUNE_ADAPTIVE_TIMEOUT=0 to disable. "
"Applies only to a Triton-style precompile pool: CuTe render workers keep "
"the configured autotune_compile_timeout (a worker past it is rendered "
"in-process), and a timeout fitted on one shape's render times would be "
"shared by every shape of the kernel. The benchmark worker's compile "
"deadline follows the lowered setting."
),
"pallas_collective_id": (
"Optional Pallas collective namespace override. Distributed kernels "
"derive a stable ID automatically; set this only when otherwise "
"identical kernels use incompatible runtime communication groups."
),
"pallas_interpret": (
"If True, run Pallas kernels in interpret mode on CPU (no TPU needed). "
"Defaults to HELION_PALLAS_INTERPRET env var."
),
"pallas_topk_recall_target": (
"Recall target for the Pallas approximate top-k lowering. Must be in "
"(0, 1]; use 1.0 when exact top-k results are required. Default 0.99."
),
"cute_region_fission": (
"If True, distribute proven row-independent materialized regions "
"into ordered CuTe launches with independently tuned tile axes. "
"Defaults to HELION_CUTE_REGION_FISSION (False)."
),
"cute_materialize_transformed_operands": (
"If True, materialize proven pure reused matrix operands before "
"a CuTe GEMM, with independent producer and consumer tile axes. "
"Input dimensions used by the proof are specialized with normal "
"runtime guards. Defaults to "
"HELION_CUTE_MATERIALIZE_TRANSFORMED_OPERANDS (False)."
),
"cute_full_slice_matmul_tiling": (
"If True, give eligible direct full-slice half/BF16 matmuls an "
"explicit, tunable contraction tile loop on CuTe. Defaults to "
"HELION_CUTE_FULL_SLICE_MATMUL_TILING (False)."
),
"cute_segmented_matmul_tiling": (
"If True, expose proven offset-delimited independent matrix rows "
"to the CuTe grouped scheduler. Defaults to "
"HELION_CUTE_SEGMENTED_MATMUL_TILING (False)."
),
"cute_flatten_nested_reductions": (
"If True, flatten proven FP32 joint jagged/feature reductions and "
"independent fresh-output stores on CuTe. Defaults to "
"HELION_CUTE_FLATTEN_NESTED_REDUCTIONS (False)."
),
"cute_graph_cse": (
"If True (the default), merge duplicate reductions and the pure "
"chains between them in the traced device IR before CuTe code "
"generation (one amax/exp/sum per operand for a softmax and a "
"log_softmax of it). Results are bitwise unchanged; set False "
"(HELION_CUTE_GRAPH_CSE=0) to bisect a structural-matcher "
"regression against the traced graph."
),
"cute_stage_unaligned_operands": (
"If True (the default), the CuTe backend copies a static-shape "
"matmul operand whose outer strides are not 16-byte multiples "
"(e.g. odd-sized bf16 rows; a misaligned base offset is not "
"visible to the pass) into a padded row-major buffer on the host "
"before the launch, so the tcgen05 pipeline keeps its TMA "
"producers instead of the scalar shared-memory producers. Set False "
"(HELION_CUTE_STAGE_UNALIGNED_OPERANDS=0) to run such operands "
"in place."
),
"cute_flat_edge_tma": (
"If True (the default), flat one-CTA tcgen05 matmul kernels load "
"partial output tiles and the K tail through TMA (zero-filled past "
"the extents) and store them with predicated vector copies. Set "
"False (HELION_CUTE_FLAT_EDGE_TMA=0) for the scalar shared-memory "
"edge staging those tiles used before."
),
"cute_exact_division": (
"If True (the default), an fp32 division by a loop-invariant scalar "
"(a row sum, a scalar kernel argument) on CuTe hoists the divisor's "
"correctly rounded reciprocal and admission window out of the "
"element loop and divides with the Markstein sequence, re-running "
"the loop with div.rn when a dividend left the window: the IEEE "
"round-to-nearest quotient bit for bit, without one div.rn "
"expansion per element. Set False (HELION_CUTE_EXACT_DIVISION=0) "
"to keep the plain per-element division; the config knob "
"cute_exact_div_form picks the form (redo / plain) per kernel. "
"Independent of fast_math, whose fp32 division is approximate."
),
"cute_promote_half_atomic_outputs": (
"If True, a fresh float16 allocation that the CuTe kernel only "
"ever writes through hl.atomic_add is accumulated in a float32 "
"twin and cast to float16 once on return (one extra elementwise "
"kernel per call). False (the default, "
"HELION_CUTE_PROMOTE_HALF_ATOMIC_OUTPUTS=0) adds into the float16 "
"storage itself, rounding to float16 on every add as the program "
"and the Triton backend do; the vector collective epilogues use "
"native packed half atomics, the scalar half atomic is a "
"compare-and-swap loop on sm_100."
),
"triton_do_not_specialize": (
"If True, pass do_not_specialize for every dynamic size/stride/symbol "
"arg so a single Triton binary handles any value, avoiding Triton-level "
"recompiles when shapes cross 0/1 or alignment boundaries. Disables "
"Triton's value/alignment specializations and can significantly slow "
"memory-bound kernels (vectorized loads rely on inner stride being "
"specialized to constexpr 1). Only takes effect when static_shapes=False. "
"Set HELION_TRITON_DO_NOT_SPECIALIZE=1 to enable globally."
),
"print_output_code": "If True, print the output code of the kernel to stderr.",
"print_repro": "If True, print Helion kernel code, config, and caller code to stderr as a standalone repro script.",
"output_origin_lines": (
"If True, annotate generated Triton code with source-origin comments. "
"Set HELION_OUTPUT_ORIGIN_LINES=0 to disable."
),
"force_autotune": (
"If True, force autotuning even if a config is provided. "
"The result is still written to the cache so subsequent runs "
"can reuse it. Set HELION_SKIP_CACHE=1 instead to skip both "
"reading and writing the cache."
),
"autotune_config_overrides": (
"Dictionary of config key/value pairs forced during autotuning. "
"Accepts HELION_AUTOTUNE_CONFIG_OVERRIDES='{\"num_warps\":4}'."
),
"autotune_seed_configs": (
"A Config or sequence of Configs to seed the autotuner initial population "
"without constraining the search space."
),
"disable_autotuner_heuristics": (
"If True, disable compiler seed generation and promotion. Correctness "
"fact hooks still run so explicit configs normalize safely. User-provided "
"autotune_seed_configs are unaffected. "
"Set HELION_DISABLE_AUTOTUNER_HEURISTICS=1 to disable globally."
),
"allow_warp_specialize": "If True, allow warp specialization for tl.range calls on CUDA devices.",
"debug_dtype_asserts": "If True, emit tl.static_assert checks for dtype after each device node.",
"ref_mode": "Reference mode for kernel execution. Can be RefMode.OFF or RefMode.EAGER.",
"autotuner_fn": (
"Function to create an autotuner. "
"Override by passing a callable to @helion.kernel(..., autotuner_fn=...)."
),
"autotune_effort": "Autotuning effort preset. One of 'none', 'quick', 'full'.",
"autotune_baseline_fn": (
"Custom baseline function for computing baseline output during autotuning. "
"If provided, this function will be called instead of running the default config. "
"Should have the same signature as the kernel function. "
"Pass as @helion.kernel(..., autotune_baseline_fn=my_baseline_fn)."
),
"autotune_baseline_atol": (
"Absolute tolerance for baseline output comparison during autotuning accuracy checks. "
"When unset, defaults to 1e-2 scaled per output tensor by max(1, rms(baseline)) "
"so large-magnitude reductions tolerate accumulation-order noise, or 0.0 for fp8 "
"dtypes under autotune_fp8_accuracy='bitwise'. Setting an explicit value disables the "
"RMS scaling and is applied exactly; an explicit 0 together with "
"autotune_baseline_rtol=0 compares fp8 outputs bitwise. "
"Pass as @helion.kernel(..., autotune_baseline_atol=1e-3)."
),
"autotune_baseline_rtol": (
"Relative tolerance for baseline output comparison during autotuning accuracy checks. "
"Defaults to 1e-2, or 0.0 for fp8 dtypes under autotune_fp8_accuracy='bitwise'. "
"Pass as @helion.kernel(..., autotune_baseline_rtol=1e-3). "
"See autotune_baseline_atol for how the absolute-tolerance floor scales when unset."
),
"autotune_fp8_accuracy": (
"How autotuning accuracy checks compare fp8 output tensors against the baseline. "
"'ulp' (default) compares dequantized values: every element must lie within one fp8 "
"ulp of max(|actual|, |baseline|) plus the atol/rtol tolerances (atol scaled by the "
"baseline RMS as usual), and the one-ulp differences must not be systematically "
"one-directional. This admits the one-bucket flips a kernel that quantizes an "
"intermediate to fp8 (e.g. an online-softmax P) produces when its reduction blocking "
"changes, while still rejecting multi-ulp and biased differences. "
"'bitwise' requires identical bytes whenever every output is fp8 and no explicit "
"tolerance is set (the previous behavior). "
"Explicit zero tolerances (autotune_baseline_atol=0 with autotune_baseline_rtol=0) "
"ask for exact outputs and select 'bitwise' whatever this setting says. "
"Set HELION_AUTOTUNE_FP8_ACCURACY=bitwise to override."
),
"autotune_baseline_accuracy_check_fn": (
"Custom accuracy check function for comparing autotuning candidate outputs against the baseline. "
"Signature: (actual: object, expected: object) -> None. Should raise AssertionError on mismatch. "
"When set, replaces the default torch.testing.assert_close-based check (atol/rtol settings are ignored). "
"Useful for scenarios where a small fraction of elements may have large relative differences, "
"e.g. checking that mismatch percentage < X AND max relative diff < Y. "
"A built-in utility ``helion._testing.assert_close_with_mismatch_tolerance`` is provided "
"for this common pattern; use ``functools.partial(assert_close_with_mismatch_tolerance, ...)`` "
"to customize thresholds. "
"Pass as @helion.kernel(..., autotune_baseline_accuracy_check_fn=my_check_fn)."
),
"autotune_cache": (
"The name of the autotuner cache class to use. "
"Set HELION_AUTOTUNE_CACHE=StrictLocalAutotuneCache to enable strict caching. "
"Set HELION_AUTOTUNE_CACHE=RemoteAutotuneCache (or StrictRemoteAutotuneCache) "
"and HELION_REMOTE_CACHE_BACKEND=mypackage.module.MyBackend to enable "
"remote caching via a user-provided RemoteCacheBackend subclass. "
"Defaults to 'LocalAutotuneCache'."
),
"autotune_benchmark_fn": (
"Custom benchmark function for rebenchmarking during autotuning. "
"Should have the following signature: "
"(fns: list[Callable[[], object]], *, repeat: int, desc: str | None = None) -> list[float]. "
"If None (default), uses the built-in benchmark function."
),
"autotune_best_available_max_configs": (
"Maximum number of cached configs to use for FROM_BEST_AVAILABLE initial population "
"and for helion.from_cache() warm-start in FiniteSearch. "
"Set HELION_BEST_AVAILABLE_MAX_CONFIGS=N to override. Default is 20."
),
"autotune_best_available_max_cache_scan": (
"Maximum number of cache files to scan when searching for matching configs in FROM_BEST_AVAILABLE strategy. "
"Set HELION_BEST_AVAILABLE_MAX_CACHE_SCAN=N to override. Default is 500."
),
"autotune_initial_population_strategy": (
"Override the initial population strategy for autotuning. "
"Valid values: 'from_random', 'from_best_available'. "
"When set, takes precedence over the HELION_AUTOTUNER_INITIAL_POPULATION env var "
"and the effort profile default."
),
"torch_compile_fusion": (
"If True, allow torch.compile to fuse this Helion kernel with surrounding Inductor ops "
"(prologue/epilogue) when used inside torch.compile. Default False. "
"Set HELION_TORCH_COMPILE_FUSION=1 to enable globally."
),
"autotune_config_filter": (
"Optional callable ``(config: Config) -> Config | None`` that the autotuner calls on every "
"candidate config before compiling or benchmarking it. If the callable returns None, "
"the config is skipped entirely (no compilation, no benchmarking). If it returns a Config "
"(which may be a modified copy of the original), that config is used for benchmarking. "
"Also filters the explicit ``configs=[...]`` list when one is provided. "
"Pass as @helion.kernel(..., autotune_config_filter=my_filter_fn)."
),
"autotune_with_torch_compile_fusion": (
"If True, autotuning benchmarks the fused kernel (with epilogue/prologue) "
"to pick configs optimal for the actual fused workload. Default False. "
"Has no effect unless torch_compile_fusion is also True. "
"Set HELION_AUTOTUNE_WITH_TORCH_COMPILE_FUSION=1 to enable globally."
),
"autotune_log_search_space": (
"If True, log search space analysis after autotuning including which features "
"were enabled/disabled, total search space size, and coverage metrics. "
"Off by default; set HELION_AUTOTUNE_LOG_SEARCH_SPACE=1 to enable."
),
"autotune_log_search_space_verbose": (
"If True, additionally log each search-space restriction (disabled pid_types, "
"tcgen05 narrowing, etc.) live at INFO the moment it is applied during "
"compilation, on top of the end-of-run summary. Implies "
"autotune_log_search_space. Off by default; set "
"HELION_AUTOTUNE_LOG_SEARCH_SPACE_VERBOSE=1 to enable."
),
"autotune_log_search_space_path": (
"Optional path to save search space analysis JSON. "
"Set HELION_AUTOTUNE_LOG_SEARCH_SPACE_PATH=/path/to/analysis.json to save."
),
}
[docs]
def __init__(self, **settings: object) -> None:
"""
Initialize the Settings object with the provided dictionary of settings.
"""
if "backend" not in settings:
settings["backend"] = _get_backend()
if "cute_rng_stream" not in settings:
backend = get_backend(cast("str", settings["backend"]))
settings["cute_rng_stream"] = _env_get_literal(
"HELION_CUTE_RNG_STREAM",
backend.rng_streams()[0],
mapping={"auto": "auto", "word0": "word0", "philox4": "philox4"},
)
origins: dict[str, StructuralSettingOrigin] = {}
for name in CUTE_STRUCTURAL_SETTINGS:
if name in settings:
origins[name] = "explicit"
else:
var_name = f"HELION_{name.upper()}"
value = os.environ.get(var_name)
# Capture value and origin from the same environment read.
# The five defaults remain False; policy selection is separate.
settings[name] = _parse_env_bool(var_name, value, False)
origins[name] = (
"environment" if value is not None and value.strip() else "default"
)
# pyrefly: ignore [bad-argument-type]
super().__init__(**settings)
self._cute_structural_origins = CuteStructuralOrigins(**origins)
if self.backend == "tileir" and os.environ.get("ENABLE_TILE", "0") != "1":
raise exc.MissingEnableTile
if self.autotune_best_of_k < 1:
raise ValueError(
f"autotune_best_of_k must be >= 1, got {self.autotune_best_of_k!r}; "
f"set HELION_AUTOTUNE_BEST_OF_K to a positive integer "
f"(default 1 = single-trial behavior)"
)
if self.autotune_handoff_budget_seconds < 1:
raise ValueError("autotune_handoff_budget_seconds must be positive")
if (
self.autotune_worker_start_timeout is not None
and self.autotune_worker_start_timeout <= 0
):
raise ValueError(
"autotune_worker_start_timeout must be a positive number of "
f"seconds, got {self.autotune_worker_start_timeout!r}; a "
"non-positive timeout would kill every fresh benchmark worker "
"before it reports ready (leave it None for the default, or set "
"HELION_AUTOTUNE_WORKER_START_TIMEOUT to a positive integer)"
)
for name, value in self.cute_flash_env_overrides.items():
if not name.startswith("HELION_CUTE_FLASH_") or not isinstance(value, str):
raise ValueError(
"cute_flash_env_overrides maps HELION_CUTE_FLASH_* names to "
f"strings, got {name!r}={value!r}"
)
if self.autotune_fp8_accuracy not in ("ulp", "bitwise"):
raise ValueError(
"autotune_fp8_accuracy must be 'ulp' or 'bitwise', "
f"got {self.autotune_fp8_accuracy!r}"
)
self._check_ref_eager_mode_before_print_output_code()
def __setattr__(self, name: str, value: object) -> None:
super().__setattr__(name, value)
if (
name in CUTE_STRUCTURAL_SETTINGS
and "_cute_structural_origins" in self.__dict__
):
# Assigning the same value is still an explicit request. Initial
# dataclass writes precede creation of the construction origins.
self._cute_structural_origins = self._cute_structural_origins.with_explicit(
name
)
def __setstate__(
self, state: tuple[dict[str, object] | None, dict[str, object]]
) -> None:
# copy.copy/deepcopy and pickle restore slots after the instance dict.
# Those restoration writes must not become explicit user overrides.
namespace, slots = state
if namespace is not None:
self.__dict__.update(namespace)
for name, value in slots.items():
object.__setattr__(self, name, value)
if "_cute_structural_origins" not in self.__dict__:
# Older serialized Settings contain only effective field values,
# just like reconstruction from to_dict(). Their flags are explicit.
self._cute_structural_origins = CuteStructuralOrigins().with_explicit(
*CUTE_STRUCTURAL_SETTINGS
)
[docs]
def copy(self, **overrides: object) -> Settings:
"""Copy values and origins, marking supplied structural overrides explicit.
Like dataclasses.replace, ordinary mutable field values are shared.
In contrast, dataclasses.replace and Settings(**settings.to_dict())
supply every effective field to the constructor and therefore record
every structural value as explicit, even when its value is unchanged.
"""
result = dataclasses.replace(self, **overrides)
result._cute_structural_origins = self._cute_structural_origins.with_explicit(
*(name for name in CUTE_STRUCTURAL_SETTINGS if name in overrides)
)
return result
@property
def cute_structural_origins(self) -> CuteStructuralOrigins:
"""Immutable origins captured at construction and explicit writes."""
return self._cute_structural_origins
[docs]
def get_cute_structural_policy(self) -> CuteStructuralPolicy:
"""Snapshot effective booleans without resolving or activating a policy."""
return CuteStructuralPolicy.from_settings(self)
[docs]
def to_dict(self) -> dict[str, object]:
"""
Convert the Settings object to a dictionary.
Returns:
dict[str, object]: A dictionary representation of the Settings object.
"""
def shallow_copy(x: object) -> object:
if isinstance(x, (list, dict)):
return x.copy()
return x
# Only include fields that are meant to be public (repr=True)
public_fields = {f.name for f in dataclasses.fields(self) if f.repr}
return {
k: shallow_copy(v)
for k, v in dataclasses.asdict(self).items()
if k in public_fields
}
[docs]
def check_autotuning_disabled(self) -> None:
msg = None
if os.environ.get("HELION_DISALLOW_AUTOTUNING", "0") == "1":
msg = "by HELION_DISALLOW_AUTOTUNING=1"
if is_fbcode():
from aiplatform.runtime_environment.runtime_environment_pybind import ( # type: ignore[import-untyped]
RuntimeEnvironment,
)
if RuntimeEnvironment().get_mast_job_name() is not None:
msg = "because autotuning is not allowed in MAST environment"
if msg:
raise exc.AutotuningDisallowedInEnvironment(msg)
[docs]
def get_rebenchmark_threshold(self) -> float:
"""
Get the effective rebenchmark threshold.
Uses the explicit setting if provided, otherwise falls back to the effort profile default.
Returns:
float: The rebenchmark threshold value.
"""
if self.autotune_rebenchmark_threshold is not None:
return self.autotune_rebenchmark_threshold
return get_effort_profile(self.autotune_effort).rebenchmark_threshold
[docs]
def get_suspicious_rebenchmark_ratio(self) -> float | None:
if self.autotune_suspicious_rebenchmark_ratio is not None:
return self.autotune_suspicious_rebenchmark_ratio
if self.autotune_benchmark_subprocess:
return 0.9
return None
def _check_ref_eager_mode_before_print_output_code(self) -> None:
"""
Check if ref eager mode is enabled before printing output code. If ref eager mode is enabled, raise an error.
"""
if self.ref_mode == RefMode.EAGER and self.print_output_code:
raise exc.RefEagerModeCodePrintError
