task#

Defines the Task class and related classes for managing kernel generation tasks, including build and test results.

Classes

`BuildResult`([worker_info, result, artifacts])	The result of a build process.
`ProcessResult`(returncode, stdout, stderr[, ...])	This collects the output of a subprocess execution with a custom message and output data.
`Task`(task_data, has_build_step, ...)	A Task represents a custom kernel generation problem, including the input code, expected output code, and associated tests.
`TestResult`(worker_info, correctness_result, ...)	The result of executing the correctness and performance tests.

class kernelfoundry.eval_pipeline.task.ProcessResult(returncode: int, stdout: str, stderr: str, message: str | None = None, output_data: dict | None = None)[source]#

This collects the output of a subprocess execution with a custom message and output data.

returncode: int#

stdout: str#

stderr: str#

message: str | None = None#

output_data: dict | None = None#

combine_output(include_message: bool = True) → str[source]#: Combines stdout and stderr into a single string.

classmethod create(subprocess_result, message: str | None = None, output_data: str | None = None) → ProcessResult[source]#: Convenience method to create a ProcessResult from a subprocess.CompletedProcess instance.

classmethod merge(results: dict[str, ProcessResult]) → ProcessResult[source]#: Merges multiple ProcessResult instances into a single ProcessResult by concatenating their outputs.

__init__(returncode: int, stdout: str, stderr: str, message: str | None = None, output_data: dict | None = None) → None#

class kernelfoundry.eval_pipeline.task.BuildResult(worker_info: dict | None = None, result: ProcessResult | None = None, artifacts: MemoryFileMap | None = None)[source]#

The result of a build process.

worker_info: dict | None = None#

result: ProcessResult | None = None#

artifacts: MemoryFileMap | None = None#

classmethod decode(data: dict) → BuildResult[source]#: Decodes a BuildResult from a dictionary.

__init__(worker_info: dict | None = None, result: ProcessResult | None = None, artifacts: MemoryFileMap | None = None) → None#

class kernelfoundry.eval_pipeline.task.TestResult(worker_info: dict | None = None, correctness_result: ~kernelfoundry.eval_pipeline.task.ProcessResult | None = None, performance_result: ~kernelfoundry.eval_pipeline.task.ProcessResult | None = None, trace_results: dict[str, ~kernelfoundry.eval_pipeline.task.ProcessResult] = <factory>, results_per_gpu: dict[str, TestResult] | None = None)[source]#

The result of executing the correctness and performance tests.

worker_info: dict | None = None#

correctness_result: ProcessResult | None = None#

performance_result: ProcessResult | None = None#

trace_results: dict[str, ProcessResult]#

results_per_gpu: dict[str, TestResult] | None = None#

classmethod decode(data: dict) → TestResult[source]#: Decodes a TestResult from a dictionary.

classmethod merge(results: dict[str, TestResult]) → TestResult[source]#: Merges multiple TestResult instances into a single TestResult.

__init__(worker_info: dict | None = None, correctness_result: ~kernelfoundry.eval_pipeline.task.ProcessResult | None = None, performance_result: ~kernelfoundry.eval_pipeline.task.ProcessResult | None = None, trace_results: dict[str, ~kernelfoundry.eval_pipeline.task.ProcessResult] = <factory>, results_per_gpu: dict[str, TestResult] | None = None) → None#

class kernelfoundry.eval_pipeline.task.Task(task_data: ~kernelfoundry.eval_pipeline.utils.memory_file_map.MemoryFileMap, has_build_step: bool, has_reference_build_step: bool, correctness_tests: list[str], profile_tests: list[str], config: dict, blocks: dict[str, dict[str, list[str]]] = <factory>, extract_code_result: ~kernelfoundry.eval_pipeline.task.ProcessResult | None = None, build_result: ~kernelfoundry.eval_pipeline.task.BuildResult | None = None, build_result_reference: ~kernelfoundry.eval_pipeline.task.BuildResult | None = None, test_result: ~kernelfoundry.eval_pipeline.task.TestResult | None = None, test_result_reference: ~kernelfoundry.eval_pipeline.task.TestResult | None = None, hyperparameters: dict | None = None)[source]#

A Task represents a custom kernel generation problem, including the input code, expected output code, and associated tests.

task_data: MemoryFileMap#

has_build_step: bool#

has_reference_build_step: bool#

correctness_tests: list[str]#

profile_tests: list[str]#

config: dict#

blocks: dict[str, dict[str, list[str]]]#

extract_code_result: ProcessResult | None = None#

build_result: BuildResult | None = None#

build_result_reference: BuildResult | None = None#

test_result: TestResult | None = None#

test_result_reference: TestResult | None = None#

hyperparameters: dict | None = None#

print_info(print_fn: Callable | None = None) → None[source]#: Prints information about the Task.

apply_config(config: DictConfig) → Task[source]#: Attaches an execution config to the Task and transfers config-driven attributes.

validate()[source]#: Checks if the Task is valid.

classmethod create(path_or_bytes: str | Path | bytes) → tuple[Task, dict][source]#

Creates a Task from a path to a directory, a tar file, zip file or from bytes of a zip/tar file.

Parameters:: path_or_bytes (str | Path | bytes) – Path to the task directory, tar file, zip file or bytes of a tar/zip file.
Returns:: The created Task instance. dict: Dictionary with additional information such as overrides.
Return type:: Task

with_blocks(blocks: dict[str, dict[str, list[str] | None]] | dict[str, str | None], keep_test_result_reference: bool = False) → Task[source]#

Returns a new Task with the given blocks updated.

Note that the newly created Task will not contain build or test results as those are not valid anymore after changing the blocks! If you want to keep the reference test result, set keep_test_result_reference=True.

Parameters:

blocks (dict[str, dict[str, list[str] | None]] | dict[str, str | None]) – The blocks to update. If the task has only a single block for a key, a simple dict can be provided mapping key->content. If multiple blocks exist for a key, a dict mapping key->file_path->content must be provided. Use a value of None to indicate that the block could not be generated, e.g., {‘EVOLVE’: None} or {‘EVOLVE’: {‘src/kernel.sycl’: None}}
keep_test_result_reference (bool) – If True, the reference test result will be kept in the new task.

Returns:

A new Task with the updated blocks.

Return type:

Task

encode() → dict[source]#: Encodes the Task to a serializable dictionary.

classmethod decode(data: dict) → Task[source]#: Decodes a Task from a serialized dictionary.

to_database_task()[source]#: Convert this Task object into a db.Task object that can be written to the detabase

__init__(task_data: ~kernelfoundry.eval_pipeline.utils.memory_file_map.MemoryFileMap, has_build_step: bool, has_reference_build_step: bool, correctness_tests: list[str], profile_tests: list[str], config: dict, blocks: dict[str, dict[str, list[str]]] = <factory>, extract_code_result: ~kernelfoundry.eval_pipeline.task.ProcessResult | None = None, build_result: ~kernelfoundry.eval_pipeline.task.BuildResult | None = None, build_result_reference: ~kernelfoundry.eval_pipeline.task.BuildResult | None = None, test_result: ~kernelfoundry.eval_pipeline.task.TestResult | None = None, test_result_reference: ~kernelfoundry.eval_pipeline.task.TestResult | None = None, hyperparameters: dict | None = None) → None#