wraps a cacheeffect that is actually used in context effect parsing cacheeffect offset int forbiddennamesinuops nextinstr oparg1 proxy for superinstructions like loadfastloadfast jumpby dispatch tieroneonly interpreter tiers tierone typing final 1 specializing adaptive interpreter pep 659 tiertwo typing final 2 experimental tracing interpreter tiers typing typealias typing literal1 2 dataclasses dataclass class instruction parts of the underlying instruction definition computed by constructor set later whether this instruction is viable as a uop dprint typing callable none lambda args kwargs none if frame in self name dprint print if self name exittrace return true this has return frame but it s okay if self name savereturnoffset return true adjusts nextinstr but only in tier 1 code if self alwaysexits dprintfskipping self name because it always exits self alwaysexits return false if lenself activecaches 1 printfskipping self name because it has 1 cache entries return false res true for forbidden in forbiddennamesinuops note to disallow unspecialized uops use if variableusedself inst forbidden if variableusedunspecializedself inst forbidden dprintfskipping self name because it uses forbidden res false return res def writebody self out formatter dedent int activecaches listactivecacheeffect tier tiers family parsing family none none write cache effect variable declarations and initializations note we assume that 64bit data in the cache is always an object pointer if this becomes false we need a way to specify syntactically what type the cache data is write the body substituting a goto for errorif and other stuff errorif must pop the inputs from the stack the code block is responsible for decrefing them note if the label doesn t exist just add it to ceval c don t pop common inputoutput effects at the bottom these aren t decref ed so they can stay instruction used for abstract interpretation write one abstract instruction sans prologue and epilogue stacking writesingleinstrforabstractinterpself out def writebody self out formatter dedent int activecaches listactivecacheeffect tier tiers family parsing family none none pass dataclasses dataclass class component instr instruction activecaches listactivecacheeffect macroparts listcomponent parsing cacheeffect dataclasses dataclass class macroinstruction set later a pseudo instruction name str targets listinstruction macroinstruction instrflags instructionflags anyinstruction instruction macroinstruction pseudoinstruction def extractblocktextblock parsing block tupleliststr bool int get lines of text with proper dedent blocklines block text splitlinestrue firsttoken lx token block tokens0 indexerror means the context is broken blockline firsttoken begin0 remove blank lines from both ends while blocklines and not blocklines0 strip blocklines pop0 blockline 1 while blocklines and not blocklines1 strip blocklines pop remove leading and trailing braces assert blocklines and blocklines0 strip assert blocklines and blocklines1 strip blocklines pop blocklines pop0 blockline 1 remove trailing blank lines while blocklines and not blocklines1 strip blocklines pop separate checkevalbreaker macro from end checkevalbreaker blocklines and blocklines1 strip checkevalbreaker if checkevalbreaker del blocklines1 return blocklines checkevalbreaker blockline def alwaysexitslines liststr str indent must match exactly todo do something better wraps a cacheeffect that is actually used in context proxy for super instructions like load_fast_load_fast interpreter tiers specializing adaptive interpreter pep 659 experimental tracing interpreter an instruction with additional data and code parts of the underlying instruction definition block text less curlies less predict calls first line of block in original code computed by constructor if the block always exits its last line else set later for consistency completeness whether this instruction is viable as a uop this has return frame but it s okay adjusts next_instr but only in tier 1 code print f skipping self name because it has 1 cache entries note to disallow unspecialized uops use if variable_used self inst forbidden write the instruction body write cache effect variable declarations and initializations note we assume that 64 bit data in the cache is always an object pointer if this becomes false we need a way to specify syntactically what type the cache data is write the body substituting a goto for error_if and other stuff error_if must pop the inputs from the stack the code block is responsible for decref ing them note if the label doesn t exist just add it to ceval c don t pop common input output effects at the bottom these aren t decref ed so they can stay instruction used for abstract interpretation write one abstract instruction sans prologue and epilogue a macro instruction set later a pseudo instruction get lines of text with proper dedent indexerror means the context is broken remove blank lines from both ends remove leading and trailing braces remove trailing blank lines separate check_eval_breaker macro from end determine whether a block always ends in a return goto etc indent must match exactly todo do something better
import dataclasses
import re
import typing

from flags import InstructionFlags, variable_used, variable_used_unspecialized
from formatting import (
    Formatter,
    UNUSED,
    list_effect_size,
)
import lexer as lx
import parsing
from parsing import StackEffect
import stacking

BITS_PER_CODE_UNIT = 16


@dataclasses.dataclass
class ActiveCacheEffect:
    

    effect: parsing.CacheEffect
    offset: int


