Arm backend: Add partial vulkan runtime support for VgfPipeline (#13471)

- Add XfailIfNoVKMLEmulationLayer for VGF unit tests
- Introduce run_target_board() to unify FVP and VKML execution paths
- Implement run_vkml_emulation_layer() with NotImplementedError for
output parsing, as the VGF runtime doesn't dump the output tensors in a
usable format at the moment.

Signed-off-by: Yufeng Shi <[email protected]>

Author: YufengShi-dudu

backends/arm/test/common.py

@@ -19,6 +19,7 @@
     corstone300_installed,
     corstone320_installed,
     model_converter_installed,
+    vkml_emulation_layer_installed,
 )
 from executorch.backends.arm.tosa_specification import TosaSpecification
 from executorch.exir.backend.compile_spec_schema import CompileSpec
@@ -90,39 +91,6 @@ def get_tosa_compile_spec_unbuilt(
     return compile_spec_builder
 
 
-def get_vgf_compile_spec(
-    tosa_spec: str | TosaSpecification,
-    compiler_flags: Optional[str] = "",
-    custom_path=None,
-) -> list[CompileSpec]:
-    """
-    Default compile spec for VGF tests.
-    """
-    return get_vgf_compile_spec_unbuilt(tosa_spec, compiler_flags, custom_path).build()
-
-
-def get_vgf_compile_spec_unbuilt(
-    tosa_spec: str | TosaSpecification,
-    compiler_flags: Optional[str] = "",
-    custom_path=None,
-) -> ArmCompileSpecBuilder:
-    """Get the ArmCompileSpecBuilder for the default VGF tests, to modify
-    the compile spec before calling .build() to finalize it.
-    """
-    if not custom_path:
-        custom_path = maybe_get_tosa_collate_path()
-
-    if custom_path is not None:
-        os.makedirs(custom_path, exist_ok=True)
-    compile_spec_builder = (
-        ArmCompileSpecBuilder()
-        .vgf_compile_spec(tosa_spec, compiler_flags)
-        .dump_intermediate_artifacts_to(custom_path)
-    )
-
-    return compile_spec_builder
-
-
 def get_u55_compile_spec(
     macs: int = 128,
     system_config: str = "Ethos_U55_High_End_Embedded",
@@ -165,6 +133,17 @@ def get_u85_compile_spec(
     ).build()
 
 
+def get_vgf_compile_spec(
+    tosa_spec: str | TosaSpecification,
+    compiler_flags: Optional[str] = "",
+    custom_path=None,
+) -> list[CompileSpec]:
+    """
+    Default compile spec for VGF tests.
+    """
+    return get_vgf_compile_spec_unbuilt(tosa_spec, compiler_flags, custom_path).build()
+
+
 def get_u55_compile_spec_unbuilt(
     macs: int,
     system_config: str,
@@ -228,6 +207,33 @@ def get_u85_compile_spec_unbuilt(
     return compile_spec  # type: ignore[return-value]
 
 
+def get_vgf_compile_spec_unbuilt(
+    tosa_spec: str | TosaSpecification,
+    compiler_flags: Optional[str] = "",
+    custom_path=None,
+) -> ArmCompileSpecBuilder:
+    """Get the ArmCompileSpecBuilder for the default VGF tests, to modify
+    the compile spec before calling .build() to finalize it.
+    """
+    if "FP" in repr(tosa_spec):
+        artifact_path = custom_path or tempfile.mkdtemp(prefix="arm_vgf_fp_")
+    elif "INT" in repr(tosa_spec):
+        artifact_path = custom_path or tempfile.mkdtemp(prefix="arm_vgf_int_")
+    else:
+        raise ValueError(f"Unsupported vgf compile_spec: {repr(tosa_spec)}")
+
+    if not os.path.exists(artifact_path):
+        os.makedirs(artifact_path, exist_ok=True)
+
+    compile_spec_builder = (
+        ArmCompileSpecBuilder()
+        .vgf_compile_spec(tosa_spec, compiler_flags)
+        .dump_intermediate_artifacts_to(artifact_path)
+    )
+
+    return compile_spec_builder
+
+
 XfailIfNoCorstone300 = pytest.mark.xfail(
     condition=not (
         corstone300_installed() and arm_executor_runner_exists("corstone-300")
@@ -251,7 +257,14 @@ def get_u85_compile_spec_unbuilt(
     raises=FileNotFoundError,
     reason="Did not find model-converter on path",
 )
-"""Xfails a test if model-converter is not installed"""
+"""Skips a test if model-converter is not installed"""
+
+XfailfNoVKMLEmulationLayer = pytest.mark.xfail(
+    condition=not (vkml_emulation_layer_installed()),
+    raises=TypeError,
+    reason="VKML environment is not set properly or executor_runner path is misused",
+)
+"""Xfails a test if VKML Emulation Layer is not installed"""
 
 xfail_type = str | tuple[str, type[Exception]]
 

backends/arm/test/ops/test_add.py

@@ -7,6 +7,7 @@
 
 from typing import Tuple
 
+import pytest
 import torch
 from executorch.backends.arm.quantizer import arm_quantizer
 from executorch.backends.arm.test import common, conftest
@@ -187,9 +188,19 @@ def test_add_tensor_u85_INT_2(test_data: input_t2):
 
 @common.parametrize("test_data", Add.test_data)
 @common.SkipIfNoModelConverter
+@common.XfailfNoVKMLEmulationLayer
+@pytest.mark.xfail(
+    reason="VGF runtime is not yet fully supported for FP pipeline (MLETORCH-1234)",
+    strict=True,
+)
 def test_add_tensor_vgf_FP(test_data: input_t1):
     pipeline = VgfPipeline[input_t1](
-        Add(), test_data(), aten_op, exir_op, tosa_version="TOSA-1.0+FP"
+        Add(),
+        test_data(),
+        aten_op,
+        exir_op,
+        tosa_version="TOSA-1.0+FP",
+        run_on_vulkan_runtime=True,
     )
     pipeline.run()
 

backends/arm/test/runner_utils.py

@@ -18,7 +18,7 @@
 import numpy as np
 import torch
 
-from executorch.backends.arm.arm_backend import is_tosa
+from executorch.backends.arm.arm_backend import is_tosa, is_vgf
 from executorch.backends.arm.test.conftest import is_option_enabled
 from executorch.backends.arm.tosa_specification import (
     get_tosa_spec,
@@ -57,6 +57,8 @@
     torch.complex128: np.complex128,
 }
 
+VALID_TARGET = {"corstone-300", "corstone-320", "vkml_emulation_layer"}
+
 
 class QuantizationParams:
     __slots__ = ["node_name", "zp", "scale", "qmin", "qmax", "dtype"]
@@ -218,6 +220,69 @@ def __torch_function__(self, func, types, args=..., kwargs=None):
         return func(*args, **kwargs)
 
 
+def run_target(
+    executorch_program_manager: ExecutorchProgramManager,
+    inputs: Tuple[torch.Tensor],
+    intermediate_path: str | Path,
+    target_board: Literal["corestone-300", "corestone-320", "vkml_emulation_layer"],
+    elf_path: str | Path,
+    timeout: int = 120,  # s
+):
+    if target_board not in VALID_TARGET:
+        raise ValueError(f"Unsupported target: {target_board}")
+
+    if target_board in ("corstone-300", "corstone-320"):
+        return run_corstone(
+            executorch_program_manager,
+            inputs,
+            intermediate_path,
+            target_board,
+            elf_path,
+            timeout,
+        )
+    elif target_board == "vkml_emulation_layer":
+        return run_vkml_emulation_layer(
+            executorch_program_manager,
+            intermediate_path,
+            elf_path,
+        )
+
+
+def run_vkml_emulation_layer(
+    executorch_program_manager: ExecutorchProgramManager,
+    intermediate_path: str | Path,
+    elf_path: str | Path,
+):
+    """Executes an inference of the exported_program on ML Emulation Layer for Vulkan
+    Args:
+        `executorch_program_manager`: The executorch program to run.
+        `intermediate_path`: Directory to save the .pte and capture outputs.
+        `elf_path`: Path to the Vulkan-capable executor_runner binary.
+    """
+
+    intermediate_path = Path(intermediate_path)
+    intermediate_path.mkdir(exist_ok=True)
+    elf_path = Path(elf_path)
+    if not elf_path.exists():
+        raise FileNotFoundError(f"Did not find elf file {elf_path}")
+
+    # Save pte to file
+    pte_path = os.path.join(intermediate_path, "program.pte")
+    with open(pte_path, "wb") as f:
+        f.write(executorch_program_manager.buffer)
+
+    cmd_line = [elf_path, "-model_path", pte_path]
+    result = _run_cmd(cmd_line)
+
+    result_stdout = result.stdout.decode()  # noqa: F841
+    # TODO: MLETORCH-1234: Support VGF e2e tests in VgfPipeline
+    # TODO: Add regex to check for error or fault messages in stdout from Emulation Layer
+    # TODO: Retrieve and return the output tensors once VGF runtime is able to dump them.
+    raise NotImplementedError(
+        "Output parsing from VKML Emulation Layer is not yet implemented. "
+    )
+
+
 def run_corstone(
     executorch_program_manager: ExecutorchProgramManager,
     inputs: Tuple[torch.Tensor],
@@ -229,7 +294,7 @@ def run_corstone(
     """Executes an inference of the exported_program on FVP.
     Returns a list of tensors with the output.
     Args:
-        `executorch_program_manager`: the executorch program to run.
+        `executorch_program_manager`: The executorch program to run.
         The output of a EdgeProgramManager.to_executorch() call.
         `inputs`: A list of tensors with the inputs of the inference.
         `dump_path`: A directory where the .pte and inputs are saved to file.
@@ -558,18 +623,52 @@ def model_converter_installed() -> bool:
     return True
 
 
-def get_elf_path(target_board):
-    elf_path = os.path.join(
-        "arm_test",
-        f"arm_semihosting_executor_runner_{target_board}",
-        "arm_executor_runner",
-    )
+def vkml_emulation_layer_installed() -> bool:
+    # Check VK_INSTANCE_LAYERS
+    vk_instance_layers = os.environ.get("VK_INSTANCE_LAYERS", "")
+    required_layers = {
+        "VK_LAYER_ML_Graph_Emulation",
+        "VK_LAYER_ML_Tensor_Emulation",
+    }
+    existing_layers = set(vk_instance_layers.split(":"))
+    layers_exists = required_layers.issubset(existing_layers)
+
+    # Check LD_LIBRARY_PATH for "emulation-layer/deploy"
+    ld_library_path = os.environ.get("LD_LIBRARY_PATH", "")
+    deploy_exists = False
+    for path in ld_library_path.split(os.path.pathsep):
+        if "emulation-layer/deploy" in path and os.path.isdir(path):
+            deploy_exists = True
+
+    return layers_exists and deploy_exists
+
+
+def assert_elf_path_exists(elf_path):
     if not os.path.exists(elf_path):
         raise FileNotFoundError(
-            f"Did not find build arm_executor_runner in path {elf_path}, run setup_testing.sh?"
+            f"Did not find build arm_executor_runner or executor_runner in path {elf_path}, run setup_testing.sh?"
         )
-    else:
-        return elf_path
+
+
+def get_elf_path(target_board):
+    if target_board not in VALID_TARGET:
+        raise ValueError(f"Unsupported target: {target_board}")
+
+    if target_board in ("corstone-300", "corstone-320"):
+        elf_path = os.path.join(
+            "arm_test",
+            f"arm_semihosting_executor_runner_{target_board}",
+            "arm_executor_runner",
+        )
+        assert_elf_path_exists(elf_path)
+    elif target_board == "vkml_emulation_layer":
+        elf_path = os.path.join(
+            "cmake-out",
+            "executor_runner",
+        )
+        assert_elf_path_exists(elf_path)
+
+    return elf_path
 
 
 def arm_executor_runner_exists(target_board):
@@ -629,6 +728,8 @@ def transpose_data_format(data: list[np.ndarray], to: Literal["NHWC", "NCHW"]):
 
 
 def get_target_board(compile_spec: list[CompileSpec]) -> str | None:
+    if is_vgf(compile_spec):
+        return "vkml_emulation_layer"
     for spec in compile_spec:
         if spec.key == "compile_flags":
             flags = spec.value.decode()

backends/arm/test/tester/arm_tester.py

@@ -51,7 +51,7 @@
     get_output_nodes,
     get_output_quantization_params,
     get_target_board,
-    run_corstone,
+    run_target,
     TosaReferenceModelDispatch,
 )
 
@@ -212,7 +212,7 @@ def run_artifact(self, inputs):
                 f"Did not find build arm_executor_runner in path {elf_path}, run setup_testing.sh?"
             )
 
-        return run_corstone(
+        return run_target(
             self.executorch_program_manager,
             inputs_flattened,
             intermediate_path,

backends/arm/test/tester/test_pipeline.py

@@ -892,7 +892,9 @@ class VgfPipeline(BasePipelineMaker, Generic[T]):
        exir_ops: Exir dialect ops expected to be found in the graph after to_edge.
        if not using use_edge_to_transform_and_lower.
 
-       run_on_vulkan_runtime: Not yet supported.
+       run_on_vulkan_runtime: Partially supported. However, comparison between reference and model
+       outputs is expected to fail, as the VGF runtime doesn't dump the output tensors in a usable
+       format at the moment.
 
        vgf_compiler_flags: Optional compiler flags.
 
@@ -992,4 +994,11 @@ def __init__(
             )
 
         if run_on_vulkan_runtime:
-            pass
+            self.add_stage(self.tester.serialize)
+            self.add_stage(
+                self.tester.run_method_and_compare_outputs,
+                atol=atol,
+                rtol=rtol,
+                qtol=qtol,
+                inputs=self.test_data,
+            )