diff --git a/src/gt4py/next/backend.py b/src/gt4py/next/backend.py index fb649bf28f..bab9691520 100644 --- a/src/gt4py/next/backend.py +++ b/src/gt4py/next/backend.py @@ -24,7 +24,7 @@ ) from gt4py.next.ffront.past_passes import linters as past_linters from gt4py.next.iterator import ir as itir -from gt4py.next.otf import arguments, artifacts, stages, toolchain, workflow +from gt4py.next.otf import arguments, artifacts, recipes, stages, toolchain, workflow def jit_to_aot_args( @@ -171,6 +171,46 @@ def compile( ) return self.loading(artifact) + def translate( + self, definition: stages.IRDefinitionT, compile_time_args: arguments.CompileTimeArgs + ) -> artifacts.ProgramSource: + """ + Run the frontend and the translation step only. + + This is the sanctioned partial run of the toolchain for stage + inspection: the program definition goes through the full `frontend` + pipeline and the `translation` step of the compile pipeline, stopping + before bindings generation and compilation. Per-call step options are + deliberately not offered; a caller needing a variant translation step + builds a variant pipeline with `dataclasses.replace`. + + Args: + definition: A program definition in any stage the frontend accepts. + compile_time_args: Compile-time arguments for the frontend + transforms and the translation step. + + Returns: + The `ProgramSource` produced by the translation step. + + Raises: + NotImplementedError: If this toolchain's `backend` is not the + standard `OTFCompileWorkflow` pipeline shape. + """ + if not isinstance(self.backend, recipes.OTFCompileWorkflow): + raise NotImplementedError( + f"Toolchain '{self.name}' does not support partial runs: 'translate'" + " requires the standard 'OTFCompileWorkflow' compile pipeline" + " ('translation' / 'bindings' / 'compilation' steps), but this" + f" toolchain's backend is a '{type(self.backend).__name__}'." + " Monolithic backends execute in a single step and produce no" + " intermediate 'ProgramSource'." + ) + source = self.backend.translation( + self.frontend(stages.ConcreteProgramDef(definition=definition, args=compile_time_args)) + ) + workflow.stage_hook("translation", source) + return source + @property def __gt_allocator__( self, diff --git a/src/gt4py/next/config.py b/src/gt4py/next/config.py index e62e2a1892..8650491b59 100644 --- a/src/gt4py/next/config.py +++ b/src/gt4py/next/config.py @@ -207,6 +207,14 @@ def _init_dump_metrics_filename() -> str: DUMP_METRICS_AT_EXIT = _dump_metrics_at_exit_env +_dump_stages_env = os.environ.get("GT4PY_DUMP_STAGES", None) + +#: Directory where the toolchain dumps the intermediate artifact produced by +#: every pipeline stage, one subdirectory per program definition. Opt-in +#: debugging feature: disabled when unset (the default). +DUMP_STAGES: pathlib.Path | None = pathlib.Path(_dump_stages_env) if _dump_stages_env else None + + #: The default for whether to allow jit-compilation for a compiled program. #: This default can be overriden per program. ENABLE_JIT_DEFAULT: bool = env_flag_to_bool("GT4PY_ENABLE_JIT_DEFAULT", default=True) diff --git a/src/gt4py/next/instrumentation/hooks.py b/src/gt4py/next/instrumentation/hooks.py index 97d7f3a080..3d5a7563bd 100644 --- a/src/gt4py/next/instrumentation/hooks.py +++ b/src/gt4py/next/instrumentation/hooks.py @@ -16,3 +16,4 @@ compile_variant_hook as compile_variant_hook, compiled_program_call_context as compiled_program_call_context, ) +from gt4py.next.otf.workflow import stage_hook as stage_hook diff --git a/src/gt4py/next/instrumentation/stage_dump.py b/src/gt4py/next/instrumentation/stage_dump.py new file mode 100644 index 0000000000..26e51b242e --- /dev/null +++ b/src/gt4py/next/instrumentation/stage_dump.py @@ -0,0 +1,253 @@ +# GT4Py - GridTools Framework +# +# Copyright (c) 2014-2024, ETH Zurich +# All rights reserved. +# +# Please, refer to the LICENSE file in the root directory. +# SPDX-License-Identifier: BSD-3-Clause + +"""Subscriber that dumps toolchain stage artifacts to disk (`GT4PY_DUMP_STAGES`). + +Enabled automatically at import time of `gt4py.next.otf.workflow` when +`gt4py.next.config.DUMP_STAGES` is set; `enable` / `disable` are provided for +programmatic and test use (setting `config.DUMP_STAGES` after import requires +calling `enable`). Artifacts are grouped in one subdirectory per program; every +file gets an increasing index prefix, seeded from the files already in the +directory, so that the listing follows the pipeline order even when several +processes (e.g. compilation workers) write to it, and so that runs from +several processes or same-named programs never overwrite each other. + +Two limitations of the current implementation: + +- Only the environment variable dumps every stage. The compile pipeline runs in + a worker process under the default `BUILD_JOBS_MODE=PROCESS`, and a worker + registers this subscriber while importing `gt4py.next` — that is, from its own + environment, before it receives the parent's configuration. A programmatic + `config.DUMP_STAGES = ...; enable()` therefore only dumps the frontend stages; + `enable` warns about it. +- Monolithic backends (e.g. `roundtrip`) have no `OTFCompileWorkflow` and + execute in a single unnamed step, so only their frontend stages are dumped. +""" + +from __future__ import annotations + +import json +import pathlib +import re +import threading +import warnings +from typing import Any + +from gt4py.next import config + + +#: Name the stage-dump subscriber is registered under on `stage_hook`. +SUBSCRIBER_NAME = "gt4py_dump_stages" + +_index_lock = threading.Lock() +_next_index: dict[pathlib.Path, int] = {} +_index_prefix_pattern = re.compile(r"^(\d+)_") + + +def enable() -> None: + """ + Register the stage-dump subscriber on `stage_hook` (idempotent). + + Warns when the compile pipeline runs in worker processes, since those + register their own subscribers at import time and so never see a + programmatic `config.DUMP_STAGES`. + """ + from gt4py.next.otf import workflow + + if SUBSCRIBER_NAME in workflow.stage_hook.registry: + return + if config.DUMP_STAGES is not None: + pathlib.Path(config.DUMP_STAGES).mkdir(parents=True, exist_ok=True) + if config.BUILD_JOBS_MODE is not config.BuildJobsMode.SERIAL: + warnings.warn( + "Stage dumping was enabled programmatically while 'BUILD_JOBS_MODE' is" + f" '{config.BUILD_JOBS_MODE.value}', so the compile pipeline runs in worker" + " processes that do not see this subscriber: only the frontend stages will" + " be dumped. Set the 'GT4PY_DUMP_STAGES' environment variable (which the" + " workers do see) or 'GT4PY_BUILD_JOBS_MODE=serial' to dump every stage.", + stacklevel=2, + ) + workflow.stage_hook.register(dump_stage, name=SUBSCRIBER_NAME) + + +def disable() -> None: + """Remove the stage-dump subscriber from `stage_hook` (idempotent).""" + from gt4py.next.otf import workflow + + if SUBSCRIBER_NAME in workflow.stage_hook.registry: + workflow.stage_hook.remove(SUBSCRIBER_NAME) + + +def dump_stage(name: str, artifact: Any) -> None: + """ + Write a stage artifact under `config.DUMP_STAGES` (a `stage_hook` subscriber). + + Dumping is purely observational, so a failure to describe or write an + artifact warns and is otherwise ignored: it must never turn a working + compilation into a failing one (in the default `PROCESS` build-jobs mode the + pipeline runs in a worker, where such an error would surface late and out of + context, if at all). + + Args: + name: Name of the pipeline step that produced `artifact`. + artifact: The artifact to dump. Unknown artifact types fall back to + their `repr`. + """ + if (dump_dir := config.DUMP_STAGES) is None: + return + try: + program_name, text, extension = _describe(artifact) + target_dir = pathlib.Path(dump_dir) / _sanitize(program_name) + target_dir.mkdir(parents=True, exist_ok=True) + _write_unique(target_dir, name, text, extension) + except Exception as error: + try: + warnings.warn( + f"Could not dump the artifact of stage '{name}'" + f" ('{type(artifact).__name__}'): {error!s}.", + stacklevel=2, + ) + except Exception: + # Warnings can be configured to raise (`-W error`), which would defeat + # the whole point of not letting a failed dump break a compilation. + pass + + +def _describe(artifact: Any) -> tuple[str, str, str]: + """Best-effort (program_name, text, file_extension) for a stage artifact.""" + from gt4py.next.ffront import stages as ffront_stages + from gt4py.next.iterator import ir as itir + from gt4py.next.otf import artifacts, workflow + + match artifact: + case workflow.ProgramWithArgs(): + return _describe(artifact.definition) + case itir.Program(): + return str(artifact.id), str(artifact), "txt" # `str` pretty-prints the IR + case ffront_stages.DSLFieldOperatorDef() | ffront_stages.DSLProgramDef(): + return artifact.definition.__name__, _dsl_source(artifact.definition), "py" + case ffront_stages.FOASTOperatorDef(): + return str(artifact.foast_node.id), _foast_text(artifact.foast_node), "txt" + case ffront_stages.PASTProgramDef(): + # PAST has no pretty printer (unlike FOAST and ITIR), and eve nodes + # define `__str__` as `__repr__`, so `repr` is all there is. + return str(artifact.past_node.id), repr(artifact.past_node), "txt" + case artifacts.ProgramSource(): + return ( + artifact.entry_point.name, + _source_text(artifact.source_code), + artifact.code_spec.file_extension, + ) + case artifacts.ExtensionSource(): + program_name = artifact.program_source.entry_point.name + if artifact.binding_source is not None: + return ( + program_name, + _source_text(artifact.binding_source.source_code), + _binding_extension(artifact.program_source.code_spec), + ) + return program_name, "(no binding source generated)", "txt" + case _: + return _guess_name(artifact), repr(artifact), "txt" + + +def _binding_extension(code_spec: Any) -> str: + """Guess the language of a `BindingSource`, which records none of its own. + + Bindings are written in the program's own language when that language can + call into Python (the C++-like backends compile them into the extension + module), and in Python otherwise — the dace backend, for instance, binds an + SDFG with a generated Python function. + """ + from gt4py.next.otf import artifacts + + return code_spec.file_extension if isinstance(code_spec, artifacts.CPPLikeCodeSpec) else "py" + + +def _source_text(source_code: Any) -> str: + """Render a source container's code, which is not always text. + + The dace backend stores the SDFG as its deserialized JSON object, so a + `ProgramSource.source_code` can be a `dict` despite its annotation. + """ + if isinstance(source_code, str): + return source_code + return json.dumps(source_code, indent=2, default=repr) + + +def _dsl_source(definition: Any) -> str: + import inspect + + try: + return inspect.getsource(definition) + except (OSError, TypeError): + return repr(definition) + + +def _foast_text(node: Any) -> str: + from gt4py.next.ffront import foast_pretty_printer + + try: + return foast_pretty_printer.pretty_format(node) + except Exception: + return repr(node) + + +def _guess_name(artifact: Any) -> str: + if (name := getattr(artifact, "entry_point_name", None)) is not None: # CPP artifacts + return str(name) + if (path := getattr(artifact, "library_path", None)) is not None: # dace artifacts + return pathlib.Path(path).stem.removeprefix("lib") + return "unknown_program" + + +def _sanitize(name: str) -> str: + """Turn a program name into a single directory name below the dump root.""" + sanitized = re.sub(r"[^\w.-]", "_", name) + # Names made of dots only ('.', '..') are path navigation rather than names + # and would write outside the per-program directory. Program names are + # usually Python identifiers, but `_guess_name` also derives them from + # library file names. + return sanitized if sanitized.strip(".") else "unknown_program" + + +def _initial_index(target_dir: pathlib.Path) -> int: + """Continue numbering after the highest index prefix already in `target_dir`. + + The counter is per process, but under the default `BUILD_JOBS_MODE=PROCESS` + the compile pipeline runs in a spawned worker whose counter starts over. + Seeding it from the directory keeps a sorted listing in pipeline order. + """ + highest = -1 + for path in target_dir.iterdir(): + if (match := _index_prefix_pattern.match(path.name)) is not None: + highest = max(highest, int(match.group(1))) + return highest + 1 + + +def _write_unique(target_dir: pathlib.Path, stage_name: str, text: str, extension: str) -> None: + with _index_lock: + if (index := _next_index.get(target_dir)) is None: + index = _initial_index(target_dir) + while True: + path = target_dir / f"{index:03d}_{stage_name}.{extension}" + try: + stream = path.open("x", encoding="utf-8") + except FileExistsError: # another process / earlier run owns this index + index += 1 + continue + try: + with stream: + stream.write(text) + except Exception: + # Exclusive creation succeeded but writing did not: do not leave an + # empty file behind, it would look like a dumped but empty stage. + path.unlink(missing_ok=True) + raise + break + _next_index[target_dir] = index + 1 diff --git a/src/gt4py/next/otf/workflow.py b/src/gt4py/next/otf/workflow.py index 8d4c2fdec0..476ef49a0e 100644 --- a/src/gt4py/next/otf/workflow.py +++ b/src/gt4py/next/otf/workflow.py @@ -18,6 +18,7 @@ from gt4py._core import filecache from gt4py.eve.xtyping import OpaqueMutableMapping from gt4py.next import config, fingerprinting, utils +from gt4py.next.instrumentation import hook_machinery StartT = TypeVar("StartT") @@ -51,6 +52,28 @@ class ProgramWithArgs(Generic[DefT, ArgsT]): args: ArgsT +@hook_machinery.event_hook +def stage_hook(name: str, artifact: Any) -> None: + """ + Event hook emitted when a named pipeline step produces an artifact. + + It is emitted by the named step pipelines after each executed step, and by + `Toolchain.translate` for the translation step it runs directly. The step + names of the standard pipelines (`func_to_past`, `past_to_itir`, + `translation`, `bindings`, `compilation`, ...) are therefore observable. + + A subscriber must declare its parameters with exactly these names, `name` + and `artifact`: the hook machinery validates the signature on registration + and rejects a mismatch. + + Args: + name: Name of the step that just ran (its field name in the pipeline + dataclass, e.g. `past_to_itir` or `translation`). + artifact: The artifact returned by the step. It is passed through + opaquely and unformatted; subscribers decide how to inspect it. + """ + + def make_step(function: Workflow[StartT, EndT]) -> ChainableWorkflowMixin[StartT, EndT]: """ Wrap a function in the workflow step convenience wrapper. @@ -159,6 +182,7 @@ def __call__(self, inp: StartT) -> EndT: step_result: Any = inp for step_name in self.step_order: step_result = getattr(self, step_name)(step_result) + stage_hook(step_name, step_result) return step_result @functools.cached_property @@ -190,6 +214,7 @@ def __call__(self, inp: StartT) -> EndT: step_result: Any = inp for step_name in self.step_order(inp): step_result = getattr(self, step_name)(step_result) + stage_hook(step_name, step_result) return step_result @abc.abstractmethod @@ -359,3 +384,13 @@ def __call__(self, inp: StartT) -> EndT: def cache_key(self, inp: StartT) -> str: return self.step_fingerprinter((self._step_fingerprint, self.input_fingerprinter(inp))) + + +if config.DUMP_STAGES is not None: + # Register the `GT4PY_DUMP_STAGES` subscriber so plain program runs dump their + # stage artifacts without any user code. Registering directly (instead of + # calling `stage_dump.enable()`) avoids re-importing this module while it is + # still being initialized. + from gt4py.next.instrumentation import stage_dump as _stage_dump + + stage_hook.register(_stage_dump.dump_stage, name=_stage_dump.SUBSCRIBER_NAME) diff --git a/src/gt4py/next/program_processors/runners/dace/program.py b/src/gt4py/next/program_processors/runners/dace/program.py index abd88a7dad..d529937685 100644 --- a/src/gt4py/next/program_processors/runners/dace/program.py +++ b/src/gt4py/next/program_processors/runners/dace/program.py @@ -18,8 +18,9 @@ from gt4py.next.ffront import decorator from gt4py.next.iterator import ir as itir, transforms as itir_transforms from gt4py.next.iterator.transforms import extractors as extractors -from gt4py.next.otf import arguments, workflow +from gt4py.next.otf import arguments, recipes, workflow from gt4py.next.program_processors.runners.dace import sdfg_args as gtx_dace_args +from gt4py.next.program_processors.runners.dace.workflow import translation as gtx_dace_translation from gt4py.next.type_system import type_specifications as ts @@ -55,43 +56,30 @@ def __sdfg__(self, *args: Any, **kwargs: Any) -> dace.sdfg.sdfg.SDFG: ), ) ) - program = gtir_stage.definition - program = itir_transforms.apply_fieldview_transforms( # run the transforms separately because they require the runtime info - program, offset_provider=offset_provider + # Run the field-view transforms separately: they need the runtime + # connectivity tables, which the SDFG itself must not capture. The + # translation step then receives an already-transformed program whose + # args only carry the offset-provider *types*. + # TODO(ricoh): remove this workaround as soon as the mandatory GTIR + # passes do not need the connectivity tables anymore. + program = itir_transforms.apply_fieldview_transforms( + gtir_stage.definition, offset_provider=offset_provider ) - object.__setattr__( - gtir_stage, - "definition", - program, + aot_args = dataclasses.replace( + gtir_stage.args, + # `CompileTimeArgs.offset_provider` is still typed as the runtime + # mapping, even though the mandatory GTIR passes are the only + # consumers needing the tables and they already ran above. + offset_provider=gtir_stage.args.offset_provider_type, # type: ignore[arg-type] ) - object.__setattr__( - gtir_stage.args, "offset_provider", gtir_stage.args.offset_provider_type - ) # TODO(ricoh): currently this is circumventing the frozenness of CompileTimeArgs - # in order to isolate DaCe from the runtime tables in connectivities.offset_provider. - # These are needed at the time of writing for mandatory GTIR passes. - # Remove this as soon as Program does not expect connectivity tables anymore. _crosscheck_dace_parsing( dace_parsed_args=[*args, *kwargs.values()], gt4py_program_args=[p.type for p in program.params], ) - otf_workflow = self.backend.backend - assert hasattr(otf_workflow, "translation") - otf_workflow_translation = ( - otf_workflow.translation.step - if isinstance(otf_workflow.translation, workflow.CachedStep) - else otf_workflow.translation - ) # Same for the translation stage, which could be a `CachedStep` depending on backend configuration. - # TODO(ricoh): switch 'disable_itir_transforms=True' because we ran them separately previously - # and so we can ensure the SDFG does not know any runtime info it shouldn't know. Remove with - # the other parts of the workaround when possible. sdfg = dace.SDFG.from_json( - otf_workflow_translation.replace( # type: ignore[union-attr] - disable_itir_transforms=True, - disable_field_origin_on_program_arguments=True, - use_metrics=False, - )(gtir_stage).source_code + _translation_only_toolchain(self.backend).translate(program, aot_args).source_code ) self.sdfg_closure_cache["arrays"] = sdfg.arrays @@ -200,6 +188,59 @@ def __sdfg_signature__(self) -> tuple[Sequence[str], Sequence[str]]: return [p.id for p in self.past_stage.past_node.params], [] +def _translation_only_toolchain(backend: gtx_backend.Toolchain) -> gtx_backend.Toolchain: + """ + Derive the translate-only toolchain variant used for SDFG conversion. + + The variant reuses the translation settings of `backend` but disables the + ITIR transforms (the caller runs them separately, with the runtime + connectivity tables), the field origins and the metrics instrumentation, + and bypasses the persistent translation cache. + + Args: + backend: The dace toolchain the program is bound to. + + Returns: + A toolchain whose `translate` produces an SDFG source for an + already-transformed program. + + Raises: + NotImplementedError: If `backend` is not shaped like the standard dace + toolchain, i.e. an 'OTFCompileWorkflow' whose translation step is a + 'DaCeTranslator' (optionally wrapped in a 'CachedStep'). + """ + pipeline = backend.backend + if not isinstance(pipeline, recipes.OTFCompileWorkflow): + raise NotImplementedError( + f"Toolchain '{backend.name}' cannot be converted to an SDFG: SDFG" + " conversion requires the standard 'OTFCompileWorkflow' compile" + f" pipeline, but this toolchain's backend is a '{type(pipeline).__name__}'." + ) + translation: workflow.Workflow[Any, Any] = pipeline.translation + if isinstance(translation, workflow.CachedStep): + # The persistent translation cache is keyed on the untransformed program, + # so it must not see the pre-transformed one handed to this variant. + translation = translation.step + if not isinstance(translation, gtx_dace_translation.DaCeTranslator): + raise NotImplementedError( + f"Toolchain '{backend.name}' cannot be converted to an SDFG: SDFG" + " conversion requires a 'DaCeTranslator' translation step, but this" + f" toolchain's is a '{type(translation).__name__}'." + ) + return dataclasses.replace( + backend, + backend=dataclasses.replace( + pipeline, + translation=dataclasses.replace( + translation, + disable_itir_transforms=True, + disable_field_origin_on_program_arguments=True, + use_metrics=False, + ), + ), + ) + + def _crosscheck_dace_parsing(dace_parsed_args: list[Any], gt4py_program_args: list[Any]) -> None: for dace_parsed_arg, gt4py_program_arg in zip( dace_parsed_args, diff --git a/tests/next_tests/integration_tests/feature_tests/dace_tests/test_program.py b/tests/next_tests/integration_tests/feature_tests/dace_tests/test_program.py index 7221f33ad6..293b3fd75c 100644 --- a/tests/next_tests/integration_tests/feature_tests/dace_tests/test_program.py +++ b/tests/next_tests/integration_tests/feature_tests/dace_tests/test_program.py @@ -11,10 +11,12 @@ from gt4py import next as gtx from gt4py.next import common +from gt4py.next.otf import arguments, workflow from next_tests.integration_tests import cases from next_tests.integration_tests.cases_utils import ( Cell, + E2V, Edge, IDim, JDim, @@ -100,3 +102,84 @@ def testee_dace( assert testee.gt4py_program_input_fields == {"a": Vertex, "b": Vertex} assert testee.gt4py_program_output_fields == {"b": Vertex, "c": Vertex} + + +@gtx.field_operator +def _shift_over_e2v(a: gtx.Field[gtx.Dims[Vertex], gtx.float64]): + return a(E2V[0]) + + +@gtx.program +def _unstructured_prog( + a: gtx.Field[gtx.Dims[Vertex], gtx.float64], b: gtx.Field[gtx.Dims[Edge], gtx.float64] +): + _shift_over_e2v(a, out=b) + + +@gtx.field_operator +def _add_ten(a: gtx.Field[[IDim, KDim], gtx.float64]) -> gtx.Field[[IDim, KDim], gtx.float64]: + return a + 10.0 + + +@gtx.program +def _cartesian_prog( + a: gtx.Field[[IDim, KDim], gtx.float64], b: gtx.Field[[IDim, KDim], gtx.float64] +): + _add_ten(a, out=b) + + +@pytest.mark.parametrize("with_connectivities", [False, True], ids=["cartesian", "unstructured"]) +def test_sdfg_conversion_does_not_mutate_gtir_cache( + exec_alloc_descriptor, + mesh_descriptor, # noqa: F811 + with_connectivities, +): + """Regression test: `__sdfg__` must leave the `past_to_itir` cache entry pristine. + + The SDFG conversion runs the field-view transforms itself (they need the runtime + connectivity tables) and used to write the result back into the frozen stage + returned by the in-memory-cached `past_to_itir` step, together with replacing + the runtime connectivity tables in its args by the mere offset provider *types*. + Because a plain `compile()` of the same program hits the very same cache entry, + every later consumer saw an already-transformed program without its neighbor + tables, which breaks domain inference for unstructured programs. + """ + if with_connectivities: + program = _unstructured_prog.with_compilation_options( + connectivities=mesh_descriptor.offset_provider + ) + else: + program = _cartesian_prog + program = program.with_backend(exec_alloc_descriptor) + offset_provider = program.compilation_options.connectivities or {} + + # The very input `__sdfg__` builds internally, so it shares its cache entry. + stage_input = workflow.ProgramWithArgs( + definition=program.past_stage, + args=arguments.CompileTimeArgs( + args=tuple(p.type for p in program.past_stage.past_node.params), + kwargs={}, + column_axis=None, + offset_provider=offset_provider, + argument_descriptor_contexts={}, + ), + ) + past_to_itir = program.backend.frontend.past_to_itir + cached_stage = past_to_itir(stage_input) + definition_before = str(cached_stage.definition) + offset_provider_before = cached_stage.args.offset_provider + + program.__sdfg__() + program.__sdfg__() + + assert past_to_itir(stage_input) is cached_stage # the entry is really cached + # Neither the lowered program nor the connectivity tables in its args were + # replaced: the entry is still what an unrelated `compile()` of the same + # program expects to get back. + assert str(cached_stage.definition) == definition_before + assert cached_stage.args.offset_provider is offset_provider_before + if with_connectivities: + assert all( + common.is_neighbor_table(connectivity) + for connectivity in cached_stage.args.offset_provider.values() + ) diff --git a/tests/next_tests/integration_tests/feature_tests/instrumentation_tests/test_hooks.py b/tests/next_tests/integration_tests/feature_tests/instrumentation_tests/test_hooks.py index c7ef03a844..f27ab5f5de 100644 --- a/tests/next_tests/integration_tests/feature_tests/instrumentation_tests/test_hooks.py +++ b/tests/next_tests/integration_tests/feature_tests/instrumentation_tests/test_hooks.py @@ -15,8 +15,9 @@ import pytest import gt4py.next as gtx -from gt4py.next import common, Dims, gtfn_cpu, typing as gtx_typing +from gt4py.next import common, config, Dims, gtfn_cpu, typing as gtx_typing from gt4py.next.instrumentation import gpu_profiler, hooks +from gt4py.next.otf import runners try: from gt4py.next.program_processors.runners import dace as dace_backends @@ -241,3 +242,42 @@ def custom_compile_variant_hook( assert hook_call_info["program_definition"] == prog.definition_stage assert hook_call_info["backend"] == backend.name assert hook_call_info["argument_descriptors"] == {"StaticArg": ["cond"]} + + +@pytest.mark.parametrize( + "backend", [b for b in BACKENDS if b is not None], ids=lambda b: getattr(b, "name", str(b)) +) +def test_stage_hook(backend: gtx_typing.Toolchain, monkeypatch): + emitted: list[tuple[str, str]] = [] + + def stage_callback(name: str, artifact: Any) -> None: + emitted.append((name, type(artifact).__name__)) + + # Compile in the calling thread, so the compile pipeline runs in this process + # (and therefore emits into this process' hook registry). + monkeypatch.setattr(config, "BUILD_JOBS_MODE", config.BuildJobsMode.SERIAL) + runners.reset_default_runner() + try: + hooks.stage_hook.register(stage_callback) + try: + prog.with_backend(backend).compile(cond=[True], offset_provider={}) + finally: + hooks.stage_hook.remove(stage_callback) + + assert [name for name, _ in emitted] == [ + "func_to_past", + "past_lint", + "field_view_prog_args_transform", + "past_to_itir", + "translation", + "bindings", + "compilation", + ] + assert ("translation", "ProgramSource") in emitted + + # After removing the subscriber nothing is emitted anymore. + emitted.clear() + prog.with_backend(backend).compile(cond=[False], offset_provider={}) + assert emitted == [] + finally: + runners.reset_default_runner() diff --git a/tests/next_tests/integration_tests/feature_tests/instrumentation_tests/test_stage_dump.py b/tests/next_tests/integration_tests/feature_tests/instrumentation_tests/test_stage_dump.py new file mode 100644 index 0000000000..1a23bb9639 --- /dev/null +++ b/tests/next_tests/integration_tests/feature_tests/instrumentation_tests/test_stage_dump.py @@ -0,0 +1,113 @@ +# GT4Py - GridTools Framework +# +# Copyright (c) 2014-2024, ETH Zurich +# All rights reserved. +# +# Please, refer to the LICENSE file in the root directory. +# SPDX-License-Identifier: BSD-3-Clause + +"""End-to-end tests of the `GT4PY_DUMP_STAGES` stage dumping.""" + +from __future__ import annotations + +import warnings + +import pytest + +import gt4py.next as gtx +from gt4py.next import config, gtfn_cpu, typing as gtx_typing +from gt4py.next.instrumentation import stage_dump +from gt4py.next.otf import runners, workflow + + +try: + from gt4py.next.program_processors.runners import dace as dace_backends + + BACKENDS = [gtfn_cpu, dace_backends.run_dace_cpu] +except ImportError: + BACKENDS = [gtfn_cpu] + + +#: The step names the standard pipelines announce for a DSL program definition, +#: in pipeline order (`Transforms.step_order`, then the `OTFCompileWorkflow` fields). +EXPECTED_STAGES = [ + "func_to_past", + "past_lint", + "field_view_prog_args_transform", + "past_to_itir", + "translation", + "bindings", + "compilation", +] + +IDim = gtx.Dimension("IDim") + + +@gtx.field_operator +def dump_op(a: gtx.Field[gtx.Dims[IDim], gtx.float64]) -> gtx.Field[gtx.Dims[IDim], gtx.float64]: + return a + 1.0 + + +@gtx.program +def dump_prog( + a: gtx.Field[gtx.Dims[IDim], gtx.float64], out: gtx.Field[gtx.Dims[IDim], gtx.float64] +): + dump_op(a, out=out) + + +@pytest.fixture +def clean_stage_hook(monkeypatch): + """Compile in the calling process with no stage subscriber registered. + + The subscriber may already be registered at import time when the test session + itself runs under `GT4PY_DUMP_STAGES`, so it is explicitly removed and restored. + """ + monkeypatch.setattr(config, "BUILD_JOBS_MODE", config.BuildJobsMode.SERIAL) + runners.reset_default_runner() + was_enabled = stage_dump.SUBSCRIBER_NAME in workflow.stage_hook.registry + stage_dump.disable() + yield + stage_dump.disable() + if was_enabled: + stage_dump.enable() + runners.reset_default_runner() + + +@pytest.mark.parametrize("backend", BACKENDS, ids=lambda b: getattr(b, "name", str(b))) +def test_dump_stages_end_to_end( + backend: gtx_typing.Toolchain, tmp_path, monkeypatch, clean_stage_hook +): + monkeypatch.setattr(config, "DUMP_STAGES", tmp_path) + + stage_dump.enable() + try: + # `dump_stage` only warns when it cannot serialize an artifact, so the + # warnings are inspected: a stage that cannot be dumped must fail here. + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + dump_prog.with_backend(backend).compile(offset_provider={}) + finally: + stage_dump.disable() + + dump_failures = [str(w.message) for w in recorded if "Could not dump" in str(w.message)] + assert not dump_failures + + program_dir = tmp_path / "dump_prog" + assert program_dir.is_dir() + + dumped = sorted(program_dir.iterdir()) + # The index prefixes are unique and reflect the pipeline order. + assert [path.name.split("_", 1)[1].rsplit(".", 1)[0] for path in dumped] == EXPECTED_STAGES + for path in dumped: + assert path.stat().st_size > 0, f"empty stage dump: {path.name}" + + (translation,) = [path for path in dumped if "translation" in path.name] + assert "dump_prog" in translation.read_text() + + +def test_no_dump_without_subscriber(tmp_path, monkeypatch, clean_stage_hook): + monkeypatch.setattr(config, "DUMP_STAGES", tmp_path) + + dump_prog.with_backend(gtfn_cpu).compile(offset_provider={}) + + assert list(tmp_path.iterdir()) == [] diff --git a/tests/next_tests/unit_tests/instrumentation_tests/test_stage_dump.py b/tests/next_tests/unit_tests/instrumentation_tests/test_stage_dump.py new file mode 100644 index 0000000000..06216c64e5 --- /dev/null +++ b/tests/next_tests/unit_tests/instrumentation_tests/test_stage_dump.py @@ -0,0 +1,245 @@ +# GT4Py - GridTools Framework +# +# Copyright (c) 2014-2024, ETH Zurich +# All rights reserved. +# +# Please, refer to the LICENSE file in the root directory. +# SPDX-License-Identifier: BSD-3-Clause + +"""Tests of the `GT4PY_DUMP_STAGES` stage-dump subscriber.""" + +from __future__ import annotations + +import dataclasses +import json + +import pytest + +from gt4py.next import config +from gt4py.next.instrumentation import stage_dump +from gt4py.next.iterator import ir as itir +from gt4py.next.otf import arguments, artifacts, workflow +from gt4py.next.otf.binding import interface + + +@pytest.fixture +def dump_dir(tmp_path, monkeypatch): + monkeypatch.setattr(config, "DUMP_STAGES", tmp_path) + yield tmp_path + + +@pytest.fixture +def serial_jobs(monkeypatch): + """Silence the `enable` warning about worker processes not dumping.""" + monkeypatch.setattr(config, "BUILD_JOBS_MODE", config.BuildJobsMode.SERIAL) + + +@pytest.fixture +def restore_subscriber(): + """Restore the subscriber a session started under `GT4PY_DUMP_STAGES` came with.""" + was_enabled = stage_dump.SUBSCRIBER_NAME in workflow.stage_hook.registry + yield + stage_dump.disable() + if was_enabled: + stage_dump.enable() + + +def make_program_source(name: str = "prog", source_code: str = "int x;") -> artifacts.ProgramSource: + return artifacts.ProgramSource( + entry_point=interface.Function(name, ()), + source_code=source_code, + library_deps=(), + code_spec=artifacts.CPPCodeSpec(), + ) + + +def test_enable_disable_idempotent(serial_jobs, restore_subscriber): + stage_dump.disable() + + stage_dump.enable() + stage_dump.enable() + assert list(workflow.stage_hook.registry) == [stage_dump.SUBSCRIBER_NAME] + assert workflow.stage_hook.callbacks == (stage_dump.dump_stage,) + + stage_dump.disable() + stage_dump.disable() + assert stage_dump.SUBSCRIBER_NAME not in workflow.stage_hook.registry + assert workflow.stage_hook.callbacks == () + + +def test_enable_warns_when_compiling_in_worker_processes(monkeypatch, restore_subscriber): + monkeypatch.setattr(config, "BUILD_JOBS_MODE", config.BuildJobsMode.PROCESS) + stage_dump.disable() + + with pytest.warns(UserWarning, match="only the frontend stages will be dumped"): + stage_dump.enable() + + +def test_dump_disabled_writes_nothing(monkeypatch): + monkeypatch.setattr(config, "DUMP_STAGES", None) + writes: list[tuple] = [] + monkeypatch.setattr(stage_dump, "_write_unique", lambda *args: writes.append(args)) + + stage_dump.dump_stage("translation", make_program_source()) + + assert writes == [] + + +def test_dump_program_source(dump_dir): + stage_dump.dump_stage("translation", make_program_source()) + + target = dump_dir / "prog" / "000_translation.cpp" + assert target.exists() + assert target.read_text() == "int x;" + + +def test_dump_index_collision(dump_dir): + stage_dump.dump_stage("translation", make_program_source(source_code="first")) + stage_dump.dump_stage("translation", make_program_source(source_code="second")) + + assert (dump_dir / "prog" / "000_translation.cpp").read_text() == "first" + assert (dump_dir / "prog" / "001_translation.cpp").read_text() == "second" + + +def test_dump_index_skips_preexisting_file(dump_dir, monkeypatch): + # A file written by another process (or an earlier run) must never be overwritten, + # even though this process' index counter still starts at zero. + monkeypatch.setattr(stage_dump, "_next_index", {}) + (dump_dir / "prog").mkdir(parents=True) + (dump_dir / "prog" / "000_translation.cpp").write_text("foreign") + + stage_dump.dump_stage("translation", make_program_source(source_code="mine")) + + assert (dump_dir / "prog" / "000_translation.cpp").read_text() == "foreign" + assert (dump_dir / "prog" / "001_translation.cpp").read_text() == "mine" + + +def test_dump_index_continues_after_other_processes(dump_dir, monkeypatch): + # A worker process compiling the second half of the pipeline starts with an + # empty counter; it must continue the numbering of the frontend dumps the + # parent already wrote, otherwise a sorted listing loses the pipeline order. + monkeypatch.setattr(stage_dump, "_next_index", {}) + program_dir = dump_dir / "prog" + program_dir.mkdir(parents=True) + for index, stage in enumerate(["func_to_past", "past_lint", "past_to_itir"]): + (program_dir / f"{index:03d}_{stage}.txt").write_text("earlier") + (program_dir / "not_indexed.txt").write_text("ignored") + + stage_dump.dump_stage("translation", make_program_source(source_code="mine")) + stage_dump.dump_stage("bindings", make_program_source(source_code="also mine")) + + assert (program_dir / "003_translation.cpp").read_text() == "mine" + assert (program_dir / "004_bindings.cpp").read_text() == "also mine" + assert [ + path.name.split("_", 1)[1].rsplit(".", 1)[0] + for path in sorted(program_dir.iterdir()) + if path.name[:3].isdigit() + ] == ["func_to_past", "past_lint", "past_to_itir", "translation", "bindings"] + + +@pytest.mark.parametrize( + "code_spec, expected_extension", + [ + (artifacts.CPPCodeSpec(), "cpp"), # gtfn: bindings compile with the program + (artifacts.CUDACodeSpec(), "cu"), + (artifacts.SDFGCodeSpec(), "py"), # dace: `bind_sdfg` emits Python + ], + ids=["cpp", "cuda", "sdfg"], +) +def test_dump_bindings_extension_follows_binding_language(dump_dir, code_spec, expected_extension): + program_source = artifacts.ProgramSource( + entry_point=interface.Function("prog", ()), + source_code="", + library_deps=(), + code_spec=code_spec, + ) + extension_source = artifacts.ExtensionSource( + program_source=program_source, + binding_source=artifacts.BindingSource(source_code="", library_deps=()), + ) + + stage_dump.dump_stage("bindings", extension_source) + + target = dump_dir / "prog" / f"000_bindings.{expected_extension}" + assert target.read_text() == "" + + +def test_dump_bindings_without_binding_source(dump_dir): + extension_source = artifacts.ExtensionSource( + program_source=make_program_source(), binding_source=None + ) + + stage_dump.dump_stage("bindings", extension_source) + + assert (dump_dir / "prog" / "000_bindings.txt").read_text() == "(no binding source generated)" + + +@pytest.mark.parametrize("name", [".", "..", "...", ""]) +def test_sanitize_rejects_path_navigation(name): + assert stage_dump._sanitize(name) == "unknown_program" + + +def test_dump_opaque_fallback(dump_dir): + @dataclasses.dataclass + class Opaque: + value: int + + artifact = Opaque(value=42) + stage_dump.dump_stage("mystery", artifact) + + target = dump_dir / "unknown_program" / "000_mystery.txt" + assert target.read_text() == repr(artifact) + + +def test_dump_envelope_unwraps(dump_dir): + program = itir.Program( + id="my_prog", function_definitions=[], params=[], declarations=[], body=[] + ) + stage_dump.dump_stage( + "past_to_itir", + workflow.ProgramWithArgs(definition=program, args=arguments.CompileTimeArgs.empty()), + ) + + target = dump_dir / "my_prog" / "000_past_to_itir.txt" + assert target.read_text() == str(program) + + +def test_dump_sanitizes_program_name(dump_dir): + stage_dump.dump_stage("translation", make_program_source(name="weird/name")) + + assert (dump_dir / "weird_name" / "000_translation.cpp").exists() + + +def test_dump_non_text_source_code(dump_dir): + # The dace backend stores the SDFG as its deserialized JSON object, so + # `ProgramSource.source_code` is not always a string. + source = artifacts.ProgramSource( + entry_point=interface.Function("prog", ()), + source_code={"type": "SDFG", "nodes": []}, + library_deps=(), + code_spec=artifacts.SDFGCodeSpec(), + ) + + stage_dump.dump_stage("translation", source) + + target = dump_dir / "prog" / "000_translation.sdfg" + assert json.loads(target.read_text()) == {"type": "SDFG", "nodes": []} + + +def test_dump_failure_warns_but_does_not_raise(dump_dir): + class Unprintable: + def __repr__(self) -> str: + raise RuntimeError("boom") + + with pytest.warns(UserWarning, match="Could not dump the artifact of stage 'mystery'"): + stage_dump.dump_stage("mystery", Unprintable()) + + +def test_dump_write_failure_leaves_no_empty_file(dump_dir, monkeypatch): + # A failing write must not leave a zero-byte file that looks like an empty stage. + monkeypatch.setattr(stage_dump, "_source_text", lambda source_code: None) + + with pytest.warns(UserWarning, match="Could not dump"): + stage_dump.dump_stage("translation", make_program_source()) + + assert list((dump_dir / "prog").iterdir()) == [] diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/test_dace_backend.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/test_dace_backend.py index f3cc628a84..b8ff99b9ff 100644 --- a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/test_dace_backend.py +++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/test_dace_backend.py @@ -21,7 +21,7 @@ from gt4py import next as gtx from gt4py._core import definitions as core_defs from gt4py.next import config -from gt4py.next.otf import runners, stages +from gt4py.next.otf import arguments, artifacts, runners, stages from gt4py.next.otf import workflow as gtx_workflow from gt4py.next.program_processors.runners.dace import transformations as gtx_transformations from gt4py.next.program_processors.runners.dace.transformations import ( @@ -34,6 +34,7 @@ factory as dace_wf_factory, translation as dace_wf_translation, ) +from gt4py.next.type_system import type_specifications as ts from next_tests.integration_tests import cases, cases_utils from next_tests.integration_tests.cases_utils import KDim @@ -570,3 +571,30 @@ def no_op_top_level_map_processing(*, sdfg: dace.SDFG, **kwargs) -> dace.SDFG: assert any(marker in generated_code for marker in ("delete ", "free")) assert np.allclose(out.asnumpy(), a.asnumpy() + b.asnumpy() + 1) + + +def test_translate_produces_sdfg_source(): + """`Toolchain.translate` is the sanctioned partial run: frontend + translation only.""" + + @gtx.field_operator + def testee_op(a: cases.IField) -> cases.IField: + return a + + @gtx.program + def testee(a: cases.IField, out: cases.IField): + testee_op(a, out=out) + + int_field = ts.FieldType(dims=[cases.IDim], dtype=ts.ScalarType(kind=ts.ScalarKind.INT32)) + compile_time_args = arguments.CompileTimeArgs( + args=(int_field, int_field), + kwargs={}, + offset_provider={}, + column_axis=None, + argument_descriptor_contexts={}, + ) + + source = dace_wf_backend.run_dace_cpu.translate(testee.definition_stage, compile_time_args) + + assert isinstance(source, artifacts.ProgramSource) + assert source.code_spec.file_extension == "sdfg" + assert isinstance(dace.SDFG.from_json(source.source_code), dace.SDFG) diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/test_gtfn.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/test_gtfn.py index 3265af0841..b4c6d53397 100644 --- a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/test_gtfn.py +++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/test_gtfn.py @@ -26,11 +26,13 @@ import pytest import gt4py._core.definitions as core_defs +import gt4py.next as gtx from gt4py.next import config, custom_layout_allocators -from gt4py.next.otf import workflow +from gt4py.next.otf import arguments, artifacts, workflow from gt4py.next.otf.compilation import build_data, cache, compiler, importer from gt4py.next.program_processors.codegens.gtfn import gtfn_module from gt4py.next.program_processors.runners import gtfn +from gt4py.next.type_system import type_specifications as ts def test_make_gtfn_toolchain_device(): @@ -204,3 +206,37 @@ def test_compile_workflow_without_config_caches_translation(): assert isinstance(workflow_.translation, workflow.CachedStep) assert isinstance(workflow_.translation.step, gtfn_module.GTFNTranslationStep) + + +IDim = gtx.Dimension("IDim") + + +@gtx.field_operator +def _copy_op(a: gtx.Field[gtx.Dims[IDim], gtx.float64]) -> gtx.Field[gtx.Dims[IDim], gtx.float64]: + return a + + +@gtx.program +def _copy_prog( + a: gtx.Field[gtx.Dims[IDim], gtx.float64], out: gtx.Field[gtx.Dims[IDim], gtx.float64] +): + _copy_op(a, out=out) + + +def test_translate_produces_cpp_source(): + """`Toolchain.translate` is the sanctioned partial run: frontend + translation only.""" + field_type = ts.FieldType(dims=[IDim], dtype=ts.ScalarType(kind=ts.ScalarKind.FLOAT64)) + compile_time_args = arguments.CompileTimeArgs( + args=(field_type, field_type), + kwargs={}, + offset_provider={}, + column_axis=None, + argument_descriptor_contexts={}, + ) + + source = gtfn.run_gtfn.translate(_copy_prog.definition_stage, compile_time_args) + + assert isinstance(source, artifacts.ProgramSource) + assert source.code_spec.file_extension == "cpp" + assert source.entry_point.name == "_copy_prog" + assert "_copy_prog" in source.source_code diff --git a/tests/next_tests/unit_tests/test_backend.py b/tests/next_tests/unit_tests/test_backend.py new file mode 100644 index 0000000000..58337fb5d4 --- /dev/null +++ b/tests/next_tests/unit_tests/test_backend.py @@ -0,0 +1,141 @@ +# GT4Py - GridTools Framework +# +# Copyright (c) 2014-2024, ETH Zurich +# All rights reserved. +# +# Please, refer to the LICENSE file in the root directory. +# SPDX-License-Identifier: BSD-3-Clause + +"""Backend-free tests of the `Toolchain` partial runs.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +import gt4py.next as gtx +from gt4py.next import backend as next_backend, custom_layout_allocators as next_allocators +from gt4py.next.iterator import ir as itir +from gt4py.next.otf import arguments, artifacts, recipes, stages, workflow +from gt4py.next.otf.binding import interface +from gt4py.next.type_system import type_specifications as ts + + +IDim = gtx.Dimension("IDim") + + +@gtx.field_operator +def copy_op(a: gtx.Field[gtx.Dims[IDim], gtx.float64]) -> gtx.Field[gtx.Dims[IDim], gtx.float64]: + return a + + +@gtx.program +def copy_prog( + a: gtx.Field[gtx.Dims[IDim], gtx.float64], out: gtx.Field[gtx.Dims[IDim], gtx.float64] +): + copy_op(a, out=out) + + +SENTINEL_SOURCE = artifacts.ProgramSource( + entry_point=interface.Function("copy_prog", ()), + source_code="// sentinel", + library_deps=(), + code_spec=artifacts.CPPCodeSpec(), +) + + +@pytest.fixture +def compile_time_args() -> arguments.CompileTimeArgs: + field_type = ts.FieldType(dims=[IDim], dtype=ts.ScalarType(kind=ts.ScalarKind.FLOAT64)) + return arguments.CompileTimeArgs( + args=(field_type, field_type), + kwargs={}, + offset_provider={}, + column_axis=None, + argument_descriptor_contexts={}, + ) + + +def _unreachable_step(inp: Any) -> Any: + raise AssertionError("This step must not run in a partial toolchain run.") + + +def test_translate_returns_program_source(compile_time_args): + seen: list[stages.CompilableProgram] = [] + + def fake_translation(inp: stages.CompilableProgram) -> artifacts.ProgramSource: + seen.append(inp) + return SENTINEL_SOURCE + + toolchain = next_backend.Toolchain( + name="fake", + backend=recipes.OTFCompileWorkflow( + translation=fake_translation, + bindings=_unreachable_step, + compilation=_unreachable_step, + ), + allocator=next_allocators.StandardCPUFieldBufferAllocator(), + frontend=next_backend.DEFAULT_TRANSFORMS, + ) + + result = toolchain.translate(copy_prog.definition_stage, compile_time_args) + + assert result is SENTINEL_SOURCE + assert len(seen) == 1 + assert isinstance(seen[0], workflow.ProgramWithArgs) + assert isinstance(seen[0].definition, itir.Program) + assert seen[0].args == compile_time_args + + +def test_translate_emits_stage_hook(compile_time_args): + emitted: list[tuple[str, Any]] = [] + + def stage_callback(name: str, artifact: Any) -> None: + emitted.append((name, artifact)) + + toolchain = next_backend.Toolchain( + name="fake", + backend=recipes.OTFCompileWorkflow( + translation=lambda inp: SENTINEL_SOURCE, + bindings=_unreachable_step, + compilation=_unreachable_step, + ), + allocator=next_allocators.StandardCPUFieldBufferAllocator(), + frontend=next_backend.DEFAULT_TRANSFORMS, + ) + + workflow.stage_hook.register(stage_callback) + try: + toolchain.translate(copy_prog.definition_stage, compile_time_args) + finally: + workflow.stage_hook.remove(stage_callback) + + translation_events = [(name, artifact) for name, artifact in emitted if name == "translation"] + assert translation_events == [("translation", SENTINEL_SOURCE)] + # The frontend steps are announced too, and the translation comes last. + assert emitted[-1][0] == "translation" + assert "past_to_itir" in [name for name, _ in emitted] + + +def test_translate_rejects_monolithic_backend(compile_time_args): + frontend_calls: list[Any] = [] + + def recording_frontend(inp: Any) -> Any: + frontend_calls.append(inp) + raise AssertionError("The frontend must not run for an unsupported compile pipeline.") + + toolchain = next_backend.Toolchain( + name="monolithic", + backend=_unreachable_step, + allocator=next_allocators.StandardCPUFieldBufferAllocator(), + frontend=recording_frontend, + ) + + with pytest.raises(NotImplementedError, match="OTFCompileWorkflow") as exc_info: + toolchain.translate(copy_prog.definition_stage, compile_time_args) + + message = str(exc_info.value) + assert "monolithic" in message # names the toolchain + assert "function" in message # names the offending backend type + assert frontend_calls == [] # fails fast, before running the frontend