32. Executing SSA, Destroying Phis, and Allocating Registers
A verified optimized graph still needs an execution strategy. This chapter implements three: an SSA interpreter for debugging, a constrained teaching machine for register allocation, and a native C backend. Each exposes different errors and delegates different responsibilities.
32.1 Phi evaluation is simultaneous
On entry to a block, select each phi operand using the predecessor actually taken. Read all selected
values before writing any phi result. For a loop that swaps a and b, sequentially assigning a=b
and then b=a loses the old a. A dictionary of pending results followed by one environment update
preserves the parallel semantics.
The interpreter bounds-checks accesses before indexing so Python's negative-index convention cannot leak into the language. It normalizes loaded and stored integers to i32 and executes instructions in source-effect order. A block-step budget catches nontermination during experiments; this is a debugging facility, not a proof of termination.
"""Execute mutable or SSA IR with ordered memory effects and simultaneous phi updates."""
from minicc.analysis import verify
from minicc.ir import Kind, alu, i32
def bind(function, arguments):
if len(arguments) != len(function.parameters): raise ValueError("wrong argument count")
result = {}
for parameter, value in zip(function.parameters, arguments):
if parameter.kind == Kind.I32:
if not isinstance(value, int): raise ValueError("scalar parameters require Python integers")
result[parameter.name] = i32(value)
else:
if not hasattr(value, "__len__") or not hasattr(value, "__getitem__") or not hasattr(value, "__setitem__"):
raise ValueError("buffer parameter requires mutable indexed storage")
result[parameter.name] = value
return result
def memory(op, args):
buffer, index = args[:2]
if not 0 <= index < len(buffer): raise IndexError(f"buffer index {index} outside [0,{len(buffer)})")
if op == "load": return i32(buffer[index])
buffer[index] = i32(args[2])
return None
def execute(function, *arguments, max_steps=100000):
verify(function)
env = bind(function, arguments)
current, previous = function.entry, None
def read(value): return value if isinstance(value, int) else env[value]
for _ in range(max_steps):
block = function.blocks[current]
incoming = {phi.dest: read(phi.incoming[previous]) for phi in block.phis}
env.update(incoming)
for instruction in block.instructions:
args = [read(arg) for arg in instruction.args]
value = memory(instruction.op, args) if instruction.op in ("load", "store") else alu(instruction.op, args)
if instruction.dest is not None: env[instruction.dest] = value
term = block.terminator
if term.op == "return": return read(term.args[0])
target = term.targets[0] if term.op == "jump" else term.targets[0 if read(term.args[0]) else 1]
previous, current = current, target
raise RuntimeError("execution step limit exceeded")
Experiment: compile a loop initialized with a=1 and b=2 that swaps them each iteration. Return
10*a+b. Even iteration counts must return 12, odd counts 21. This small program exposes phi-copy
bugs that ordinary sums rarely detect.
32.2 Phi destruction belongs on edges
An SSA phi in block merge says which assignment happens for each incoming edge. Lower it to copies on those edges. Putting a predecessor-specific copy unconditionally at the end of a branching predecessor can change the value on the wrong path. Putting all copies at the merge loses the identity of the incoming edge.
The implementation creates an edge block for every incoming edge to a phi-bearing block, redirects that branch or jump, saves all phi sources into fresh temporaries, then writes all destinations. Saving every source is conservative but handles arbitrary copy cycles. A later parallel-copy scheduler could emit acyclic copies directly and use one scratch value only when breaking a cycle.
This transformation leaves the graph in a non-SSA form: the same destination may be assigned on several incoming edges. Re-run definite assignment and structural verification under that form. Do not incorrectly apply the single-definition verifier after intentionally destroying SSA.
32.3 Liveness gives an interference graph
At each program point, two distinct live values need different locations unless a separate value- equality argument permits sharing. Compute block live-in/live-out sets, then scan instructions backward. For a definition d, connect d to every value live after its definition, remove d from that live set, and add the instruction's operands.
Parameters are installed before entering the function. This implementation connects all scalar parameters to one another, including unused ones, so initialization cannot overwrite a live parameter. A better allocator could avoid binding dead parameters, but omitting both protections is incorrect. Buffers remain separate typed resources and never receive scalar-register colors.
Color the interference graph greedily in descending degree order. Pick an available register color; if none is available, assign a distinct spill slot. This policy is simple and deterministic but not optimal. It does not perform coalescing, live-range splitting, rematerialization, or spill-slot reuse. Those are subsequent compiler projects with measurable effects on this baseline.
32.4 Execute the allocation, not just inspect it
The teaching machine reads and writes the allocated physical locations. If the allocator assigns interfering values the same register, the computation is actually corrupted and the source-oracle test fails. A test that only checks the allocator's own graph would be weaker: the graph and allocator could agree on the same incorrect assumptions.
Spilled values live in a frame and their reads and writes are counted. The teaching instruction set permits spill-slot operands directly. A register-only hardware ISA would require explicit reload and spill instructions, scratch-register management, stack layout, and calling-convention support. The counts are algorithmic traffic, not measured CPU or GPU cycles.
"""Phi destruction, graph coloring, and a bounded-register interpreter with spills.
This teaching machine permits spill-slot operands directly. It counts their traffic,
not host CPU cycles. It is not a physical ISA encoder or a hardware cost model.
"""
from copy import deepcopy
from dataclasses import dataclass
from itertools import combinations
from minicc.analysis import liveness, verify
from minicc.interpreter import bind, memory
from minicc.ir import Block, Instruction, Kind, Terminator, alu, operands
def destroy_phis(original):
if not original.ssa: raise ValueError("phi destruction requires SSA")
verify(original)
function = deepcopy(original)
counter = 0
for parent, block in list(function.blocks.items()):
targets = list(block.terminator.targets)
for index, target in enumerate(targets):
phis = function.blocks[target].phis
if not phis: continue
counter += 1
edge = f"$edge{counter}"
reads, writes = [], []
# Save every source before writing any destination: parallel-copy semantics.
for i, phi in enumerate(phis):
temporary = f"$edge{counter}_copy{i}"
reads.append(Instruction("copy", temporary, (phi.incoming[parent],)))
writes.append(Instruction("copy", phi.dest, (temporary,)))
function.blocks[edge] = Block(reads + writes, Terminator("jump", targets=(target,)))
targets[index] = edge
block.terminator.targets = tuple(targets)
for block in function.blocks.values(): block.phis.clear()
function.ssa = False
verify(function)
return function
def interference(function):
if function.ssa or any(b.phis for b in function.blocks.values()): raise ValueError("destroy phis first")
verify(function)
_, outgoing = liveness(function)
buffers = {p.name for p in function.parameters if p.kind == Kind.BUFFER}
neighbors = {}
def ensure(values):
for value in values - buffers: neighbors.setdefault(value, set())
def connect(a, b):
if a == b or a in buffers or b in buffers: return
ensure({a, b})
neighbors[a].add(b); neighbors[b].add(a)
parameters = {p.name for p in function.parameters if p.kind == Kind.I32}
ensure(parameters)
# Arguments are installed together before entry. Protect even unused parameters
# from clobbering another parameter during that initial binding.
for a, b in combinations(sorted(parameters), 2): connect(a, b)
for name, block in function.blocks.items():
live = (outgoing[name] | operands(block.terminator.args)) - buffers
ensure(live)
for instruction in reversed(block.instructions):
if instruction.dest is not None:
ensure({instruction.dest})
live.discard(instruction.dest)
for value in live: connect(instruction.dest, value)
live |= operands(instruction.args) - buffers
ensure(live)
return neighbors
@dataclass(frozen=True)
class Location:
kind: str
index: int
@dataclass
class Allocation:
function: object
locations: dict[str, Location]
registers: int
slots: int
neighbors: dict[str, set[str]]
def allocate(function, registers=4):
if registers < 1: raise ValueError("at least one register is required")
neighbors = interference(function)
locations, spills = {}, 0
# Deterministic largest-degree-first greedy coloring; not optimal coloring.
for value in sorted(neighbors, key=lambda v: (-len(neighbors[v]), v)):
unavailable = {locations[n].index for n in neighbors[value] if n in locations and locations[n].kind == "register"}
available = next((r for r in range(registers) if r not in unavailable), None)
if available is None:
locations[value] = Location("slot", spills)
spills += 1
else: locations[value] = Location("register", available)
allocation = Allocation(function, locations, registers, spills, neighbors)
for value, adjacent in neighbors.items():
for other in adjacent:
if locations[value] == locations[other]: raise ValueError("interfering values share storage")
return allocation
def execute_allocated(allocation, *arguments, max_steps=100000):
function = allocation.function
env = bind(function, arguments)
buffers = {p.name: env[p.name] for p in function.parameters if p.kind == Kind.BUFFER}
storage = {"register": [0]*allocation.registers, "slot": [0]*allocation.slots}
counts = {"spill_reads": 0, "spill_writes": 0, "instructions": 0}
def write(name, value):
location = allocation.locations[name]
storage[location.kind][location.index] = value
if location.kind == "slot": counts["spill_writes"] += 1
def read(value):
if isinstance(value, int): return value
if value in buffers: return buffers[value]
location = allocation.locations[value]
if location.kind == "slot": counts["spill_reads"] += 1
return storage[location.kind][location.index]
for parameter in function.parameters:
if parameter.kind == Kind.I32: write(parameter.name, env[parameter.name])
current = function.entry
for _ in range(max_steps):
block = function.blocks[current]
for instruction in block.instructions:
counts["instructions"] += 1
args = [read(value) for value in instruction.args]
result = memory(instruction.op, args) if instruction.op in ("load", "store") else alu(instruction.op, args)
if instruction.dest is not None: write(instruction.dest, result)
term = block.terminator
if term.op == "return": return read(term.args[0]), counts
current = term.targets[0] if term.op == "jump" else term.targets[0 if read(term.args[0]) else 1]
raise RuntimeError("execution step limit exceeded")
Experiment: run the same compiled function with register budgets 1, 2, 4, and 16. Check semantic agreement before comparing spill slots and traffic. Then implement a copy-propagation or coalescing pass and determine whether it reduces actual dynamic spill traffic, not only graph size.
32.5 Native C needs a declared numerical and memory contract
A naive C emitter prints signed a+b, but overflowing signed arithmetic is undefined in C. The
backend converts operands to uint32, computes modulo arithmetic there, and copies the resulting bit
pattern into int32 with memcpy. Comparisons still use signed operands. The minimum i32 literal is
emitted without depending on a positive 2147483648 having int type.
Every buffer parameter has a pointer and explicit length. Every access checks its signed index
before dereferencing. An error pointer communicates a trap to the wrapper, which raises an
IndexError after the native function returns. Earlier writes remain visible. Scalar results are
not mistaken for error codes; a legitimate result of zero remains distinguishable from failure.
The wrapper declares every ctypes argument type and return type. It accepts only aligned, writable,
contiguous rank-one int32 buffers. Rejecting a noncontiguous view avoids treating a logical stride
as physical adjacency. Writable buffers simplify the contract even for functions that only read;
relaxing that guard requires per-parameter effect information.
Phi assignments in C are again emitted with parallel-copy semantics on branch edges. All SSA scalar names receive safe generated C identifiers, so source identifiers cannot inject generated code. The original source is parsed structurally, never interpolated as C statements.
"""Render verified SSA as bounds-checked C and execute it through a declared ABI."""
import ctypes
import shutil
import subprocess
from pathlib import Path
from tempfile import TemporaryDirectory
import numpy as np
from minicc.analysis import verify
from minicc.ir import Kind
def render_c(function):
if not function.ssa: raise ValueError("C backend expects SSA")
verify(function)
names = {}
def symbol(name):
if name not in names: names[name] = f"v{len(names)}"
return names[name]
def operand(value):
if isinstance(value, int): return f"INT32_C({value})" if value != -(2**31) else "(-INT32_C(2147483647)-1)"
return symbol(value)
arguments = []
parameters = {p.name for p in function.parameters}
for parameter in function.parameters:
name = symbol(parameter.name)
arguments.append(f"int32_t *{name}" if parameter.kind == Kind.BUFFER else f"int32_t {name}")
if parameter.kind == Kind.BUFFER: arguments.append(f"int32_t {name}_length")
arguments.append("int32_t *error")
labels = {name: f"block_{i}" for i, name in enumerate(function.blocks)}
for block in function.blocks.values():
for phi in block.phis: symbol(phi.dest)
for instruction in block.instructions:
if instruction.dest is not None: symbol(instruction.dest)
lines = ["#include <stdint.h>", "#include <string.h>",
# Bit-preserving conversion avoids signed-overflow UB and avoids
# relying on an out-of-range unsigned-to-signed C cast.
"static int32_t signed_bits(uint32_t x) { int32_t y; memcpy(&y, &x, sizeof(y)); return y; }",
f"int32_t compiled({', '.join(arguments)}) {{", " *error = 0;"]
for name, c_name in names.items():
if name not in parameters: lines.append(f" int32_t {c_name};")
lines.append(f" goto {labels[function.entry]};")
def edge(parent, target, indent):
lines.append(indent + "{")
phis = function.blocks[target].phis
for i, phi in enumerate(phis): lines.append(indent + f" int32_t edge_{i} = {operand(phi.incoming[parent])};")
for i, phi in enumerate(phis): lines.append(indent + f" {symbol(phi.dest)} = edge_{i};")
lines.append(indent + f" goto {labels[target]};")
lines.append(indent + "}")
comparisons = {"lt": "<", "le": "<=", "eq": "==", "ne": "!="}
arithmetic = {"add": "+", "sub": "-", "mul": "*"}
for name, block in function.blocks.items():
lines.append(labels[name] + ": ;")
for instruction in block.instructions:
args = [operand(arg) for arg in instruction.args]
if instruction.op in ("load", "store"):
base, index = args[:2]
lines.append(f" if ({index} < 0 || {index} >= {base}_length) {{ *error = 1; return 0; }}")
statement = f"{base}[{index}] = {args[2]}" if instruction.op == "store" else f"{symbol(instruction.dest)} = {base}[{index}]"
else:
if instruction.op == "copy": expression = args[0]
elif instruction.op in comparisons: expression = f"({args[0]} {comparisons[instruction.op]} {args[1]})"
else: expression = f"signed_bits((uint32_t){args[0]} {arithmetic[instruction.op]} (uint32_t){args[1]})"
statement = f"{symbol(instruction.dest)} = {expression}"
lines.append(f" {statement};")
term = block.terminator
if term.op == "return": lines.append(f" return {operand(term.args[0])};")
elif term.op == "jump": edge(name, term.targets[0], " ")
else:
lines.append(f" if ({operand(term.args[0])})")
edge(name, term.targets[0], " ")
lines.append(" else")
edge(name, term.targets[1], " ")
lines.append("}")
return "\n".join(lines)
class NativeFunction:
def __init__(self, function):
compiler = shutil.which("gcc")
if compiler is None: raise RuntimeError("gcc is required for native execution")
self.function, self.source = function, render_c(function)
self.directory = TemporaryDirectory(prefix="minicc-")
source, library = Path(self.directory.name)/"kernel.c", Path(self.directory.name)/"kernel.so"
source.write_text(self.source)
subprocess.run([compiler, "-std=c11", "-O2", "-shared", "-fPIC", str(source), "-o", str(library)],
check=True, capture_output=True, text=True)
self.library = ctypes.CDLL(str(library))
self.call = self.library.compiled
types = []
for parameter in function.parameters:
types.extend([ctypes.POINTER(ctypes.c_int32), ctypes.c_int32] if parameter.kind == Kind.BUFFER else [ctypes.c_int32])
self.call.argtypes, self.call.restype = [*types, ctypes.POINTER(ctypes.c_int32)], ctypes.c_int32
def __call__(self, *arguments):
if len(arguments) != len(self.function.parameters): raise ValueError("wrong argument count")
native = []
for parameter, value in zip(self.function.parameters, arguments):
if parameter.kind == Kind.I32:
if not isinstance(value, int): raise ValueError("scalar parameters require Python integers")
native.append(ctypes.c_int32(value))
else:
if not isinstance(value, np.ndarray) or value.dtype != np.int32 or value.ndim != 1 or not value.flags.c_contiguous or not value.flags.aligned or not value.flags.writeable:
raise ValueError("native buffers require aligned, writable, contiguous rank-one int32 arrays")
if len(value) >= 2**31: raise ValueError("buffer exceeds i32 length domain")
native.extend([value.ctypes.data_as(ctypes.POINTER(ctypes.c_int32)), ctypes.c_int32(len(value))])
error = ctypes.c_int32()
result = self.call(*native, ctypes.byref(error))
if error.value: raise IndexError("compiled buffer access out of bounds")
return result
32.6 What these backends establish
The native backend delegates instruction selection, scheduling, register allocation, assembly, and linking to GCC. The register-machine backend exposes allocation but does not emit a physical ISA. Together they make two useful boundaries visible instead of pretending the C emitter performs every backend task.
Native execution has no interpreter step budget; do not run a possibly nonterminating test in-process
without a process timeout. The generated library and its temporary source remain owned by the
NativeFunction object for its lifetime. This is an experimental JIT, not a persistent executable
cache or a sandbox for adversarial programs.
Acceptance: execute a program through mutable IR, SSA, optimized SSA, the allocated machine, and native C. Compare returns, final buffers, and failures against independent expected behavior. Keep the generated C when diagnosing a native-only mismatch, and identify the first stage that changes the specified semantics.