Public API#

The handful of things you use when writing a task package. Everything here is what a task author touches directly; the rest of the package is internal machinery, browsable under Full module index.

See Writing tests for how these fit together.

Task base class#

Every task defines a class deriving from TestBase. It provides the build hooks and the compilation helper; you add the tests.

class kernelfoundry.TestBase[source]

Bases: ABC

Base class from which kernel task tests must derive.

Example

The following shows how to define a test by deriving from TestBase:

# This is a partial example; see templates for a complete task.
from pathlib import Path
import torch
import pytest
from kernelfoundry import TestBase

# ... pytest fixtures for device/kernel/data are omitted for brevity.

class TestRelu(TestBase):
    def build(self, gpu_arch) -> list[str]:
        return self.compile_torch_extension(
            extension_name="relu_kernel",
            src="relu_kernel.sycl",
            output_dir=Path(__file__).parent,
            gpu_arch=gpu_arch,
        )

    def test_correctness(self, data, kernel, device):
        x, y = data
        assert torch.allclose(kernel(x), y, rtol=1e-4, atol=1e-4)

    @pytest.mark.performance
    def test_benchmark(self, data, kernel, device, measure_runtime_torch):
        # measure_runtime_torch fixture is provided by kernelfoundry/conftest.py
        x, _ = data
        measure_runtime_torch(kernel, device, args=(x,))

build(gpu_arch, **buildtime_params) → list[str][source]

Builds the kernel and returns a list of build artifacts required for running the tests.

Returns:

List of paths to the build artifacts. The artifacts must not: be outside of the task folder structure.

Return type:

list[str]

build_reference(gpu_arch, **buildtime_params) → list[str][source]

Builds the reference code and returns a list of build artifacts required for running the tests.

Returns:

list of paths to the build artifacts. The artifacts must not: be outside of the task folder structure.

Return type:

list[str]

static compile_torch_extension(extension_name: str, src: str | Path, output_dir: str | Path, gpu_arch: str, timeout: int = 120, backend: str = 'torch') → list[str][source]

Compiles the source file to a PyTorch extension.

Parameters:

extension_name (str) – Name of the PyTorch extension to build.
src (str) – Path to the source file.
output_dir (str) – Directory to store the compiled outputs.
gpu_arch (str) – GPU architecture string.
timeout (int) – Timeout for each compilation step in seconds.
backend (str) – The backend to use for compilation. This is either ‘torch’ (default) or ‘icpx’.

Returns:

List of paths to the compiled extensions.

Return type:

list[str]

static get_machine_gpu_arch() → str[source]

Returns the GPU architecture string of the local machine.

Returns:: GPU architecture string.
Return type:: str

static validate()[source]

Hook for task-specific validation of the task definition.

Override in a subclass to assert anything that must hold before a job starts – for example that required data files are present, or that a build script is executable. The default implementation does nothing.

Assertions#

Comparison helpers with tolerances appropriate to GPU floating point. assert_allclose is the one to reach for by default.

kernelfoundry.testing.assert_allclose(actual, expected, *, epsilon: float = 1e-07, rtol: float = 0.01, ratio_below_max_err: float = 0.99, msg: str | Callable[[str], str] | None = None, err_stats: bool = True) → None[source]

Asserts that two arrays are close within a given relative tolerance.

This function computes the absolute relative error between the new and original outputs, and determines if the proportion of elements within a specified maximum relative error is above a given ratio.

This function behaves like all_close_with_slack, but raises an AssertionError with a detailed message

Parameters:

actual (np.ndarray|torch.Tensor) – The output tensor to validate.
expected (np.ndarray|torch.Tensor) – The reference output tensor.
epsilon (float, optional) – A small constant to avoid division by zero. Default is 1e-7.
rtol (float, optional) – The maximum relative error allowed. Default is 0.01.
ratio_below_max_err (float, optional) – The minimum required ratio of elements with error below the maximum relative error. Default is 0.99.
msg (str | Callable[[str], str] | None, optional) – Optional custom error message.
err_stats (bool, optional) – Whether to include error statistics. Default is True.

Raises:

AssertionError – If the arrays are not close enough.

kernelfoundry.testing.all_close_with_slack(output_reference: torch.Tensor, output_kernel: torch.Tensor, epsilon: float = 1e-07, max_rel_err: float = 0.01, ratio_below_max_err: float = 0.99) → bool[source]

Check the accuracy of the kernel output compared to the reference output.

