diff --git a/scratchv/analysis/ir_verifier.py b/scratchv/analysis/ir_verifier.py index 0226f8d..50b9cc2 100644 --- a/scratchv/analysis/ir_verifier.py +++ b/scratchv/analysis/ir_verifier.py @@ -15,6 +15,7 @@ be unreachable. Conditional branches must have exactly two targets specified. 6. SSA validity: Each value must be assigned exactly once (SSA). + 7. Entry existence: Every function must contain a basic block. Usage:: @@ -39,6 +40,7 @@ from scratchv.ir.types import ( OpCode, Function, + Instruction, Program, ) @@ -136,6 +138,7 @@ def verify(self) -> list[VerificationError]: the program passed all checks. """ self._errors = [] + self._check_global_ssa_validity() for func in self.program.functions: self._verify_function(func) @@ -153,7 +156,17 @@ def _verify_function(self, func: Function) -> None: func: The function to verify. """ # Collect all block names for label checks - block_names: set[str] = {b.name for b in func.blocks} + block_names: set[str] = set() + for block in func.blocks: + if block.name in block_names: + self._add_error( + ErrorLevel.ERROR, + f"duplicate block label '{block.name}'", + func_name=func.name, + block_name=block.name, + rule="label-existence", + ) + block_names.add(block.name) # Check 1: Def-before-use per function self._check_def_before_use(func) @@ -189,46 +202,144 @@ def _verify_function(self, func: Function) -> None: def _check_def_before_use(self, func: Function) -> None: """Ensure all value operands are defined before use. - Uses a two-pass approach: - 1. First pass: collect all values that are assigned (appear as - instruction destinations) across all blocks. - 2. Second pass: flag operands that are never assigned and aren't - constants or function params. - - Values that appear as operands but are never assigned are treated - as implicit input variables (not flagged as errors). + Function parameters, program globals, and constants are available at + function entry. Instruction results are available only after their + defining instruction and only in blocks dominated by that definition. Args: func: The function to check. """ - # Pass 1: collect all defined names (instruction destinations) - defined: set[str] = set() - - # Function parameters are pre-defined - for param in func.params: - defined.add(param.name) + entry_values = {param.name for param in func.params} + entry_values.update(value.name for value in self.program.global_values) + definitions: dict[str, list[tuple[str, int]]] = {} for block in func.blocks: - for instr in block.instructions: + for i, instr in enumerate(block.instructions): if instr.dest is not None: - defined.add(instr.dest.name) + definitions.setdefault(instr.dest.name, []).append( + (block.name, i), + ) + + dominators = self._compute_dominators(func) - # Pass 2: flag uses of undefined values for block in func.blocks: - for instr in block.instructions: + for i, instr in enumerate(block.instructions): for op in instr.operands: - if op.name not in defined: - # Allow constants (auto-defined) and implicit inputs - if op.is_constant: - continue - # Treat as implicit input (not an error) - # Mark so it's not flagged again - defined.add(op.name) + if op.is_constant or op.name in entry_values: continue - # Also track values created mid-block for intra-block checks - if instr.dest is not None: - defined.add(instr.dest.name) + sites = definitions.get(op.name, []) + is_defined = any( + ( + def_block == block.name + and def_index < i + ) + or ( + def_block != block.name + and def_block in dominators.get(block.name, set()) + ) + for def_block, def_index in sites + ) + if is_defined: + continue + + self._add_error( + ErrorLevel.ERROR, + f"value '{op.name}' is used before a dominating " + "definition", + func_name=func.name, + block_name=block.name, + instruction_index=i, + value_name=op.name, + rule="def-before-use", + ) + + def _compute_dominators(self, func: Function) -> dict[str, set[str]]: + """Compute block dominators for the reachable control-flow graph.""" + if not func.blocks: + return {} + + block_names = {block.name for block in func.blocks} + successors: dict[str, set[str]] = { + block.name: set() for block in func.blocks + } + + for index, block in enumerate(func.blocks): + if not block.instructions: + if index + 1 < len(func.blocks): + successors[block.name].add(func.blocks[index + 1].name) + continue + terminator = block.instructions[-1] + for target in self._branch_targets(terminator): + if target in block_names: + successors[block.name].add(target) + if ( + terminator.opcode not in { + OpCode.BR, OpCode.BR_IF, OpCode.RETURN, + } + and index + 1 < len(func.blocks) + ): + successors[block.name].add(func.blocks[index + 1].name) + + entry = func.blocks[0].name + reachable = {entry} + worklist = [entry] + while worklist: + current = worklist.pop() + for successor in successors[current]: + if successor not in reachable: + reachable.add(successor) + worklist.append(successor) + + predecessors: dict[str, set[str]] = { + name: set() for name in block_names + } + for source, targets in successors.items(): + for target in targets: + predecessors[target].add(source) + + dominators = { + name: ({entry} if name == entry else set(reachable)) + for name in reachable + } + changed = True + while changed: + changed = False + for name in reachable: + if name == entry: + continue + reachable_predecessors = predecessors[name] & reachable + if reachable_predecessors: + common = set.intersection( + *(dominators[pred] for pred in reachable_predecessors) + ) + updated = {name} | common + else: + updated = {name} + if updated != dominators[name]: + dominators[name] = updated + changed = True + + for name in block_names - reachable: + dominators[name] = {name} + return dominators + + @staticmethod + def _branch_targets(instr: Instruction) -> list[str]: + """Return normalized targets for a branch instruction.""" + if instr.opcode == OpCode.BR: + return [(instr.target or "").strip()] + if instr.opcode == OpCode.BR_IF: + return [ + target.strip() + for target in (instr.target or "").split(",") + ] + return [] + + @staticmethod + def _has_two_branch_targets(targets: list[str]) -> bool: + """Return whether a conditional branch has two usable targets.""" + return len(targets) == 2 and all(targets) # ------------------------------------------------------------------- # Rule 2: Block termination @@ -282,29 +393,37 @@ def _check_label_existence( """ for block in func.blocks: for i, instr in enumerate(block.instructions): - target = instr.target - if target is None: + if instr.opcode not in {OpCode.BR, OpCode.BR_IF}: continue - # BR_IF has comma-separated targets + targets = self._branch_targets(instr) if instr.opcode == OpCode.BR_IF: - parts = target.split(",") - for part in parts: - part = part.strip() - if part and part not in block_names: + if not self._has_two_branch_targets(targets): + self._add_error( + ErrorLevel.ERROR, + "conditional branch must specify two non-empty " + "targets", + func_name=func.name, + block_name=block.name, + instruction_index=i, + rule="label-existence", + ) + for target in targets: + if not target: + if instr.opcode == OpCode.BR: self._add_error( ErrorLevel.ERROR, - f"branch target '{part}' does not exist", + "branch target is missing", func_name=func.name, block_name=block.name, instruction_index=i, rule="label-existence", ) - else: + continue if target not in block_names: self._add_error( ErrorLevel.ERROR, - f"jump target '{target}' does not exist", + f"branch target '{target}' does not exist", func_name=func.name, block_name=block.name, instruction_index=i, @@ -330,28 +449,19 @@ def _check_type_consistency(self, func: Function) -> None: for block in func.blocks: for i, instr in enumerate(block.instructions): - if instr.opcode in binary_ops and len(instr.operands) >= 2: - lhs, rhs = instr.operands[0], instr.operands[1] - if lhs.dtype != rhs.dtype: - self._add_error( - ErrorLevel.WARNING, - f"operand type mismatch: '{lhs.name}' is " - f"{lhs.dtype.value}, '{rhs.name}' is " - f"{rhs.dtype.value}", - func_name=func.name, - block_name=block.name, - instruction_index=i, - rule="type-consistency", - ) - - if instr.opcode in nn_ops and len(instr.operands) >= 2: - lhs, rhs = instr.operands[0], instr.operands[1] - if lhs.dtype != rhs.dtype: + if ( + instr.opcode in binary_ops | nn_ops + and len(instr.operands) >= 2 + ): + expected = instr.operands[0] + for operand in instr.operands[1:]: + if expected.dtype == operand.dtype: + continue self._add_error( ErrorLevel.WARNING, - f"NN op operand type mismatch: '{lhs.name}' is " - f"{lhs.dtype.value}, '{rhs.name}' is " - f"{rhs.dtype.value}", + f"operand type mismatch: '{expected.name}' is " + f"{expected.dtype.value}, '{operand.name}' is " + f"{operand.dtype.value}", func_name=func.name, block_name=block.name, instruction_index=i, @@ -376,43 +486,41 @@ def _check_control_flow_integrity( func: The function to check. block_names: Valid block names. """ + terminators = {OpCode.BR, OpCode.BR_IF, OpCode.RETURN} + for block in func.blocks: for i, instr in enumerate(block.instructions): - if instr.opcode == OpCode.BR: - # Cannot have instructions after unconditional jump - if i < len(block.instructions) - 1: - self._add_error( - ErrorLevel.ERROR, - "unreachable instructions after unconditional " - "branch", - func_name=func.name, - block_name=block.name, - instruction_index=i, - rule="control-flow-integrity", - ) + if ( + instr.opcode in terminators + and i < len(block.instructions) - 1 + ): + self._add_error( + ErrorLevel.ERROR, + f"unreachable instructions after " + f"{instr.opcode.value}", + func_name=func.name, + block_name=block.name, + instruction_index=i, + rule="control-flow-integrity", + ) - elif instr.opcode == OpCode.BR_IF: + if instr.opcode == OpCode.BR_IF: # Must have exactly two targets - target = instr.target or "" - targets = [ - t.strip() for t in target.split(",") if t.strip() - ] - if len(targets) != 2: + targets = self._branch_targets(instr) + if not self._has_two_branch_targets(targets): self._add_error( ErrorLevel.ERROR, - f"conditional branch has {len(targets)} " - f"targets, expected 2", + "conditional branch must have exactly two " + "non-empty targets", func_name=func.name, block_name=block.name, instruction_index=i, rule="control-flow-integrity", ) - - elif instr.opcode == OpCode.RETURN: - if i < len(block.instructions) - 1: + if not instr.operands: self._add_error( ErrorLevel.ERROR, - "unreachable instructions after return", + "conditional branch has no condition operand", func_name=func.name, block_name=block.name, instruction_index=i, @@ -423,13 +531,41 @@ def _check_control_flow_integrity( # Rule 6: SSA validity # ------------------------------------------------------------------- + def _check_global_ssa_validity(self) -> None: + """Ensure program-global values have unique SSA names.""" + assigned: set[str] = set() + for value in self.program.global_values: + if value.name in assigned: + self._add_error( + ErrorLevel.ERROR, + f"global value '{value.name}' assigned multiple times " + "(SSA violation)", + value_name=value.name, + rule="ssa-validity", + ) + else: + assigned.add(value.name) + def _check_ssa_validity(self, func: Function) -> None: """Check SSA validity: each value must be assigned exactly once. Args: func: The function to check. """ - assigned: dict[str, int] = {} # value name -> first assignment index + assigned = {value.name for value in self.program.global_values} + + for param in func.params: + if param.name in assigned: + self._add_error( + ErrorLevel.ERROR, + f"value '{param.name}' assigned multiple times " + "(SSA violation)", + func_name=func.name, + value_name=param.name, + rule="ssa-validity", + ) + else: + assigned.add(param.name) for block in func.blocks: for i, instr in enumerate(block.instructions): @@ -446,7 +582,7 @@ def _check_ssa_validity(self, func: Function) -> None: rule="ssa-validity", ) else: - assigned[instr.dest.name] = i + assigned.add(instr.dest.name) # ------------------------------------------------------------------- # Helper diff --git a/scratchv/compiler.py b/scratchv/compiler.py index a3484d2..365f647 100644 --- a/scratchv/compiler.py +++ b/scratchv/compiler.py @@ -53,6 +53,7 @@ class CompilerConfig: cycle_stats: Run 5-stage pipeline cycle estimation (detailed). enable_forwarding: Enable forwarding in cycle estimator. branch_predictor: Branch predictor mode for cycle estimator. + verify_ir: Validate IR throughout the compilation pipeline. """ backend: str = "riscv" @@ -73,6 +74,7 @@ class CompilerConfig: cycle_stats: bool = False enable_forwarding: bool = True branch_predictor: str = "always_not_taken" + verify_ir: bool = False # ═══════════════════════════════════════════════════════════════════════════════ @@ -253,15 +255,30 @@ def compile(self, input_path: str, output_path: str | None = None, from scratchv.ir.printer import IRPrinter ir_dump_before = IRPrinter(program).dump() - # --- 2. Verify IR (if configured) --- - if self.config.use_logger: - self._verify_ir(program, warnings) + # --- 2. Verify parsed IR (if configured) --- + if self.config.verify_ir: + verify_result = self._verify_ir(program, "after parsing") + warnings.extend(verify_result.warnings) + if not verify_result.success: + return CompileResult( + success=False, + errors=[verify_result.message], + warnings=warnings, + ) # --- 3. Optimize --- opt_message = "" if self.config.optimize_level != "none": opt_result = self._run_optimizations(program) opt_message = opt_result.message + warnings.extend(opt_result.warnings) + if not opt_result.success: + return CompileResult( + success=False, + errors=[opt_result.message], + warnings=warnings, + ) + program = opt_result.data ir_dump_after = "" if self.config.dump_ir: @@ -277,13 +294,27 @@ def compile(self, input_path: str, output_path: str | None = None, ") ---\n" + ir_dump_after ) - # --- 4. Code generation --- + # --- 4. Verify final IR, then generate code --- + if self.config.verify_ir: + verify_result = self._verify_ir( + program, "before code generation", + ) + warnings.extend(verify_result.warnings) + if not verify_result.success: + return CompileResult( + success=False, + errors=[verify_result.message], + warnings=warnings, + ir_dump=ir_dump, + ) + try: asm_text = self._generate_code(program) except Exception as e: return CompileResult( success=False, errors=[f"Codegen error: {e}"], ir_dump=ir_dump, + warnings=warnings, ) # --- 5. Post-codegen passes --- @@ -347,17 +378,9 @@ def _parse(self, input_path: str, dsl_source: str | None = None): # ── Internal: verify IR ───────────────────────────────────────────────── - def _verify_ir(self, program, warnings: list[str]) -> None: - """Run IR verifier and collect warnings.""" - from scratchv.analysis.ir_verifier import IRVerifier - verifier = IRVerifier(program) - issues = verifier.verify() - for issue in issues: - msg = str(issue) - if issue.level.value == "error": - warnings.append(f"IR: {msg}") - else: - warnings.append(f"IR(warning): {msg}") + def _verify_ir(self, program, phase: str) -> PassResult: + """Run the same verifier pass used between optimizations.""" + return _IRVerificationPass(phase).run(program) # ── Internal: optimizations ───────────────────────────────────────────── @@ -367,17 +390,23 @@ def _run_optimizations(self, program) -> PassResult: from scratchv.optimizer.dead_code import DeadCodeEliminator pm = PassManager("optimizer") - pm.add(_PassAdapter("constant-folding", ConstantFolder(program))) - pm.add(_PassAdapter("dead-code-elim", DeadCodeEliminator(program))) + + def add_optimizer(name: str, legacy_pass: Any) -> None: + pm.add(_PassAdapter(name, legacy_pass)) + if self.config.verify_ir: + pm.add(_IRVerificationPass(f"after pass '{name}'")) + + add_optimizer("constant-folding", ConstantFolder(program)) + add_optimizer("dead-code-elim", DeadCodeEliminator(program)) if self.config.optimize_level == "all": from scratchv.optimizer.peephole import IRPeepholeOptimizer from scratchv.optimizer.muladd_fusion import MulAddFusion from scratchv.optimizer.licm import LICM - pm.add(_PassAdapter("ir-peephole", IRPeepholeOptimizer(program))) - pm.add(_PassAdapter("muladd-fusion", MulAddFusion(program))) - pm.add(_PassAdapter("licm", LICM(program))) + add_optimizer("ir-peephole", IRPeepholeOptimizer(program)) + add_optimizer("muladd-fusion", MulAddFusion(program)) + add_optimizer("licm", LICM(program)) return pm.run(program) @@ -486,6 +515,38 @@ def _run_asm_passes(self, asm_text: str, warnings: list[str]) -> str: # _PassAdapter — wraps legacy passes that don't implement CompilerPass # ═══════════════════════════════════════════════════════════════════════════════ +class _IRVerificationPass(CompilerPass): + """Validate IR at a named pipeline boundary.""" + + def __init__(self, phase: str): + self._phase = phase + + @property + def name(self) -> str: + return f"verify-ir {self._phase}" + + def run(self, input_data: Any) -> PassResult: + from scratchv.analysis.ir_verifier import ErrorLevel, IRVerifier + + issues = IRVerifier(input_data).verify() + errors = [issue for issue in issues if issue.level == ErrorLevel.ERROR] + warnings = [ + f"IR verification {self._phase}: {issue}" + for issue in issues + if issue.level == ErrorLevel.WARNING + ] + if errors: + details = "\n".join(str(error) for error in errors) + return PassResult( + data=None, + message=( + f"IR verification failed {self._phase}:\n{details}" + ), + warnings=warnings, + ) + return PassResult(data=input_data, warnings=warnings) + + class _PassAdapter(CompilerPass): """Adapter that wraps a legacy pass object into the ``CompilerPass`` API. diff --git a/scratchv/frontend/dsl_parser.py b/scratchv/frontend/dsl_parser.py index 19aa04b..3cdc3a9 100644 --- a/scratchv/frontend/dsl_parser.py +++ b/scratchv/frontend/dsl_parser.py @@ -102,9 +102,11 @@ def _resolve(self, name: str) -> Value: return self.builder.load_const(val) except ValueError: pass - # Create a variable on first access + # A name first seen as an operand is a function input. v = self.builder.make_value(name=name) self._vars[name] = v + if self.builder.current_func is not None: + self.builder.current_func.params.append(v) return v def _parse_kwargs( diff --git a/scratchv/frontend/onnx_parser.py b/scratchv/frontend/onnx_parser.py index 222b403..64c701e 100644 --- a/scratchv/frontend/onnx_parser.py +++ b/scratchv/frontend/onnx_parser.py @@ -56,6 +56,7 @@ def parse(self, model_path: str) -> Program: # Multi-element tensor: store pointer info in attrs val.is_constant = False val.shape = tuple(arr.shape) + self.builder.program.global_values.append(val) self._value_map[init.name] = val # Map graph inputs to function params diff --git a/scratchv/main.py b/scratchv/main.py index 25eaf33..5224bbd 100644 --- a/scratchv/main.py +++ b/scratchv/main.py @@ -74,7 +74,7 @@ def build_arg_parser() -> argparse.ArgumentParser: ) parser.add_argument( "--verify-ir", action="store_true", - help="Run IR verifier before and after optimization (Topic 21)", + help="Run IR verifier throughout the compiler pipeline (Topic 21)", ) parser.add_argument( "--beautify", action="store_true", @@ -153,6 +153,7 @@ def args_to_config(args: argparse.Namespace) -> CompilerConfig: cycle_stats=args.cycle_stats, enable_forwarding=not args.no_forwarding, branch_predictor=args.branch_predictor, + verify_ir=args.verify_ir, ) diff --git a/tests/test_ir_verifier.py b/tests/test_ir_verifier.py index 62e5c75..4990c2e 100644 --- a/tests/test_ir_verifier.py +++ b/tests/test_ir_verifier.py @@ -1,19 +1,94 @@ """Tests for the IR verifier module.""" +from __future__ import annotations + +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Optional + +import pytest + from scratchv.frontend.dsl_parser import DSLParser from scratchv.frontend.dsl_extended import ExtendedDSLParser from scratchv.analysis.ir_verifier import ( IRVerifier, VerificationError, ErrorLevel, verify_ir, ) +from scratchv.compiler import CompilerConfig, CompilerDriver from scratchv.ir.types import ( - Program, Function, Instruction, OpCode, Value, DataType, + Program, Function, Instruction, BasicBlock, OpCode, Value, DataType, ) +from scratchv.main import args_to_config, build_arg_parser +from scratchv.pass_interface import PassResult + + +class StubCompilerDriver(CompilerDriver): + """Compiler driver with deterministic parsing and code generation.""" + + def __init__(self, config: CompilerConfig, program: Program) -> None: + super().__init__(config) + self.program = program + self.codegen_called = False + + def _parse( + self, input_path: str, dsl_source: Optional[str] = None, + ) -> Program: + return self.program + + def _generate_code(self, program: Program) -> str: + self.codegen_called = True + return "stub assembly\n" + + +def make_valid_program() -> Program: + program = Program() + param = Value(name="x") + func = Function(name="main", params=[param]) + program.add_function(func) + block = func.new_block("entry") + block.add(Instruction(opcode=OpCode.RETURN, operands=[param])) + return program + + +def make_undefined_program() -> Program: + program = Program() + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + block.add(Instruction( + opcode=OpCode.RETURN, + operands=[Value(name="missing")], + )) + return program + + +def make_warning_only_program() -> Program: + program = Program() + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + left = Value( + name="left", dtype=DataType.FLOAT32, + is_constant=True, const_value=1.0, + ) + right = Value( + name="right", dtype=DataType.INT32, + is_constant=True, const_value=2, + ) + result = Value(name="result") + block.add(Instruction( + opcode=OpCode.ADD, + dest=result, + operands=[left, right], + )) + block.add(Instruction(opcode=OpCode.RETURN, operands=[result])) + return program class TestVerificationError: """Tests for VerificationError dataclass.""" - def test_create_error(self): + def test_create_error(self) -> None: err = VerificationError( level=ErrorLevel.ERROR, message="value used before definition", @@ -27,7 +102,7 @@ def test_create_error(self): assert "main" in err.function_name or err.function_name == "main" assert err.rule == "def-before-use" - def test_create_warning(self): + def test_create_warning(self) -> None: err = VerificationError( level=ErrorLevel.WARNING, message="type mismatch", @@ -35,7 +110,7 @@ def test_create_warning(self): ) assert err.level == ErrorLevel.WARNING - def test_str_representation(self): + def test_str_representation(self) -> None: err = VerificationError( level=ErrorLevel.ERROR, message="test message", @@ -55,7 +130,7 @@ class TestIRVerifier: # Simple valid programs # ------------------------------------------------------------------ - def test_valid_simple_program(self): + def test_valid_simple_program(self) -> None: dsl = """ c = add(a, b) return c @@ -65,8 +140,11 @@ def test_valid_simple_program(self): verifier = IRVerifier(program) errors = verifier.verify() assert len(errors) == 0 # Should be valid + assert {value.name for value in program.functions[0].params} == { + "a", "b", + } - def test_valid_nn_pipeline(self): + def test_valid_nn_pipeline(self) -> None: dsl = """ t1 = relu(x) t2 = softmax(t1, axis:-1) @@ -82,15 +160,14 @@ def test_valid_nn_pipeline(self): # def-before-use # ------------------------------------------------------------------ - def test_def_before_use_implicit_input(self): - """Value without definition is treated as implicit input.""" + def test_function_parameter_is_defined_at_entry(self) -> None: + """Function parameters are available before the first instruction.""" program = Program() - func = Function(name="main") + v_x = Value(name="x", dtype=DataType.FLOAT32) + func = Function(name="main", params=[v_x]) program.add_function(func) block = func.new_block("entry") - v_x = Value(name="x", dtype=DataType.FLOAT32) - # Use x without defining it -- treated as implicit input by verifier use_instr = Instruction( opcode=OpCode.ADD, dest=Value(name="c"), @@ -105,16 +182,165 @@ def test_def_before_use_implicit_input(self): verifier = IRVerifier(program) errors = verifier.verify() - # Verifier treats undefined values as implicit inputs (not errors) - # Only block-termination check applies here def_errors = [e for e in errors if e.rule == "def-before-use"] - assert len(def_errors) == 0 # implicit inputs are allowed + assert def_errors == [] + + def test_undefined_value_is_rejected(self) -> None: + """A non-constant operand must be declared or produced.""" + program = Program() + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + missing = Value(name="missing") + result = Value(name="result") + block.add(Instruction( + opcode=OpCode.ADD, + dest=result, + operands=[missing, missing], + )) + block.add(Instruction(opcode=OpCode.RETURN, operands=[result])) + + errors = IRVerifier(program).verify() + def_errors = [e for e in errors if e.rule == "def-before-use"] + + assert len(def_errors) == 2 + assert all(e.value_name == "missing" for e in def_errors) + + def test_use_before_later_definition_is_rejected(self) -> None: + """A definition later in the block cannot satisfy an earlier use.""" + program = Program() + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + value = Value(name="later") + result = Value(name="result") + one = Value(name="one", is_constant=True, const_value=1.0) + block.add(Instruction( + opcode=OpCode.ADD, + dest=result, + operands=[value, one], + )) + block.add(Instruction( + opcode=OpCode.ADD, + dest=value, + operands=[one, one], + )) + block.add(Instruction(opcode=OpCode.RETURN, operands=[result])) + + errors = IRVerifier(program).verify() + def_errors = [e for e in errors if e.rule == "def-before-use"] + + assert len(def_errors) == 1 + assert def_errors[0].instruction_index == 0 + assert def_errors[0].value_name == "later" + + def test_definition_must_dominate_cross_block_use(self) -> None: + """A value defined on one branch is unavailable at the merge.""" + program = Program() + func = Function(name="main") + program.add_function(func) + entry = func.new_block("entry") + left = func.new_block("left") + right = func.new_block("right") + merge = func.new_block("merge") + cond = Value(name="cond", is_constant=True, const_value=1) + one = Value(name="one", is_constant=True, const_value=1.0) + branch_value = Value(name="branch_value") + + entry.add(Instruction( + opcode=OpCode.BR_IF, + operands=[cond], + target="left,right", + )) + left.add(Instruction( + opcode=OpCode.ADD, + dest=branch_value, + operands=[one, one], + )) + left.add(Instruction(opcode=OpCode.BR, target="merge")) + right.add(Instruction(opcode=OpCode.BR, target="merge")) + merge.add(Instruction( + opcode=OpCode.RETURN, + operands=[branch_value], + )) + + errors = IRVerifier(program).verify() + def_errors = [e for e in errors if e.rule == "def-before-use"] + + assert len(def_errors) == 1 + assert def_errors[0].block_name == "merge" + assert def_errors[0].value_name == "branch_value" + + def test_entry_definition_dominates_successor_use(self) -> None: + """A value defined in entry is available in a successor block.""" + program = Program() + func = Function(name="main") + program.add_function(func) + entry = func.new_block("entry") + exit_block = func.new_block("exit") + one = Value(name="one", is_constant=True, const_value=1.0) + value = Value(name="value") + entry.add(Instruction( + opcode=OpCode.ADD, + dest=value, + operands=[one, one], + )) + entry.add(Instruction(opcode=OpCode.BR, target="exit")) + exit_block.add(Instruction(opcode=OpCode.RETURN, operands=[value])) + + errors = IRVerifier(program).verify() + def_errors = [e for e in errors if e.rule == "def-before-use"] + + assert def_errors == [] + + def test_empty_block_falls_through_for_dominance(self) -> None: + """Empty warning-only blocks preserve sequential CFG flow.""" + program = Program() + func = Function(name="main") + program.add_function(func) + entry = func.new_block("entry") + func.new_block("gap") + exit_block = func.new_block("exit") + one = Value(name="one", is_constant=True, const_value=1.0) + value = Value(name="value") + entry.add(Instruction( + opcode=OpCode.ADD, + dest=value, + operands=[one, one], + )) + entry.add(Instruction(opcode=OpCode.BR, target="gap")) + exit_block.add(Instruction(opcode=OpCode.RETURN, operands=[value])) + + passed, errors = verify_ir(program) + + assert passed is True + assert [e for e in errors if e.rule == "def-before-use"] == [] + + def test_program_global_is_defined_at_entry(self) -> None: + """Program globals are available to every function.""" + program = Program() + weight = Value(name="weight") + program.global_values.append(weight) + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + result = Value(name="result") + block.add(Instruction( + opcode=OpCode.ADD, + dest=result, + operands=[weight, weight], + )) + block.add(Instruction(opcode=OpCode.RETURN, operands=[result])) + + errors = IRVerifier(program).verify() + + assert [e for e in errors if e.rule == "def-before-use"] == [] # ------------------------------------------------------------------ # Block termination # ------------------------------------------------------------------ - def test_block_termination_missing(self): + def test_block_termination_missing(self) -> None: """Block without terminator should error.""" program = Program() func = Function(name="main") @@ -133,7 +359,7 @@ def test_block_termination_missing(self): term_errors = [e for e in errors if e.rule == "block-termination"] assert len(term_errors) >= 1 - def test_empty_block_warning(self): + def test_empty_block_warning(self) -> None: """Empty block should be a warning.""" program = Program() func = Function(name="main") @@ -149,7 +375,7 @@ def test_empty_block_warning(self): # Label existence # ------------------------------------------------------------------ - def test_label_existence(self): + def test_label_existence(self) -> None: """Branch to nonexistent label should error.""" program = Program() func = Function(name="main") @@ -167,11 +393,59 @@ def test_label_existence(self): label_errors = [e for e in errors if e.rule == "label-existence"] assert len(label_errors) >= 1 + def test_branch_requires_a_target(self) -> None: + """A branch without a target is not a valid label reference.""" + program = Program() + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + block.add(Instruction(opcode=OpCode.BR)) + + errors = IRVerifier(program).verify() + label_errors = [e for e in errors if e.rule == "label-existence"] + + assert len(label_errors) == 1 + + def test_label_check_ignores_non_branch_target_metadata(self) -> None: + """Only control-flow instructions carry block label targets.""" + program = Program() + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + one = Value(name="one", is_constant=True, const_value=1.0) + block.add(Instruction( + opcode=OpCode.ADD, + dest=Value(name="result"), + operands=[one, one], + target="not-a-block", + )) + block.add(Instruction(opcode=OpCode.RETURN)) + + errors = IRVerifier(program).verify() + + assert [e for e in errors if e.rule == "label-existence"] == [] + + def test_duplicate_block_labels_are_rejected(self) -> None: + """A branch label must identify exactly one basic block.""" + program = Program() + func = Function(name="main") + program.add_function(func) + first = BasicBlock("duplicate") + second = BasicBlock("duplicate") + first.add(Instruction(opcode=OpCode.RETURN)) + second.add(Instruction(opcode=OpCode.RETURN)) + func.blocks.extend([first, second]) + + errors = IRVerifier(program).verify() + label_errors = [e for e in errors if e.rule == "label-existence"] + + assert len(label_errors) == 1 + # ------------------------------------------------------------------ # Type consistency # ------------------------------------------------------------------ - def test_type_consistency_warning(self): + def test_type_consistency_warning(self) -> None: """Operands with different types should be a warning.""" program = Program() func = Function(name="main") @@ -201,11 +475,38 @@ def test_type_consistency_warning(self): # WARNING, not ERROR assert all(e.level == ErrorLevel.WARNING for e in type_errors) + def test_nn_op_checks_every_operand_type(self) -> None: + """A mismatched third NN operand also produces a warning.""" + program = Program() + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + inputs = [ + Value(name="x", is_constant=True, const_value=1.0), + Value(name="w", is_constant=True, const_value=2.0), + Value( + name="bias", dtype=DataType.INT32, + is_constant=True, const_value=3, + ), + ] + block.add(Instruction( + opcode=OpCode.CONV, + dest=Value(name="result"), + operands=inputs, + )) + block.add(Instruction(opcode=OpCode.RETURN)) + + errors = IRVerifier(program).verify() + type_errors = [e for e in errors if e.rule == "type-consistency"] + + assert len(type_errors) == 1 + assert type_errors[0].level == ErrorLevel.WARNING + # ------------------------------------------------------------------ # Control flow integrity # ------------------------------------------------------------------ - def test_control_flow_unreachable_after_br(self): + def test_control_flow_unreachable_after_br(self) -> None: """Instructions after unconditional branch should error.""" program = Program() func = Function(name="main") @@ -229,7 +530,7 @@ def test_control_flow_unreachable_after_br(self): cf_errors = [e for e in errors if e.rule == "control-flow-integrity"] assert len(cf_errors) >= 1 - def test_br_if_target_count(self): + def test_br_if_target_count(self) -> None: """BR_IF must have exactly 2 targets.""" program = Program() func = Function(name="main") @@ -248,11 +549,74 @@ def test_br_if_target_count(self): cf_errors = [e for e in errors if e.rule == "control-flow-integrity"] assert len(cf_errors) >= 1 + def test_control_flow_unreachable_after_br_if(self) -> None: + """A conditional branch is also a block terminator.""" + program = Program() + func = Function(name="main") + program.add_function(func) + entry = func.new_block("entry") + true_block = func.new_block("true") + false_block = func.new_block("false") + cond = Value(name="cond", is_constant=True, const_value=1) + entry.add(Instruction( + opcode=OpCode.BR_IF, + operands=[cond], + target="true,false", + )) + entry.add(Instruction(opcode=OpCode.RETURN)) + true_block.add(Instruction(opcode=OpCode.RETURN)) + false_block.add(Instruction(opcode=OpCode.RETURN)) + + errors = IRVerifier(program).verify() + cf_errors = [e for e in errors if e.rule == "control-flow-integrity"] + + assert len(cf_errors) == 1 + + def test_br_if_rejects_an_empty_target(self) -> None: + """Two comma slots are invalid when either label is empty.""" + program = Program() + func = Function(name="main") + program.add_function(func) + entry = func.new_block("entry") + target = func.new_block("target") + cond = Value(name="cond", is_constant=True, const_value=1) + entry.add(Instruction( + opcode=OpCode.BR_IF, + operands=[cond], + target="target,", + )) + target.add(Instruction(opcode=OpCode.RETURN)) + + errors = IRVerifier(program).verify() + + assert [e for e in errors if e.rule == "label-existence"] + assert [e for e in errors if e.rule == "control-flow-integrity"] + + def test_br_if_requires_a_condition_operand(self) -> None: + """A conditional branch without a condition cannot be selected.""" + program = Program() + func = Function(name="main") + program.add_function(func) + entry = func.new_block("entry") + left = func.new_block("left") + right = func.new_block("right") + entry.add(Instruction( + opcode=OpCode.BR_IF, + target="left,right", + )) + left.add(Instruction(opcode=OpCode.RETURN)) + right.add(Instruction(opcode=OpCode.RETURN)) + + errors = IRVerifier(program).verify() + cf_errors = [e for e in errors if e.rule == "control-flow-integrity"] + + assert len(cf_errors) == 1 + # ------------------------------------------------------------------ # Entry existence # ------------------------------------------------------------------ - def test_entry_existence(self): + def test_entry_existence(self) -> None: """Function with no blocks should error.""" program = Program() func = Function(name="main") # no blocks @@ -267,7 +631,7 @@ def test_entry_existence(self): # SSA validity # ------------------------------------------------------------------ - def test_ssa_validity(self): + def test_ssa_validity(self) -> None: """Multiple assignments to same value should error.""" program = Program() func = Function(name="main") @@ -295,11 +659,62 @@ def test_ssa_validity(self): ssa_errors = [e for e in errors if e.rule == "ssa-validity"] assert len(ssa_errors) >= 1 + def test_ssa_rejects_redefining_a_parameter(self) -> None: + """Function parameters count as their value's SSA definition.""" + program = Program() + param = Value(name="x") + func = Function(name="main", params=[param]) + program.add_function(func) + block = func.new_block("entry") + one = Value(name="one", is_constant=True, const_value=1.0) + block.add(Instruction( + opcode=OpCode.ADD, + dest=Value(name="x"), + operands=[one, one], + )) + block.add(Instruction(opcode=OpCode.RETURN)) + + errors = IRVerifier(program).verify() + ssa_errors = [e for e in errors if e.rule == "ssa-validity"] + + assert len(ssa_errors) == 1 + + def test_ssa_rejects_duplicate_parameters(self) -> None: + """Two entry definitions cannot share the same SSA name.""" + program = Program() + func = Function( + name="main", + params=[Value(name="x"), Value(name="x")], + ) + program.add_function(func) + block = func.new_block("entry") + block.add(Instruction(opcode=OpCode.RETURN)) + + errors = IRVerifier(program).verify() + ssa_errors = [e for e in errors if e.rule == "ssa-validity"] + + assert len(ssa_errors) == 1 + assert ssa_errors[0].value_name == "x" + + def test_ssa_rejects_duplicate_globals_without_functions(self) -> None: + """Program globals are checked even when the program has no function.""" + program = Program() + program.global_values.extend([ + Value(name="weight"), + Value(name="weight"), + ]) + + errors = IRVerifier(program).verify() + ssa_errors = [e for e in errors if e.rule == "ssa-validity"] + + assert len(ssa_errors) == 1 + assert ssa_errors[0].value_name == "weight" + # ------------------------------------------------------------------ # Convenience function # ------------------------------------------------------------------ - def test_verify_ir_function(self): + def test_verify_ir_function(self) -> None: dsl = """ c = add(a, b) return c @@ -310,11 +725,38 @@ def test_verify_ir_function(self): assert passed is True assert len(errors) == 0 + def test_verify_ir_allows_warning_only_program(self) -> None: + """Warnings are reported but do not fail the convenience API.""" + program = Program() + func = Function(name="main") + program.add_function(func) + block = func.new_block("entry") + left = Value( + name="left", dtype=DataType.FLOAT32, + is_constant=True, const_value=1.0, + ) + right = Value( + name="right", dtype=DataType.INT32, + is_constant=True, const_value=2, + ) + block.add(Instruction( + opcode=OpCode.ADD, + dest=Value(name="result"), + operands=[left, right], + )) + block.add(Instruction(opcode=OpCode.RETURN)) + + passed, errors = verify_ir(program) + + assert passed is True + assert len(errors) == 1 + assert errors[0].level == ErrorLevel.WARNING + class TestIRVerifierWithExtendedParser: """Verify IR generated by the extended parser.""" - def test_if_else_ir_valid(self): + def test_if_else_without_phi_is_rejected(self) -> None: dsl = """ if (a > b): c = add(a, b) @@ -327,11 +769,12 @@ def test_if_else_ir_valid(self): program = parser.parse(dsl) verifier = IRVerifier(program) errors = verifier.verify() - # The extended parser should generate valid IR - real_errors = [e for e in errors if e.level == ErrorLevel.ERROR] - assert len(real_errors) == 0, f"IR verification failed: {real_errors}" + def_errors = [e for e in errors if e.rule == "def-before-use"] - def test_while_ir_valid(self): + assert len(def_errors) == 1 + assert def_errors[0].block_name == "if_end3" + + def test_while_without_phi_is_rejected(self) -> None: dsl = """ while (i < 10): acc = add(acc, x) @@ -342,5 +785,220 @@ def test_while_ir_valid(self): program = parser.parse(dsl) verifier = IRVerifier(program) errors = verifier.verify() - real_errors = [e for e in errors if e.level == ErrorLevel.ERROR] - assert len(real_errors) == 0, f"IR verification failed: {real_errors}" + def_errors = [e for e in errors if e.rule == "def-before-use"] + + assert len(def_errors) == 1 + assert def_errors[0].block_name == "while_exit3" + + +class TestIRVerifierCompilerPipeline: + """Exercise IR verification through the public compiler entry point.""" + + def test_cli_flag_is_mapped_to_compiler_config(self) -> None: + parser = build_arg_parser() + + enabled = args_to_config(parser.parse_args([ + "input.dsl", "--verify-ir", + ])) + disabled = args_to_config(parser.parse_args(["input.dsl"])) + + assert enabled.verify_ir is True + assert disabled.verify_ir is False + + def test_error_after_parse_stops_the_pipeline( + self, tmp_path: Path, + ) -> None: + config = CompilerConfig(verify_ir=True, optimize_level="basic") + driver = StubCompilerDriver(config, make_undefined_program()) + output = tmp_path / "output.s" + + result = driver.compile("input.dsl", str(output)) + + assert result.success is False + assert driver.codegen_called is False + assert output.exists() is False + assert "after parsing" in result.errors[0] + assert "def-before-use" in result.errors[0] + + def test_logger_does_not_enable_ir_verification( + self, tmp_path: Path, + ) -> None: + config = CompilerConfig(use_logger=True, verify_ir=False) + driver = StubCompilerDriver(config, make_undefined_program()) + + result = driver.compile( + "input.dsl", str(tmp_path / "output.s"), + ) + + assert result.success is True + assert driver.codegen_called is True + + @pytest.mark.parametrize( + ("optimize_level", "expected_checks"), + [("basic", 4), ("all", 7)], + ) + def test_verifies_after_every_optimization_pass( + self, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + optimize_level: str, + expected_checks: int, + ) -> None: + original_verify = IRVerifier.verify + checks = [] + + def counted_verify( + verifier: IRVerifier, + ) -> list[VerificationError]: + checks.append(verifier.program) + return original_verify(verifier) + + monkeypatch.setattr(IRVerifier, "verify", counted_verify) + config = CompilerConfig( + verify_ir=True, + optimize_level=optimize_level, + ) + driver = StubCompilerDriver(config, make_valid_program()) + + result = driver.compile( + "input.dsl", str(tmp_path / "output.s"), + ) + + assert result.success is True + assert len(checks) == expected_checks + + def test_error_introduced_by_a_pass_stops_immediately( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + ) -> None: + from scratchv.optimizer.constant_folding import ConstantFolder + from scratchv.optimizer.dead_code import DeadCodeEliminator + + dead_code_called = False + + def corrupt_ir(folder: ConstantFolder) -> int: + return_instr = ( + folder.program.functions[0].blocks[0].instructions[-1] + ) + return_instr.operands = [Value(name="missing")] + return 1 + + def track_dead_code(_eliminator: DeadCodeEliminator) -> int: + nonlocal dead_code_called + dead_code_called = True + return 0 + + monkeypatch.setattr(ConstantFolder, "run", corrupt_ir) + monkeypatch.setattr(DeadCodeEliminator, "run", track_dead_code) + config = CompilerConfig(verify_ir=True, optimize_level="basic") + driver = StubCompilerDriver(config, make_valid_program()) + + result = driver.compile( + "input.dsl", str(tmp_path / "output.s"), + ) + + assert result.success is False + assert dead_code_called is False + assert driver.codegen_called is False + assert "constant-folding" in result.errors[0] + assert "def-before-use" in result.errors[0] + + def test_verifies_again_before_codegen(self, tmp_path: Path) -> None: + class CorruptingDriver(StubCompilerDriver): + def _run_optimizations(self, program: Program) -> PassResult: + return_instr = program.functions[0].blocks[0].instructions[-1] + return_instr.operands = [Value(name="missing")] + return PassResult(data=program) + + config = CompilerConfig(verify_ir=True, optimize_level="basic") + driver = CorruptingDriver(config, make_valid_program()) + + result = driver.compile( + "input.dsl", str(tmp_path / "output.s"), + ) + + assert result.success is False + assert driver.codegen_called is False + assert "before code generation" in result.errors[0] + + def test_warnings_are_reported_without_stopping_codegen( + self, tmp_path: Path, + ) -> None: + config = CompilerConfig(verify_ir=True) + driver = StubCompilerDriver(config, make_warning_only_program()) + + result = driver.compile( + "input.dsl", str(tmp_path / "output.s"), + ) + + assert result.success is True + assert driver.codegen_called is True + assert any( + "type-consistency" in warning + for warning in result.warnings + ) + + def test_codegen_failure_preserves_ir_warnings( + self, tmp_path: Path, + ) -> None: + class FailingCodegenDriver(StubCompilerDriver): + def _generate_code(self, program: Program) -> str: + raise RuntimeError("boom") + + config = CompilerConfig(verify_ir=True) + driver = FailingCodegenDriver(config, make_warning_only_program()) + + result = driver.compile( + "input.dsl", str(tmp_path / "output.s"), + ) + + assert result.success is False + assert "Codegen error: boom" in result.errors + assert any( + "type-consistency" in warning + for warning in result.warnings + ) + + +class TestIRVerifierFrontendInputs: + """Verify that frontends declare external IR values explicitly.""" + + def test_onnx_tensor_initializer_is_a_program_global( + self, monkeypatch: pytest.MonkeyPatch, + ) -> None: + tensor_type = SimpleNamespace( + elem_type=1, + shape=SimpleNamespace(dim=[]), + ) + graph = SimpleNamespace( + name="main", + initializer=[SimpleNamespace( + name="weight", + data_type=1, + array=SimpleNamespace(size=2, shape=(2,)), + )], + input=[SimpleNamespace( + name="x", + type=SimpleNamespace(tensor_type=tensor_type), + )], + output=[SimpleNamespace(name="y")], + node=[SimpleNamespace( + op_type="Add", + input=["x", "weight"], + output=["y"], + )], + ) + fake_onnx = SimpleNamespace( + load=lambda _path: SimpleNamespace(graph=graph), + numpy_helper=SimpleNamespace( + to_array=lambda initializer: initializer.array, + ), + ) + monkeypatch.setitem(sys.modules, "onnx", fake_onnx) + + from scratchv.frontend.onnx_parser import ONNXParser + program = ONNXParser().parse("model.onnx") + passed, errors = verify_ir(program) + + assert [value.name for value in program.global_values] == ["weight"] + assert passed is True + assert errors == []