| """Utilities for emitting C code.""" | |
| from __future__ import annotations | |
| import sys | |
| from collections.abc import Callable | |
| from typing import Final | |
| from mypyc.codegen.cstring import c_string_initializer | |
| from mypyc.codegen.literals import Literals, literal_sort_key | |
| from mypyc.common import ( | |
| ATTR_PREFIX, | |
| BITMAP_BITS, | |
| FAST_ISINSTANCE_MAX_SUBCLASSES, | |
| HAVE_IMMORTAL, | |
| MODULE_PREFIX, | |
| NATIVE_PREFIX, | |
| PREFIX, | |
| REG_PREFIX, | |
| STATIC_PREFIX, | |
| TYPE_PREFIX, | |
| TYPE_VAR_PREFIX, | |
| ) | |
| from mypyc.ir.class_ir import ClassIR, all_concrete_classes | |
| from mypyc.ir.func_ir import ( | |
| FUNC_CLASSMETHOD, | |
| FUNC_STATICMETHOD, | |
| FuncDecl, | |
| FuncIR, | |
| get_text_signature, | |
| ) | |
| from mypyc.ir.ops import ( | |
| NAMESPACE_MODULE, | |
| NAMESPACE_STATIC, | |
| NAMESPACE_TYPE, | |
| NAMESPACE_TYPE_VAR, | |
| BasicBlock, | |
| Value, | |
| ) | |
| from mypyc.ir.rtypes import ( | |
| RInstance, | |
| RPrimitive, | |
| RTuple, | |
| RType, | |
| RUnion, | |
| RVec, | |
| int_rprimitive, | |
| is_bool_or_bit_rprimitive, | |
| is_bytearray_rprimitive, | |
| is_bytes_rprimitive, | |
| is_dict_rprimitive, | |
| is_fixed_width_rtype, | |
| is_float_rprimitive, | |
| is_frozenset_rprimitive, | |
| is_int16_rprimitive, | |
| is_int32_rprimitive, | |
| is_int64_rprimitive, | |
| is_int_rprimitive, | |
| is_list_rprimitive, | |
| is_native_rprimitive, | |
| is_none_rprimitive, | |
| is_object_rprimitive, | |
| is_optional_type, | |
| is_range_rprimitive, | |
| is_set_rprimitive, | |
| is_short_int_rprimitive, | |
| is_str_rprimitive, | |
| is_tuple_rprimitive, | |
| is_uint8_rprimitive, | |
| object_rprimitive, | |
| optional_value_type, | |
| vec_api_by_item_type, | |
| vec_item_type_tags, | |
| ) | |
| from mypyc.namegen import NameGenerator, exported_name | |
| from mypyc.primitives.registry import builtin_names | |
| from mypyc.sametype import is_same_type | |
| # Whether to insert debug asserts for all error handling, to quickly | |
| # catch errors propagating without exceptions set. | |
| DEBUG_ERRORS: Final = False | |
| PREFIX_MAP: Final = { | |
| NAMESPACE_STATIC: STATIC_PREFIX, | |
| NAMESPACE_TYPE: TYPE_PREFIX, | |
| NAMESPACE_MODULE: MODULE_PREFIX, | |
| NAMESPACE_TYPE_VAR: TYPE_VAR_PREFIX, | |
| } | |
| # Map from RVec._ctype to C macro prefix for VEC_*_INCREF/DECREF/BUF macros | |
| VEC_MACRO_PREFIX: Final = { | |
| "VecI64": "VEC_I64", | |
| "VecI32": "VEC_I32", | |
| "VecI16": "VEC_I16", | |
| "VecU8": "VEC_U8", | |
| "VecFloat": "VEC_FLOAT", | |
| "VecBool": "VEC_BOOL", | |
| "VecT": "VEC_T", | |
| "VecNested": "VEC_NESTED", | |
| } | |
| class HeaderDeclaration: | |
| """A representation of a declaration in C. | |
| This is used to generate declarations in header files and | |
| (optionally) definitions in source files. | |
| Attributes: | |
| decl: C source code for the declaration. | |
| defn: Optionally, C source code for a definition. | |
| dependencies: The names of any objects that must be declared prior. | |
| is_type: Whether the declaration is of a C type. (C types will be declared in | |
| external header files and not marked 'extern'.) | |
| needs_export: Whether the declared object needs to be exported to | |
| other modules in the linking table. | |
| """ | |
| def __init__( | |
| self, | |
| decl: str | list[str], | |
| defn: list[str] | None = None, | |
| *, | |
| dependencies: set[str] | None = None, | |
| is_type: bool = False, | |
| needs_export: bool = False, | |
| ) -> None: | |
| self.decl = [decl] if isinstance(decl, str) else decl | |
| self.defn = defn | |
| self.dependencies = dependencies or set() | |
| self.is_type = is_type | |
| self.needs_export = needs_export | |
| class EmitterContext: | |
| """Shared emitter state for a compilation group.""" | |
| def __init__( | |
| self, | |
| names: NameGenerator, | |
| strict_traceback_checks: bool, | |
| group_name: str | None = None, | |
| group_map: dict[str, str | None] | None = None, | |
| ) -> None: | |
| """Setup shared emitter state. | |
| Args: | |
| names: The name generator to use | |
| group_map: Map from module names to group name | |
| group_name: Current group name | |
| """ | |
| self.temp_counter = 0 | |
| self.names = names | |
| self.group_name = group_name | |
| self.group_map = group_map or {} | |
| # Groups that this group depends on | |
| self.group_deps: set[str] = set() | |
| # The map below is used for generating declarations and | |
| # definitions at the top of the C file. The main idea is that they can | |
| # be generated at any time during the emit phase. | |
| # A map of a C identifier to whatever the C identifier declares. Currently this is | |
| # used for declaring structs and the key corresponds to the name of the struct. | |
| # The declaration contains the body of the struct. | |
| self.declarations: dict[str, HeaderDeclaration] = {} | |
| self.literals = Literals() | |
| # See mypyc/options.py for context. | |
| self.strict_traceback_checks = strict_traceback_checks | |
| class ErrorHandler: | |
| """Describes handling errors in unbox/cast operations.""" | |
| class AssignHandler(ErrorHandler): | |
| """Assign an error value on error.""" | |
| class GotoHandler(ErrorHandler): | |
| """Goto label on error.""" | |
| def __init__(self, label: str) -> None: | |
| self.label = label | |
| class TracebackAndGotoHandler(ErrorHandler): | |
| """Add traceback item and goto label on error.""" | |
| def __init__( | |
| self, label: str, source_path: str, module_name: str, traceback_entry: tuple[str, int] | |
| ) -> None: | |
| self.label = label | |
| self.source_path = source_path | |
| self.module_name = module_name | |
| self.traceback_entry = traceback_entry | |
| class ReturnHandler(ErrorHandler): | |
| """Return a constant value on error.""" | |
| def __init__(self, value: str) -> None: | |
| self.value = value | |
| class Emitter: | |
| """Helper for C code generation.""" | |
| def __init__( | |
| self, | |
| context: EmitterContext, | |
| value_names: dict[Value, str] | None = None, | |
| capi_version: tuple[int, int] | None = None, | |
| filepath: str | None = None, | |
| ) -> None: | |
| self.context = context | |
| self.capi_version = capi_version or sys.version_info[:2] | |
| self.names = context.names | |
| self.value_names = value_names or {} | |
| self.fragments: list[str] = [] | |
| self._indent = 0 | |
| self.filepath = filepath | |
| # Low-level operations | |
| def indent(self) -> None: | |
| self._indent += 4 | |
| def dedent(self) -> None: | |
| self._indent -= 4 | |
| assert self._indent >= 0 | |
| def label(self, label: BasicBlock) -> str: | |
| return "CPyL%s" % label.label | |
| def reg(self, reg: Value) -> str: | |
| return REG_PREFIX + self.value_names[reg] | |
| def attr(self, name: str) -> str: | |
| return ATTR_PREFIX + name | |
| def object_annotation(self, obj: object, line: str) -> str: | |
| """Build a C comment with a literal value's string representation. | |
| This is a debugging aid that makes generated C easier to read. | |
| If it contains illegal characters or is too long, return an empty string. | |
| """ | |
| formatted = stable_literal_repr(obj) | |
| if any(x in formatted for x in ("/*", "*/", "\0")) or len(formatted) >= 256: | |
| return "" | |
| return f" /* {formatted} */" | |
| def emit_line(self, line: str = "", *, ann: object = None) -> None: | |
| if line.startswith("}"): | |
| self.dedent() | |
| comment = self.object_annotation(ann, line) if ann is not None else "" | |
| self.fragments.append(self._indent * " " + line + comment + "\n") | |
| if line.endswith("{"): | |
| self.indent() | |
| def emit_lines(self, *lines: str) -> None: | |
| for line in lines: | |
| self.emit_line(line) | |
| def emit_label(self, label: BasicBlock | str) -> None: | |
| if isinstance(label, str): | |
| text = label | |
| else: | |
| if label.label == 0 or not label.referenced: | |
| return | |
| text = self.label(label) | |
| # Extra semicolon prevents an error when the next line declares a tempvar | |
| self.fragments.append(f"{text}: ;\n") | |
| def emit_from_emitter(self, emitter: Emitter) -> None: | |
| self.fragments.extend(emitter.fragments) | |
| def emit_printf(self, fmt: str, *args: str) -> None: | |
| fmt = fmt.replace("\n", "\\n") | |
| self.emit_line("printf(%s);" % ", ".join(['"%s"' % fmt] + list(args))) | |
| self.emit_line("fflush(stdout);") | |
| def temp_name(self) -> str: | |
| self.context.temp_counter += 1 | |
| return "__tmp%d" % self.context.temp_counter | |
| def new_label(self) -> str: | |
| self.context.temp_counter += 1 | |
| return "__LL%d" % self.context.temp_counter | |
| def get_module_group_prefix(self, module_name: str) -> str: | |
| """Get the group prefix for a module (relative to the current group). | |
| The prefix should be prepended to the object name whenever | |
| accessing an object from this module. | |
| If the module lives is in the current compilation group, there is | |
| no prefix. But if it lives in a different group (and hence a separate | |
| extension module), we need to access objects from it indirectly via an | |
| export table. | |
| For example, for code in group `a` to call a function `bar` in group `b`, | |
| it would need to do `exports_b.CPyDef_bar(...)`, while code that is | |
| also in group `b` can simply do `CPyDef_bar(...)`. | |
| Thus the prefix for a module in group `b` is 'exports_b.' if the current | |
| group is *not* b and just '' if it is. | |
| """ | |
| groups = self.context.group_map | |
| target_group_name = groups.get(module_name) | |
| if target_group_name and target_group_name != self.context.group_name: | |
| self.context.group_deps.add(target_group_name) | |
| return f"exports_{exported_name(target_group_name)}." | |
| else: | |
| return "" | |
| def get_group_prefix(self, obj: ClassIR | FuncDecl) -> str: | |
| """Get the group prefix for an object.""" | |
| # See docs above | |
| return self.get_module_group_prefix(obj.module_name) | |
| def register_group_dep(self, cl: ClassIR) -> None: | |
| """Record `cl`'s defining group as a cross-group dep, if any. | |
| Call this when emitting code that refers to `cl`'s struct | |
| layout: the .c file consuming that layout needs the defining | |
| group's `__native_*.h` included, and group_deps drives which | |
| headers get pulled in. | |
| """ | |
| target_group = self.context.group_map.get(cl.module_name) | |
| if target_group and target_group != self.context.group_name: | |
| self.context.group_deps.add(target_group) | |
| def static_name(self, id: str, module: str | None, prefix: str = STATIC_PREFIX) -> str: | |
| """Create name of a C static variable. | |
| These are used for literals and imported modules, among other | |
| things. | |
| The caller should ensure that the (id, module) pair cannot | |
| overlap with other calls to this method within a compilation | |
| group. | |
| """ | |
| lib_prefix = "" if not module else self.get_module_group_prefix(module) | |
| # If we are accessing static via the export table, we need to dereference | |
| # the pointer also. | |
| star_maybe = "*" if lib_prefix else "" | |
| suffix = self.names.private_name(module or "", id) | |
| return f"{star_maybe}{lib_prefix}{prefix}{suffix}" | |
| def type_struct_name(self, cl: ClassIR) -> str: | |
| return self.static_name(cl.name, cl.module_name, prefix=TYPE_PREFIX) | |
| def ctype(self, rtype: RType) -> str: | |
| return rtype._ctype | |
| def ctype_spaced(self, rtype: RType) -> str: | |
| """Adds a space after ctype for non-pointers.""" | |
| ctype = self.ctype(rtype) | |
| if ctype[-1] == "*": | |
| return ctype | |
| else: | |
| return ctype + " " | |
| def set_undefined_value(self, target: str, rtype: RType) -> None: | |
| if isinstance(rtype, RVec): | |
| self.emit_line(f"{target}.len = -1;") | |
| self.emit_line(f"{target}.items = NULL;") | |
| else: | |
| self.emit_line(f"{target} = {self.c_undefined_value(rtype)};") | |
| def c_undefined_value(self, rtype: RType) -> str: | |
| if not rtype.is_unboxed: | |
| return "NULL" | |
| elif isinstance(rtype, RPrimitive): | |
| return rtype.c_undefined | |
| elif isinstance(rtype, RTuple): | |
| return self.tuple_undefined_value(rtype) | |
| elif isinstance(rtype, RVec): | |
| return f"({self.ctype(rtype)}) {{ -1, NULL }}" | |
| assert False, rtype | |
| def c_error_value(self, rtype: RType) -> str: | |
| return self.c_undefined_value(rtype) | |
| def native_function_name(self, fn: FuncDecl) -> str: | |
| return f"{NATIVE_PREFIX}{fn.cname(self.names)}" | |
| def native_function_call(self, fn: FuncDecl) -> str: | |
| """Return the C expression for a call to `fn`'s native (CPyDef_) entry. | |
| For cross-group references under `separate=True`, this prepends the | |
| exports-table indirection (e.g. `exports_other.CPyDef_foo`). Same as | |
| `native_function_name()` for in-group calls. | |
| """ | |
| return f"{self.get_group_prefix(fn)}{self.native_function_name(fn)}" | |
| def wrapper_function_call(self, fn: FuncDecl) -> str: | |
| """Return the C expression for a call to `fn`'s Python-wrapper (CPyPy_) entry. | |
| Like `native_function_call`, but for the PyObject-level wrapper that | |
| boxes/unboxes arguments. Used from slot generators (tp_init, etc.). | |
| """ | |
| return f"{self.get_group_prefix(fn)}{PREFIX}{fn.cname(self.names)}" | |
| def tuple_c_declaration(self, rtuple: RTuple) -> list[str]: | |
| result = [ | |
| f"#ifndef MYPYC_DECLARED_{rtuple.struct_name}", | |
| f"#define MYPYC_DECLARED_{rtuple.struct_name}", | |
| f"typedef struct {rtuple.struct_name} {{", | |
| ] | |
| if len(rtuple.types) == 0: # empty tuple | |
| # Empty tuples contain a flag so that they can still indicate | |
| # error values. | |
| result.append("int empty_struct_error_flag;") | |
| else: | |
| i = 0 | |
| for typ in rtuple.types: | |
| result.append(f"{self.ctype_spaced(typ)}f{i};") | |
| i += 1 | |
| result.append(f"}} {rtuple.struct_name};") | |
| result.append("#endif") | |
| result.append("") | |
| return result | |
| def bitmap_field(self, index: int) -> str: | |
| """Return C field name used for attribute bitmap.""" | |
| n = index // BITMAP_BITS | |
| if n == 0: | |
| return "bitmap" | |
| return f"bitmap{n + 1}" | |
| def attr_bitmap_expr(self, obj: str, cl: ClassIR, index: int) -> str: | |
| """Return reference to the attribute definedness bitmap.""" | |
| cast = f"({cl.struct_name(self.names)} *)" | |
| attr = self.bitmap_field(index) | |
| return f"({cast}{obj})->{attr}" | |
| def emit_attr_bitmap_set( | |
| self, value: str, obj: str, rtype: RType, cl: ClassIR, attr: str | |
| ) -> None: | |
| """Mark an attribute as defined in the attribute bitmap. | |
| Assumes that the attribute is tracked in the bitmap (only some attributes | |
| use the bitmap). If 'value' is not equal to the error value, do nothing. | |
| """ | |
| self._emit_attr_bitmap_update(value, obj, rtype, cl, attr, clear=False) | |
| def emit_attr_bitmap_clear(self, obj: str, rtype: RType, cl: ClassIR, attr: str) -> None: | |
| """Mark an attribute as undefined in the attribute bitmap. | |
| Unlike emit_attr_bitmap_set, clear unconditionally. | |
| """ | |
| self._emit_attr_bitmap_update("", obj, rtype, cl, attr, clear=True) | |
| def _emit_attr_bitmap_update( | |
| self, value: str, obj: str, rtype: RType, cl: ClassIR, attr: str, clear: bool | |
| ) -> None: | |
| if value: | |
| check = self.error_value_check(rtype, value, "==") | |
| self.emit_line(f"if (unlikely({check})) {{") | |
| index = cl.bitmap_attrs.index(attr) | |
| mask = 1 << (index & (BITMAP_BITS - 1)) | |
| bitmap = self.attr_bitmap_expr(obj, cl, index) | |
| if clear: | |
| self.emit_line(f"{bitmap} &= ~{mask};") | |
| else: | |
| self.emit_line(f"{bitmap} |= {mask};") | |
| if value: | |
| self.emit_line("}") | |
| def emit_undefined_attr_check( | |
| self, | |
| rtype: RType, | |
| attr_expr: str, | |
| compare: str, | |
| obj: str, | |
| attr: str, | |
| cl: ClassIR, | |
| *, | |
| unlikely: bool = False, | |
| ) -> None: | |
| check = self.error_value_check(rtype, attr_expr, compare) | |
| if unlikely: | |
| check = f"unlikely({check})" | |
| if rtype.error_overlap: | |
| index = cl.bitmap_attrs.index(attr) | |
| bit = 1 << (index & (BITMAP_BITS - 1)) | |
| attr = self.bitmap_field(index) | |
| obj_expr = f"({cl.struct_name(self.names)} *){obj}" | |
| check = f"{check} && !(({obj_expr})->{attr} & {bit})" | |
| self.emit_line(f"if ({check}) {{") | |
| def error_value_check(self, rtype: RType, value: str, compare: str) -> str: | |
| if isinstance(rtype, RTuple): | |
| return self.tuple_undefined_check_cond( | |
| rtype, value, self.c_error_value, compare, check_exception=False | |
| ) | |
| elif isinstance(rtype, RVec): | |
| if compare == "==": | |
| return f"{value}.len < 0" | |
| elif compare == "!=": | |
| return f"{value}.len >= 0" | |
| assert False, compare | |
| else: | |
| return f"{value} {compare} {self.c_error_value(rtype)}" | |
| def tuple_undefined_check_cond( | |
| self, | |
| rtuple: RTuple, | |
| tuple_expr_in_c: str, | |
| c_type_compare_val: Callable[[RType], str], | |
| compare: str, | |
| *, | |
| check_exception: bool = True, | |
| ) -> str: | |
| if len(rtuple.types) == 0: | |
| # empty tuple | |
| return "{}.empty_struct_error_flag {} {}".format( | |
| tuple_expr_in_c, compare, c_type_compare_val(int_rprimitive) | |
| ) | |
| if rtuple.error_overlap: | |
| i = 0 | |
| item_type = rtuple.types[0] | |
| else: | |
| for i, typ in enumerate(rtuple.types): | |
| if not typ.error_overlap: | |
| item_type = rtuple.types[i] | |
| break | |
| else: | |
| assert False, "not expecting tuple with error overlap" | |
| if isinstance(item_type, RTuple): | |
| return self.tuple_undefined_check_cond( | |
| item_type, tuple_expr_in_c + f".f{i}", c_type_compare_val, compare | |
| ) | |
| elif isinstance(item_type, RVec): | |
| return f"{tuple_expr_in_c}.f{i}.len {compare} -1" | |
| else: | |
| check = f"{tuple_expr_in_c}.f{i} {compare} {c_type_compare_val(item_type)}" | |
| if rtuple.error_overlap and check_exception: | |
| check += " && PyErr_Occurred()" | |
| return check | |
| def tuple_undefined_value(self, rtuple: RTuple) -> str: | |
| """Undefined tuple value suitable in an expression.""" | |
| return f"({rtuple.struct_name}) {self.c_initializer_undefined_value(rtuple)}" | |
| def c_initializer_undefined_value(self, rtype: RType) -> str: | |
| """Undefined value represented in a form suitable for variable initialization.""" | |
| if isinstance(rtype, RTuple): | |
| if not rtype.types: | |
| # Empty tuples contain a flag so that they can still indicate | |
| # error values. | |
| return f"{{ {int_rprimitive.c_undefined} }}" | |
| items = ", ".join([self.c_initializer_undefined_value(t) for t in rtype.types]) | |
| return f"{{ {items} }}" | |
| elif isinstance(rtype, RVec): | |
| return "{ -1, NULL }" | |
| else: | |
| return self.c_undefined_value(rtype) | |
| # Higher-level operations | |
| def declare_tuple_struct(self, tuple_type: RTuple) -> None: | |
| if tuple_type.struct_name not in self.context.declarations: | |
| dependencies = set() | |
| for typ in tuple_type.types: | |
| # XXX other types might eventually need similar behavior | |
| if isinstance(typ, RTuple): | |
| dependencies.add(typ.struct_name) | |
| self.context.declarations[tuple_type.struct_name] = HeaderDeclaration( | |
| self.tuple_c_declaration(tuple_type), dependencies=dependencies, is_type=True | |
| ) | |
| def emit_inc_ref(self, dest: str, rtype: RType, *, rare: bool = False) -> None: | |
| """Increment reference count of C expression `dest`. | |
| For composite unboxed structures (e.g. tuples) recursively | |
| increment reference counts for each component. | |
| If rare is True, optimize for code size and compilation speed. | |
| """ | |
| if is_int_rprimitive(rtype): | |
| if rare: | |
| self.emit_line("CPyTagged_IncRef(%s);" % dest) | |
| else: | |
| self.emit_line("CPyTagged_INCREF(%s);" % dest) | |
| elif isinstance(rtype, RTuple): | |
| for i, item_type in enumerate(rtype.types): | |
| self.emit_inc_ref(f"{dest}.f{i}", item_type) | |
| elif isinstance(rtype, RVec): | |
| prefix = VEC_MACRO_PREFIX[rtype._ctype] | |
| self.emit_line(f"{prefix}_INCREF({dest});") | |
| elif not rtype.is_unboxed: | |
| # Always inline, since this is a simple but very hot op | |
| if rtype.may_be_immortal or not HAVE_IMMORTAL: | |
| self.emit_line("CPy_INCREF(%s);" % dest) | |
| else: | |
| self.emit_line("CPy_INCREF_NO_IMM(%s);" % dest) | |
| # Otherwise assume it's an unboxed, pointerless value and do nothing. | |
| def emit_dec_ref( | |
| self, dest: str, rtype: RType, *, is_xdec: bool = False, rare: bool = False | |
| ) -> None: | |
| """Decrement reference count of C expression `dest`. | |
| For composite unboxed structures (e.g. tuples) recursively | |
| decrement reference counts for each component. | |
| If rare is True, optimize for code size and compilation speed. | |
| """ | |
| x = "X" if is_xdec else "" | |
| if is_int_rprimitive(rtype): | |
| if rare: | |
| self.emit_line(f"CPyTagged_{x}DecRef({dest});") | |
| else: | |
| # Inlined | |
| self.emit_line(f"CPyTagged_{x}DECREF({dest});") | |
| elif isinstance(rtype, RTuple): | |
| for i, item_type in enumerate(rtype.types): | |
| self.emit_dec_ref(f"{dest}.f{i}", item_type, is_xdec=is_xdec, rare=rare) | |
| elif isinstance(rtype, RVec): | |
| prefix = VEC_MACRO_PREFIX[rtype._ctype] | |
| self.emit_line(f"{prefix}_DECREF({dest});") | |
| elif not rtype.is_unboxed: | |
| if rare: | |
| self.emit_line(f"CPy_{x}DecRef({dest});") | |
| else: | |
| # Inlined | |
| if rtype.may_be_immortal or not HAVE_IMMORTAL: | |
| self.emit_line(f"CPy_{x}DECREF({dest});") | |
| else: | |
| self.emit_line(f"CPy_{x}DECREF_NO_IMM({dest});") | |
| elif rtype.is_refcounted: | |
| assert False, f"dec_ref not implemented for {rtype}" | |
| # Otherwise assume it's an unboxed, pointerless value and do nothing. | |
| def pretty_name(self, typ: RType) -> str: | |
| value_type = optional_value_type(typ) | |
| if value_type is not None: | |
| return "%s or None" % self.pretty_name(value_type) | |
| return str(typ) | |
| def emit_cast( | |
| self, | |
| src: str, | |
| dest: str, | |
| typ: RType, | |
| *, | |
| declare_dest: bool = False, | |
| error: ErrorHandler | None = None, | |
| raise_exception: bool = True, | |
| optional: bool = False, | |
| src_type: RType | None = None, | |
| likely: bool = True, | |
| ) -> None: | |
| """Emit code for casting a value of given type. | |
| Somewhat strangely, this supports unboxed types but only | |
| operates on boxed versions. This is necessary to properly | |
| handle types such as Optional[int] in compatibility glue. | |
| By default, assign NULL (error value) to dest if the value has | |
| an incompatible type and raise TypeError. These can be customized | |
| using 'error' and 'raise_exception'. | |
| Always copy/steal the reference in 'src'. | |
| Args: | |
| src: Name of source C variable | |
| dest: Name of target C variable | |
| typ: Type of value | |
| declare_dest: If True, also declare the variable 'dest' | |
| error: What happens on error | |
| raise_exception: If True, also raise TypeError on failure | |
| likely: If the cast is likely to succeed (can be False for unions) | |
| """ | |
| error = error or AssignHandler() | |
| # Special case casting *from* optional | |
| if src_type and is_optional_type(src_type) and not is_object_rprimitive(typ): | |
| value_type = optional_value_type(src_type) | |
| assert value_type is not None | |
| if is_same_type(value_type, typ): | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| check = "({} != Py_None)" | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check.format(src), optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| return | |
| # TODO: Verify refcount handling. | |
| if ( | |
| is_list_rprimitive(typ) | |
| or is_dict_rprimitive(typ) | |
| or is_set_rprimitive(typ) | |
| or is_frozenset_rprimitive(typ) | |
| or is_str_rprimitive(typ) | |
| or is_range_rprimitive(typ) | |
| or is_float_rprimitive(typ) | |
| or is_int_rprimitive(typ) | |
| or is_bool_or_bit_rprimitive(typ) | |
| or is_fixed_width_rtype(typ) | |
| ): | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| if is_list_rprimitive(typ): | |
| prefix = "PyList" | |
| elif is_dict_rprimitive(typ): | |
| prefix = "PyDict" | |
| elif is_set_rprimitive(typ): | |
| prefix = "PySet" | |
| elif is_frozenset_rprimitive(typ): | |
| prefix = "PyFrozenSet" | |
| elif is_str_rprimitive(typ): | |
| prefix = "PyUnicode" | |
| elif is_range_rprimitive(typ): | |
| prefix = "PyRange" | |
| elif is_float_rprimitive(typ): | |
| prefix = "CPyFloat" | |
| elif is_int_rprimitive(typ) or is_fixed_width_rtype(typ): | |
| # TODO: Range check for fixed-width types? | |
| prefix = "PyLong" | |
| elif is_bool_or_bit_rprimitive(typ): | |
| prefix = "PyBool" | |
| else: | |
| assert False, f"unexpected primitive type: {typ}" | |
| check = "({}_Check({}))" | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check.format(prefix, src), optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| elif is_bytes_rprimitive(typ): | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| check = "(PyBytes_Check({}))" | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check.format(src, src), optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| elif is_bytearray_rprimitive(typ): | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| check = "(PyByteArray_Check({}))" | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check.format(src, src), optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| elif is_tuple_rprimitive(typ): | |
| if declare_dest: | |
| self.emit_line(f"{self.ctype(typ)} {dest};") | |
| check = "(PyTuple_Check({}))" | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check.format(src), optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| elif isinstance(typ, RInstance): | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| concrete = all_concrete_classes(typ.class_ir) | |
| # If there are too many concrete subclasses or we can't find any | |
| # (meaning the code ought to be dead or we aren't doing global opts), | |
| # fall back to a normal typecheck. | |
| # Otherwise check all the subclasses. | |
| if not concrete or len(concrete) > FAST_ISINSTANCE_MAX_SUBCLASSES + 1: | |
| check = "(PyObject_TypeCheck({}, {}))".format( | |
| src, self.type_struct_name(typ.class_ir) | |
| ) | |
| else: | |
| full_str = "(Py_TYPE({src}) == {targets[0]})" | |
| for i in range(1, len(concrete)): | |
| full_str += " || (Py_TYPE({src}) == {targets[%d]})" % i | |
| if len(concrete) > 1: | |
| full_str = "(%s)" % full_str | |
| check = full_str.format( | |
| src=src, targets=[self.type_struct_name(ir) for ir in concrete] | |
| ) | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check, optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| elif is_none_rprimitive(typ): | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| check = "({} == Py_None)" | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check.format(src), optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| elif is_object_rprimitive(typ): | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| self.emit_arg_check(src, dest, typ, "", optional) | |
| self.emit_line(f"{dest} = {src};") | |
| if optional: | |
| self.emit_line("}") | |
| elif is_native_rprimitive(typ): | |
| # Native primitive types have type check functions of form "CPy<Name>_Check(...)". | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| short_name = typ.name.rsplit(".", 1)[-1] | |
| check = f"(CPy{short_name}_Check({src}))" | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check, optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| elif isinstance(typ, RUnion): | |
| self.emit_union_cast( | |
| src, dest, typ, declare_dest, error, optional, src_type, raise_exception | |
| ) | |
| elif isinstance(typ, RTuple): | |
| assert not optional | |
| self.emit_tuple_cast(src, dest, typ, declare_dest, error, src_type) | |
| elif isinstance(typ, RVec): | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| # Build type check expression based on vec kind | |
| api_name = vec_api_by_item_type.get(typ.item_type) | |
| depth = typ.depth() | |
| if api_name: | |
| # Specialized vec types (vec[i64], vec[i32], etc.) | |
| check = f"(Py_TYPE({src}) == {api_name}.boxed_type)" | |
| elif depth == 0: | |
| # Generic vec types (vec[T], vec[T | None]) with reference type items | |
| item_type_c = self.vec_item_type_c(typ) | |
| check = ( | |
| f"(Py_TYPE({src}) == VecTApi.boxed_type && " | |
| f"VEC_T_BUF(((VecTObject *){src})->vec)->item_type == {item_type_c})" | |
| ) | |
| else: | |
| # Nested vec types (vec[vec[...]]). Check boxed type, item type, and depth. | |
| unwrapped = typ.unwrap_item_type() | |
| if unwrapped in vec_item_type_tags: | |
| type_value = str(vec_item_type_tags[unwrapped]) | |
| else: | |
| type_value = self.vec_item_type_c(typ) | |
| check = ( | |
| f"(Py_TYPE({src}) == VecNestedApi.boxed_type && " | |
| f"VEC_NESTED_BUF(((VecNestedObject *){src})->vec)->item_type == {type_value} && " | |
| f"VEC_NESTED_BUF(((VecNestedObject *){src})->vec)->depth == {depth})" | |
| ) | |
| if likely: | |
| check = f"(likely{check})" | |
| self.emit_arg_check(src, dest, typ, check, optional) | |
| self.emit_lines(f" {dest} = {src};", "else {") | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_line("}") | |
| else: | |
| assert False, "Cast not implemented: %s" % typ | |
| def emit_cast_error_handler( | |
| self, error: ErrorHandler, src: str, dest: str, typ: RType, raise_exception: bool | |
| ) -> None: | |
| if raise_exception: | |
| if isinstance(error, TracebackAndGotoHandler): | |
| # Merge raising and emitting traceback entry into a single call. | |
| self.emit_type_error_traceback( | |
| error.source_path, error.module_name, error.traceback_entry, typ=typ, src=src | |
| ) | |
| self.emit_line("goto %s;" % error.label) | |
| return | |
| self.emit_line(f'CPy_TypeError("{self.pretty_name(typ)}", {src}); ') | |
| if isinstance(error, AssignHandler): | |
| self.emit_line("%s = NULL;" % dest) | |
| elif isinstance(error, GotoHandler): | |
| self.emit_line("goto %s;" % error.label) | |
| elif isinstance(error, TracebackAndGotoHandler): | |
| self.emit_line("%s = NULL;" % dest) | |
| self.emit_traceback(error.source_path, error.module_name, error.traceback_entry) | |
| self.emit_line("goto %s;" % error.label) | |
| else: | |
| assert isinstance(error, ReturnHandler), error | |
| self.emit_line("return %s;" % error.value) | |
| def emit_union_cast( | |
| self, | |
| src: str, | |
| dest: str, | |
| typ: RUnion, | |
| declare_dest: bool, | |
| error: ErrorHandler, | |
| optional: bool, | |
| src_type: RType | None, | |
| raise_exception: bool, | |
| ) -> None: | |
| """Emit cast to a union type. | |
| The arguments are similar to emit_cast. | |
| """ | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| good_label = self.new_label() | |
| if optional: | |
| self.emit_line(f"if ({src} == NULL) {{") | |
| self.emit_line(f"{dest} = {self.c_error_value(typ)};") | |
| self.emit_line(f"goto {good_label};") | |
| self.emit_line("}") | |
| for item in typ.items: | |
| self.emit_cast( | |
| src, | |
| dest, | |
| item, | |
| declare_dest=False, | |
| raise_exception=False, | |
| optional=False, | |
| likely=False, | |
| ) | |
| self.emit_line(f"if ({dest} != NULL) goto {good_label};") | |
| # Handle cast failure. | |
| self.emit_cast_error_handler(error, src, dest, typ, raise_exception) | |
| self.emit_label(good_label) | |
| def emit_tuple_cast( | |
| self, | |
| src: str, | |
| dest: str, | |
| typ: RTuple, | |
| declare_dest: bool, | |
| error: ErrorHandler, | |
| src_type: RType | None, | |
| ) -> None: | |
| """Emit cast to a tuple type. | |
| The arguments are similar to emit_cast. | |
| """ | |
| if declare_dest: | |
| self.emit_line(f"PyObject *{dest};") | |
| # This reuse of the variable is super dodgy. We don't even | |
| # care about the values except to check whether they are | |
| # invalid. | |
| out_label = self.new_label() | |
| self.emit_lines( | |
| "if (unlikely(!(PyTuple_Check({r}) && PyTuple_GET_SIZE({r}) == {size}))) {{".format( | |
| r=src, size=len(typ.types) | |
| ), | |
| f"{dest} = NULL;", | |
| f"goto {out_label};", | |
| "}", | |
| ) | |
| for i, item in enumerate(typ.types): | |
| # Since we did the checks above this should never fail | |
| self.emit_cast( | |
| f"PyTuple_GET_ITEM({src}, {i})", | |
| dest, | |
| item, | |
| declare_dest=False, | |
| raise_exception=False, | |
| optional=False, | |
| ) | |
| self.emit_line(f"if ({dest} == NULL) goto {out_label};") | |
| self.emit_line(f"{dest} = {src};") | |
| self.emit_label(out_label) | |
| def emit_arg_check(self, src: str, dest: str, typ: RType, check: str, optional: bool) -> None: | |
| if optional: | |
| self.emit_line(f"if ({src} == NULL) {{") | |
| self.emit_line(f"{dest} = {self.c_error_value(typ)};") | |
| if check != "": | |
| self.emit_line("{}if {}".format("} else " if optional else "", check)) | |
| elif optional: | |
| self.emit_line("else {") | |
| def emit_unbox( | |
| self, | |
| src: str, | |
| dest: str, | |
| typ: RType, | |
| *, | |
| declare_dest: bool = False, | |
| error: ErrorHandler | None = None, | |
| raise_exception: bool = True, | |
| optional: bool = False, | |
| borrow: bool = False, | |
| ) -> None: | |
| """Emit code for unboxing a value of given type (from PyObject *). | |
| By default, assign error value to dest if the value has an | |
| incompatible type and raise TypeError. These can be customized | |
| using 'error' and 'raise_exception'. | |
| Generate a new reference unless 'borrow' is True. | |
| Args: | |
| src: Name of source C variable | |
| dest: Name of target C variable | |
| typ: Type of value | |
| declare_dest: If True, also declare the variable 'dest' | |
| error: What happens on error | |
| raise_exception: If True, also raise TypeError on failure | |
| optional: If True, NULL src value is allowed and will map to error value | |
| borrow: If True, create a borrowed reference | |
| """ | |
| error = error or AssignHandler() | |
| # TODO: Verify refcount handling. | |
| if isinstance(error, AssignHandler): | |
| error_action = f"{dest} = {self.c_error_value(typ)};" | |
| elif isinstance(error, GotoHandler): | |
| error_action = "goto %s;" % error.label | |
| else: | |
| assert isinstance(error, ReturnHandler), error | |
| error_action = "return %s;" % error.value | |
| failure = error_action | |
| if raise_exception: | |
| raise_exc = f'CPy_TypeError("{self.pretty_name(typ)}", {src}); ' | |
| failure = raise_exc + failure | |
| if is_int_rprimitive(typ) or is_short_int_rprimitive(typ): | |
| if declare_dest: | |
| self.emit_line(f"CPyTagged {dest};") | |
| self.emit_arg_check(src, dest, typ, f"(likely(PyLong_Check({src})))", optional) | |
| if borrow: | |
| self.emit_line(f" {dest} = CPyTagged_BorrowFromObject({src});") | |
| else: | |
| self.emit_line(f" {dest} = CPyTagged_FromObject({src});") | |
| self.emit_line("else {") | |
| self.emit_line(failure) | |
| self.emit_line("}") | |
| elif is_bool_or_bit_rprimitive(typ): | |
| # Whether we are borrowing or not makes no difference. | |
| if declare_dest: | |
| self.emit_line(f"char {dest};") | |
| self.emit_arg_check(src, dest, typ, f"(unlikely(!PyBool_Check({src}))) {{", optional) | |
| self.emit_line(failure) | |
| self.emit_line("} else") | |
| conversion = f"{src} == Py_True" | |
| self.emit_line(f" {dest} = {conversion};") | |
| elif is_none_rprimitive(typ): | |
| # Whether we are borrowing or not makes no difference. | |
| if declare_dest: | |
| self.emit_line(f"char {dest};") | |
| self.emit_arg_check(src, dest, typ, f"(unlikely({src} != Py_None)) {{", optional) | |
| self.emit_line(failure) | |
| self.emit_line("} else") | |
| self.emit_line(f" {dest} = 1;") | |
| elif is_int64_rprimitive(typ): | |
| # Whether we are borrowing or not makes no difference. | |
| assert not optional # Not supported for overlapping error values | |
| if declare_dest: | |
| self.emit_line(f"int64_t {dest};") | |
| self.emit_line(f"{dest} = CPyLong_AsInt64({src});") | |
| if not isinstance(error, AssignHandler): | |
| self.emit_unbox_failure_with_overlapping_error_value(dest, typ, failure) | |
| elif is_int32_rprimitive(typ): | |
| # Whether we are borrowing or not makes no difference. | |
| assert not optional # Not supported for overlapping error values | |
| if declare_dest: | |
| self.emit_line(f"int32_t {dest};") | |
| self.emit_line(f"{dest} = CPyLong_AsInt32({src});") | |
| if not isinstance(error, AssignHandler): | |
| self.emit_unbox_failure_with_overlapping_error_value(dest, typ, failure) | |
| elif is_int16_rprimitive(typ): | |
| # Whether we are borrowing or not makes no difference. | |
| assert not optional # Not supported for overlapping error values | |
| if declare_dest: | |
| self.emit_line(f"int16_t {dest};") | |
| self.emit_line(f"{dest} = CPyLong_AsInt16({src});") | |
| if not isinstance(error, AssignHandler): | |
| self.emit_unbox_failure_with_overlapping_error_value(dest, typ, failure) | |
| elif is_uint8_rprimitive(typ): | |
| # Whether we are borrowing or not makes no difference. | |
| assert not optional # Not supported for overlapping error values | |
| if declare_dest: | |
| self.emit_line(f"uint8_t {dest};") | |
| self.emit_line(f"{dest} = CPyLong_AsUInt8({src});") | |
| if not isinstance(error, AssignHandler): | |
| self.emit_unbox_failure_with_overlapping_error_value(dest, typ, failure) | |
| elif is_float_rprimitive(typ): | |
| assert not optional # Not supported for overlapping error values | |
| if declare_dest: | |
| self.emit_line(f"double {dest};") | |
| # TODO: Don't use __float__ and __index__ | |
| self.emit_line(f"{dest} = PyFloat_AsDouble({src});") | |
| self.emit_lines(f"if ({dest} == -1.0 && PyErr_Occurred()) {{", failure, "}") | |
| elif isinstance(typ, RTuple): | |
| self.declare_tuple_struct(typ) | |
| if declare_dest: | |
| self.emit_line(f"{self.ctype(typ)} {dest};") | |
| if optional: | |
| self.emit_line(f"if ({src} == NULL) {{") | |
| self.emit_line(f"{dest} = {self.c_error_value(typ)};") | |
| self.emit_line("} else {") | |
| item_error_labels: list[tuple[str, int]] = [] | |
| cast_temp = self.temp_name() | |
| self.emit_tuple_cast( | |
| src, cast_temp, typ, declare_dest=True, error=error, src_type=None | |
| ) | |
| self.emit_line(f"if (unlikely({cast_temp} == NULL)) {{") | |
| # self.emit_arg_check(src, dest, typ, | |
| # '(!PyTuple_Check({}) || PyTuple_Size({}) != {}) {{'.format( | |
| # src, src, len(typ.types)), optional) | |
| self.emit_line(failure) # TODO: Decrease refcount? | |
| self.emit_line("} else {") | |
| if not typ.types: | |
| self.emit_line(f"{dest}.empty_struct_error_flag = 0;") | |
| for i, item_type in enumerate(typ.types): | |
| temp = self.temp_name() | |
| # emit_tuple_cast above checks the size, so this should not fail | |
| self.emit_line(f"PyObject *{temp} = PyTuple_GET_ITEM({src}, {i});") | |
| temp2 = self.temp_name() | |
| # Unbox or check the item. | |
| if item_type.is_unboxed: | |
| item_error_label = self.new_label() | |
| item_error_labels.append((item_error_label, i)) | |
| self.emit_unbox( | |
| temp, | |
| temp2, | |
| item_type, | |
| # The tuple cast has already checked the item type. In the | |
| # normal AssignHandler case, preserve conversion errors such | |
| # as an integer overflow instead of replacing them with a | |
| # TypeError. | |
| raise_exception=raise_exception and not isinstance(error, AssignHandler), | |
| error=GotoHandler(item_error_label), | |
| declare_dest=True, | |
| borrow=borrow, | |
| ) | |
| else: | |
| if not borrow: | |
| self.emit_inc_ref(temp, object_rprimitive) | |
| self.emit_cast(temp, temp2, item_type, declare_dest=True) | |
| self.emit_line(f"{dest}.f{i} = {temp2};") | |
| self.emit_line("}") | |
| if item_error_labels: | |
| done_label = self.new_label() | |
| self.emit_line(f"goto {done_label};") | |
| for item_error_label, failed_item in item_error_labels: | |
| self.emit_label(item_error_label) | |
| # The failed field has not been assigned yet, but earlier fields | |
| # may own references that must be released before propagating. | |
| if not borrow: | |
| for previous_index, previous_type in enumerate(typ.types[:failed_item]): | |
| self.emit_dec_ref(f"{dest}.f{previous_index}", previous_type) | |
| self.emit_line(error_action) | |
| if isinstance(error, AssignHandler): | |
| self.emit_line(f"goto {done_label};") | |
| self.emit_label(done_label) | |
| if optional: | |
| self.emit_line("}") | |
| elif isinstance(typ, RVec): | |
| if declare_dest: | |
| self.emit_line(f"{self.ctype(typ)} {dest};") | |
| if optional: | |
| self.emit_line(f"if ({src} == NULL) {{") | |
| self.emit_line(f"{dest} = {self.c_error_value(typ)};") | |
| self.emit_line("} else {") | |
| specialized_api_name = vec_api_by_item_type.get(typ.item_type) | |
| if specialized_api_name is not None: | |
| self.emit_line(f"{dest} = {specialized_api_name}.unbox({src});") | |
| else: | |
| depth = typ.depth() | |
| unwrapped = typ.unwrap_item_type() | |
| if unwrapped in vec_item_type_tags: | |
| type_value = str(vec_item_type_tags[unwrapped]) | |
| else: | |
| type_value = self.vec_item_type_c(typ) | |
| if depth == 0: | |
| self.emit_line(f"{dest} = VecTApi.unbox({src}, {type_value});") | |
| else: | |
| self.emit_line(f"{dest} = VecNestedApi.unbox({src}, {type_value}, {depth});") | |
| self.emit_line(f"if (VEC_IS_ERROR({dest})) {{") | |
| self.emit_line(failure) | |
| self.emit_line("}") | |
| if optional: | |
| self.emit_line("}") | |
| else: | |
| assert False, "Unboxing not implemented: %s" % typ | |
| def vec_item_type_c(self, typ: RVec) -> str: | |
| item_type = typ.unwrap_item_type() | |
| type_c_ptr = self.type_c_ptr(item_type) | |
| # Can never be None, since we unwrapped the item type above | |
| assert type_c_ptr is not None | |
| type_value = f"(size_t){type_c_ptr}" | |
| if typ.is_optional(): | |
| type_value = f"({type_value} | 1)" | |
| return type_value | |
| def type_c_ptr(self, typ: RPrimitive | RInstance) -> str | None: | |
| if isinstance(typ, RPrimitive) and typ.is_refcounted: | |
| return "&" + builtin_names[typ.name][1] | |
| elif isinstance(typ, RInstance): | |
| return self.type_struct_name(typ.class_ir) | |
| return None | |
| def emit_box( | |
| self, src: str, dest: str, typ: RType, declare_dest: bool = False, can_borrow: bool = False | |
| ) -> None: | |
| """Emit code for boxing a value of given type. | |
| Generate a simple assignment if no boxing is needed. | |
| The source reference count is stolen for the result (no need to decref afterwards). | |
| """ | |
| # TODO: Always generate a new reference (if a reference type) | |
| if declare_dest: | |
| declaration = "PyObject *" | |
| else: | |
| declaration = "" | |
| if is_int_rprimitive(typ) or is_short_int_rprimitive(typ): | |
| # Steal the existing reference if it exists. | |
| self.emit_line(f"{declaration}{dest} = CPyTagged_StealAsObject({src});") | |
| elif is_bool_or_bit_rprimitive(typ): | |
| # N.B: bool is special cased to produce a borrowed value | |
| # after boxing, so we don't need to increment the refcount | |
| # when this comes directly from a Box op. | |
| self.emit_lines(f"{declaration}{dest} = {src} ? Py_True : Py_False;") | |
| if not can_borrow: | |
| self.emit_inc_ref(dest, object_rprimitive) | |
| elif is_none_rprimitive(typ): | |
| # N.B: None is special cased to produce a borrowed value | |
| # after boxing, so we don't need to increment the refcount | |
| # when this comes directly from a Box op. | |
| self.emit_lines(f"{declaration}{dest} = Py_None;") | |
| if not can_borrow: | |
| self.emit_inc_ref(dest, object_rprimitive) | |
| elif is_int32_rprimitive(typ) or is_int16_rprimitive(typ) or is_uint8_rprimitive(typ): | |
| self.emit_line(f"{declaration}{dest} = PyLong_FromLong({src});") | |
| elif is_int64_rprimitive(typ): | |
| self.emit_line(f"{declaration}{dest} = PyLong_FromLongLong({src});") | |
| elif is_float_rprimitive(typ): | |
| self.emit_line(f"{declaration}{dest} = PyFloat_FromDouble({src});") | |
| elif isinstance(typ, RTuple): | |
| self.declare_tuple_struct(typ) | |
| if not typ.types: | |
| self.emit_line(f"{declaration}{dest} = CPyTuple_LoadEmptyTupleConstant();") | |
| else: | |
| self.emit_line(f"{declaration}{dest} = PyTuple_New({len(typ.types)});") | |
| self.emit_line(f"if (unlikely({dest} == NULL))") | |
| self.emit_line(" CPyError_OutOfMemory();") | |
| # TODO: Fail if dest is None | |
| for i in range(len(typ.types)): | |
| if not typ.is_unboxed: | |
| self.emit_line(f"PyTuple_SET_ITEM({dest}, {i}, {src}.f{i}") | |
| else: | |
| inner_name = self.temp_name() | |
| self.emit_box(f"{src}.f{i}", inner_name, typ.types[i], declare_dest=True) | |
| self.emit_line(f"PyTuple_SET_ITEM({dest}, {i}, {inner_name});") | |
| elif isinstance(typ, RVec): | |
| specialized_api_name = vec_api_by_item_type.get(typ.item_type) | |
| if specialized_api_name is not None: | |
| api = specialized_api_name | |
| elif typ.depth() > 0: | |
| api = "VecNestedApi" | |
| else: | |
| api = "VecTApi" | |
| # Empty vecs of this sort don't describe item type, so it needs to be | |
| # passed explicitly. | |
| item_type = self.vec_item_type_c(typ) | |
| self.emit_line(f"{declaration}{dest} = {api}.box({src}, {item_type});") | |
| return | |
| self.emit_line(f"{declaration}{dest} = {api}.box({src});") | |
| else: | |
| assert not typ.is_unboxed | |
| # Type is boxed -- trivially just assign. | |
| self.emit_line(f"{declaration}{dest} = {src};") | |
| def emit_error_check(self, value: str, rtype: RType, failure: str) -> None: | |
| """Emit code for checking a native function return value for uncaught exception.""" | |
| if isinstance(rtype, RTuple): | |
| if len(rtype.types) == 0: | |
| return # empty tuples can't fail. | |
| else: | |
| cond = self.tuple_undefined_check_cond(rtype, value, self.c_error_value, "==") | |
| self.emit_line(f"if ({cond}) {{") | |
| elif isinstance(rtype, RVec): | |
| self.emit_line(f"if ({value}.len < 0) {{") | |
| elif rtype.error_overlap: | |
| # The error value is also valid as a normal value, so we need to also check | |
| # for a raised exception. | |
| self.emit_line(f"if ({value} == {self.c_error_value(rtype)} && PyErr_Occurred()) {{") | |
| else: | |
| self.emit_line(f"if ({value} == {self.c_error_value(rtype)}) {{") | |
| self.emit_lines(failure, "}") | |
| def emit_gc_visit(self, target: str, rtype: RType) -> None: | |
| """Emit code for GC visiting a C variable reference. | |
| Assume that 'target' represents a C expression that refers to a | |
| struct member, such as 'self->x'. | |
| """ | |
| if not rtype.is_refcounted: | |
| # Not refcounted -> no pointers -> no GC interaction. | |
| return | |
| elif isinstance(rtype, RPrimitive) and rtype.name == "builtins.int": | |
| self.emit_line(f"if (CPyTagged_CheckLong({target})) {{") | |
| self.emit_line(f"Py_VISIT(CPyTagged_LongAsObject({target}));") | |
| self.emit_line("}") | |
| elif isinstance(rtype, RTuple): | |
| for i, item_type in enumerate(rtype.types): | |
| self.emit_gc_visit(f"{target}.f{i}", item_type) | |
| elif isinstance(rtype, RVec): | |
| prefix = VEC_MACRO_PREFIX[rtype._ctype] | |
| self.emit_line(f"if ({target}.items) {{ Py_VISIT({prefix}_BUF({target})); }}") | |
| elif self.ctype(rtype) == "PyObject *": | |
| # The simplest case. | |
| self.emit_line(f"Py_VISIT({target});") | |
| else: | |
| assert False, "emit_gc_visit() not implemented for %s" % repr(rtype) | |
| def emit_gc_clear(self, target: str, rtype: RType) -> None: | |
| """Emit code for clearing a C attribute reference for GC. | |
| Assume that 'target' represents a C expression that refers to a | |
| struct member, such as 'self->x'. | |
| """ | |
| if not rtype.is_refcounted: | |
| # Not refcounted -> no pointers -> no GC interaction. | |
| return | |
| elif isinstance(rtype, RPrimitive) and rtype.name == "builtins.int": | |
| self.emit_line(f"if (CPyTagged_CheckLong({target})) {{") | |
| self.emit_line(f"CPyTagged __tmp = {target};") | |
| self.emit_line(f"{target} = {self.c_undefined_value(rtype)};") | |
| self.emit_line("Py_XDECREF(CPyTagged_LongAsObject(__tmp));") | |
| self.emit_line("}") | |
| elif isinstance(rtype, RTuple): | |
| for i, item_type in enumerate(rtype.types): | |
| self.emit_gc_clear(f"{target}.f{i}", item_type) | |
| elif isinstance(rtype, RVec): | |
| prefix = VEC_MACRO_PREFIX[rtype._ctype] | |
| self.emit_line(f"if ({target}.items) {{") | |
| self.emit_line(f" Py_DECREF({prefix}_BUF({target}));") | |
| self.emit_line(f" {target}.items = NULL;") | |
| self.emit_line("}") | |
| elif self.ctype(rtype) == "PyObject *" and self.c_undefined_value(rtype) == "NULL": | |
| # The simplest case. | |
| self.emit_line(f"Py_CLEAR({target});") | |
| else: | |
| assert False, "emit_gc_clear() not implemented for %s" % repr(rtype) | |
| def emit_reuse_clear(self, target: str, rtype: RType) -> None: | |
| """Emit attribute clear before object is added into freelist. | |
| Assume that 'target' represents a C expression that refers to a | |
| struct member, such as 'self->x'. | |
| Unlike emit_gc_clear(), initialize attribute value to match a freshly | |
| allocated object. | |
| """ | |
| if isinstance(rtype, RTuple): | |
| for i, item_type in enumerate(rtype.types): | |
| self.emit_reuse_clear(f"{target}.f{i}", item_type) | |
| elif not rtype.is_refcounted: | |
| self.emit_line(f"{target} = {rtype.c_undefined};") | |
| elif isinstance(rtype, RPrimitive) and rtype.name == "builtins.int": | |
| self.emit_line(f"if (CPyTagged_CheckLong({target})) {{") | |
| self.emit_line(f"CPyTagged __tmp = {target};") | |
| self.emit_line(f"{target} = {self.c_undefined_value(rtype)};") | |
| self.emit_line("Py_XDECREF(CPyTagged_LongAsObject(__tmp));") | |
| self.emit_line("} else {") | |
| self.emit_line(f"{target} = {self.c_undefined_value(rtype)};") | |
| self.emit_line("}") | |
| else: | |
| self.emit_gc_clear(target, rtype) | |
| def emit_traceback( | |
| self, source_path: str, module_name: str, traceback_entry: tuple[str, int] | |
| ) -> None: | |
| return self._emit_traceback("CPy_AddTraceback", source_path, module_name, traceback_entry) | |
| def emit_type_error_traceback( | |
| self, | |
| source_path: str, | |
| module_name: str, | |
| traceback_entry: tuple[str, int], | |
| *, | |
| typ: RType, | |
| src: str, | |
| ) -> None: | |
| func = "CPy_TypeErrorTraceback" | |
| type_str = f'"{self.pretty_name(typ)}"' | |
| return self._emit_traceback( | |
| func, source_path, module_name, traceback_entry, type_str=type_str, src=src | |
| ) | |
| def _emit_traceback( | |
| self, | |
| func: str, | |
| source_path: str, | |
| module_name: str, | |
| traceback_entry: tuple[str, int], | |
| type_str: str = "", | |
| src: str = "", | |
| ) -> None: | |
| if self.context.strict_traceback_checks: | |
| assert traceback_entry[1] >= 0, "Traceback cannot have a negative line number" | |
| globals_static = self.static_name("globals", module_name) | |
| line = '%s("%s", "%s", %d, %s' % ( | |
| func, | |
| source_path.replace("\\", "\\\\"), | |
| traceback_entry[0], | |
| traceback_entry[1], | |
| globals_static, | |
| ) | |
| if type_str: | |
| assert src | |
| line += f", {type_str}, {src}" | |
| line += ");" | |
| self.emit_line(line) | |
| if DEBUG_ERRORS: | |
| self.emit_line('assert(PyErr_Occurred() != NULL && "failure w/o err!");') | |
| def emit_unbox_failure_with_overlapping_error_value( | |
| self, dest: str, typ: RType, failure: str | |
| ) -> None: | |
| self.emit_line(f"if ({dest} == {self.c_error_value(typ)} && PyErr_Occurred()) {{") | |
| self.emit_line(failure) | |
| self.emit_line("}") | |
| def emit_cpyfunction_instance( | |
| self, fn: FuncIR, name: str, filepath: str, error_stmt: str | |
| ) -> str: | |
| module = self.static_name(fn.decl.module_name, None, prefix=MODULE_PREFIX) | |
| module = f"CPyImport_GetModuleCache(&{module})" | |
| cname = f"{PREFIX}{fn.cname(self.names)}" | |
| wrapper_name = f"wrapper_{cname}" | |
| cfunc = f"(PyCFunction){cname}" | |
| func_flags = ["METH_FASTCALL", "METH_KEYWORDS"] | |
| if fn.class_name and fn.decl.kind == FUNC_STATICMETHOD: | |
| func_flags.append("METH_STATIC") | |
| elif fn.class_name and fn.decl.kind == FUNC_CLASSMETHOD: | |
| func_flags.append("METH_CLASS") | |
| doc = f"PyDoc_STR({native_function_doc_initializer(fn)})" | |
| has_self_arg = "true" if fn.class_name and fn.decl.kind != FUNC_STATICMETHOD else "false" | |
| code_flags = "CO_COROUTINE" | |
| self.emit_line( | |
| f'PyObject* {wrapper_name} = CPyFunction_New({module}, "{filepath}", "{name}", {cfunc}, {" | ".join(func_flags)}, {doc}, {fn.line}, {code_flags}, {has_self_arg});' | |
| ) | |
| self.emit_line(f"if (unlikely(!{wrapper_name}))") | |
| self.emit_line(error_stmt) | |
| return wrapper_name | |
| def emit_base_tp_function_call( | |
| self, derived_cl: ClassIR, tp_func: str, args: str, *, prefix: str = "" | |
| ) -> None: | |
| # Walk past intermediate heap types (Python or mypyc classes) to reach a | |
| # static C-level ancestor. Calling a heap type's tp_dealloc/tp_traverse/ | |
| # tp_clear would dispatch through subtype_dealloc, which uses Py_TYPE(self) | |
| # (still our subtype) and re-enters our own function — infinite recursion. | |
| type_obj = self.type_struct_name(derived_cl) | |
| base_var = f"_base_{tp_func}" | |
| self.emit_line(f"PyTypeObject *{base_var} = {type_obj}->tp_base;") | |
| self.emit_line(f"while ({base_var}->tp_flags & Py_TPFLAGS_HEAPTYPE) {{") | |
| self.emit_line(f" {base_var} = {base_var}->tp_base;") | |
| self.emit_line("}") | |
| self.emit_line(f"{prefix}{base_var}->{tp_func}({args});") | |
| def c_array_initializer(components: list[str], *, indented: bool = False) -> str: | |
| """Construct an initializer for a C array variable. | |
| Components are C expressions valid in an initializer. | |
| For example, if components are ["1", "2"], the result | |
| would be "{1, 2}", which can be used like this: | |
| int a[] = {1, 2}; | |
| If the result is long, split it into multiple lines. | |
| """ | |
| indent = " " * 4 if indented else "" | |
| res = [] | |
| current: list[str] = [] | |
| cur_len = 0 | |
| for c in components: | |
| if not current or cur_len + 2 + len(indent) + len(c) < 70: | |
| current.append(c) | |
| cur_len += len(c) + 2 | |
| else: | |
| res.append(indent + ", ".join(current)) | |
| current = [c] | |
| cur_len = len(c) | |
| if not res: | |
| # Result fits on a single line | |
| return "{%s}" % ", ".join(current) | |
| # Multi-line result | |
| res.append(indent + ", ".join(current)) | |
| return "{\n " + ",\n ".join(res) + "\n" + indent + "}" | |
| def native_function_doc_initializer(func: FuncIR) -> str: | |
| text_sig = get_text_signature(func) | |
| if text_sig is None: | |
| return "NULL" | |
| docstring = f"{text_sig}\n--\n\n" | |
| return c_string_initializer(docstring.encode("ascii", errors="backslashreplace")) | |
| def stable_literal_repr(obj: object) -> str: | |
| """Return a single-line repr of a literal value. | |
| Behaves like repr() for most values, but renders frozenset members in a | |
| deterministic order (frozenset iteration order is hash-seed dependent). | |
| """ | |
| if isinstance(obj, frozenset): | |
| if not obj: | |
| return "frozenset()" | |
| items = ", ".join(stable_literal_repr(item) for item in sorted(obj, key=literal_sort_key)) | |
| return "frozenset({" + items + "})" | |
| elif isinstance(obj, tuple): | |
| if len(obj) == 1: | |
| return "(" + stable_literal_repr(obj[0]) + ",)" | |
| return "(" + ", ".join(stable_literal_repr(item) for item in obj) + ")" | |
| return repr(obj) |