FORBIDDEN_NAMES_IN_UOPS = (
    "next_instr",
    "oparg1",  
    "JUMPBY",
    "DISPATCH",
    "TIER_ONE_ONLY",
)



TIER_ONE: typing.Final = 1  
TIER_TWO: typing.Final = 2  
Tiers: typing.TypeAlias = typing.Literal[1, 2]


@dataclasses.dataclass
class Instruction:
    

    
    inst: parsing.InstDef
    name: str
    annotations: list[str]
    block: parsing.Block
    block_text: list[str]  
    block_line: int  

    
    always_exits: str  
    has_deopt: bool
    needs_this_instr: bool
    cache_offset: int
    cache_effects: list[parsing.CacheEffect]
    input_effects: list[StackEffect]
    output_effects: list[StackEffect]
    unmoved_names: frozenset[str]
    instr_fmt: str
    instr_flags: InstructionFlags
    active_caches: list[ActiveCacheEffect]

    
    family: parsing.Family | None = None
    predicted: bool = False

    def __init__(self, inst: parsing.InstDef):
        self.inst = inst
        self.name = inst.name
        self.annotations = inst.annotations
        self.block = inst.block
        self.block_text, self.check_eval_breaker, self.block_line = extract_block_text(
            self.block
        )
        self.always_exits = always_exits(self.block_text)
        self.has_deopt = variable_used(self.inst, "DEOPT_IF")
        self.cache_effects = [
            effect for effect in inst.inputs if isinstance(effect, parsing.CacheEffect)
        ]
        self.cache_offset = sum(c.size for c in self.cache_effects)
        self.needs_this_instr = variable_used(self.inst, "this_instr") or any(c.name != UNUSED for c in self.cache_effects)
        self.input_effects = [
            effect for effect in inst.inputs if isinstance(effect, StackEffect)
        ]
        self.output_effects = inst.outputs  
        unmoved_names: set[str] = set()
        for ieffect, oeffect in zip(self.input_effects, self.output_effects):
            if ieffect == oeffect and ieffect.name == oeffect.name:
                unmoved_names.add(ieffect.name)
            else:
                break
        self.unmoved_names = frozenset(unmoved_names)

        self.instr_flags = InstructionFlags.fromInstruction(inst)

        self.active_caches = []
        offset = 0
        for effect in self.cache_effects:
            if effect.name != UNUSED:
                self.active_caches.append(ActiveCacheEffect(effect, offset))
            offset += effect.size

        if self.instr_flags.HAS_ARG_FLAG:
            fmt = "IB"
        else:
            fmt = "IX"
        if offset:
            fmt += "C" + "0" * (offset - 1)
        self.instr_fmt = fmt

    def is_viable_uop(self) -> bool:
        
        dprint: typing.Callable[..., None] = lambda *args, **kwargs: None
        if "FRAME" in self.name:
            dprint = print

        if self.name == "_EXIT_TRACE":
            return True  
        if self.name == "_SAVE_RETURN_OFFSET":
            return True  
        if self.always_exits:
            dprint(f"Skipping {self.name} because it always exits: {self.always_exits}")
            return False
        if len(self.active_caches) > 1:
            
            return False
        res = True
        for forbidden in FORBIDDEN_NAMES_IN_UOPS:
            
            
            if variable_used_unspecialized(self.inst, forbidden):
                dprint(f"Skipping {self.name} because it uses {forbidden}")
                res = False
        return res

    def write_body(
        self,
        out: Formatter,
        dedent: int,
        active_caches: list[ActiveCacheEffect],
        tier: Tiers,
        family: parsing.Family | None,
    ) -> None:
        
        
        for active in active_caches:
            ceffect = active.effect
            bits = ceffect.size * BITS_PER_CODE_UNIT
            if bits == 64:
                
                
                
                
                typ = "PyObject *"
                func = "read_obj"
            else:
                typ = f"uint{bits}_t "
                func = f"read_u{bits}"
            if tier == TIER_ONE:
                out.emit(
                    f"{typ}{ceffect.name} = "
                    f"{func}(&this_instr[{active.offset + 1}].cache);"
                )
            else:
                out.emit(f"{typ}{ceffect.name} = ({typ.strip()})CURRENT_OPERAND();")

        
        assert dedent <= 0
        extra = " " * -dedent
        names_to_skip = self.unmoved_names | frozenset({UNUSED, "null"})
        offset = 0
        context = self.block.context
        assert context is not None and context.owner is not None
        filename = context.owner.filename
        for line in self.block_text:
            out.set_lineno(self.block_line + offset, filename)
            offset += 1
            if m := re.match(r"(\s*)ERROR_IF\((.+), (\w+)\);\s*(?://.*)?$", line):
                space, cond, label = m.groups()
                space = extra + space
                
                
                

                
                
                ieffs = list(self.input_effects)
                oeffs = list(self.output_effects)
                while (
                    ieffs
                    and oeffs
                    and ieffs[0] == oeffs[0]
                    and ieffs[0].name == oeffs[0].name
                ):
                    ieffs.pop(0)
                    oeffs.pop(0)
                ninputs, symbolic = list_effect_size(ieffs)
                if ninputs:
                    label = f"pop_{ninputs}_{label}"
                if tier == TIER_TWO:
                    label = label + "_tier_two"
                if symbolic:
                    out.write_raw(
                        f"{space}if ({cond}) {{ STACK_SHRINK({symbolic}); goto {label}; }}\n"
                    )
                else:
                    out.write_raw(f"{space}if ({cond}) goto {label};\n")
            elif m := re.match(r"(\s*)DEOPT_IF\((.+)\);\s*(?://.*)?$", line):
                space, cond = m.groups()
                space = extra + space
                target = family.name if family else self.name
                out.write_raw(f"{space}DEOPT_IF({cond}, {target});\n")
            elif "DEOPT" in line:
                filename = context.owner.filename
                lineno = context.owner.tokens[context.begin].line
                print(f"{filename}:{lineno}: ERROR: DEOPT_IF() must be all on one line")
                out.write_raw(extra + line)
            elif m := re.match(r"(\s*)DECREF_INPUTS\(\);\s*(?://.*)?$", line):
                out.reset_lineno()
                space = extra + m.group(1)
                for ieff in self.input_effects:
                    if ieff.name in names_to_skip:
                        continue
                    if ieff.size:
                        out.write_raw(
                            f"{space}for (int _i = {ieff.size}; --_i >= 0;) {{\n"
                        )
                        out.write_raw(f"{space}    Py_DECREF({ieff.name}[_i]);\n")
                        out.write_raw(f"{space}}}\n")
                    else:
                        decref = "XDECREF" if ieff.cond else "DECREF"
                        out.write_raw(f"{space}Py_{decref}({ieff.name});\n")
            else:
                out.write_raw(extra + line)
        out.reset_lineno()


