parens -= 1
out.emit(tkn)
+ReplacementFunctionType = Callable[
+ [Token, Iterator[Token], Uop, Stack, Instruction | None], None
+]
-def replace_deopt(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- unused: Stack,
- inst: Instruction | None,
-) -> None:
- out.emit_at("DEOPT_IF", tkn)
- out.emit(next(tkn_iter))
- emit_to(out, tkn_iter, "RPAREN")
- next(tkn_iter) # Semi colon
- out.emit(", ")
- assert inst is not None
- assert inst.family is not None
- out.emit(inst.family.name)
- out.emit(");\n")
-
+class Emitter:
-def replace_error(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- stack: Stack,
- inst: Instruction | None,
-) -> None:
- out.emit_at("if ", tkn)
- out.emit(next(tkn_iter))
- emit_to(out, tkn_iter, "COMMA")
- label = next(tkn_iter).text
- next(tkn_iter) # RPAREN
- next(tkn_iter) # Semi colon
- out.emit(") ")
- c_offset = stack.peek_offset()
- try:
- offset = -int(c_offset)
- except ValueError:
- offset = -1
- if offset > 0:
- out.emit(f"goto pop_{offset}_")
- out.emit(label)
- out.emit(";\n")
- elif offset == 0:
- out.emit("goto ")
- out.emit(label)
- out.emit(";\n")
- else:
- out.emit("{\n")
- stack.flush_locally(out)
- out.emit("goto ")
- out.emit(label)
- out.emit(";\n")
- out.emit("}\n")
+ 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,
+ "CHECK_EVAL_BREAKER": self.check_eval_breaker,
+ "SYNC_SP": self.sync_sp,
+ }
+ self.out = out
-def replace_error_no_pop(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- stack: Stack,
- inst: Instruction | None,
-) -> None:
- next(tkn_iter) # LPAREN
- next(tkn_iter) # RPAREN
- next(tkn_iter) # Semi colon
- out.emit_at("goto error;", tkn)
+ def deopt_if(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ unused: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ self.out.emit_at("DEOPT_IF", tkn)
+ self.out.emit(next(tkn_iter))
+ 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")
+ exit_if = deopt_if
-def replace_decrefs(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- stack: Stack,
- inst: Instruction | None,
-) -> None:
- next(tkn_iter)
- next(tkn_iter)
- next(tkn_iter)
- 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:
- out.emit(f"for (int _i = {var.size}; --_i >= 0;) {{\n")
- out.emit(f"PyStackRef_CLOSE({var.name}[_i]);\n")
- out.emit("}\n")
- elif var.condition:
- if var.condition == "1":
- out.emit(f"PyStackRef_CLOSE({var.name});\n")
- elif var.condition != "0":
- out.emit(f"PyStackRef_XCLOSE({var.name});\n")
+ def error_if(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ stack: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ self.out.emit_at("if ", tkn)
+ self.out.emit(next(tkn_iter))
+ emit_to(self.out, tkn_iter, "COMMA")
+ label = next(tkn_iter).text
+ next(tkn_iter) # RPAREN
+ next(tkn_iter) # Semi colon
+ self.out.emit(") ")
+ c_offset = 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:
- out.emit(f"PyStackRef_CLOSE({var.name});\n")
+ self.out.emit("{\n")
+ stack.flush_locally(self.out)
+ self.out.emit("goto ")
+ self.out.emit(label)
+ self.out.emit(";\n")
+ self.out.emit("}\n")
+ def error_no_pop(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ stack: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ next(tkn_iter) # LPAREN
+ next(tkn_iter) # RPAREN
+ next(tkn_iter) # Semi colon
+ self.out.emit_at("goto error;", tkn)
-def replace_sync_sp(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- stack: Stack,
- inst: Instruction | None,
-) -> None:
- next(tkn_iter)
- next(tkn_iter)
- next(tkn_iter)
- stack.flush(out)
-
+ def decref_inputs(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ stack: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ 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:
+ 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")
-def replace_check_eval_breaker(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- stack: Stack,
- inst: Instruction | None,
-) -> None:
- next(tkn_iter)
- next(tkn_iter)
- next(tkn_iter)
- if not uop.properties.ends_with_eval_breaker:
- out.emit_at("CHECK_EVAL_BREAKER();", tkn)
+ def sync_sp(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ stack: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ next(tkn_iter)
+ next(tkn_iter)
+ next(tkn_iter)
+ stack.flush(self.out)
-REPLACEMENT_FUNCTIONS = {
- "EXIT_IF": replace_deopt,
- "DEOPT_IF": replace_deopt,
- "ERROR_IF": replace_error,
- "ERROR_NO_POP": replace_error_no_pop,
- "DECREF_INPUTS": replace_decrefs,
- "CHECK_EVAL_BREAKER": replace_check_eval_breaker,
- "SYNC_SP": replace_sync_sp,
-}
-
-ReplacementFunctionType = Callable[
- [CWriter, Token, Iterator[Token], Uop, Stack, Instruction | None], None
-]
+ def check_eval_breaker(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ stack: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ next(tkn_iter)
+ next(tkn_iter)
+ next(tkn_iter)
+ if not uop.properties.ends_with_eval_breaker:
+ self.out.emit_at("CHECK_EVAL_BREAKER();", tkn)
-def emit_tokens(
- out: CWriter,
- uop: Uop,
- stack: Stack,
- inst: Instruction | None,
- replacement_functions: Mapping[
- str, ReplacementFunctionType
- ] = REPLACEMENT_FUNCTIONS,
-) -> None:
- tkns = uop.body[1:-1]
- if not tkns:
- return
- tkn_iter = iter(tkns)
- out.start_line()
- for tkn in tkn_iter:
- if tkn.kind == "IDENTIFIER" and tkn.text in replacement_functions:
- replacement_functions[tkn.text](out, tkn, tkn_iter, uop, stack, inst)
- else:
- out.emit(tkn)
+ def emit_tokens(
+ self,
+ uop: Uop,
+ stack: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ tkns = uop.body[1:-1]
+ if not tkns:
+ return
+ tkn_iter = iter(tkns)
+ self.out.start_line()
+ for tkn in tkn_iter:
+ if tkn.kind == "IDENTIFIER" and tkn.text in self._replacers:
+ self._replacers[tkn.text](tkn, tkn_iter, uop, stack, inst)
+ else:
+ self.out.emit(tkn)
+ def emit(self, txt: str | Token) -> None:
+ self.out.emit(txt)
def cflags(p: Properties) -> str:
flags: list[str] = []
DEFAULT_INPUT,
ROOT,
write_header,
- emit_tokens,
- replace_sync_sp,
+ Emitter,
)
from cwriter import CWriter
from typing import TextIO, Iterator
else:
out.emit(f"{var.name} = sym_new_not_null(ctx);\n")
+class OptimizerEmitter(Emitter):
+
+ pass
+
def write_uop(
override: Uop | None,
cast = f"uint{cache.size*16}_t"
out.emit(f"{type}{cache.name} = ({cast})this_instr->operand;\n")
if override:
- replacement_funcs = {
- "DECREF_INPUTS": decref_inputs,
- "SYNC_SP": replace_sync_sp,
- }
- emit_tokens(out, override, stack, None, replacement_funcs)
+ emitter = OptimizerEmitter(out)
+ emitter.emit_tokens(override, stack, None)
else:
emit_default(out, uop)
DEFAULT_INPUT,
ROOT,
write_header,
- emit_tokens,
type_and_null,
+ Emitter,
)
from cwriter import CWriter
from typing import TextIO
declare_variable(var, out)
def write_uop(
- uop: Part, out: CWriter, offset: int, stack: Stack, inst: Instruction, braces: bool
+ uop: Part, emitter: Emitter, offset: int, stack: Stack, inst: Instruction, braces: bool
) -> int:
# out.emit(stack.as_comment() + "\n")
if isinstance(uop, Skip):
entries = "entries" if uop.size > 1 else "entry"
- out.emit(f"/* Skip {uop.size} cache {entries} */\n")
+ emitter.emit(f"/* Skip {uop.size} cache {entries} */\n")
return offset + uop.size
if isinstance(uop, Flush):
- out.emit(f"// flush\n")
- stack.flush(out)
+ emitter.emit(f"// flush\n")
+ stack.flush(emitter.out)
return offset
try:
locals: dict[str, Local] = {}
- out.start_line()
+ emitter.out.start_line()
if braces:
- out.emit(f"// {uop.name}\n")
+ emitter.out.emit(f"// {uop.name}\n")
peeks: list[Local] = []
for var in reversed(uop.stack.inputs):
code, local = stack.pop(var)
- out.emit(code)
+ emitter.emit(code)
if var.peek:
peeks.append(local)
if local.defined:
while peeks:
stack.push(peeks.pop())
if braces:
- out.emit("{\n")
- out.emit(stack.define_output_arrays(uop.stack.outputs))
+ emitter.emit("{\n")
+ emitter.out.emit(stack.define_output_arrays(uop.stack.outputs))
for cache in uop.caches:
if cache.name != "unused":
else:
type = f"uint{cache.size*16}_t "
reader = f"read_u{cache.size*16}"
- out.emit(
+ emitter.emit(
f"{type}{cache.name} = {reader}(&this_instr[{offset}].cache);\n"
)
if inst.family is None:
- out.emit(f"(void){cache.name};\n")
+ emitter.emit(f"(void){cache.name};\n")
offset += cache.size
- emit_tokens(out, uop, stack, inst)
+ emitter.emit_tokens(uop, stack, inst)
for i, var in enumerate(uop.stack.outputs):
if not var.peek:
if var.name in locals:
local = Local.unused(var)
else:
local = Local.local(var)
- out.emit(stack.push(local))
+ emitter.emit(stack.push(local))
if braces:
- out.start_line()
- out.emit("}\n")
- # out.emit(stack.as_comment() + "\n")
+ emitter.out.start_line()
+ emitter.emit("}\n")
+ # emitter.emit(stack.as_comment() + "\n")
return offset
except StackError as ex:
raise analysis_error(ex.args[0], uop.body[0])
"""
)
out = CWriter(outfile, 2, lines)
+ emitter = Emitter(out)
out.emit("\n")
for name, inst in sorted(analysis.instructions.items()):
needs_this = uses_this(inst)
for part in inst.parts:
# Only emit braces if more than one uop
insert_braces = len([p for p in inst.parts if isinstance(p, Uop)]) > 1
- offset = write_uop(part, out, offset, stack, inst, insert_braces)
+ offset = write_uop(part, emitter, offset, stack, inst, insert_braces)
out.start_line()
if not inst.parts[-1].properties.always_exits:
stack.flush(out)
from generators_common import (
DEFAULT_INPUT,
ROOT,
- write_header,
- emit_tokens,
emit_to,
- REPLACEMENT_FUNCTIONS,
+ write_header,
type_and_null,
+ Emitter
)
from cwriter import CWriter
from typing import TextIO, Iterator
for var in uop.stack.outputs:
declare_variable(var, uop, required, out)
-def tier2_replace_error(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- stack: Stack,
- inst: Instruction | None,
-) -> None:
- out.emit_at("if ", tkn)
- out.emit(next(tkn_iter))
- emit_to(out, tkn_iter, "COMMA")
- label = next(tkn_iter).text
- next(tkn_iter) # RPAREN
- next(tkn_iter) # Semi colon
- out.emit(") JUMP_TO_ERROR();\n")
+class Tier2Emitter(Emitter):
-def tier2_replace_error_no_pop(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- stack: Stack,
- inst: Instruction | None,
-) -> None:
- next(tkn_iter) # LPAREN
- next(tkn_iter) # RPAREN
- next(tkn_iter) # Semi colon
- out.emit_at("JUMP_TO_ERROR();", tkn)
+ def __init__(self, out: CWriter):
+ super().__init__(out)
+ self._replacers["oparg"] = self.oparg
-def tier2_replace_deopt(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- unused: Stack,
- inst: Instruction | None,
-) -> None:
- out.emit_at("if ", tkn)
- out.emit(next(tkn_iter))
- emit_to(out, tkn_iter, "RPAREN")
- next(tkn_iter) # Semi colon
- out.emit(") {\n")
- out.emit("UOP_STAT_INC(uopcode, miss);\n")
- out.emit("JUMP_TO_JUMP_TARGET();\n");
- out.emit("}\n")
+ def error_if(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ stack: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ self.out.emit_at("if ", tkn)
+ self.emit(next(tkn_iter))
+ emit_to(self.out, tkn_iter, "COMMA")
+ label = next(tkn_iter).text
+ next(tkn_iter) # RPAREN
+ next(tkn_iter) # Semi colon
+ self.emit(") JUMP_TO_ERROR();\n")
+ def error_no_pop(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ stack: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ next(tkn_iter) # LPAREN
+ next(tkn_iter) # RPAREN
+ next(tkn_iter) # Semi colon
+ self.out.emit_at("JUMP_TO_ERROR();", tkn)
-def tier2_replace_exit_if(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- unused: Stack,
- inst: Instruction | None,
-) -> None:
- out.emit_at("if ", tkn)
- out.emit(next(tkn_iter))
- emit_to(out, tkn_iter, "RPAREN")
- next(tkn_iter) # Semi colon
- out.emit(") {\n")
- out.emit("UOP_STAT_INC(uopcode, miss);\n")
- out.emit("JUMP_TO_JUMP_TARGET();\n")
- out.emit("}\n")
+ def deopt_if(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ unused: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ self.out.emit_at("if ", tkn)
+ self.emit(next(tkn_iter))
+ emit_to(self.out, tkn_iter, "RPAREN")
+ next(tkn_iter) # Semi colon
+ self.emit(") {\n")
+ self.emit("UOP_STAT_INC(uopcode, miss);\n")
+ self.emit("JUMP_TO_JUMP_TARGET();\n");
+ self.emit("}\n")
+ def exit_if( # type: ignore[override]
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ unused: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ self.out.emit_at("if ", tkn)
+ self.emit(next(tkn_iter))
+ emit_to(self.out, tkn_iter, "RPAREN")
+ next(tkn_iter) # Semi colon
+ self.emit(") {\n")
+ self.emit("UOP_STAT_INC(uopcode, miss);\n")
+ self.emit("JUMP_TO_JUMP_TARGET();\n")
+ self.emit("}\n")
-def tier2_replace_oparg(
- out: CWriter,
- tkn: Token,
- tkn_iter: Iterator[Token],
- uop: Uop,
- unused: Stack,
- inst: Instruction | None,
-) -> None:
- if not uop.name.endswith("_0") and not uop.name.endswith("_1"):
- out.emit(tkn)
- return
- amp = next(tkn_iter)
- if amp.text != "&":
- out.emit(tkn)
- out.emit(amp)
- return
- one = next(tkn_iter)
- assert one.text == "1"
- out.emit_at(uop.name[-1], tkn)
-
+ def oparg(
+ self,
+ tkn: Token,
+ tkn_iter: Iterator[Token],
+ uop: Uop,
+ unused: Stack,
+ inst: Instruction | None,
+ ) -> None:
+ if not uop.name.endswith("_0") and not uop.name.endswith("_1"):
+ self.emit(tkn)
+ return
+ amp = next(tkn_iter)
+ if amp.text != "&":
+ self.emit(tkn)
+ self.emit(amp)
+ return
+ one = next(tkn_iter)
+ assert one.text == "1"
+ self.out.emit_at(uop.name[-1], tkn)
-TIER2_REPLACEMENT_FUNCTIONS = REPLACEMENT_FUNCTIONS.copy()
-TIER2_REPLACEMENT_FUNCTIONS["ERROR_IF"] = tier2_replace_error
-TIER2_REPLACEMENT_FUNCTIONS["ERROR_NO_POP"] = tier2_replace_error_no_pop
-TIER2_REPLACEMENT_FUNCTIONS["DEOPT_IF"] = tier2_replace_deopt
-TIER2_REPLACEMENT_FUNCTIONS["oparg"] = tier2_replace_oparg
-TIER2_REPLACEMENT_FUNCTIONS["EXIT_IF"] = tier2_replace_exit_if
-
-
-def write_uop(uop: Uop, out: CWriter, stack: Stack) -> None:
+def write_uop(uop: Uop, emitter: Emitter, stack: Stack) -> None:
locals: dict[str, Local] = {}
try:
- out.start_line()
+ emitter.out.start_line()
if uop.properties.oparg:
- out.emit("oparg = CURRENT_OPARG();\n")
+ emitter.emit("oparg = CURRENT_OPARG();\n")
assert uop.properties.const_oparg < 0
elif uop.properties.const_oparg >= 0:
- out.emit(f"oparg = {uop.properties.const_oparg};\n")
- out.emit(f"assert(oparg == CURRENT_OPARG());\n")
+ emitter.emit(f"oparg = {uop.properties.const_oparg};\n")
+ emitter.emit(f"assert(oparg == CURRENT_OPARG());\n")
for var in reversed(uop.stack.inputs):
code, local = stack.pop(var)
- out.emit(code)
+ emitter.emit(code)
if local.defined:
locals[local.name] = local
- out.emit(stack.define_output_arrays(uop.stack.outputs))
+ emitter.emit(stack.define_output_arrays(uop.stack.outputs))
for cache in uop.caches:
if cache.name != "unused":
if cache.size == 4:
else:
type = f"uint{cache.size*16}_t "
cast = f"uint{cache.size*16}_t"
- out.emit(f"{type}{cache.name} = ({cast})CURRENT_OPERAND();\n")
- emit_tokens(out, uop, stack, None, TIER2_REPLACEMENT_FUNCTIONS)
+ emitter.emit(f"{type}{cache.name} = ({cast})CURRENT_OPERAND();\n")
+ emitter.emit_tokens(uop, stack, None)
for i, var in enumerate(uop.stack.outputs):
if var.name in locals:
local = locals[var.name]
else:
local = Local.local(var)
- out.emit(stack.push(local))
+ emitter.emit(stack.push(local))
except StackError as ex:
raise analysis_error(ex.args[0], uop.body[0]) from None
"""
)
out = CWriter(outfile, 2, lines)
+ emitter = Tier2Emitter(out)
out.emit("\n")
for name, uop in analysis.uops.items():
if uop.properties.tier == 1:
out.emit(f"case {uop.name}: {{\n")
declare_variables(uop, out)
stack = Stack()
- write_uop(uop, out, stack)
+ write_uop(uop, emitter, stack)
out.start_line()
if not uop.properties.always_exits:
stack.flush(out)
projects / thirdparty / Python / cpython.git / commitdiff
