diff --git a/backends/qualcomm/qnn_preprocess.py b/backends/qualcomm/qnn_preprocess.py index a267dc2f763..0d8fcd6fe3b 100644 --- a/backends/qualcomm/qnn_preprocess.py +++ b/backends/qualcomm/qnn_preprocess.py @@ -8,6 +8,7 @@ from collections import defaultdict from typing import Dict, final, List +import executorch.backends.qualcomm.python.PyQnnManagerAdaptor as PyQnnManager import torch # noqa: F401 from executorch.backends.qualcomm._passes.qnn_pass_manager import ( get_qnn_pass_manager_cls, @@ -46,6 +47,49 @@ logger.setLevel(logging.DEBUG) +def _check_io_binding(edge_program: ExportedProgram, nodes_to_wrappers) -> None: + """Fail here if QNN's graph I/O will not line up with the delegate signature. + + At runtime the delegate binds its arguments positionally: it walks the tensor + lists recovered from the context binary and consumes one argument per tensor + the name prefixes mark as bindable (QnnExecuTorchBackend::execute). Nothing + reconciles that walk with the number of arguments ExecuTorch actually passes, + so a graph that publishes extra I/O reads past the end of the argument list on + device. Catching it here costs one pass over the wrappers and reports the + offending tensor names instead of a register dump. + """ + qnn_inputs, qnn_outputs = set(), set() + for wrappers in nodes_to_wrappers.values(): + for wrapper in wrappers.values(): + name = PyQnnManager.PyQnnTensorWrapper(wrapper).GetName() + # Mutable buffers are threaded through separately and never consume a + # delegate argument; the runtime skips them by the same marker. + if "mutbuf_" in name: + continue + if name.startswith("input_"): + qnn_inputs.add(name) + elif name.startswith("output_"): + qnn_outputs.add(name) + + signature = edge_program.graph_signature + num_inputs = len(signature.user_inputs) + num_outputs = len(signature.user_outputs) + if len(qnn_inputs) == num_inputs and len(qnn_outputs) == num_outputs: + return + + raise RuntimeError( + "QNN graph I/O does not match the delegated program signature. QNN " + f"declares {len(qnn_inputs)} graph inputs and {len(qnn_outputs)} graph " + f"outputs; the signature declares {num_inputs} user inputs and " + f"{num_outputs} user outputs. The runtime binds delegate arguments " + "positionally, so this reads past the end of the argument list on device." + f"\n qnn inputs : {sorted(qnn_inputs)}" + f"\n qnn outputs : {sorted(qnn_outputs)}" + f"\n signature inputs : {list(signature.user_inputs)}" + f"\n signature outputs : {list(signature.user_outputs)}" + ) + + @final class QnnBackend(BackendDetails): @staticmethod @@ -111,6 +155,7 @@ def _build_op_wrappers( else: raise RuntimeError(f"{node.op} is not supported in Qnn") + _check_io_binding(edge_program, nodes_to_wrappers) return py_op_wrapper_list @staticmethod diff --git a/backends/qualcomm/tests/test_passes.py b/backends/qualcomm/tests/test_passes.py index 9085b4eb7ed..ea5767dd19d 100644 --- a/backends/qualcomm/tests/test_passes.py +++ b/backends/qualcomm/tests/test_passes.py @@ -848,5 +848,79 @@ def test_backend_bundle_cache_survives_an_expired_entry(self): self.assertGreaterEqual(reused, 1, "third manager must reuse the live bundle") +class TestIoBindingCheck(unittest.TestCase): + """_check_io_binding must fire when QNN publishes I/O the signature lacks. + + The runtime binds delegate arguments positionally against the tensor list in + the context binary, so a graph that declares more bindable tensors than the + program passes reads past the end of args. That has to be caught at lowering. + """ + + @staticmethod + def _edge_program(num_inputs, num_outputs): + signature = MagicMock() + signature.user_inputs = [f"arg_{i}" for i in range(num_inputs)] + signature.user_outputs = [f"out_{i}" for i in range(num_outputs)] + edge_program = MagicMock() + edge_program.graph_signature = signature + return edge_program + + def _run(self, names, num_inputs, num_outputs): + from executorch.backends.qualcomm import qnn_preprocess + + wrappers = {f"n{i}": {0: object()} for i in range(len(names))} + by_id = { + id(next(iter(v.values()))): name + for v, name in zip(wrappers.values(), names) + } + + class _StubWrapper: + def __init__(self, wrapper): + self._name = by_id[id(wrapper)] + + def GetName(self): + return self._name + + with unittest.mock.patch.object( + qnn_preprocess.PyQnnManager, "PyQnnTensorWrapper", _StubWrapper + ): + qnn_preprocess._check_io_binding( + self._edge_program(num_inputs, num_outputs), wrappers + ) + + def test_matching_io_passes(self): + self._run( + ["input_0_x", "output_y", "conv2d_3", "_frozen_param0"], + num_inputs=1, + num_outputs=1, + ) + + def test_mutable_buffers_do_not_consume_arguments(self): + # A mutable buffer is threaded through separately and is not a user input. + self._run( + ["input_0_x", "input_1_mutbuf_0_cache", "output_y"], + num_inputs=1, + num_outputs=1, + ) + + def test_extra_published_inputs_are_rejected(self): + with self.assertRaises(RuntimeError) as ctx: + self._run( + ["input_0_x", "input_1_stray", "input_2_stray", "output_y"], + num_inputs=1, + num_outputs=1, + ) + message = str(ctx.exception) + self.assertIn("QNN declares 3 graph inputs", message) + self.assertIn("input_1_stray", message) + + def test_extra_published_outputs_are_rejected(self): + with self.assertRaises(RuntimeError) as ctx: + self._run( + ["input_0_x", "output_y", "output_stray"], num_inputs=1, num_outputs=1 + ) + self.assertIn("2 graph outputs", str(ctx.exception)) + + if __name__ == "__main__": unittest.main()