ContentsFoundations to FrontierBezhig

graduate_code/solutions/compiler.py

All lab code. Download this file.

"""GP1: pure expressions -> typed AST -> verified, forward-only bytecode.
This is an educational compiler, not a sandbox for hostile programs.
"""
import ast
import sys
from .language import infer, evaluate


def parse(source):
    if type(source) is not str or not 1 <= len(source) <= 4096:
        raise ValueError("source must contain 1..4096 characters")
    try:
        host = ast.parse(source, mode="eval")
    except (SyntaxError, RecursionError) as error:
        raise ValueError("invalid source syntax") from error

    def convert(node, depth=0):
        if depth > 48:
            raise ValueError("syntax depth exceeds 48")
        if isinstance(node, ast.Constant) and type(node.value) in (int, bool):
            if type(node.value) is int and node.value.bit_length() > 256:
                raise ValueError("integer literal exceeds 256 bits")
            return node.value
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
            if isinstance(node.operand, ast.Constant) and type(node.operand.value) is int:
                return -convert(node.operand, depth+1)
        if isinstance(node, ast.Name):
            return node.id
        if isinstance(node, ast.BinOp) and type(node.op) in (ast.Add, ast.Mult):
            tag = "+" if isinstance(node.op, ast.Add) else "*"
            return tag, convert(node.left,depth+1), convert(node.right,depth+1)
        if isinstance(node, ast.Compare) and len(node.ops) == 1 and isinstance(node.ops[0],ast.Lt):
            return "<",convert(node.left,depth+1),convert(node.comparators[0],depth+1)
        if isinstance(node, ast.IfExp):
            return ("if",convert(node.test,depth+1),convert(node.body,depth+1),
                    convert(node.orelse,depth+1))
        if (isinstance(node, ast.Call) and isinstance(node.func,ast.Name)
                and node.func.id == "let" and len(node.args) == 3 and not node.keywords):
            name = node.args[0]
            if not isinstance(name,ast.Constant) or type(name.value) is not str or not name.value.isidentifier():
                raise ValueError("let requires a quoted identifier")
            return "let",name.value,convert(node.args[1],depth+1),convert(node.args[2],depth+1)
        raise ValueError("unsupported source construct")

    expression = convert(host.body)
    infer(expression)  # both branches checked, including dead branches
    return expression


def simplify(expression):
    infer(expression)

    def visit(e):
        if type(e) is not tuple:
            return e
        tag = e[0]
        if tag == "let":
            return tag,e[1],visit(e[2]),visit(e[3])
        if tag == "if":
            c,a,b = visit(e[1]),visit(e[2]),visit(e[3])
            return (a if c else b) if type(c) is bool else (tag,c,a,b)
        a,b = visit(e[1]),visit(e[2])
        if type(a) is int and type(b) is int:
            value = a+b if tag == "+" else a*b if tag == "*" else a < b
            if type(value) is bool or value.bit_length() <= 256:
                return value
        if tag == "+":
            if type(a) is int and a == 0: return b
            if type(b) is int and b == 0: return a
        if tag == "*":
            if (type(a) is int and a == 0) or (type(b) is int and b == 0): return 0
            if type(a) is int and a == 1: return b
            if type(b) is int and b == 1: return a
        return tag,a,b

    return visit(expression)


def compile_expr(expression):
    infer(expression)
    instructions, next_slot = [],0

    def emit(instruction):
        if len(instructions) >= 1023:  # reserve HALT
            raise ValueError("bytecode exceeds 1024 instructions")
        instructions.append(instruction)
        return len(instructions)-1

    def visit(e, scope):
        nonlocal next_slot
        if type(e) in (int,bool):
            if type(e) is int and e.bit_length() > 256:
                raise ValueError("literal exceeds 256 bits")
            emit(("PUSH",e))
        elif type(e) is str:
            emit(("LOAD",scope[e]))
        elif e[0] == "let":
            visit(e[2],scope)
            if next_slot >= 128:
                raise ValueError("local slots exceed 128")
            slot = next_slot
            next_slot += 1
            emit(("STORE",slot))
            visit(e[3],{**scope,e[1]:slot})
        elif e[0] == "if":
            visit(e[1],scope)
            branch = emit(("JIF",None))
            visit(e[2],scope)
            jump = emit(("JMP",None))
            instructions[branch] = ("JIF",len(instructions))
            visit(e[3],scope)
            instructions[jump] = ("JMP",len(instructions))
        else:
            visit(e[1],scope)
            visit(e[2],scope)
            emit(({"+":"ADD","*":"MUL","<":"LT"}[e[0]],))

    visit(expression,{})
    instructions.append(("HALT",))
    verify(instructions)
    return instructions