InstructionOrCacheEffect = Instruction | parsing.CacheEffect



class AbstractInstruction(Instruction):
    def __init__(self, inst: parsing.InstDef):
        super().__init__(inst)

    def write(self, out: Formatter, tier: Tiers = TIER_ONE) -> None:
        
        stacking.write_single_instr_for_abstract_interp(self, out)

    def write_body(
        self,
        out: Formatter,
        dedent: int,
        active_caches: list[ActiveCacheEffect],
        tier: Tiers,
        family: parsing.Family | None,
    ) -> None:
        pass


@dataclasses.dataclass
class Component:
    instr: Instruction
    active_caches: list[ActiveCacheEffect]


MacroParts = list[Component | parsing.CacheEffect]


@dataclasses.dataclass
class MacroInstruction:
    

    name: str
    instr_fmt: str
    instr_flags: InstructionFlags
    macro: parsing.Macro
    parts: MacroParts
    cache_offset: int
    
    predicted: bool = False
    family: parsing.Family | None = None


@dataclasses.dataclass
class PseudoInstruction:
    

    name: str
    targets: list[Instruction | MacroInstruction]
    instr_flags: InstructionFlags


AnyInstruction = Instruction | MacroInstruction | PseudoInstruction


def extract_block_text(block: parsing.Block) -> tuple[list[str], bool, int]:
    
    blocklines = block.text.splitlines(True)
    first_token: lx.Token = block.tokens[0]  
    block_line = first_token.begin[0]

    
    while blocklines and not blocklines[0].strip():
        blocklines.pop(0)
        block_line += 1
    while blocklines and not blocklines[-1].strip():
        blocklines.pop()

    
    assert blocklines and blocklines[0].strip() == "{"
    assert blocklines and blocklines[-1].strip() == "}"
    blocklines.pop()
    blocklines.pop(0)
    block_line += 1

    
    while blocklines and not blocklines[-1].strip():
        blocklines.pop()

    
    check_eval_breaker = (
        blocklines != [] and blocklines[-1].strip() == "CHECK_EVAL_BREAKER();"
    )
    if check_eval_breaker:
        del blocklines[-1]

    return blocklines, check_eval_breaker, block_line


def always_exits(lines: list[str]) -> str:
    
    if not lines:
        return ""
    line = lines[-1].rstrip()
    
    if line[:12] != " " * 12:
        return ""
    line = line[12:]
    if line.startswith(
        (
            "goto ",
            "return ",
            "DISPATCH",
            "GO_TO_",
            "Py_UNREACHABLE()",
            "ERROR_IF(true, ",
        )
    ):
        return line
    return ""