| from pathlib import Path |
| from typing import TextIO |
| |
| from analyzer import ( |
| Instruction, |
| Uop, |
| Properties, |
| StackItem, |
| analysis_error, |
| ) |
| from cwriter import CWriter |
| from typing import Callable, TextIO, Iterator, Iterable |
| from lexer import Token |
| from stack import Storage, StackError |
| |
| # Set this to true for voluminous output showing state of stack and locals |
| PRINT_STACKS = False |
| |
| class TokenIterator: |
| |
| look_ahead: Token | None |
| iterator: Iterator[Token] |
| |
| def __init__(self, tkns: Iterable[Token]): |
| self.iterator = iter(tkns) |
| self.look_ahead = None |
| |
| def __iter__(self) -> "TokenIterator": |
| return self |
| |
| def __next__(self) -> Token: |
| if self.look_ahead is None: |
| return next(self.iterator) |
| else: |
| res = self.look_ahead |
| self.look_ahead = None |
| return res |
| |
| def peek(self) -> Token | None: |
| if self.look_ahead is None: |
| for tkn in self.iterator: |
| self.look_ahead = tkn |
| break |
| return self.look_ahead |
| |
| ROOT = Path(__file__).parent.parent.parent.resolve() |
| DEFAULT_INPUT = (ROOT / "Python/bytecodes.c").as_posix() |
| |
| |
| def root_relative_path(filename: str) -> str: |
| try: |
| return Path(filename).resolve().relative_to(ROOT).as_posix() |
| except ValueError: |
| # Not relative to root, just return original path. |
| return filename |
| |
| |
| def type_and_null(var: StackItem) -> tuple[str, str]: |
| if var.type: |
| return var.type, "NULL" |
| elif var.is_array(): |
| return "_PyStackRef *", "NULL" |
| else: |
| return "_PyStackRef", "PyStackRef_NULL" |
| |
| |
| def write_header( |
| generator: str, sources: list[str], outfile: TextIO, comment: str = "//" |
| ) -> None: |
| outfile.write( |
| f"""{comment} This file is generated by {root_relative_path(generator)} |
| {comment} from: |
| {comment} {", ".join(root_relative_path(src) for src in sources)} |
| {comment} Do not edit! |
| """ |
| ) |
| |
| |
| def emit_to(out: CWriter, tkn_iter: TokenIterator, end: str) -> Token: |
| parens = 0 |
| for tkn in tkn_iter: |
| if tkn.kind == end and parens == 0: |
| return tkn |
| if tkn.kind == "LPAREN": |
| parens += 1 |
| if tkn.kind == "RPAREN": |
| parens -= 1 |
| out.emit(tkn) |
| raise analysis_error(f"Expecting {end}. Reached end of file", tkn) |
| |
| |
| ReplacementFunctionType = Callable[ |
| [Token, TokenIterator, Uop, Storage, Instruction | None], bool |
| ] |
| |
| def always_true(tkn: Token | None) -> bool: |
| if tkn is None: |
| return False |
| return tkn.text in {"true", "1"} |
| |
| |
| class Emitter: |
| out: CWriter |
| _replacers: dict[str, ReplacementFunctionType] |
| |
| def __init__(self, out: CWriter): |
| self._replacers = { |
| "EXIT_IF": self.exit_if, |
| "DEOPT_IF": self.deopt_if, |
| "ERROR_IF": self.error_if, |
| "ERROR_NO_POP": self.error_no_pop, |
| "DECREF_INPUTS": self.decref_inputs, |
| "DEAD": self.kill, |
| "INPUTS_DEAD": self.kill_inputs, |
| "SYNC_SP": self.sync_sp, |
| "SAVE_STACK": self.save_stack, |
| "RELOAD_STACK": self.reload_stack, |
| "PyStackRef_CLOSE": self.stackref_close, |
| "PyStackRef_CLOSE_SPECIALIZED": self.stackref_close, |
| "PyStackRef_AsPyObjectSteal": self.stackref_steal, |
| "DISPATCH": self.dispatch, |
| "INSTRUCTION_SIZE": self.instruction_size, |
| } |
| self.out = out |
| |
| def dispatch( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| self.emit(tkn) |
| return False |
| |
| def deopt_if( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| self.out.emit_at("DEOPT_IF", tkn) |
| lparen = next(tkn_iter) |
| self.emit(lparen) |
| assert lparen.kind == "LPAREN" |
| first_tkn = tkn_iter.peek() |
| emit_to(self.out, tkn_iter, "RPAREN") |
| next(tkn_iter) # Semi colon |
| self.out.emit(", ") |
| assert inst is not None |
| assert inst.family is not None |
| self.out.emit(inst.family.name) |
| self.out.emit(");\n") |
| return not always_true(first_tkn) |
| |
| exit_if = deopt_if |
| |
| def error_if( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| lparen = next(tkn_iter) |
| assert lparen.kind == "LPAREN" |
| first_tkn = tkn_iter.peek() |
| unconditional = always_true(first_tkn) |
| if unconditional: |
| next(tkn_iter) |
| comma = next(tkn_iter) |
| if comma.kind != "COMMA": |
| raise analysis_error(f"Expected comma, got '{comma.text}'", comma) |
| self.out.start_line() |
| else: |
| self.out.emit_at("if ", tkn) |
| self.emit(lparen) |
| emit_to(self.out, tkn_iter, "COMMA") |
| self.out.emit(") ") |
| label = next(tkn_iter).text |
| next(tkn_iter) # RPAREN |
| next(tkn_iter) # Semi colon |
| storage.clear_inputs("at ERROR_IF") |
| c_offset = storage.stack.peek_offset() |
| try: |
| offset = -int(c_offset) |
| except ValueError: |
| offset = -1 |
| if offset > 0: |
| self.out.emit(f"goto pop_{offset}_") |
| self.out.emit(label) |
| self.out.emit(";\n") |
| elif offset == 0: |
| self.out.emit("goto ") |
| self.out.emit(label) |
| self.out.emit(";\n") |
| else: |
| self.out.emit("{\n") |
| storage.copy().flush(self.out) |
| self.out.emit("goto ") |
| self.out.emit(label) |
| self.out.emit(";\n") |
| self.out.emit("}\n") |
| return not unconditional |
| |
| def error_no_pop( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| next(tkn_iter) # LPAREN |
| next(tkn_iter) # RPAREN |
| next(tkn_iter) # Semi colon |
| self.out.emit_at("goto error;", tkn) |
| return False |
| |
| def decref_inputs( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| next(tkn_iter) |
| next(tkn_iter) |
| next(tkn_iter) |
| self.out.emit_at("", tkn) |
| for var in uop.stack.inputs: |
| if var.name == "unused" or var.name == "null" or var.peek: |
| continue |
| if var.size: |
| if var.size == "1": |
| self.out.emit(f"PyStackRef_CLOSE({var.name}[0]);\n") |
| else: |
| self.out.emit(f"for (int _i = {var.size}; --_i >= 0;) {{\n") |
| self.out.emit(f"PyStackRef_CLOSE({var.name}[_i]);\n") |
| self.out.emit("}\n") |
| elif var.condition: |
| if var.condition == "1": |
| self.out.emit(f"PyStackRef_CLOSE({var.name});\n") |
| elif var.condition != "0": |
| self.out.emit(f"PyStackRef_XCLOSE({var.name});\n") |
| else: |
| self.out.emit(f"PyStackRef_CLOSE({var.name});\n") |
| for input in storage.inputs: |
| input.defined = False |
| return True |
| |
| def kill_inputs( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| next(tkn_iter) |
| next(tkn_iter) |
| next(tkn_iter) |
| for var in storage.inputs: |
| var.defined = False |
| return True |
| |
| def kill( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| next(tkn_iter) |
| name_tkn = next(tkn_iter) |
| name = name_tkn.text |
| next(tkn_iter) |
| next(tkn_iter) |
| for var in storage.inputs: |
| if var.name == name: |
| var.defined = False |
| break |
| else: |
| raise analysis_error(f"'{name}' is not a live input-only variable", name_tkn) |
| return True |
| |
| def stackref_close( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| self.out.emit(tkn) |
| tkn = next(tkn_iter) |
| assert tkn.kind == "LPAREN" |
| self.out.emit(tkn) |
| name = next(tkn_iter) |
| self.out.emit(name) |
| if name.kind == "IDENTIFIER": |
| for var in storage.inputs: |
| if var.name == name.text: |
| var.defined = False |
| rparen = emit_to(self.out, tkn_iter, "RPAREN") |
| self.emit(rparen) |
| return True |
| |
| stackref_steal = stackref_close |
| |
| def sync_sp( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| next(tkn_iter) |
| next(tkn_iter) |
| next(tkn_iter) |
| storage.clear_inputs("when syncing stack") |
| storage.flush(self.out) |
| self._print_storage(storage) |
| return True |
| |
| def emit_save(self, storage: Storage) -> None: |
| storage.save(self.out) |
| self._print_storage(storage) |
| |
| def save_stack( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| next(tkn_iter) |
| next(tkn_iter) |
| next(tkn_iter) |
| self.emit_save(storage) |
| return True |
| |
| def emit_reload(self, storage: Storage) -> None: |
| storage.reload(self.out) |
| self._print_storage(storage) |
| |
| def reload_stack( |
| self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| next(tkn_iter) |
| next(tkn_iter) |
| next(tkn_iter) |
| self.emit_reload(storage) |
| return True |
| |
| def instruction_size(self, |
| tkn: Token, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> bool: |
| """Replace the INSTRUCTION_SIZE macro with the size of the current instruction.""" |
| if uop.instruction_size is None: |
| raise analysis_error("The INSTRUCTION_SIZE macro requires uop.instruction_size to be set", tkn) |
| self.out.emit(f" {uop.instruction_size} ") |
| return True |
| |
| def _print_storage(self, storage: Storage) -> None: |
| if PRINT_STACKS: |
| self.out.start_line() |
| self.emit(storage.as_comment()) |
| self.out.start_line() |
| |
| def _emit_if( |
| self, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> tuple[bool, Token, Storage]: |
| """Returns (reachable?, closing '}', stack).""" |
| tkn = next(tkn_iter) |
| assert tkn.kind == "LPAREN" |
| self.out.emit(tkn) |
| rparen = emit_to(self.out, tkn_iter, "RPAREN") |
| self.emit(rparen) |
| if_storage = storage.copy() |
| reachable, rbrace, if_storage = self._emit_block(tkn_iter, uop, if_storage, inst, True) |
| try: |
| maybe_else = tkn_iter.peek() |
| if maybe_else and maybe_else.kind == "ELSE": |
| self._print_storage(storage) |
| self.emit(rbrace) |
| self.emit(next(tkn_iter)) |
| maybe_if = tkn_iter.peek() |
| if maybe_if and maybe_if.kind == "IF": |
| #Emit extra braces around the if to get scoping right |
| self.emit(" {\n") |
| self.emit(next(tkn_iter)) |
| else_reachable, rbrace, else_storage = self._emit_if(tkn_iter, uop, storage, inst) |
| self.out.start_line() |
| self.emit("}\n") |
| else: |
| else_reachable, rbrace, else_storage = self._emit_block(tkn_iter, uop, storage, inst, True) |
| if not reachable: |
| # Discard the if storage |
| reachable = else_reachable |
| storage = else_storage |
| elif not else_reachable: |
| # Discard the else storage |
| storage = if_storage |
| reachable = True |
| else: |
| if PRINT_STACKS: |
| self.emit("/* Merge */\n") |
| else_storage.merge(if_storage, self.out) |
| storage = else_storage |
| self._print_storage(storage) |
| else: |
| if reachable: |
| if PRINT_STACKS: |
| self.emit("/* Merge */\n") |
| if_storage.merge(storage, self.out) |
| storage = if_storage |
| self._print_storage(storage) |
| else: |
| # Discard the if storage |
| reachable = True |
| except StackError as ex: |
| self._print_storage(if_storage) |
| raise analysis_error(ex.args[0], rbrace) # from None |
| return reachable, rbrace, storage |
| |
| def _emit_block( |
| self, |
| tkn_iter: TokenIterator, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| emit_first_brace: bool |
| ) -> tuple[bool, Token, Storage]: |
| """ Returns (reachable?, closing '}', stack).""" |
| braces = 1 |
| out_stores = set(uop.output_stores) |
| tkn = next(tkn_iter) |
| reload: Token | None = None |
| try: |
| reachable = True |
| line : int = -1 |
| if tkn.kind != "LBRACE": |
| raise analysis_error(f"PEP 7: expected '{{', found: {tkn.text}", tkn) |
| escaping_calls = uop.properties.escaping_calls |
| if emit_first_brace: |
| self.emit(tkn) |
| self._print_storage(storage) |
| for tkn in tkn_iter: |
| if PRINT_STACKS and tkn.line != line: |
| self.out.start_line() |
| self.emit(storage.as_comment()) |
| self.out.start_line() |
| line = tkn.line |
| if tkn in escaping_calls: |
| if tkn != reload: |
| self.emit_save(storage) |
| _, reload = escaping_calls[tkn] |
| elif tkn == reload: |
| self.emit_reload(storage) |
| if tkn.kind == "LBRACE": |
| self.out.emit(tkn) |
| braces += 1 |
| elif tkn.kind == "RBRACE": |
| self._print_storage(storage) |
| braces -= 1 |
| if braces == 0: |
| return reachable, tkn, storage |
| self.out.emit(tkn) |
| elif tkn.kind == "GOTO": |
| reachable = False; |
| self.out.emit(tkn) |
| elif tkn.kind == "IDENTIFIER": |
| if tkn.text in self._replacers: |
| if not self._replacers[tkn.text](tkn, tkn_iter, uop, storage, inst): |
| reachable = False |
| else: |
| if tkn in out_stores: |
| for out in storage.outputs: |
| if out.name == tkn.text: |
| out.defined = True |
| out.in_memory = False |
| break |
| if tkn.text.startswith("DISPATCH"): |
| self._print_storage(storage) |
| reachable = False |
| self.out.emit(tkn) |
| elif tkn.kind == "IF": |
| self.out.emit(tkn) |
| if_reachable, rbrace, storage = self._emit_if(tkn_iter, uop, storage, inst) |
| if reachable: |
| reachable = if_reachable |
| self.out.emit(rbrace) |
| else: |
| self.out.emit(tkn) |
| except StackError as ex: |
| raise analysis_error(ex.args[0], tkn) from None |
| raise analysis_error("Expecting closing brace. Reached end of file", tkn) |
| |
| |
| def emit_tokens( |
| self, |
| uop: Uop, |
| storage: Storage, |
| inst: Instruction | None, |
| ) -> Storage: |
| tkn_iter = TokenIterator(uop.body) |
| self.out.start_line() |
| _, rbrace, storage = self._emit_block(tkn_iter, uop, storage, inst, False) |
| try: |
| self._print_storage(storage) |
| storage.push_outputs() |
| self._print_storage(storage) |
| except StackError as ex: |
| raise analysis_error(ex.args[0], rbrace) |
| return storage |
| |
| def emit(self, txt: str | Token) -> None: |
| self.out.emit(txt) |
| |
| |
| def cflags(p: Properties) -> str: |
| flags: list[str] = [] |
| if p.oparg: |
| flags.append("HAS_ARG_FLAG") |
| if p.uses_co_consts: |
| flags.append("HAS_CONST_FLAG") |
| if p.uses_co_names: |
| flags.append("HAS_NAME_FLAG") |
| if p.jumps: |
| flags.append("HAS_JUMP_FLAG") |
| if p.has_free: |
| flags.append("HAS_FREE_FLAG") |
| if p.uses_locals: |
| flags.append("HAS_LOCAL_FLAG") |
| if p.eval_breaker: |
| flags.append("HAS_EVAL_BREAK_FLAG") |
| if p.deopts: |
| flags.append("HAS_DEOPT_FLAG") |
| if p.side_exit: |
| flags.append("HAS_EXIT_FLAG") |
| if not p.infallible: |
| flags.append("HAS_ERROR_FLAG") |
| if p.error_without_pop: |
| flags.append("HAS_ERROR_NO_POP_FLAG") |
| if p.escapes: |
| flags.append("HAS_ESCAPES_FLAG") |
| if p.pure: |
| flags.append("HAS_PURE_FLAG") |
| if p.oparg_and_1: |
| flags.append("HAS_OPARG_AND_1_FLAG") |
| if flags: |
| return " | ".join(flags) |
| else: |
| return "0" |