def verify(code):
    """Abstractly execute all acyclic paths; intersect definitely assigned locals."""
    if not isinstance(code,(list,tuple)) or not 1 <= len(code) <= 1024:
        raise ValueError("bounded nonempty instruction sequence required")
    states = {0:((),{})}
    result_type = None

    def merge(target, stack, slots):
        incoming = tuple(stack)
        if not 0 <= target < len(code):
            raise ValueError("fallthrough or jump beyond bytecode")
        if target not in states:
            states[target] = incoming,dict(slots)
        else:
            old_stack,old_slots = states[target]
            if incoming != old_stack:
                raise ValueError("incompatible join stacks")
            states[target] = incoming,{k:v for k,v in old_slots.items() if slots.get(k) == v}

    for pc,instruction in enumerate(code):
        if pc not in states:
            raise ValueError("unreachable instruction rejected by policy")
        if type(instruction) is not tuple or not instruction or type(instruction[0]) is not str:
            raise ValueError("malformed instruction")
        op = instruction[0]
        arity = {"PUSH":2,"LOAD":2,"STORE":2,"JIF":2,"JMP":2,
                 "ADD":1,"MUL":1,"LT":1,"HALT":1}.get(op)
        if arity is None or len(instruction) != arity:
            raise ValueError("unknown instruction or arity")
        stack,slots = states[pc]
        stack,slots = list(stack),dict(slots)

        def pop(expected=None):
            if not stack:
                raise ValueError("stack underflow")
            typ = stack.pop()
            if expected is not None and typ != expected:
                raise ValueError("wrong operand type")
            return typ

        if op == "PUSH":
            value = instruction[1]
            if type(value) not in (int,bool) or (type(value) is int and value.bit_length() > 256):
                raise ValueError("invalid literal")
            stack.append("bool" if type(value) is bool else "int")
        elif op in ("LOAD","STORE"):
            slot = instruction[1]
            if type(slot) is not int or not 0 <= slot < 128:
                raise ValueError("invalid slot")
            if op == "LOAD":
                if slot not in slots:
                    raise ValueError("load before definite assignment")
                stack.append(slots[slot])
            else:
                slots[slot] = pop()
        elif op in ("ADD","MUL","LT"):
            pop("int"); pop("int")
            stack.append("bool" if op == "LT" else "int")
        elif op in ("JIF","JMP"):
            target = instruction[1]
            if type(target) is not int or not pc < target < len(code):
                raise ValueError("jumps must target strictly later instructions")
            if op == "JIF": pop("bool")
            merge(target,stack,slots)
            if op == "JMP": continue
        elif op == "HALT":
            if len(stack) != 1:
                raise ValueError("HALT requires exactly one result")
            if result_type is not None and result_type != stack[0]:
                raise ValueError("incompatible result types")
            result_type = stack[0]
            continue
        merge(pc+1,stack,slots)
    if result_type is None:
        raise ValueError("no valid halt")
    return result_type


def run(code):
    verify(code)
    stack,slots,pc = [],{},0
    while True:
        instruction = code[pc]
        op = instruction[0]
        if op == "PUSH": stack.append(instruction[1])
        elif op == "LOAD": stack.append(slots[instruction[1]])
        elif op == "STORE": slots[instruction[1]] = stack.pop()
        elif op in ("ADD","MUL","LT"):
            b,a = stack.pop(),stack.pop()
            value = a+b if op == "ADD" else a*b if op == "MUL" else a < b
            if type(value) is int and value.bit_length() > 4096:
                raise ValueError("arithmetic result exceeds 4096 bits")
            stack.append(value)
        elif op == "JIF":
            if not stack.pop():
                pc = instruction[1]
                continue
        elif op == "JMP":
            pc = instruction[1]
            continue
        elif op == "HALT": return stack.pop()
        pc += 1


if __name__ == "__main__":
    source = sys.argv[1] if len(sys.argv)>1 else 'let("x",3,x*x if x<4 else 0)'
    expression = parse(source)
    code = compile_expr(expression)
    print("AST:",expression,"type:",infer(expression))
    for pc,instruction in enumerate(code): print(pc,instruction)
    print("interpreter:",evaluate(expression),"VM:",run(code))