This function computes the absolute relative error between the new and original outputs, and determines if the proportion of elements within a specified maximum relative error is above a given ratio.

Parameters:

output_reference (torch.Tensor) – The reference output tensor.
output_kernel (torch.Tensor) – The kernel output tensor to compare.
epsilon (float, optional) – A small constant to avoid division by zero. Default is 1e-7.
max_rel_err (float, optional) – The maximum relative error allowed. Default is 0.01.
ratio_below_max_err (float, optional) – The minimum required ratio of elements with error below the maximum relative error. Default is 0.99.

Returns:

True if the ratio of elements with a relative error below max_rel_err: is greater than ratio_below_max_err, False otherwise.

Return type:

bool

kernelfoundry.testing.cosine_similarity(output_reference: torch.Tensor, output_kernel: torch.Tensor, min_sim: float = 0.99985)[source]

Compute cosine similarity of flattened output tensors.

Returns:: True if similarity meets the threshold, False otherwise.
Return type:: bool

Benchmarking#

measure_runtime_torch is the simplest correct benchmark for torch workloads; measure_runtime is the general form. If you write a custom benchmark, the timed trials must run inside profiler_session or no profiler data is collected.

kernelfoundry.eval_pipeline.utils.performance.measure_runtime_torch(target: Callable, device: str | torch.device, args: tuple | list | None = None, kwargs: dict | None = None, warmup_min_time: float = 1.0, warmup_min_iters: int = 10, inner_loop_min_time: float = 0.01, perf_trials_min_iters: int = 10, perf_trials_min_time: float = 1.0, use_itt: bool = False, reduce_iterations_for_external_profiler: bool = True, auto_replicate_inputs_size: int = 134217728, output: list[float] | None = None, profiler_label: str | None = None) → list[float][source]

Measures the runtime of the target callable on the specified torch device.

Parameters:

target (Callable) – The kernel function to be measured.
device (Union[str, torch.device]) – The device to use for synchronization.
args – Positional arguments to pass to the target function. This can be a list of positional arguments to iterate over different inputs for each call to the target. Note that you must provide a list of kwargs of the same length if you provide a list of args. Use kwargs=len(args)*[{}] if there are no kwargs to pass. Note that arguments for each list entry should have the same shape and structure.
kwargs – Keyword arguments to pass to the target function. If this is a list of keyword argument dictionaries, then this function will iterate through the list to use a different set of kwargs for each call to the target with the intent to avoid caching effects. Note that args must be a list of tuples/lists of the same length as kwargs in this case. Note that arguments for each list entry should have the same shape and structure.
warmup_min_time (float) – Minimum total time for warmup phase in seconds.
warmup_min_iters (int) – Minimum number of iterations for warmup phase.
inner_loop_min_time (float) – Minimum time for inner loop trials in seconds.
perf_trials_min_iters (int) – Minimum number of performance trials.
perf_trials_min_time (float) – Minimum total time for performance trials in seconds.
use_itt (bool) – Whether to use ITT annotations during profiling.
reduce_iterations_for_external_profiler (bool) – If an external profiler is detected, reduce the number of iterations to avoid long profiling sessions.
auto_replicate_inputs_size (int) – Replicate the inputs, args and kwargs, to this size to avoid caching effects for very small inputs. Set to 0 to disable replication. This option has no effect if args and kwargs are lists of arguments.
output (list[float]) – Optional list to store the measured runtimes.
profiler_label (str | None) – Optional label embedded into ITT model run loop markers.

Returns:

List of measured runtimes in milliseconds.

Return type:

list[float]

kernelfoundry.eval_pipeline.utils.performance.measure_runtime(target: Callable, sync_fn: Callable, args: list[tuple] | list[list] | tuple | list | None = None, kwargs: list[dict] | dict | None = None, warmup_min_time: float = 1.0, warmup_min_iters: int = 10, inner_loop_min_time: float = 0.01, perf_trials_min_iters: int = 10, perf_trials_min_time: float = 1.0, use_itt: bool = False, reduce_iterations_for_external_profiler: bool = True, auto_replicate_inputs_size: int = 134217728, info_str: str = '', output: list[float] | None = None, profiler_label: str | None = None) → list[float][source]

Measures the runtime of the target callable.

The function assumes that all invocations of the target are run in order on the same device, and that the sync_fn function will synchronize the device to ensure all operations are complete.

Parameters:

target (Callable) – The kernel function to be measured.
sync_fn (Callable) – The synchronization function to be called after target execution.
args – Positional arguments to pass to the target function. This can be a list of positional arguments to iterate over different inputs for each call to the target. Note that you must provide a list of kwargs of the same length if you provide a list of args. Use kwargs=len(args)*[{}] if there are no kwargs to pass. Note that arguments for each list entry should have the same shape and structure.
kwargs – Keyword arguments to pass to the target function. If this is a list of keyword argument dictionaries, then this function will iterate through the list to use a different set of kwargs for each call to the target with the intent to avoid caching effects. Note that args must be a list of tuples/lists of the same length as kwargs in this case. Note that arguments for each list entry should have the same shape and structure.
warmup_min_time (float) – Minimum total time for warmup phase in seconds.
warmup_min_iters (int) – Minimum number of iterations for warmup phase.
inner_loop_min_time (float) – Minimum time for inner loop trials in seconds.
perf_trials_min_iters (int) – Minimum number of performance trials.
perf_trials_min_time (float) – Minimum total time for performance trials in seconds.
use_itt (bool) – Whether to use ITT annotations during profiling.
reduce_iterations_for_external_profiler (bool) – If an external profiler is detected, reduce the number of iterations to avoid long profiling sessions.
auto_replicate_inputs_size (int) – Replicate the inputs, args and kwargs, to this size to avoid caching effects for very small inputs. Set to 0 to disable replication. This option has no effect if args and kwargs are lists of arguments.
info_str (str) – Additional info string added before the timing info about warmup and test iterations. Useful for adding information about the device.
output (list[float]) – Optional list to store the measured runtimes.
profiler_label (str | None) – Optional label embedded into ITT model run loop markers. If omitted, the current pytest node id is read from the environment when available.

Returns:

List of measured runtimes in milliseconds.

kernelfoundry.eval_pipeline.utils.performance.profiler_session(profiler_name: str, stop_on_exit: bool = False)[source]

Context manager to resume, pause and stop profiler sessions.

Supports unitrace and vtune.

Parameters:

profiler_name (str) – The name of the profiler to manage.
stop_on_exit (bool) – Whether to stop the profiler session on exit.
False (If)
stopped. (the session will be paused instead of)

kernelfoundry.eval_pipeline.utils.performance.detect_profiler() → str | None[source]

Detect whether the process is running under a profiler.

Returns:: The detected profiler name, or None if not detected.
Return type:: str | None

Pytest fixtures#

Available in any task that keeps the shipped conftest.py. Request them as test arguments.

Fixture	Purpose
`measure_runtime_torch`	Benchmark a torch callable: moves inputs to the device, runs warmup, records runtimes.
`measure_runtime`	Benchmark a general callable, for non-torch workloads.
`torch_profile`	Thin wrapper over `measure_runtime_torch` for profiling runs.
`profile_store`	Where measured runtimes are recorded. Required if you write a custom benchmark.
`use_reference`	True when running under `pytest --ref`, so the reference is exercised instead of the kernel.
`template_args_wrapper`	Supports templated kernels, where each parameter combination is benchmarked.
`profiler_test_label`	Labels the profiler region for the current test.

Compilers#

Chosen through eval_config.kernel_compiler. TorchCompiler is the default and builds a PyTorch extension; IcpxCompiler invokes icpx directly.

class kernelfoundry.compiler.TorchCompiler(extension_name: str, src: str, build_dir: str, gpu_arch: str, timeout: int = 120, verbose: bool = False)[source]

Bases: BaseKernelCompiler

Compiler class using the torch cpp_extension compiler.

LOAD_FAILED_RETURNCODE = 87: Exit code the build subprocess uses when the sources compiled but the module would not import. Distinct from 1 so the caller can tell the two failures apart; see compile().

classmethod failed_phase(returncode: int) → str | None[source]: Which phase failed: "compile", "load", or None when nothing did.

compile()[source]

Compile the source via torch.utils.cpp_extension.

Returns:: Build artifacts and captured compiler output.
Return type:: dict

class kernelfoundry.compiler.IcpxCompiler(extension_name: str, src: str, build_dir: str, gpu_arch: str, timeout: int = 120, verbose: bool = False)[source]

Bases: BaseKernelCompiler

Compiler class for SYCL programs using the Intel icpx compiler.

compile()[source]: Compile the SYCL program into a PyTorch extension.

MCP server#

The MCP server exposes one tool, build_and_test(folder_path), which builds and benchmarks a task package and returns the outcome as structured data a coding agent can act on. Setup and the full return contract are in the MCP server README.